Commit 890a40db authored by Xavier Thompson's avatar Xavier Thompson

software/theia: Adapts tests for embedded SR

parent 8cb2f7d7
...@@ -52,12 +52,55 @@ setUpModule, SlapOSInstanceTestCase = makeModuleSetUpAndTestCaseClass(theia_soft ...@@ -52,12 +52,55 @@ setUpModule, SlapOSInstanceTestCase = makeModuleSetUpAndTestCaseClass(theia_soft
class TheiaTestCase(SlapOSInstanceTestCase): class TheiaTestCase(SlapOSInstanceTestCase):
__partition_reference__ = 'T' # for supervisord sockets in included slapos __partition_reference__ = 'T' # for supervisord sockets in included slapos
@classmethod
def _getPath(cls, *components):
return os.path.join(cls.computer_partition_root_path, *components)
@classmethod @classmethod
def _getSlapos(cls): def _getSlapos(cls):
partition_root = cls.computer_partition_root_path try:
slapos = os.path.join(partition_root, 'srv', 'runner', 'bin', 'slapos') return cls._theia_slapos
except AttributeError:
cls._theia_slapos = slapos = cls._getPath('srv', 'runner', 'bin', 'slapos')
return slapos return slapos
@classmethod
def callSlapos(cls, *command, **kwargs):
return subprocess.call((cls._getSlapos(),) + command, **kwargs)
@classmethod
def checkSlapos(cls, *command, **kwargs):
kwargs['universal_newlines'] = True
return subprocess.check_call((cls._getSlapos(),) + command, **kwargs)
@classmethod
def captureSlapos(cls, *command, **kwargs):
kwargs['universal_newlines'] = True
return subprocess.check_output((cls._getSlapos(),) + command, **kwargs)
@classmethod
def requestInstance(cls, parameter_dict=None, state='started'):
cls.slap.request(
software_release=cls.getSoftwareURL(),
software_type=cls.getInstanceSoftwareType(),
partition_reference=cls.default_partition_reference,
partition_parameter_kw=parameter_dict,
state=state
)
@classmethod
def restartService(cls, service):
with cls.slap.instance_supervisor_rpc as supervisor:
for process_info in supervisor.getAllProcessInfo():
service_name = process_info['name']
if service in service_name:
service_id = '%s:%s' % (process_info['group'], service_name)
supervisor.stopProcess(service_id)
supervisor.startProcess(service_id)
break
else:
raise Exception("Service %s not found" % service)
class TestTheia(TheiaTestCase): class TestTheia(TheiaTestCase):
def setUp(self): def setUp(self):
...@@ -187,7 +230,7 @@ class TestTheia(TheiaTestCase): ...@@ -187,7 +230,7 @@ class TestTheia(TheiaTestCase):
self.computer_partition_root_path, self.computer_partition_root_path,
'srv', 'srv',
'project', 'project',
'request-script-template.sh', 'request-script-example.sh',
) )
self.assertTrue(os.path.exists(script_path)) self.assertTrue(os.path.exists(script_path))
...@@ -231,82 +274,70 @@ class TestTheiaEmbeddedSlapOSShutdown(TheiaTestCase): ...@@ -231,82 +274,70 @@ class TestTheiaEmbeddedSlapOSShutdown(TheiaTestCase):
self.assertFalse(embedded_slapos_process.is_running()) self.assertFalse(embedded_slapos_process.is_running())
class ReRequestMixin(object): class TestTheiaWithEmbeddedInstance(TheiaTestCase):
def rerequest(self, parameter_dict=None, state='started'): sr_url = '~/bogus/sr/url.cfg'
software_url = self.getSoftwareURL() sr_type = 'bogus-type'
software_type = self.getInstanceSoftwareType() sr_config = {"bogus": "yes"}
name = self.default_partition_reference regexpr = re.compile(r"([\w/\-\.]+)\s+slaprunner\s+available")
self.slap.request(
software_release=software_url,
software_type=software_type,
partition_reference=name,
partition_parameter_kw=parameter_dict,
state=state)
def reinstantiate(self):
# Process at least twice to propagate parameter changes
try:
self.slap.waitForInstance()
except SlapOSNodeCommandError:
pass
self.slap.waitForInstance(self.instance_max_retry)
class TestTheiaWithSR(TheiaTestCase, ReRequestMixin):
sr_url = '~/bogus/software.cfg'
sr_type = 'bogus_type'
instance_parameters = '{\n"bogus_param": "bogus_value",\n"bogus_param2": "bogus_value2"\n}'
def proxy_show(self, slapos):
return subprocess.check_output((slapos, 'proxy', 'show'), universal_newlines=True)
def test(self): @classmethod
slapos = self._getSlapos() def getInstanceParameterDict(cls, sr_url=None, sr_type=None, sr_config=None):
home = self.computer_partition_root_path return {
'initial-embedded-instance': json.dumps({
# Check that no request script was generated 'software-url': sr_url or cls.sr_url,
request_script = os.path.join(home, 'srv', 'project', 'request_embedded.sh') 'software-type': sr_type or cls.sr_type,
self.assertFalse(os.path.exists(request_script)) 'instance-parameters': sr_config or cls.sr_config,
}),
# Manually request old-name 'Embedded Instance'
old_instance_name = "Embedded Instance"
subprocess.check_call((slapos, 'request', old_instance_name, 'bogus_url'))
self.assertIn(old_instance_name, self.proxy_show(slapos))
# Update Theia instance parameters
embedded_request_parameters = {
'embedded-sr': self.sr_url,
'embedded-sr-type': self.sr_type,
'embedded-instance-parameters': self.instance_parameters
} }
self.rerequest(embedded_request_parameters)
self.reinstantiate()
# Check that embedded instance was requested def expandUrl(self, url):
instance_name = "embedded_instance" if url.startswith('~/'):
info = self.proxy_show(slapos) url = os.path.join(self.getPath(), url[2:])
try: return url
self.assertIn(instance_name, info)
except AssertionError: def assertSupplied(self, sr_url, info=None):
for filename in os.listdir(home): info = info or self.captureSlapos('proxy', 'show', text=True)
if 'standalone' in filename and '.log' in filename: self.assertIn(sr_url, info)
filepath = os.path.join(home, filename) self.assertIn(sr_url, self.regexpr.findall(info))
with open(filepath) as f:
print("Contents of filepath: " + filepath) def assertNotSupplied(self, sr_url, info=None):
print(f.read()) info = info or self.captureSlapos('proxy', 'show', text=True)
raise self.assertNotIn(sr_url, info)
def assertEmbedded(self, sr_url, sr_type, config):
proxy_info = self.captureSlapos('proxy', 'show', text=True)
self.assertSupplied(sr_url, info=proxy_info)
name = 'embedded_instance'
self.assertIn(name, self.captureSlapos('service', 'list', text=True))
info = self.captureSlapos('service', 'info', name, text=True)
self.assertIn(sr_url, info)
self.assertIn(sr_type, proxy_info)
self.assertIn(repr(config).replace("u'", "'"), info)
def assertNotEmbedded(self, sr_url, sr_type, config):
sr_url = self.expandUrl(sr_url)
proxy_info = self.captureSlapos('proxy', 'show', text=True)
self.assertNotSupplied(sr_url, info=proxy_info)
self.assertNotIn(sr_type, proxy_info)
# Check that old-name instance was renamed def test(self):
self.assertNotIn(old_instance_name, info) # Check that embedded instance is supplied and requested
initial_sr_url = self.expandUrl(self.sr_url)
self.assertEmbedded(initial_sr_url, self.sr_type, self.sr_config)
# Check embedded instance parameters # Change parameters for embedded instance
bogus_sr = os.path.join(home, self.sr_url[2:]) sr_url = '/bogus/sr/url-2.cfg'
sr_type = 'bogus-type-2'
sr_config = {"bogus-2": "true"}
self.requestInstance(
self.getInstanceParameterDict(sr_url, sr_type, sr_config))
self.waitForInstance()
self.assertIsNotNone(re.search(r"%s\s+slaprunner\s+available" % (bogus_sr,), info), info) # Check that parameters have not been taken into account
self.assertIsNotNone(re.search(r"%s\s+%s\s+%s" % (bogus_sr, self.sr_type, instance_name), info), info) self.assertNotEmbedded(sr_url, sr_type, sr_config)
service_info = subprocess.check_output((slapos, 'service', 'info', instance_name), universal_newlines=True) # Check that previous instance has not been changed
self.assertIn("{'bogus_param': 'bogus_value', 'bogus_param2': 'bogus_value2'}", service_info) self.assertEmbedded(initial_sr_url, self.sr_type, self.sr_config)
class TestTheiaFrontend(TheiaTestCase): class TestTheiaFrontend(TheiaTestCase):
...@@ -331,7 +362,9 @@ class TestTheiaEnv(TheiaTestCase): ...@@ -331,7 +362,9 @@ class TestTheiaEnv(TheiaTestCase):
@classmethod @classmethod
def getInstanceParameterDict(cls): def getInstanceParameterDict(cls):
return { return {
'embedded-sr': cls.dummy_software_path, 'initial-embedded-instance': json.dumps({
'software-url': cls.dummy_software_path,
}),
'autorun': 'stopped', 'autorun': 'stopped',
} }
...@@ -444,6 +477,11 @@ class ResilientTheiaMixin(object): ...@@ -444,6 +477,11 @@ class ResilientTheiaMixin(object):
def getInstanceSoftwareType(cls): def getInstanceSoftwareType(cls):
return 'resilient' return 'resilient'
def waitForinstance(self, *args, **kwargs):
# process twice to propagate to all instances
for _ in range(2):
super(ResilientTheiaMixin, self).waitForinstance(*args, **kwargs)
class TestTheiaResilientInterface(ResilientTheiaMixin, TestTheia): class TestTheiaResilientInterface(ResilientTheiaMixin, TestTheia):
@classmethod @classmethod
...@@ -453,9 +491,9 @@ class TestTheiaResilientInterface(ResilientTheiaMixin, TestTheia): ...@@ -453,9 +491,9 @@ class TestTheiaResilientInterface(ResilientTheiaMixin, TestTheia):
cls.computer_partition_root_path = cls._getPartitionPath('export') cls.computer_partition_root_path = cls._getPartitionPath('export')
class TestTheiaResilientWithSR(ResilientTheiaMixin, TestTheiaWithSR): class TestTheiaResilientWithEmbeddedInstance(ResilientTheiaMixin, TestTheiaWithEmbeddedInstance):
@classmethod @classmethod
def setUpClass(cls): def setUpClass(cls):
super(TestTheiaResilientWithSR, cls).setUpClass() super(TestTheiaResilientWithEmbeddedInstance, cls).setUpClass()
# Patch the computer root path to that of the export theia instance # Patch the computer root path to that of the export theia instance
cls.computer_partition_root_path = cls._getPartitionPath('export') cls.computer_partition_root_path = cls._getPartitionPath('export')
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment