Commit 30dd0182 authored by Xavier Thompson's avatar Xavier Thompson

software/theia: Refactor tests

parent 890a40db
......@@ -62,10 +62,9 @@ class ERP5Mixin(object):
_test_software_url = erp5_software_release_url
_connexion_parameters_regex = re.compile(r"{.*}", re.DOTALL)
def _getERP5ConnexionParameters(self, software_type='export'):
slapos = self._getSlapos(software_type)
out = subprocess.check_output(
(slapos, 'request', 'test_instance', self._test_software_url),
def _getERP5ConnexionParameters(self, instance_type='export'):
out = self.captureSlapos(
'request', 'test_instance', self._test_software_url,
stderr=subprocess.STDOUT,
)
print(out)
......@@ -110,10 +109,10 @@ class ERP5Mixin(object):
raise Exception("Found several partitions for ERP5 %s" % servicename)
return found.pop()
def _getERP5PartitionPath(self, software_type, servicename, *paths):
def _getERP5PartitionPath(self, instance_type, servicename, *paths):
partition = self._getERP5Partition(servicename)
return self._getPartitionPath(
software_type, 'srv', 'runner', 'instance', partition, *paths)
return self.getPartitionPath(
instance_type, 'srv', 'runner', 'instance', partition, *paths)
class TestTheiaResilienceERP5(ERP5Mixin, test_resiliency.TestTheiaResilience):
......@@ -161,16 +160,15 @@ class TestTheiaResilienceERP5(ERP5Mixin, test_resiliency.TestTheiaResilience):
# Update ERP5 parameters
print('Requesting ERP5 with parameters %s' % params)
slapos = self._getSlapos()
subprocess.check_call((slapos, 'request', 'test_instance', self._test_software_url, '--parameters', params))
self.checkSlapos('request', 'test_instance', self._test_software_url, '--parameters', params)
# Process twice to propagate parameter changes
for _ in range(2):
subprocess.check_call((slapos, 'node', 'instance'))
self.checkSlapos('node', 'instance')
# Restart cron (actually all) services to let them take the new date into account
# XXX this should not be required, updating ERP5 parameters should be enough
subprocess.call((slapos, 'node', 'restart', 'all'))
self.callSlapos('node', 'restart', 'all')
# Wait until after the programmed backup date, and a bit more
t = (soon - datetime.now()).total_seconds()
......@@ -213,9 +211,8 @@ class TestTheiaResilienceERP5(ERP5Mixin, test_resiliency.TestTheiaResilience):
self.assertNotIn(self._erp5_new_title, out)
# Stop all services
slapos = self._getSlapos()
print("Stop all services")
subprocess.call((slapos, 'node', 'stop', 'all'))
self.callSlapos('node', 'stop', 'all')
# Manually restore mariadb from backup
mariadb_restore_script = os.path.join(mariadb_partition, 'bin', 'restore-from-backup')
......
......@@ -53,7 +53,7 @@ class TheiaTestCase(SlapOSInstanceTestCase):
__partition_reference__ = 'T' # for supervisord sockets in included slapos
@classmethod
def _getPath(cls, *components):
def getPath(cls, *components):
return os.path.join(cls.computer_partition_root_path, *components)
@classmethod
......@@ -61,7 +61,7 @@ class TheiaTestCase(SlapOSInstanceTestCase):
try:
return cls._theia_slapos
except AttributeError:
cls._theia_slapos = slapos = cls._getPath('srv', 'runner', 'bin', 'slapos')
cls._theia_slapos = slapos = cls.getPath('srv', 'runner', 'bin', 'slapos')
return slapos
@classmethod
......@@ -70,12 +70,11 @@ class TheiaTestCase(SlapOSInstanceTestCase):
@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
kwargs.setdefault('universal_newlines', kwargs.pop('text', None))
return subprocess.check_output((cls._getSlapos(),) + command, **kwargs)
@classmethod
......@@ -147,7 +146,7 @@ class TestTheia(TheiaTestCase):
# there's a public folder to serve file
with open('{}/srv/frontend-static/public/test_file'.format(
self.computer_partition_root_path), 'w') as f:
self.getPath()), 'w') as f:
f.write("hello")
resp = self.get(urljoin(authenticated_url, '/public/'))
self.assertIn('test_file', resp.text)
......@@ -171,10 +170,9 @@ class TestTheia(TheiaTestCase):
self.assertIn('ipv6', self.connection_parameters)
def test_theia_slapos(self):
home = self.getPath()
# Make sure we can use the shell and the integrated slapos command
process = pexpect.spawnu(
'{}/bin/theia-shell'.format(self.computer_partition_root_path),
env={'HOME': self.computer_partition_root_path})
process = pexpect.spawnu(home + '/bin/theia-shell', env={'HOME': home})
# use a large enough terminal so that slapos proxy show table fit in the screen
process.setwinsize(5000, 5000)
......@@ -216,10 +214,11 @@ class TestTheia(TheiaTestCase):
process.wait()
def test_theia_shell_execute_tasks(self):
home = self.getPath()
# shell needs to understand -c "command" arguments for theia tasks feature
test_file = '{}/test file'.format(self.computer_partition_root_path)
test_file = home + '/test file'
subprocess.check_call([
'{}/bin/theia-shell'.format(self.computer_partition_root_path),
home + '/bin/theia-shell',
'-c',
'touch "{}"'.format(test_file)
])
......@@ -227,7 +226,7 @@ class TestTheia(TheiaTestCase):
def test_theia_request_script(self):
script_path = os.path.join(
self.computer_partition_root_path,
self.getPath(),
'srv',
'project',
'request-script-example.sh',
......@@ -235,18 +234,15 @@ class TestTheia(TheiaTestCase):
self.assertTrue(os.path.exists(script_path))
def test_slapos_cli(self):
slapos = self._getSlapos()
proxy_show_output = subprocess.check_output((slapos, 'proxy', 'show'))
self.assertIn(b'slaprunner', proxy_show_output)
computer_list_output = subprocess.check_output((slapos, 'computer', 'list'))
self.assertIn(b'slaprunner', computer_list_output)
self.assertIn(b'slaprunner', self.captureSlapos('proxy', 'show'))
self.assertIn(b'slaprunner', self.captureSlapos('computer', 'list'))
class TestTheiaEmbeddedSlapOSShutdown(TheiaTestCase):
def test_stopping_instance_stops_embedded_slapos(self):
embedded_slapos_supervisord_socket = _getSupervisordSocketPath(
os.path.join(
self.computer_partition_root_path,
self.getPath(),
'srv',
'runner',
'instance',
......@@ -372,7 +368,7 @@ class TestTheiaEnv(TheiaTestCase):
"""Make sure environment variables are the same whether we use shell or supervisor services.
"""
# The path of the env.json file expected to be generated by building the dummy software release
env_json_path = os.path.join(self.computer_partition_root_path, 'srv', 'runner', 'software', 'env.json')
env_json_path = self.getPath('srv', 'runner', 'software', 'env.json')
# Get the pid of the theia process from the test node's instance-supervisord
with self.slap.instance_supervisor_rpc as supervisor:
......@@ -390,7 +386,7 @@ class TestTheiaEnv(TheiaTestCase):
# Start a theia shell that inherits the environment of the theia process
# This simulates the environment of a shell launched from the browser application
theia_shell_process = pexpect.spawnu('{}/bin/theia-shell'.format(self.computer_partition_root_path), env=theia_env)
theia_shell_process = pexpect.spawnu('{}/bin/theia-shell'.format(self.getPath()), env=theia_env)
try:
theia_shell_process.expect_exact('Standalone SlapOS for computer `slaprunner` activated')
......@@ -410,7 +406,7 @@ class TestTheiaEnv(TheiaTestCase):
# Note that we have two services, slapos-node-software and slapos-node-software-all
# The later uses --all which is what we want to use here, because the software
# is already installed and we want to install it again, this time from supervisor
embedded_run_path = os.path.join(self.computer_partition_root_path, 'srv', 'runner', 'var', 'run')
embedded_run_path = self.getPath('srv', 'runner', 'var', 'run')
embedded_supervisord_socket_path = _getSupervisordSocketPath(embedded_run_path, self.logger)
with getSupervisorRPC(embedded_supervisord_socket_path) as embedded_supervisor:
previous_stop_time = embedded_supervisor.getProcessInfo('slapos-node-software-all')['stop']
......@@ -443,6 +439,8 @@ class ResilientTheiaMixin(object):
@classmethod
def setUpClass(cls):
super(ResilientTheiaMixin, cls).setUpClass()
# Patch the computer root path to that of the export theia instance
cls.computer_partition_root_path = cls.getPartitionPath('export')
# Add resiliency files to snapshot patterns
cls._save_instance_file_pattern_list += (
'*/srv/export-exitcode-file',
......@@ -452,48 +450,53 @@ class ResilientTheiaMixin(object):
)
@classmethod
def _getPartition(cls, software_type):
def getPartitionId(cls, instance_type):
software_url = cls.getSoftwareURL()
for computer_partition in cls.slap.computer.getComputerPartitionList():
partition_url = computer_partition.getSoftwareRelease()._software_release
partition_type = computer_partition.getType()
if partition_url == software_url and partition_type == software_type:
return computer_partition
raise Exception("Theia %s partition not found" % software_type)
if partition_url == software_url and partition_type == instance_type:
return computer_partition.getId()
raise Exception("Theia %s partition not found" % instance_type)
@classmethod
def _getPartitionId(cls, software_type):
return cls._getPartition(software_type).getId()
def getPartitionPath(cls, instance_type='export', *paths):
return os.path.join(cls.slap._instance_root, cls.getPartitionId(instance_type), *paths)
@classmethod
def _getPartitionPath(cls, software_type, *paths):
return os.path.join(cls.slap._instance_root, cls._getPartitionId(software_type), *paths)
def _getSlapos(cls, instance_type='export'):
return cls.getPartitionPath(instance_type, 'srv', 'runner', 'bin', 'slapos')
@classmethod
def _getSlapos(cls, software_type='export'):
return cls._getPartitionPath(software_type, 'srv', 'runner', 'bin', 'slapos')
def callSlapos(cls, *command, **kwargs):
instance_type = kwargs.pop('instance_type', 'export')
return subprocess.call((cls._getSlapos(instance_type),) + command, **kwargs)
@classmethod
def checkSlapos(cls, *command, **kwargs):
instance_type = kwargs.pop('instance_type', 'export')
return subprocess.check_call((cls._getSlapos(instance_type),) + command, **kwargs)
@classmethod
def captureSlapos(cls, *command, **kwargs):
kwargs.setdefault('universal_newlines', kwargs.pop('text', None))
instance_type = kwargs.pop('instance_type', 'export')
return subprocess.check_output((cls._getSlapos(instance_type),) + command, **kwargs)
@classmethod
def getInstanceSoftwareType(cls):
return 'resilient'
def waitForinstance(self, *args, **kwargs):
@classmethod
def waitForInstance(cls):
# process twice to propagate to all instances
for _ in range(2):
super(ResilientTheiaMixin, self).waitForinstance(*args, **kwargs)
super(ResilientTheiaMixin, cls).waitForInstance()
class TestTheiaResilientInterface(ResilientTheiaMixin, TestTheia):
@classmethod
def setUpClass(cls):
super(TestTheiaResilientInterface, cls).setUpClass()
# Patch the computer root path to that of the export theia instance
cls.computer_partition_root_path = cls._getPartitionPath('export')
pass
class TestTheiaResilientWithEmbeddedInstance(ResilientTheiaMixin, TestTheiaWithEmbeddedInstance):
@classmethod
def setUpClass(cls):
super(TestTheiaResilientWithEmbeddedInstance, cls).setUpClass()
# Patch the computer root path to that of the export theia instance
cls.computer_partition_root_path = cls._getPartitionPath('export')
pass
This diff is collapsed.
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