software/theia: Refactor tests

software/theia/test/project_tests.py

@@ -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')

software/theia/test/test.py

@@ -54,7 +54,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
@@ -62,7 +62,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
@@ -71,12 +71,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
@@ -148,7 +147,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)
@@ -169,10 +168,9 @@ class TestTheia(TheiaTestCase):
       self.assertTrue(resp.raw)
 
   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)
@@ -214,10 +212,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)
     ])
@@ -225,7 +224,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-template.sh',
@@ -233,18 +232,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',
@@ -359,7 +355,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:
@@ -377,7 +373,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')
 
@@ -397,7 +393,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']
@@ -430,6 +426,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',
@@ -439,48 +437,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 TestTheiaResilientWithEmbeddedSR(ResilientTheiaMixin, TestTheiaWithEmbeddedSR):
-  @classmethod
-  def setUpClass(cls):
-    super(TestTheiaResilientWithEmbeddedSR, cls).setUpClass()
-    # Patch the computer root path to that of the export theia instance
-    cls.computer_partition_root_path = cls._getPartitionPath('export')
+  pass