software/*/test: update tests now that computer_partition_root_path is a pathlib.Path

software/erp5/test/test/test_balancer.py

-import glob
 import ipaddress
 import json
 import logging
@@ -267,17 +266,13 @@ class TestLog(BalancerTestCase, CrontabMixin):
     # crontab for apachedex is executed
     self._executeCrontabAtDate('generate-apachedex-report', '23:59')
     # it creates a report for the day
-    apachedex_report, = glob.glob(
-        os.path.join(
-            self.computer_partition_root_path,
-            'srv',
-            'monitor',
-            'private',
-            'apachedex',
-            'ApacheDex-*.html',
-        ))
-    with open(apachedex_report) as f:
-      report_text = f.read()
+    apachedex_report, = (
+      self.computer_partition_root_path
+        / 'srv'
+        / 'monitor'
+        / 'private'
+        / 'apachedex').glob('ApacheDex-*.html')
+    report_text = apachedex_report.read_text()
     self.assertIn('APacheDEX', report_text)
     # having this table means that apachedex could parse some lines.
     self.assertIn('<h2>Hits per status code</h2>', report_text)
@@ -318,8 +313,8 @@ class TestLog(BalancerTestCase, CrontabMixin):
     self.assertEqual(
         requests.get(self.default_balancer_zope_url, verify=False).status_code,
         requests.codes.service_unavailable)
-    with open(os.path.join(self.computer_partition_root_path, 'var', 'log', 'apache-error.log')) as error_log_file:
-      error_line = error_log_file.read().splitlines()[-1]
+    error_log_file = self.computer_partition_root_path / 'var' / 'log' / 'apache-error.log'
+    error_line = error_log_file.read_text().splitlines()[-1]
     self.assertIn('backend default has no server available!', error_line)
     # this log also include a timestamp
     self.assertRegex(error_line, r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}')
@@ -416,7 +411,7 @@ class TestBalancer(BalancerTestCase):
     # real time statistics can be obtained by using the stats socket and there
     # is a wrapper which makes this a bit easier.
     socat_process = subprocess.Popen(
-        [self.computer_partition_root_path + '/bin/haproxy-socat-stats'],
+        [self.computer_partition_root_path / 'bin' / 'haproxy-socat-stats'],
         stdin=subprocess.PIPE,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT
@@ -604,14 +599,10 @@ class TestServerTLSEmbeddedCaucase(BalancerTestCase):
       balancer_parsed_url.port)
 
     # run caucase updater in the future, so that certificate is renewed
-    caucase_updater, = glob.glob(
-      os.path.join(
-        self.computer_partition_root_path,
-        'etc',
-        'service',
-        'caucase-updater-haproxy-certificate-*',
-      ))
-    process = pexpect.spawnu("faketime +90days " + caucase_updater)
+    caucase_updater, = (
+      self.computer_partition_root_path / 'etc' / 'service'
+    ).glob('caucase-updater-haproxy-certificate-*')
+    process = pexpect.spawnu(f"faketime +90days {caucase_updater}")
     logger = self.logger
     class DebugLogFile:
       def write(self, msg):
@@ -953,21 +944,16 @@ class TestClientTLS(BalancerTestCase):
 
       # We have two services in charge of updating CRL and CA certificates for
       # each frontend CA, plus the one for the balancer's own certificate
-      caucase_updater_list = glob.glob(
-          os.path.join(
-              self.computer_partition_root_path,
-              'etc',
-              'service',
-              'caucase-updater-*',
-          ))
+      caucase_updater_list = list((
+        self.computer_partition_root_path / 'etc' / 'service'
+      ).glob('caucase-updater-*'))
       self.assertEqual(len(caucase_updater_list), 3)
 
       # find the one corresponding to this caucase
       for caucase_updater_candidate in caucase_updater_list:
-        with open(caucase_updater_candidate) as f:
-          if caucase.url in f.read():
-            caucase_updater = caucase_updater_candidate
-            break
+        if caucase.url in caucase_updater_candidate.read_text():
+          caucase_updater = caucase_updater_candidate
+          break
       else:
         self.fail("Could not find caucase updater script for %s" % caucase.url)
 

software/erp5/test/test/test_erp5.py

@@ -477,8 +477,8 @@ class TestSeleniumTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMi
     }
 
   def test_test_runner_configuration_json_file(self):
-    runUnitTest_script, = glob.glob(
-        self.computer_partition_root_path + "/../*/bin/runUnitTest.real")
+    runUnitTest_script, = self.computer_partition_root_path.glob(
+      "../*/bin/runUnitTest.real")
     config_file = None
     with open(runUnitTest_script) as f:
       for line in f:
