Update CDP Mode

Author: mdmintz

help_docs/method_summary.md

@@ -250,8 +250,10 @@ self.save_screenshot_to_logs(name=None, selector=None, by="css selector")
 self.save_as_pdf_to_logs(name=None)
 self.save_data_to_logs(data, file_name=None)
 self.append_data_to_logs(data, file_name=None)
-self.save_page_source(name, folder=None)
 self.save_page_source_to_logs(name=None)
+self.save_page_source(name, folder=None)
+# Duplicates:
+# self.save_as_html(name, folder=None)
 self.save_cookies(name="cookies.txt")
 self.load_cookies(name="cookies.txt", expiry=False)
 self.delete_all_cookies()

seleniumbase/core/browser_launcher.py

@@ -756,12 +756,16 @@ def uc_open_with_cdp_mode(driver, url=None, **kwargs):
     cdp.set_value = CDPM.set_value
     cdp.submit = CDPM.submit
     cdp.evaluate = CDPM.evaluate
+    cdp.execute_script = CDPM.execute_script
     cdp.js_dumps = CDPM.js_dumps
     cdp.maximize = CDPM.maximize
     cdp.minimize = CDPM.minimize
     cdp.medimize = CDPM.medimize
     cdp.set_window_rect = CDPM.set_window_rect
     cdp.reset_window_size = CDPM.reset_window_size
+    cdp.activate_messenger = CDPM.activate_messenger
+    cdp.set_messenger_theme = CDPM.set_messenger_theme
+    cdp.post_message = CDPM.post_message
     cdp.set_locale = CDPM.set_locale
     cdp.set_local_storage_item = CDPM.set_local_storage_item
     cdp.set_session_storage_item = CDPM.set_session_storage_item
@@ -877,6 +881,8 @@ def uc_open_with_cdp_mode(driver, url=None, **kwargs):
     cdp.scroll_to_bottom = CDPM.scroll_to_bottom
     cdp.scroll_up = CDPM.scroll_up
     cdp.scroll_down = CDPM.scroll_down
+    cdp.save_page_source = CDPM.save_page_source
+    cdp.save_as_html = CDPM.save_as_html
     cdp.save_screenshot = CDPM.save_screenshot
     cdp.print_to_pdf = CDPM.print_to_pdf
     cdp.save_as_pdf = CDPM.save_as_pdf

seleniumbase/core/sb_cdp.py

@@ -1024,6 +1024,9 @@ def evaluate(self, expression):
             ).strip()
         return self.loop.run_until_complete(self.page.evaluate(expression))
 
+    def execute_script(self, expression):
+        return self.evaluate(expression)
+
     def js_dumps(self, obj_name):
         """Similar to evaluate(), but for dictionary results."""
         if obj_name.startswith("return "):
@@ -1407,6 +1410,64 @@ def enter_mfa_code(self, selector, totp_key=None, timeout=None):
         mfa_code = self.get_mfa_code(totp_key)
         self.type(selector, mfa_code + "\n", timeout=timeout)
 
+    def activate_messenger(self):
+        js_utils.activate_messenger(self)
+        self.__add_light_pause()
+
+    def set_messenger_theme(
+        self, theme="default", location="default", max_messages="default"
+    ):
+        """Sets a theme for posting messages.
+        Themes: ["flat", "future", "block", "air", "ice"]
+        Locations: ["top_left", "top_center", "top_right",
+                    "bottom_left", "bottom_center", "bottom_right"]
+        max_messages: The limit of concurrent messages to display."""
+        if not theme:
+            theme = "default"  # "flat"
+        if not location:
+            location = "default"  # "bottom_right"
+        if not max_messages:
+            max_messages = "default"  # "8"
+        else:
+            max_messages = str(max_messages)  # Value must be in string format
+        js_utils.set_messenger_theme(
+            self,
+            theme=theme,
+            location=location,
+            max_messages=max_messages,
+        )
+        self.__add_light_pause()
+
+    def post_message(self, message, duration=None, pause=True, style="info"):
+        """Post a message on the screen with Messenger.
+        Arguments:
+            message: The message to display.
+            duration: The time until the message vanishes. (Default: 2.55s)
+            pause: If True, the program waits until the message completes.
+            style: "info", "success", or "error"."""
+        driver = self.driver
+        if hasattr(driver, "cdp_base"):
+            driver = driver.cdp_base
+        if style not in ["info", "success", "error"]:
+            style = "info"
+        if not duration:
+            duration = settings.DEFAULT_MESSAGE_DURATION
+        if (
+            (
+                driver.config.headless
+                or (hasattr(sb_config, "xvfb") and sb_config.xvfb)
+            )
+            and float(duration) > 0.75
+        ):
+            duration = 0.75
+        try:
+            js_utils.post_message(self, message, duration, style=style)
+        except Exception:
+            print(" * %s message: %s" % (style.upper(), message))
+        if pause:
+            duration = float(duration) + 0.15
+            time.sleep(float(duration))
+
     def set_locale(self, locale):
         """(Settings will take effect on the next page load)"""
         self.loop.run_until_complete(self.page.set_locale(locale))
@@ -1510,7 +1571,10 @@ def __install_pyautogui_if_missing(self):
                 except Exception:
                     if (
                         shared_utils.is_linux()
-                        and (not sb_config.headed or sb_config.xvfb)
+                        and (
+                            not sb_config.headed
+                            or (hasattr(sb_config, "xvfb") and sb_config.xvfb)
+                        )
                         and not driver.config.headless
                         and (
                             not hasattr(sb_config, "_virtual_display")
@@ -2611,6 +2675,40 @@ def scroll_down(self, amount=25):
         self.loop.run_until_complete(self.page.scroll_down(amount))
         self.loop.run_until_complete(self.page.wait())
 
+    def save_page_source(self, name, folder=None):
+        import codecs
+        from seleniumbase.core import log_helper
+        if not name.endswith(".html"):
+            name = name + ".html"
+        if folder:
+            abs_path = os.path.abspath(".")
+            file_path = os.path.join(abs_path, folder)
+            if not os.path.exists(file_path):
+                os.makedirs(file_path)
+            html_file_path = os.path.join(file_path, name)
+        else:
+            html_file_path = name
+        page_source = self.get_page_source()
+        last_page = self.get_current_url()
+        meta_charset = '<meta charset="utf-8">'
+        rendered_source = ""
+        if "://" in last_page:
+            base_href_html = log_helper.get_base_href_html(last_page)
+            if ' charset="' not in page_source:
+                rendered_source = "%s\n%s\n%s" % (
+                    base_href_html, meta_charset, page_source
+                )
+            else:
+                rendered_source = "%s\n%s" % (base_href_html, page_source)
+        else:
+            rendered_source = page_source
+        html_file = codecs.open(html_file_path, "w+", "utf-8")
+        html_file.write(rendered_source)
+        html_file.close()
+
+    def save_as_html(self, *args, **kwargs):
+        self.save_page_source(*args, **kwargs)
+
     def save_screenshot(self, name, folder=None, selector=None):
         filename = name
         if folder:

seleniumbase/fixtures/base_case.py

@@ -9205,6 +9205,10 @@ def switch_to_newest_tab(self):
         """Same as self.switch_to_newest_window()"""
         self.switch_to_newest_window()
 
+    def save_as_html(self, name, folder=None):
+        """Same as self.save_page_source()"""
+        self.save_page_source(name, folder=folder)
+
     def input(
         self, selector, text, by="css selector", timeout=None, retry=False
     ):
@@ -9654,9 +9658,11 @@ def add_meta_tag(self, http_equiv=None, content=None):
 
     def activate_messenger(self):
         self.__check_scope()
-        self._check_browser()
+        if not self.__is_cdp_swap_needed():
+            self._check_browser()
         js_utils.activate_messenger(self.driver)
-        self.wait_for_ready_state_complete()
+        if not self.__is_cdp_swap_needed():
+            self.wait_for_ready_state_complete()
 
     def set_messenger_theme(
         self, theme="default", location="default", max_messages="default"

seleniumbase/fixtures/js_utils.py

@@ -29,6 +29,8 @@ def wait_for_ready_state_complete(driver, timeout=settings.LARGE_TIMEOUT):
     If the timeout is exceeded, the test will still continue
       because readyState == "interactive" may be good enough.
     (Previously, tests would fail immediately if exceeding the timeout.)"""
+    if hasattr(driver, "_swap_driver"):
+        return
     if hasattr(settings, "SKIP_JS_WAITS") and settings.SKIP_JS_WAITS:
         return
     start_ms = time.time() * 1000.0
@@ -54,8 +56,12 @@ def wait_for_ready_state_complete(driver, timeout=settings.LARGE_TIMEOUT):
 
 
 def execute_async_script(driver, script, timeout=settings.LARGE_TIMEOUT):
-    driver.set_script_timeout(timeout)
-    return driver.execute_async_script(script)
+    if hasattr(driver, "set_script_timeout"):
+        driver.set_script_timeout(timeout)
+    if hasattr(driver, "execute_async_script"):
+        return driver.execute_async_script(script)
+    else:
+        return None
 
 
 def wait_for_angularjs(driver, timeout=settings.LARGE_TIMEOUT, **kwargs):
@@ -937,7 +943,10 @@ def post_message(driver, message, msg_dur=None, style="info"):
         execute_script(driver, messenger_script)
     except TypeError as e:
         if (
-            shared_utils.is_cdp_swap_needed(driver)
+            (
+                shared_utils.is_cdp_swap_needed(driver)
+                or hasattr(driver, "_swap_driver")
+            )
             and "cannot unpack non-iterable" in str(e)
         ):
             pass

seleniumbase/fixtures/page_actions.py

@@ -1528,10 +1528,10 @@ def save_page_source(driver, name, folder=None):
         page_source = driver.cdp.get_page_source()
     else:
         page_source = driver.page_source
-    html_file = codecs.open(html_file_path, "w+", "utf-8")
     rendered_source = log_helper.get_html_source_with_base_href(
         driver, page_source
     )
+    html_file = codecs.open(html_file_path, "w+", "utf-8")
     html_file.write(rendered_source)
     html_file.close()
 

seleniumbase/undetected/cdp_driver/cdp_util.py

@@ -335,6 +335,8 @@ async def start(
             xvfb = True
         else:
             xvfb = False
+    if not hasattr(sb_config, "xvfb"):
+        sb_config.xvfb = xvfb
     if incognito is None:
         if "--incognito" in sys_argv:
             incognito = True

seleniumbase/undetected/cdp_driver/config.py

@@ -202,6 +202,7 @@ def __call__(self):
             "InsecureDownloadWarnings,DownloadBubble,DownloadBubbleV2,"
             "OptimizationTargetPrediction,OptimizationGuideModelDownloading,"
             "SidePanelPinning,UserAgentClientHint,PrivacySandboxSettings4,"
+            "OptimizationHintsFetching,InterestFeedContentSuggestions,"
             "DisableLoadExtensionCommandLineSwitch"
         ]
         if self.proxy: