[py] Add Mutation Logging support

Author: AutomatedTester

py/BUILD.bazel

@@ -29,6 +29,12 @@ copy_file(
     out = "selenium/webdriver/remote/findElements.js",
 )
 
+copy_file(
+    name = "mutation-listener",
+    src = "//javascript/cdp-support:closure",
+    out = "selenium/webdriver/remote/mutation-listener.js"
+)
+
 copy_file(
     name = "firefox-driver-prefs",
     src = "//third_party/js/selenium:webdriver_json",
@@ -46,6 +52,7 @@ py_library(
         ":firefox-driver-prefs",
         ":get-attribute",
         ":is-displayed",
+        ":mutation-listener",
     ] + [":create-cdp-srcs-" + n for n in BROWSER_VERSIONS],
     imports = ["."],
     visibility = ["//visibility:public"],

py/selenium/webdriver/remote/script_key.py

@@ -19,8 +19,8 @@
 
 class ScriptKey:
 
-    def __init__(self):
-        self._id = uuid.uuid4()
+    def __init__(self, id=None):
+        self._id = id or uuid.uuid4()
 
     @property
     def id(self):

py/selenium/webdriver/remote/webdriver.py

@@ -22,6 +22,7 @@
 import copy
 from contextlib import (contextmanager, asynccontextmanager)
 from importlib import import_module
+import json
 import pkgutil
 import warnings
 import sys
@@ -710,13 +711,16 @@ def find_elements_by_css_selector(self, css_selector):
         warnings.warn("find_elements_by_* commands are deprecated. Please use find_elements() instead")
         return self.find_elements(by=By.CSS_SELECTOR, value=css_selector)
 
-    def pin_script(self, script):
+    def pin_script(self, script, script_key=None):
         """
 
         """
-        script_key = ScriptKey()
-        self.pinned_scripts[script_key.id] = script
-        return script_key
+        if not script_key:
+            _script_key = ScriptKey()
+        else:
+            _script_key = ScriptKey(script_key)
+        self.pinned_scripts[_script_key.id] = script
+        return _script_key
 
     def unpin(self, script_key):
         """
@@ -1483,6 +1487,42 @@ def get_log(self, log_type):
         """
         return self.execute(Command.GET_LOG, {'type': log_type})['value']
 
+    @asynccontextmanager
+    async def log_mutation_events(self):
+        """
+        Listens for mutation events and emits them as it finds them
+
+        :Usage:
+             ::
+
+        """
+        _pkg = '.'.join(__name__.split('.')[:-1])
+        mutation_listener_js = pkgutil.get_data(_pkg, 'mutation-listener.js').decode('utf8').strip()
+
+        assert sys.version_info >= (3, 7)
+        global cdp
+        async with self._get_bidi_connection():
+            global devtools
+            page = cdp.get_session_context('page.enable')
+            await page.execute(devtools.page.enable())
+            runtime = cdp.get_session_context('runtime.enable')
+            await runtime.execute(devtools.runtime.enable())
+            await runtime.execute(devtools.runtime.add_binding("__webdriver_attribute"))
+            self.pin_script(mutation_listener_js)
+            script_key = await page.execute(devtools.page.add_script_to_evaluate_on_new_document(mutation_listener_js))
+            self.pin_script(mutation_listener_js, script_key)
+            self.execute_script(f"return {mutation_listener_js}")
+            event = {}
+            async with runtime.wait_for(devtools.runtime.BindingCalled) as evnt:
+                yield event
+
+            payload = json.loads(evnt.value.payload)
+            elements = self.find_elements(By.CSS_SELECTOR, "*[data-__webdriver_id={}".format(payload['target']))
+            # event["element"] = elements[0]
+            event["attribute_name"] = payload['name']
+            event["current_value"] = payload['value']
+            event["old_value"] = payload['oldValue']
+
     @asynccontextmanager
     async def add_js_error_listener(self):
         """

py/test/selenium/webdriver/common/bidi_tests.py

@@ -15,6 +15,9 @@
 # specific language governing permissions and limitations
 # under the License.
 from selenium.webdriver.common.by import By
+from selenium.webdriver.support import expected_conditions as EC
+from selenium.webdriver.support.ui import WebDriverWait
+
 import pytest
 
 
@@ -45,3 +48,16 @@ async def test_collect_js_exceptions(driver, pages):
         driver.find_element(By.ID, "throwing-mouseover").click()
     assert exceptions is not None
     assert exceptions.exception_details.stack_trace.call_frames[0].function_name == "onmouseover"
+
+
+@pytest.mark.xfail_firefox
+@pytest.mark.xfail_safari
+async def test_collect_log_mutations(driver, pages):
+    async with driver.log_mutation_events() as event:
+        pages.load("dynamic.html")
+        driver.find_element(By.ID, "reveal").click()
+        WebDriverWait(driver, 5).until(EC.visibility_of(driver.find_element(By.ID, "revealed")))
+
+    assert event["attribute_name"] == "style"
+    assert event["current_value"] == ""
+    assert event["old_value"] == "display:none;"