[feature] Added retry mechanism to SeleniumTestMixin #464

Retry selenium tests if the tests fails on the first attempt.
This prevents failng the CI build from flaky tests.

Closes #464

Author: pandafy

docs/developer/test-utilities.rst

@@ -179,6 +179,35 @@ This mixin provides the core Selenium setup logic and reusable test
 methods that must be used across all OpenWISP modules based on Django to
 enforce best practices and avoid flaky tests.
 
+It includes a built-in retry mechanism that can automatically repeat
+failing tests to identify transient (flaky) failures. You can customize
+this behavior using the following class attributes:
+
+- ``retry_max``: The maximum number of times to retry a failing test.
+  Defaults to ``5``.
+- ``retry_delay``: The number of seconds to wait between retries. Defaults
+  to ``0``.
+- ``retry_threshold``: The minimum ratio of successful retries required
+  for the test to be considered as passed. If the success ratio falls
+  below this threshold, the test is marked as failed. Defaults to ``0.8``.
+
+**Example usage:**
+
+.. code-block:: python
+
+    from openwisp_utils.tests import SeleniumTestMixin
+    from django.contrib.staticfiles.testing import StaticLiveServerTestCase
+
+
+    class MySeleniumTest(SeleniumTestMixin, StaticLiveServerTestCase):
+        retry_max = 10
+        retry_delay = 0
+        retry_threshold = 0.9
+
+        def test_something(self):
+            self.open("/some-url/")
+            # Your test logic here
+
 .. _selenium_dependencies:
 
 Selenium Dependencies

openwisp_utils/tests/selenium.py

@@ -1,4 +1,5 @@
 import os
+import time
 
 from selenium import webdriver
 from selenium.common.exceptions import TimeoutException
@@ -19,6 +20,62 @@ class SeleniumTestMixin:
     admin_password = 'password'
     browser = 'firefox'
 
+    retry_max = 5
+    retry_delay = 0
+    retry_threshold = 0.8
+
+    def _print_retry_message(self, test_name, attempt):
+        if attempt == 0:
+            return
+        print('-' * 80)
+        print(f'[Retry] Retrying "{test_name}", attempt {attempt}/{self.retry_max}. ')
+        print('-' * 80)
+
+    def _setup_and_call(self, result, debug=False):
+        """Override unittest.TestCase.run to retry flaky tests.
+
+        This method is responsible for calling setUp and tearDown methods.
+        Thus, we override this method to implement the retry mechanism
+        instead of TestCase.run().
+        """
+        original_result = result
+        test_name = self.id()
+        success_count = 0
+        failed_result = None
+        # Manually call startTest to ensure TimeLoggingTestResult can
+        # measure the execution time for the test.
+        original_result.startTest(self)
+
+        for attempt in range(self.retry_max + 1):
+            # Use a new result object to prevent writing all attempts
+            # to stdout.
+            result = self.defaultTestResult()
+            super()._setup_and_call(result, debug)
+            if result.wasSuccessful():
+                if attempt == 0:
+                    original_result.addSuccess(self)
+                    return
+                else:
+                    success_count += 1
+            else:
+                failed_result = result
+            self._print_retry_message(test_name, attempt)
+            if self.retry_delay:
+                time.sleep(self.retry_delay)
+
+        if success_count / self.retry_max < self.retry_threshold:
+            # If the success rate of retries is below the threshold then,
+            # copy errors and failures from the last failed result to the
+            # original result.
+            original_result.failures = failed_result.failures
+            original_result.errors = failed_result.errors
+            if hasattr(original_result, 'events'):
+                # Parallel tests uses RemoteTestResult which relies on events.
+                original_result.events = failed_result.events
+        else:
+            # Mark the test as passed in the original result
+            original_result.addSuccess(self)
+
     @classmethod
     def setUpClass(cls):
         super().setUpClass()

tests/test_project/tests/test_selenium.py

@@ -1,3 +1,5 @@
+from unittest import expectedFailure
+
 from django.contrib.staticfiles.testing import StaticLiveServerTestCase
 from django.urls import reverse
 from selenium.common.exceptions import JavascriptException, NoSuchElementException
@@ -753,3 +755,40 @@ def test_get_browser_logs(self):
         self.assertEqual(self.get_browser_logs(), [])
         self.web_driver.execute_script('console.log("test")')
         self.assertEqual(len(self.get_browser_logs()), 1)
+
+
+class TestSeleniumMixinRetryMechanism(SeleniumTestMixin, StaticLiveServerTestCase):
+    retry_delay = 0
+
+    @classmethod
+    def setUpClass(cls):
+        # We don't need browser instances for these tests.
+        pass
+
+    @classmethod
+    def tearDownClass(cls):
+        pass
+
+    def test_retry_mechanism_pass(self):
+        if not hasattr(self, '_test_retry_mechanism_pass_called'):
+            self._test_retry_mechanism_pass_called = 1
+            self.fail('Failing on first call')
+        else:
+            self._test_retry_mechanism_pass_called += 1
+
+    def test_retry_mechanism_not_called(self):
+        if not hasattr(self, '_test_retry_mechanism_not_called'):
+            self._test_retry_mechanism_not_called = 1
+        else:
+            # This code should not be executed because the test
+            # is called only once.
+            self._test_retry_mechanism_not_called += 1
+        self.assertEqual(self._test_retry_mechanism_not_called, 1)
+
+    @expectedFailure
+    def test_retry_mechanism_fails(self):
+        if not hasattr(self, '_test_retry_mechanism_fails_called'):
+            self._test_retry_mechanism_fails_called = 0
+        self._test_retry_mechanism_fails_called += 1
+        if self._test_retry_mechanism_fails_called < 5:
+            self.fail('Report failed test')