@@ -504,8 +504,8 @@ class TestDisableTestRunner(ERP5InstanceTestCase, TestPublishedURLIsReachableMix
     """
     # self.computer_partition_root_path is the path of root partition.
     # we want to assert that no scripts exist in any partition.
-    bin_programs = list(map(os.path.basename,
-      glob.glob(self.computer_partition_root_path + "/../*/bin/*")))
+    bin_programs = [
+      p.name for p in self.computer_partition_root_path.glob("../*/bin/*")]
 
     self.assertTrue(bin_programs) # just to check the glob was correct.
     self.assertNotIn('runUnitTest', bin_programs)

software/kvm/test/test.py

@@ -160,7 +160,7 @@ class KVMTestCase(InstanceTestCase):
     return image_list
 
   @classmethod
-  def _findTopLevelPartitionPath(cls, path):
+  def _findTopLevelPartitionPath(cls, path: str):
     index = 0
     while True:
       index = path.find(os.path.sep, index) + len(os.path.sep)
@@ -551,7 +551,7 @@ class TestAccessDefaultBootstrap(MonitorAccessMixin, KVMTestCase):
   def test(self):
     # START: mock .slapos-resource with tap.ipv4_addr
     # needed for netconfig.sh
-    partition_path = self.computer_partition_root_path
+    partition_path = str(self.computer_partition_root_path)
     top_partition_path = self._findTopLevelPartitionPath(partition_path)
 
     with open(os.path.join(top_partition_path, '.slapos-resource')) as f:

software/proftpd/test/test.py

@@ -236,7 +236,7 @@ class TestFilesAndSocketsInInstanceDir(ProFTPdTestCase):
         [],
         [
             f for f in self.proftpdProcess.open_files() if f.mode != 'r'
-            if not f.path.startswith(self.computer_partition_root_path)
+            if not f.path.startswith(str(self.computer_partition_root_path))
         ])
 
   def test_only_unix_socket_in_instance_dir(self):
@@ -244,7 +244,7 @@ class TestFilesAndSocketsInInstanceDir(ProFTPdTestCase):
         [],
         [
             s for s in self.proftpdProcess.connections('unix')
-            if not s.laddr.startswith(self.computer_partition_root_path)
+            if not s.laddr.startswith(str(self.computer_partition_root_path))
         ])
 
 

software/ssh/test/test.py

@@ -28,13 +28,11 @@
 import os
 import contextlib
 import paramiko
-import subprocess
 
 from urllib.parse import urlparse
-import socket
 
 from slapos.testing.testcase import makeModuleSetUpAndTestCaseClass
-from slapos.util import bytes2str
+
 
 setUpModule, SlapOSInstanceTestCase = makeModuleSetUpAndTestCaseClass(
     os.path.abspath(
@@ -71,5 +69,5 @@ class TestSSH(SlapOSInstanceTestCase):
         self.assertTrue(client.get_transport().is_active())
         # simple commands can also be executed ( this would be like `ssh bash -c 'pwd'` )
         # exec_command means `ssh user@host command`
-        current_path = bytes2str(client.exec_command("pwd")[1].read(1000)).strip()
-        self.assertIn(current_path, self.computer_partition_root_path)
+        current_path = client.exec_command("pwd")[1].read(1000).decode().strip()
+        self.assertIn(current_path, str(self.computer_partition_root_path))