diff --git a/tests/runtests.py b/tests/runtests.py index ffa47ede..c78f9e92 100644 --- a/tests/runtests.py +++ b/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,7 +61,7 @@ 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 @@ -68,41 +69,6 @@ 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] diff --git a/tests/utils.py b/tests/utils.py index c7def8c8..bd3f3ef5 100644 --- a/tests/utils.py +++ b/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",