Add the ability to run parallel tests when using the Dashboard

Author: mdmintz

seleniumbase/fixtures/base_case.py

@@ -7269,15 +7269,20 @@ def setUp(self, masterqa_mode=False):
             self.guest_mode = sb_config.guest_mode
             self.devtools = sb_config.devtools
             self.remote_debug = sb_config.remote_debug
+            self._multithreaded = sb_config._multithreaded
+            self._reuse_session = sb_config.reuse_session
+            self._crumbs = sb_config.crumbs
             self.dashboard = sb_config.dashboard
             self._dash_initialized = sb_config._dashboard_initialized
+            if self.dashboard and self._multithreaded:
+                import fasteners
+                self.dash_lock = fasteners.InterProcessLock(
+                    constants.Dashboard.LOCKFILE)
             self.swiftshader = sb_config.swiftshader
             self.user_data_dir = sb_config.user_data_dir
             self.extension_zip = sb_config.extension_zip
             self.extension_dir = sb_config.extension_dir
             self.maximize_option = sb_config.maximize_option
-            self._reuse_session = sb_config.reuse_session
-            self._crumbs = sb_config.crumbs
             self.save_screenshot_after_test = sb_config.save_screenshot
             self.visual_baseline = sb_config.visual_baseline
             self.timeout_multiplier = sb_config.timeout_multiplier
@@ -7391,13 +7396,21 @@ def setUp(self, masterqa_mode=False):
 
         # Dashboard pre-processing:
         if self.dashboard:
-            sb_config._sbase_detected = True
-            sb_config._only_unittest = False
-            if not self._dash_initialized:
-                sb_config._dashboard_initialized = True
+            if self._multithreaded:
+                with self.dash_lock:
+                    sb_config._sbase_detected = True
+                    sb_config._only_unittest = False
+                    if not self._dash_initialized:
+                        sb_config._dashboard_initialized = True
+                        self._dash_initialized = True
+                        self.__process_dashboard(False, init=True)
+            else:
                 sb_config._sbase_detected = True
-                self._dash_initialized = True
-                self.__process_dashboard(False, init=True)
+                sb_config._only_unittest = False
+                if not self._dash_initialized:
+                    sb_config._dashboard_initialized = True
+                    self._dash_initialized = True
+                    self.__process_dashboard(False, init=True)
 
         has_url = False
         if self._reuse_session:
@@ -7696,6 +7709,23 @@ def __create_log_path_as_needed(self, test_logpath):
 
     def __process_dashboard(self, has_exception, init=False):
         ''' SeleniumBase Dashboard Processing '''
+        existing_res = sb_config._results  # Used by multithreaded tests
+        if self._multithreaded:
+            abs_path = os.path.abspath('.')
+            dash_json_loc = constants.Dashboard.DASH_JSON
+            dash_jsonpath = os.path.join(abs_path, dash_json_loc)
+            if not init and os.path.exists(dash_jsonpath):
+                with open(dash_jsonpath, 'r') as f:
+                    dash_json = f.read().strip()
+                dash_data, d_id, dash_runtimes, d_stats = json.loads(dash_json)
+                num_passed, num_failed, num_skipped, num_untested = d_stats
+                sb_config._results = dash_data
+                sb_config._display_id = d_id
+                sb_config._duration = dash_runtimes
+                sb_config.item_count_passed = num_passed
+                sb_config.item_count_failed = num_failed
+                sb_config.item_count_skipped = num_skipped
+                sb_config.item_count_untested = num_untested
         if len(sb_config._extra_dash_entries) > 0:
             # First take care of existing entries from non-SeleniumBase tests
             for test_id in sb_config._extra_dash_entries:
@@ -7741,6 +7771,11 @@ def __process_dashboard(self, has_exception, init=False):
                 sb_config.item_count_skipped += 1
                 sb_config.item_count_untested -= 1
                 sb_config._results[test_id] = "Skipped"
