Merge branch 'master' into dependabot/pip/images/openwisp_base/boto3-gte-1.43.39-and-lt-1.44.0

Author: nemesifier

tests/runtests.py

@@ -12,10 +12,11 @@
 from selenium.webdriver.common.by import By
 from selenium.webdriver.support import expected_conditions as EC
 from selenium.webdriver.support.ui import WebDriverWait
-from utils import TestUtilities
+from utils import BaseTestUtils, FunctionalTestUtils
 
 
-class Pretest(TestUtilities, unittest.TestCase):
+# 0 in the name is on purpose for alphabetical discovery
+class Test0Preconditions(BaseTestUtils, unittest.TestCase):
     """Checks to perform before tests"""
 
     def test_wait_for_services(self):
@@ -60,49 +61,14 @@ def test_wait_for_services(self):
             self.fail(f"All celery workers are not online: {online_workers}")
 
 
-class TestServices(TestUtilities, unittest.TestCase):
+class TestServices(FunctionalTestUtils, unittest.TestCase):
     custom_static_token = None
 
     @property
     def failureException(self):
         TestServices.failed_test = True
         return super().failureException
 
-    @classmethod
-    def _execute_docker_compose_command(cls, cmd_args, use_text_mode=False):
-        """Execute a docker compose command and log output.
-
-        Args:
-            cmd_args: List of command arguments for subprocess.Popen
-            use_text_mode: If True, use text mode for subprocess output
-
-        Returns:
-            Tuple of (output, error) from command execution
-        """
-        kwargs = {
-            "stdout": subprocess.PIPE,
-            "stderr": subprocess.PIPE,
-            "cwd": cls.root_location,
-        }
-        if use_text_mode:
-            kwargs["text"] = True
-        cmd = subprocess.run(cmd_args, check=False, **kwargs)
-        if use_text_mode:
-            output, error = cmd.stdout, cmd.stderr
-        else:
-            output = cmd.stdout.decode("utf-8", errors="replace") if cmd.stdout else ""
-            error = cmd.stderr.decode("utf-8", errors="replace") if cmd.stderr else ""
-        output, error = map(str, (cmd.stdout, cmd.stderr))
-        with open(cls.config["logs_file"], "a") as logs_file:
-            logs_file.write(output)
-            logs_file.write(error)
-        if cmd.returncode != 0:
-            raise RuntimeError(
-                f"docker compose command failed "
-                f"({cmd.returncode}): {' '.join(cmd_args)}"
-            )
-        return output, error
-
     @classmethod
     def _setup_admin_theme_links(cls):
         """Configure admin theme links during tests.
@@ -125,8 +91,8 @@ def _setup_admin_theme_links(cls):
                 f"body{{--openwisp-test: {cls.custom_static_token};}}"
             )
         script = rf"""
-            grep -q OPENWISP_ADMIN_THEME_LINKS /opt/openwisp/openwisp/settings.py || \
-            printf "\nOPENWISP_ADMIN_THEME_LINKS=[{{\"type\":\"text/css\",\"href\":\"/static/admin/css/openwisp.css\",\"rel\":\"stylesheet\",\"media\":\"all\"}},{{\"type\":\"text/css\",\"href\":\"/static/{cls.config["custom_css_filename"]}\",\"rel\":\"stylesheet\",\"media\":\"all\"}},{{\"type\":\"image/x-icon\",\"href\":\"ui/openwisp/images/favicon.png\",\"rel\":\"icon\"}}]\n" >> /opt/openwisp/openwisp/settings.py &&
+            sed -i '/^OPENWISP_ADMIN_THEME_LINKS[[:space:]]*=/d' /opt/openwisp/openwisp/settings.py &&
+            printf "\nOPENWISP_ADMIN_THEME_LINKS=[{{\"type\":\"text/css\",\"href\":\"/static/admin/css/openwisp.css\",\"rel\":\"stylesheet\",\"media\":\"all\"}},{{\"type\":\"text/css\",\"href\":\"/static/{cls.config["custom_css_filename"]}\",\"rel\":\"stylesheet\",\"media\":\"all\"}},{{\"type\":\"image/svg+xml\",\"href\":\"/static/ui/openwisp/images/favicon.svg\",\"rel\":\"icon\"}}]\n" >> /opt/openwisp/openwisp/settings.py &&
             python collectstatic.py &&
             uwsgi --reload uwsgi.pid
         """  # noqa: E501
@@ -153,27 +119,7 @@ def _cleanup_stale_test_data(cls):
         avoid any Selenium dependency during setup.
         """
         try:
-            cls._execute_docker_compose_command(
-                [
-                    "docker",
-                    "compose",
-                    "exec",
-                    "-T",
-                    "dashboard",
-                    "python",
-                    "manage.py",
-                    "shell",
-                    "-c",
-                    (
-                        "from openwisp_users.models import User; "
-                        "User.objects.filter("
-                        "username__in=['signup-user', 'test_superuser', "
-                        "'test_superuser2']"
-                        ").delete()"
-                    ),
-                ],
-                use_text_mode=True,
-            )
+            cls.delete_test_users("signup-user", "test_superuser", "test_superuser2")
         except Exception as e:
             exc_type = type(e).__name__
             print(
@@ -223,6 +169,12 @@ def setUpClass(cls):
 
     @classmethod
     def tearDownClass(cls):
+        try:
+            cls.delete_test_users(*cls.test_usernames_to_delete)
+            cls.test_usernames_to_delete.clear()
+        except Exception as e:
+            exc_type = type(e).__name__
+            print(f"Unable to delete test users: {exc_type}: {e}")
         for resource_link in cls.objects_to_delete:
             try:
                 cls._delete_object(resource_link)
@@ -277,6 +229,13 @@ def test_admin_login(self):
     def test_custom_static_files_loaded(self):
         self.login()
         self.open("/admin/")
+        favicon_href = self.web_driver.find_element(
+            By.CSS_SELECTOR, 'link[rel="icon"]'
+        ).get_attribute("href")
+        self.assertRegex(
+            favicon_href,
+            r"/static/ui/openwisp/images/favicon(\.[0-9a-f]+)?\.svg$",
+        )
         # Check if the custom CSS variable is applied
         value = self.web_driver.execute_script(
             "return getComputedStyle(document.body)"
@@ -362,6 +321,7 @@ def test_add_superuser(self):
     def test_forgot_password(self):
         """Test forgot password to ensure that postfix is working properly."""
 
+        self.login()
         self.logout()
         try:
             WebDriverWait(self.base_driver, 3).until(
@@ -463,23 +423,22 @@ def _test_celery_task_registered(container_name):
     def test_radius_user_registration(self):
         """Ensure users can register using the RADIUS API."""
         url = f"{self.config['api_url']}/api/v1/radius/organization/default/account/"
-        response = requests.post(
-            url,
-            json={
-                "username": "signup-user",
-                "email": "user@signup.com",
-                "password1": "rLx6OH%[",
-                "password2": "rLx6OH%[",
-            },
-            verify=False,
-        )
-        self.assertEqual(response.status_code, 201)
-        # Delete the created user
-        self.login()
-        self.get_resource(
-            "signup-user", "/admin/openwisp_users/user/", "field-username"
-        )
-        self.objects_to_delete.append(self.base_driver.current_url)
+        username = "signup-user"
+        try:
+            response = requests.post(
+                url,
+                json={
+                    "username": username,
+                    "email": "user@signup.com",
+                    "password1": "rLx6OH%[",
+                    "password2": "rLx6OH%[",
+                },
+                verify=False,
+                timeout=10,
+            )
+            self.assertEqual(response.status_code, 201, response.text)
+        finally:
+            self.delete_test_users(username)
 
     def test_freeradius(self):
         """Ensure freeradius service is working correctly."""
@@ -532,8 +491,8 @@ def test_containers_down(self):
             )
 
 
-class TestUtils(TestUtilities, unittest.TestCase):
-    """Other tests"""
+class TestLocalUtils(BaseTestUtils, unittest.TestCase):
+    """Tests for local utilities"""
 
     def test_update_version_updates_only_version_file(self):
         repository_root = Path(__file__).resolve().parents[1]

tests/utils.py

@@ -11,13 +11,10 @@
 from selenium.webdriver.common.by import By
 
 
-class TestConfig:
-    """Configuration class for setting up test parameters and utilities."""
-
-    def shortDescription(self):
-        """Return a short description for the test."""
-        return None
+class BaseTestUtils:
+    """Base class for setting up test parameters and utilities."""
 
+    docker_compose_timeout = 120
     docker_client = docker.from_env()
     ctx = ssl.create_default_context()
     ctx.check_hostname = False
@@ -28,13 +25,104 @@ def shortDescription(self):
     with open(config_file) as json_file:
         config = json.load(json_file)
 
+    def shortDescription(self):
+        """Keep verbose unittest output focused on test names, not docstrings."""
+        return None
+
+    @classmethod
+    def _execute_docker_compose_command(cls, cmd_args, use_text_mode=False):
+        """Execute a docker compose command and log output."""
+        kwargs = {
+            "stdout": subprocess.PIPE,
+            "stderr": subprocess.PIPE,
+            "cwd": cls.root_location,
+        }
+        if use_text_mode:
+            kwargs["text"] = True
+        try:
+            cmd = subprocess.run(
+                cmd_args,
+                check=False,
+                timeout=cls.docker_compose_timeout,
+                **kwargs,
+            )
+        except subprocess.TimeoutExpired as e:
+            output = e.stdout or ""
+            error = e.stderr or ""
+            if isinstance(output, bytes):
+                output = output.decode("utf-8", errors="replace") if output else ""
+            if isinstance(error, bytes):
+                error = error.decode("utf-8", errors="replace") if error else ""
+            with open(cls.config["logs_file"], "a") as logs_file:
+                logs_file.write(output)
+                logs_file.write(error)
+            raise RuntimeError(
+                "docker compose command timed out "
+                f"after {cls.docker_compose_timeout}s: {' '.join(cmd_args)}"
+            )
+        if use_text_mode:
+            output, error = cmd.stdout, cmd.stderr
+        else:
+            output = cmd.stdout.decode("utf-8", errors="replace") if cmd.stdout else ""
+            error = cmd.stderr.decode("utf-8", errors="replace") if cmd.stderr else ""
+        with open(cls.config["logs_file"], "a") as logs_file:
+            logs_file.write(output)
+            logs_file.write(error)
+        if cmd.returncode != 0:
+            raise RuntimeError(
+                f"docker compose command failed "
+                f"({cmd.returncode}): {' '.join(cmd_args)}"
+            )
+        return output, error
+
+    def docker_compose_get_container_id(self, container_name):
+        """Get the Docker container ID for a specific container.
+
+        Parameters:
+
+        - container_name (str): The name of the Docker container.
 
-class TestUtilities(SeleniumTestMixin, TestConfig):
-    """Utility functions for testing."""
+        Returns:
+            str: The ID of the Docker container.
+        """
+        output, _ = self._execute_docker_compose_command(
+            ["docker", "compose", "ps", "--quiet", container_name]
+        )
+        return output.rstrip()
+
+
+class FunctionalTestUtils(SeleniumTestMixin, BaseTestUtils):
+    """Utilities for functional testing."""
 
     objects_to_delete = []
+    test_usernames_to_delete = set()
     browser = "chrome"
 
+    @classmethod
+    def delete_test_users(cls, *usernames):
+        """Delete test-created users without relying on browser state."""
+        usernames = sorted(set(usernames))
+        if not usernames:
+            return
+        cls._execute_docker_compose_command(
+            [
+                "docker",
+                "compose",
+                "exec",
+                "-T",
+                "dashboard",
+                "python",
+                "manage.py",
+                "shell",
+                "-c",
+                (
+                    "from openwisp_users.models import User; "
+                    f"User.objects.filter(username__in={usernames!r}).delete()"
+                ),
+            ],
+            use_text_mode=True,
+        )
+
     def setUp(self):
         # Override TestSeleniumMixin setUp which uses
         # Django methods to create superuser
@@ -111,7 +199,7 @@ def create_superuser(
         self.find_element(By.NAME, "password2", driver=driver).send_keys(password)
         self.find_element(By.NAME, "is_superuser", driver=driver).click()
         self._click_save_btn(driver)
-        self.objects_to_delete.append(driver.current_url)
+        self.test_usernames_to_delete.add(username)
         self._click_save_btn(driver)
         self._wait_until_page_ready()
         self.wait_for_visibility(By.ID, "content", driver=driver, timeout=10)
@@ -254,25 +342,6 @@ def add_mobile_location_point(self, location_name, driver=None):
         self._click_save_btn(driver)
         self.get_resource(location_name, "/admin/geo/location/", driver=driver)
 
-    def docker_compose_get_container_id(self, container_name):
-        """Get the Docker container ID for a specific container.
-
-        Parameters:
-
-        - container_name (str): The name of the Docker container.
-
-        Returns:
-            str: The ID of the Docker container.
-        """
-        services_output = subprocess.Popen(
-            ["docker", "compose", "ps", "--quiet", container_name],
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-            cwd=self.root_location,
-        )
-        output, _ = services_output.communicate()
-        return output.rstrip().decode("utf-8")
-
     def create_network_topology(
         self,
         label="automated-selenium-test-01",