software/theia: Improve tests

software/theia/test/test.py

@@ -67,12 +67,21 @@ class TestTheia(TheiaTestCase):
   def setUp(self):
     self.connection_parameters = self.computer_partition.getConnectionParameterDict()
 
+  def get(self, url, expect_code=requests.codes.ok):
+    resp = requests.get(url, verify=False)
+    self.assertEqual(
+      expect_code,
+      resp.status_code,
+      '%s returned %d instead of %d' % (url, resp.status_code, expect_code),
+    )
+    return resp
+
   def test_backend_http_get(self):
-    resp = requests.get(self.connection_parameters['backend-url'], verify=False)
-    self.assertEqual(requests.codes.unauthorized, resp.status_code)
+    backend_url = self.connection_parameters['backend-url']
+    self.get(backend_url, requests.codes.unauthorized)
 
     # with login/password, this is allowed
-    parsed_url = urlparse(self.connection_parameters['backend-url'])
+    parsed_url = urlparse(backend_url)
     authenticated_url = parsed_url._replace(
         netloc='{}:{}@[{}]:{}'.format(
             self.connection_parameters['username'],
@@ -80,12 +89,11 @@ class TestTheia(TheiaTestCase):
             parsed_url.hostname,
             parsed_url.port,
         )).geturl()
-    resp = requests.get(authenticated_url, verify=False)
-    self.assertEqual(requests.codes.ok, resp.status_code)
+    self.get(authenticated_url)
 
   def test_http_get(self):
-    resp = requests.get(self.connection_parameters['url'], verify=False)
-    self.assertEqual(requests.codes.unauthorized, resp.status_code)
+    url = self.connection_parameters['url']
+    self.get(url, requests.codes.unauthorized)
 
     # with login/password, this is allowed
     parsed_url = urlparse(self.connection_parameters['url'])
@@ -96,33 +104,28 @@ class TestTheia(TheiaTestCase):
             parsed_url.hostname,
             parsed_url.port,
         )).geturl()
-    resp = requests.get(authenticated_url, verify=False)
-    self.assertEqual(requests.codes.ok, resp.status_code)
+    self.get(authenticated_url)
 
     # 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:
       f.write("hello")
-    resp = requests.get(urljoin(authenticated_url, '/public/'), verify=False)
+    resp = self.get(urljoin(authenticated_url, '/public/'))
     self.assertIn('test_file', resp.text)
-    resp = requests.get(
-        urljoin(authenticated_url, '/public/test_file'), verify=False)
+    resp = self.get(urljoin(authenticated_url, '/public/test_file'))
     self.assertEqual('hello', resp.text)
 
     # there's a (not empty) favicon
-    resp = requests.get(
-        urljoin(authenticated_url, '/favicon.ico'), verify=False)
-    self.assertEqual(requests.codes.ok, resp.status_code)
+    resp = self.get(urljoin(authenticated_url, '/favicon.ico'))
     self.assertTrue(resp.raw)
 
     # there is a CSS referencing fonts
-    css_text = requests.get(urljoin(authenticated_url, '/css/slapos.css'), verify=False).text
+    css_text = self.get(urljoin(authenticated_url, '/css/slapos.css')).text
     css_urls = re.findall(r'url\([\'"]+([^\)]+)[\'"]+\)', css_text)
     self.assertTrue(css_urls)
     # and fonts are served
     for url in css_urls:
-      resp = requests.get(urljoin(authenticated_url, url), verify=False)
-      self.assertEqual(requests.codes.ok, resp.status_code)
+      resp = self.get(urljoin(authenticated_url, url))
       self.assertTrue(resp.raw)
 
   def test_theia_slapos(self):
@@ -281,7 +284,7 @@ class TestTheiaEnv(TheiaTestCase):
     }
 
   def test_theia_env(self):
-    """Make sure environment variables are the same wether we use shell or supervisor services.
+    """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')
@@ -303,47 +306,49 @@ 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.expect_exact('Standalone SlapOS for computer `slaprunner` activated')
-
-    # Launch slapos node software from theia shell
-    theia_shell_process.sendline('slapos node software')
-    theia_shell_process.expect('Installing software release %s' % self.dummy_software_path)
-    theia_shell_process.expect('Finished software releases.')
-
-    # Get the theia shell environment
-    with open(env_json_path) as f:
-      theia_shell_env = json.load(f)
-
-    # Remove the env.json file to later be sure that a new one has been generated
-    os.remove(env_json_path)
-
-    # Launch slapos node software service from the embedded supervisord.
-    # Note that we have two services, slapos-not-software and slapos-not-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_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']
-      embedded_supervisor.startProcess('slapos-node-software-all')
-      for _retries in range(20):
-        time.sleep(1)
-        if embedded_supervisor.getProcessInfo('slapos-node-software-all')['stop'] != previous_stop_time:
-          break
-      else:
-        self.fail("the supervisord service 'slapos-node-software-all' takes too long to finish")
-
-    # Get the supervisord environment
-    with open(env_json_path) as f:
-      supervisord_env = json.load(f)
-
-    # Compare relevant variables from both environments
-    self.maxDiff = None
-    self.assertEqual(theia_shell_env['PATH'].split(':'), supervisord_env['PATH'].split(':'))
-    self.assertEqual(theia_shell_env['SLAPOS_CONFIGURATION'], supervisord_env['SLAPOS_CONFIGURATION'])
-    self.assertEqual(theia_shell_env['SLAPOS_CLIENT_CONFIGURATION'], supervisord_env['SLAPOS_CLIENT_CONFIGURATION'])
-    self.assertEqual(theia_shell_env['HOME'], supervisord_env['HOME'])
-
-    # Cleanup the theia shell process
-    theia_shell_process.terminate()
-    theia_shell_process.wait()
+    try:
+      theia_shell_process.expect_exact('Standalone SlapOS for computer `slaprunner` activated')
+
+      # Launch slapos node software from theia shell
+      theia_shell_process.sendline('slapos node software')
+      theia_shell_process.expect('Installing software release %s' % self.dummy_software_path)
+      theia_shell_process.expect('Finished software releases.')
+
+      # Get the theia shell environment
+      with open(env_json_path) as f:
+        theia_shell_env = json.load(f)
+
+      # Remove the env.json file to later be sure that a new one has been generated
+      os.remove(env_json_path)
+
+      # Launch slapos node software service from the embedded supervisord.
+      # 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_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']
+        embedded_supervisor.startProcess('slapos-node-software-all')
+        for _retries in range(20):
+          time.sleep(1)
+          if embedded_supervisor.getProcessInfo('slapos-node-software-all')['stop'] != previous_stop_time:
+            break
+        else:
+          self.fail("the supervisord service 'slapos-node-software-all' takes too long to finish")
+
+      # Get the supervisord environment
+      with open(env_json_path) as f:
+        supervisord_env = json.load(f)
+
+      # Compare relevant variables from both environments
+      self.maxDiff = None
+      self.assertEqual(theia_shell_env['PATH'].split(':'), supervisord_env['PATH'].split(':'))
+      self.assertEqual(theia_shell_env['SLAPOS_CONFIGURATION'], supervisord_env['SLAPOS_CONFIGURATION'])
+      self.assertEqual(theia_shell_env['SLAPOS_CLIENT_CONFIGURATION'], supervisord_env['SLAPOS_CLIENT_CONFIGURATION'])
+      self.assertEqual(theia_shell_env['HOME'], supervisord_env['HOME'])
+
+    finally:
+      # Cleanup the theia shell process
+      theia_shell_process.terminate()
+      theia_shell_process.wait()