+            elif self._multithreaded and test_id in existing_res.keys() and (
+                    existing_res[test_id] == "Skipped"):
+                sb_config.item_count_skipped += 1
+                sb_config.item_count_untested -= 1
+                sb_config._results[test_id] = "Skipped"
             elif has_exception:
                 # pytest-rerunfailures may cause duplicate results
                 if test_id not in sb_config._results.keys() or (
@@ -7778,6 +7813,14 @@ def __process_dashboard(self, has_exception, init=False):
             if sb_config._using_html_report:
                 # Add the pie chart to the pytest html report
                 sb_config._saved_dashboard_pie = self.extract_chart()
+                if self._multithreaded:
+                    abs_path = os.path.abspath('.')
+                    dash_pie = json.dumps(sb_config._saved_dashboard_pie)
+                    dash_pie_loc = constants.Dashboard.DASH_PIE
+                    pie_path = os.path.join(abs_path, dash_pie_loc)
+                    pie_file = codecs.open(pie_path, "w+", encoding="utf-8")
+                    pie_file.writelines(dash_pie)
+                    pie_file.close()
         head = (
             '<head><meta charset="utf-8" />'
             '<meta property="og:image" '
@@ -7887,6 +7930,17 @@ def __process_dashboard(self, has_exception, init=False):
         out_file.writelines(the_html)
         out_file.close()
         time.sleep(0.05)  # Add time for dashboard server to process updates
+        if self._multithreaded:
+            d_stats = (num_passed, num_failed, num_skipped, num_untested)
+            _results = sb_config._results
+            _display_id = sb_config._display_id
+            _duration = sb_config._duration
+            dash_json = json.dumps((_results, _display_id, _duration, d_stats))
+            dash_json_loc = constants.Dashboard.DASH_JSON
+            dash_jsonpath = os.path.join(abs_path, dash_json_loc)
+            dash_json_file = codecs.open(dash_jsonpath, "w+", encoding="utf-8")
+            dash_json_file.writelines(dash_json)
+            dash_json_file.close()
 
     def has_exception(self):
         """ (This method should ONLY be used in custom tearDown() methods.)
@@ -8031,7 +8085,11 @@ def tearDown(self):
                                 test_logpath, self.driver,
                                 self.__last_page_source)
                 if self.dashboard:
-                    self.__process_dashboard(has_exception)
+                    if self._multithreaded:
+                        with self.dash_lock:
+                            self.__process_dashboard(has_exception)
+                    else:
+                        self.__process_dashboard(has_exception)
                 # (Pytest) Finally close all open browser windows
                 self.__quit_all_drivers()
             if self.headless:

seleniumbase/fixtures/constants.py

@@ -29,10 +29,13 @@ class Charts:
 
 class Dashboard:
     TITLE = "SeleniumBase Test Results Dashboard"
-    STYLE_CSS = 'https://seleniumbase.io/cdn/css/pytest_style.css'
+    STYLE_CSS = "https://seleniumbase.io/cdn/css/pytest_style.css"
     META_REFRESH_HTML = '<meta http-equiv="refresh" content="12">'
     # LIVE_JS = 'https://livejs.com/live.js#html'
-    LIVE_JS = 'https://seleniumbase.io/cdn/js/live.js#html'
+    LIVE_JS = "https://seleniumbase.io/cdn/js/live.js#html"
+    LOCKFILE = Files.DOWNLOADS_FOLDER + "/dashboard.lock"
+    DASH_JSON = Files.DOWNLOADS_FOLDER + "/dashboard.json"
+    DASH_PIE = Files.DOWNLOADS_FOLDER + "/dash_pie.json"
 
 
 class SavedCookies:

seleniumbase/plugins/pytest_plugin.py

@@ -582,13 +582,18 @@ def pytest_addoption(parser):
                 "\n  It's not thread-safe for WebDriver processes! "
                 "\n  Use --time-limit=s from SeleniumBase instead!\n")
 
-    # The SeleniumBase Dashboard does not yet support multi-threadeded tests.
-    if "--dashboard" in sys_argv:
-        arg_join = " ".join(sys_argv)
-        if ("-n" in sys_argv) or ("-n=" in arg_join):
-            raise Exception(
-                "\n\n  Multi-threading is not yet supported using --dashboard"
-                "\n  (You can speed up tests using --reuse-session / --rs)\n")
+    # Dashboard Mode does not support tests using forked subprocesses.
+    if "--forked" in sys_argv and "--dashboard" in sys_argv:
+        raise Exception(
+            '\n\n  Dashboard Mode does NOT support forked subprocesses!'
+            '\n  (*** DO NOT combine "--forked" with "--dashboard"! ***)\n')
+
+    # Reuse-Session Mode does not support tests using forked subprocesses.
+    if "--forked" in sys_argv and (
+            "--rs" in sys_argv or "--reuse-session" in sys_argv):
+        raise Exception(
+            '\n\n  Reuse-Session Mode does NOT support forked subprocesses!'
+            '\n  (DO NOT combine "--forked" with "--rs"/"--reuse-session"!)\n')
 
     # As a shortcut, you can use "--edge" instead of "--browser=edge", etc,
     # but you can only specify one default browser for tests. (Default: chrome)
@@ -874,6 +879,16 @@ def pytest_sessionfinish(session):
 
 
 def pytest_terminal_summary(terminalreporter):
+    latest_logs_dir = os.getcwd() + "/latest_logs/"
+    if sb_config._multithreaded:
+        if os.path.exists(latest_logs_dir) and os.listdir(latest_logs_dir):
+            sb_config._has_exception = True
+        if sb_config.dashboard:
+            abs_path = os.path.abspath('.')
+            dash_lock = constants.Dashboard.LOCKFILE
+            dash_lock_path = os.path.join(abs_path, dash_lock)
+            if os.path.exists(dash_lock_path):
+                sb_config._only_unittest = False
     if sb_config._has_exception and (
             sb_config.dashboard and not sb_config._only_unittest):
         # Print link a second time because the first one may be off-screen
@@ -882,7 +897,6 @@ def pytest_terminal_summary(terminalreporter):
             "-", "Dashboard: %s" % dashboard_file)
     if sb_config._has_exception or sb_config.save_screenshot:
         # Log files are generated during test failures and Screenshot Mode
-        latest_logs_dir = os.getcwd() + "/latest_logs/"
         terminalreporter.write_sep(
             "-", "LogPath: %s" % latest_logs_dir)
 
@@ -903,10 +917,19 @@ def pytest_unconfigure():
     if hasattr(sb_config, 'log_path'):
         log_helper.archive_logs_if_set(
             sb_config.log_path, sb_config.archive_logs)
-
     # Dashboard post-processing: Disable time-based refresh and stamp complete
+    if sb_config._multithreaded and sb_config.dashboard:
+        abs_path = os.path.abspath('.')
+        dash_lock = constants.Dashboard.LOCKFILE
+        dash_lock_path = os.path.join(abs_path, dash_lock)
+        if os.path.exists(dash_lock_path):
+            sb_config._only_unittest = False
     if hasattr(sb_config, 'dashboard') and (
             sb_config.dashboard and not sb_config._only_unittest):
+        if sb_config._multithreaded:
+            import fasteners
+            dash_lock = fasteners.InterProcessLock(
+                constants.Dashboard.LOCKFILE)
         stamp = ""
         if sb_config._dash_is_html_report:
             # (If the Dashboard URL is the same as the HTML Report URL:)
@@ -924,19 +947,35 @@ def pytest_unconfigure():
         swap_with_3 = '<td class="col-result">Unreported</td>'
         find_it_4 = 'href="https://seleniumbase.io/img/dash_pie.png"'
         swap_with_4 = 'href="https://seleniumbase.io/img/dash_pie_2.png"'
+        find_it_5 = 'content="https://seleniumbase.io/img/dash_pie.png"'
+        swap_with_5 = 'content="https://seleniumbase.io/img/dash_pie_2.png"'
         try:
+            if sb_config._multithreaded:
+                dash_lock.acquire()
             abs_path = os.path.abspath('.')
             dashboard_path = os.path.join(abs_path, "dashboard.html")
             # Part 1: Finalizing the dashboard / integrating html report
             if os.path.exists(dashboard_path):
                 the_html_d = None
                 with open(dashboard_path, 'r', encoding='utf-8') as f:
                     the_html_d = f.read()
+                if sb_config._multithreaded and "-c" in sys.argv:
+                    # Threads have "-c" in sys.argv, except for the last
+                    raise Exception('Break out of "try" block.')
+                if sb_config._multithreaded:
+                    dash_pie_loc = constants.Dashboard.DASH_PIE
+                    pie_path = os.path.join(abs_path, dash_pie_loc)
+                    if os.path.exists(pie_path):
+                        import json
+                        with open(pie_path, 'r') as f:
+                            dash_pie = f.read().strip()
+                        sb_config._saved_dashboard_pie = json.loads(dash_pie)
                 # If the test run doesn't complete by itself, stop refresh
                 the_html_d = the_html_d.replace(find_it, swap_with)
                 the_html_d = the_html_d.replace(find_it_2, swap_with_2)
                 the_html_d = the_html_d.replace(find_it_3, swap_with_3)
                 the_html_d = the_html_d.replace(find_it_4, swap_with_4)
+                the_html_d = the_html_d.replace(find_it_5, swap_with_5)
                 the_html_d += stamp
                 if sb_config._dash_is_html_report and (
                         sb_config._saved_dashboard_pie):
@@ -985,8 +1024,13 @@ def pytest_unconfigure():
                             the_html_r += sb_config._dash_final_summary
                     with open(html_report_path, "w", encoding='utf-8') as f:
                         f.write(the_html_r)  # Finalize the HTML report
+        except KeyboardInterrupt:
+            pass
         except Exception:
             pass
+        finally:
+            if sb_config._multithreaded:
+                dash_lock.release()
 
 
 @pytest.fixture()
@@ -1037,6 +1081,8 @@ def pytest_runtest_makereport(item, call):
     pytest_html = item.config.pluginmanager.getplugin('html')
     outcome = yield
     report = outcome.get_result()
+    if sb_config._multithreaded:
+        sb_config._using_html_report = True  # For Dashboard use
     if pytest_html and report.when == 'call' and (
             hasattr(sb_config, 'dashboard')):
         if sb_config.dashboard and not sb_config._sbase_detected: