From 26a029d407be480d791972afb5975cf62c9360a6 Mon Sep 17 00:00:00 2001
From: Daniel Baumann <daniel.baumann@progress-linux.org>
Date: Fri, 19 Apr 2024 02:47:55 +0200
Subject: Adding upstream version 124.0.1.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
---
 testing/webcompat/client.py | 759 ++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 759 insertions(+)
 create mode 100644 testing/webcompat/client.py

(limited to 'testing/webcompat/client.py')

diff --git a/testing/webcompat/client.py b/testing/webcompat/client.py
new file mode 100644
index 0000000000..88208e5507
--- /dev/null
+++ b/testing/webcompat/client.py
@@ -0,0 +1,759 @@
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at http://mozilla.org/MPL/2.0/.
+
+import asyncio
+import contextlib
+import time
+from urllib.parse import quote
+
+import webdriver
+from webdriver.bidi.modules.script import ContextTarget
+
+
+class Client:
+    def __init__(self, session, event_loop):
+        self.session = session
+        self.event_loop = event_loop
+        self.content_blocker_loaded = False
+
+    @property
+    def current_url(self):
+        return self.session.url
+
+    @property
+    def alert(self):
+        return self.session.alert
+
+    @property
+    def context(self):
+        return self.session.send_session_command("GET", "moz/context")
+
+    @context.setter
+    def context(self, context):
+        self.session.send_session_command("POST", "moz/context", {"context": context})
+
+    @contextlib.contextmanager
+    def using_context(self, context):
+        orig_context = self.context
+        needs_change = context != orig_context
+
+        if needs_change:
+            self.context = context
+
+        try:
+            yield
+        finally:
+            if needs_change:
+                self.context = orig_context
+
+    def wait_for_content_blocker(self):
+        if not self.content_blocker_loaded:
+            with self.using_context("chrome"):
+                self.session.execute_async_script(
+                    """
+                    const done = arguments[0],
+                          signal = "safebrowsing-update-finished";
+                    function finish() {
+                        Services.obs.removeObserver(finish, signal);
+                        done();
+                    }
+                    Services.obs.addObserver(finish, signal);
+                """
+                )
+                self.content_blocker_loaded = True
+
+    @property
+    def keyboard(self):
+        return self.session.actions.sequence("key", "keyboard_id")
+
+    @property
+    def mouse(self):
+        return self.session.actions.sequence(
+            "pointer", "pointer_id", {"pointerType": "mouse"}
+        )
+
+    @property
+    def pen(self):
+        return self.session.actions.sequence(
+            "pointer", "pointer_id", {"pointerType": "pen"}
+        )
+
+    @property
+    def touch(self):
+        return self.session.actions.sequence(
+            "pointer", "pointer_id", {"pointerType": "touch"}
+        )
+
+    @property
+    def wheel(self):
+        return self.session.actions.sequence("wheel", "wheel_id")
+
+    @property
+    def modifier_key(self):
+        if self.session.capabilities["platformName"] == "mac":
+            return "\ue03d"  # meta (command)
+        else:
+            return "\ue009"  # control
+
+    def inline(self, doc):
+        return "data:text/html;charset=utf-8,{}".format(quote(doc))
+
+    async def top_context(self):
+        contexts = await self.session.bidi_session.browsing_context.get_tree()
+        return contexts[0]
+
+    async def navigate(self, url, timeout=None, **kwargs):
+        return await asyncio.wait_for(
+            asyncio.ensure_future(self._navigate(url, **kwargs)), timeout=timeout
+        )
+
+    async def _navigate(self, url, wait="complete", await_console_message=None):
+        if self.session.test_config.get("use_pbm") or self.session.test_config.get(
+            "use_strict_etp"
+        ):
+            print("waiting for content blocker...")
+            self.wait_for_content_blocker()
+        if await_console_message is not None:
+            console_message = await self.promise_console_message_listener(
+                await_console_message
+            )
+        if wait == "load":
+            page_load = await self.promise_readystate_listener("load", url=url)
+        try:
+            await self.session.bidi_session.browsing_context.navigate(
+                context=(await self.top_context())["context"],
+                url=url,
+                wait=wait if wait != "load" else None,
+            )
+        except webdriver.bidi.error.UnknownErrorException as u:
+            m = str(u)
+            if (
+                "NS_BINDING_ABORTED" not in m
+                and "NS_ERROR_ABORT" not in m
+                and "NS_ERROR_WONT_HANDLE_CONTENT" not in m
+            ):
+                raise u
+        if wait == "load":
+            await page_load
+        if await_console_message is not None:
+            await console_message
+
+    async def promise_event_listener(self, events, check_fn=None, timeout=20):
+        if type(events) is not list:
+            events = [events]
+
+        await self.session.bidi_session.session.subscribe(events=events)
+
+        future = self.event_loop.create_future()
+
+        listener_removers = []
+
+        def remove_listeners():
+            for listener_remover in listener_removers:
+                try:
+                    listener_remover()
+                except Exception:
+                    pass
+
+        async def on_event(method, data):
+            print("on_event", method, data)
+            val = None
+            if check_fn is not None:
+                val = check_fn(method, data)
+                if val is None:
+                    return
+            future.set_result(val)
+
+        for event in events:
+            r = self.session.bidi_session.add_event_listener(event, on_event)
+            listener_removers.append(r)
+
+        async def task():
+            try:
+                return await asyncio.wait_for(future, timeout=timeout)
+            finally:
+                remove_listeners()
+                try:
+                    await asyncio.wait_for(
+                        self.session.bidi_session.session.unsubscribe(events=events),
+                        timeout=4,
+                    )
+                except asyncio.exceptions.TimeoutError:
+                    print("Unexpectedly timed out unsubscribing", events)
+                    pass
+
+        return asyncio.create_task(task())
+
+    async def promise_console_message_listener(self, msg, **kwargs):
+        def check(method, data):
+            if "text" in data:
+                if msg in data["text"]:
+                    return data
+            if "args" in data and len(data["args"]):
+                for arg in data["args"]:
+                    if "value" in arg and msg in arg["value"]:
+                        return data
+
+        return await self.promise_event_listener("log.entryAdded", check, **kwargs)
+
+    async def is_console_message(self, message):
+        try:
+            await (await self.promise_console_message_listener(message, timeout=2))
+            return True
+        except asyncio.exceptions.TimeoutError:
+            return False
+
+    async def promise_readystate_listener(self, state, url=None, **kwargs):
+        event = f"browsingContext.{state}"
+
+        def check(method, data):
+            if url is None or url in data["url"]:
+                return data
+
+        return await self.promise_event_listener(event, check, **kwargs)
+
+    async def promise_frame_listener(self, url, state="domContentLoaded", **kwargs):
+        event = f"browsingContext.{state}"
+
+        def check(method, data):
+            if url is None or url in data["url"]:
+                return Client.Context(self, data["context"])
+
+        return await self.promise_event_listener(event, check, **kwargs)
+
+    async def find_frame_context_by_url(self, url):
+        def find_in(arr, url):
+            for context in arr:
+                if url in context["url"]:
+                    return context
+            for context in arr:
+                found = find_in(context["children"], url)
+                if found:
+                    return found
+
+        return find_in([await self.top_context()], url)
+
+    class Context:
+        def __init__(self, client, id):
+            self.client = client
+            self.target = ContextTarget(id)
+
+        async def find_css(self, selector, all=False):
+            all = "All" if all else ""
+            return await self.client.session.bidi_session.script.evaluate(
+                expression=f"document.querySelector{all}('{selector}')",
+                target=self.target,
+                await_promise=False,
+            )
+
+        def timed_js(self, timeout, poll, fn, is_displayed=False):
+            return f"""() => new Promise((_good, _bad) => {{
+                        {self.is_displayed_js()}
+                        var _poll = {poll} * 1000;
+                        var _time = {timeout} * 1000;
+                        var _done = false;
+                        var resolve = val => {{
+                            if ({is_displayed}) {{
+                                if (val.length) {{
+                                    val = val.filter(v = is_displayed(v));
+                                }} else {{
+                                    val = is_displayed(val) && val;
+                                }}
+                                if (!val.length && !val.matches) {{
+                                    return;
+                                }}
+                            }}
+                            _done = true;
+                            clearInterval(_int);
+                            _good(val);
+                        }};
+                        var reject = str => {{
+                            _done = true;
+                            clearInterval(_int);
+                            _bad(val);
+                        }};
+                        var _int = setInterval(() => {{
+                            {fn};
+                            if (!_done) {{
+                                _time -= _poll;
+                                if (_time <= 0) {{
+                                    reject();
+                                }}
+                            }}
+                        }}, poll);
+                    }})"""
+
+        def is_displayed_js(self):
+            return """
+                function is_displayed(e) {
+                    const s = window.getComputedStyle(e),
+                          v = s.visibility === "visible",
+                          o = Math.abs(parseFloat(s.opacity));
+                    return e.getClientRects().length > 0 && v && (isNaN(o) || o === 1.0);
+                }
+                """
+
+        async def await_css(
+            self,
+            selector,
+            all=False,
+            timeout=10,
+            poll=0.25,
+            condition=False,
+            is_displayed=False,
+        ):
+            all = "All" if all else ""
+            condition = (
+                f"var elem=arguments[0]; if ({condition})" if condition else False
+            )
+            return await self.client.session.bidi_session.script.evaluate(
+                expression=self.timed_js(
+                    timeout,
+                    poll,
+                    f"""
+                    var ele = document.querySelector{all}('{selector}')";
+                    if (ele && (!"length" in ele || ele.length > 0)) {{
+                        '{condition}'
+                        resolve(ele);
+                    }}
+                    """,
+                ),
+                target=self.target,
+                await_promise=True,
+            )
+
+        async def await_text(self, text, **kwargs):
+            xpath = f"//*[contains(text(),'{text}')]"
+            return await self.await_xpath(self, xpath, **kwargs)
+
+        async def await_xpath(
+            self, xpath, all=False, timeout=10, poll=0.25, is_displayed=False
+        ):
+            all = "true" if all else "false"
+            return await self.client.session.bidi_session.script.evaluate(
+                expression=self.timed_js(
+                    timeout,
+                    poll,
+                    """
+                    var ret = [];
+                    var r, res = document.evaluate(`{xpath}`, document, null, 4);
+                    while (r = res.iterateNext()) {
+                        ret.push(r);
+                    }
+                    resolve({all} ? ret : ret[0]);
+                    """,
+                ),
+                target=self.target,
+                await_promise=True,
+            )
+
+    def wrap_script_args(self, args):
+        if args is None:
+            return args
+        out = []
+        for arg in args:
+            if arg is None:
+                out.append({"type": "undefined"})
+                continue
+            t = type(arg)
+            if t == int or t == float:
+                out.append({"type": "number", "value": arg})
+            elif t == bool:
+                out.append({"type": "boolean", "value": arg})
+            elif t == str:
+                out.append({"type": "string", "value": arg})
+            else:
+                if "type" in arg:
+                    out.push(arg)
+                    continue
+                raise ValueError(f"Unhandled argument type: {t}")
+        return out
+
+    class PreloadScript:
+        def __init__(self, client, script, target):
+            self.client = client
+            self.script = script
+            if type(target) == list:
+                self.target = target[0]
+            else:
+                self.target = target
+
+        def stop(self):
+            return self.client.session.bidi_session.script.remove_preload_script(
+                script=self.script
+            )
+
+        async def run(self, fn, *args, await_promise=False):
+            val = await self.client.session.bidi_session.script.call_function(
+                arguments=self.client.wrap_script_args(args),
+                await_promise=await_promise,
+                function_declaration=fn,
+                target=self.target,
+            )
+            if val and "value" in val:
+                return val["value"]
+            return val
+
+    async def make_preload_script(self, text, sandbox, args=None, context=None):
+        if not context:
+            context = (await self.top_context())["context"]
+        target = ContextTarget(context, sandbox)
+        if args is None:
+            text = f"() => {{ {text} }}"
+        script = await self.session.bidi_session.script.add_preload_script(
+            function_declaration=text,
+            arguments=self.wrap_script_args(args),
+            sandbox=sandbox,
+        )
+        return Client.PreloadScript(self, script, target)
+
+    async def await_alert(self, text):
+        if not hasattr(self, "alert_preload_script"):
+            self.alert_preload_script = await self.make_preload_script(
+                """
+                    window.__alerts = [];
+                    window.wrappedJSObject.alert = function(text) {
+                        window.__alerts.push(text);
+                    }
+                """,
+                "alert_detector",
+            )
+        return self.alert_preload_script.run(
+            """(msg) => new Promise(done => {
+                    const to = setInterval(() => {
+                        if (window.__alerts.includes(msg)) {
+                            clearInterval(to);
+                            done();
+                        }
+                    }, 200);
+               })
+            """,
+            text,
+            await_promise=True,
+        )
+
+    async def await_popup(self, url=None):
+        if not hasattr(self, "popup_preload_script"):
+            self.popup_preload_script = await self.make_preload_script(
+                """
+                    window.__popups = [];
+                    window.wrappedJSObject.open = function(url) {
+                        window.__popups.push(url);
+                    }
+                """,
+                "popup_detector",
+            )
+        return self.popup_preload_script.run(
+            """(url) => new Promise(done => {
+                    const to = setInterval(() => {
+                        if (url === undefined && window.__popups.length) {
+                            clearInterval(to);
+                            return done(window.__popups[0]);
+                        }
+                        const found = window.__popups.find(u => u.includes(url));
+                        if (found !== undefined) {
+                            clearInterval(to);
+                            done(found);
+                        }
+                    }, 1000);
+               })
+            """,
+            url,
+            await_promise=True,
+        )
+
+    async def track_listener(self, type, selector):
+        if not hasattr(self, "listener_preload_script"):
+            self.listener_preload_script = await self.make_preload_script(
+                """
+                window.__listeners = {};
+                var proto = EventTarget.wrappedJSObject.prototype;
+                var def = Object.getOwnPropertyDescriptor(proto, "addEventListener");
+                var old = def.value;
+                def.value = function(type, fn, opts) {
+                    if ("matches" in this) {
+                        if (!window.__listeners[type]) {
+                            window.__listeners[type] = new Set();
+                        }
+                        window.__listeners[type].add(this);
+                    }
+                    return old.call(this, type, fn, opts)
+                };
+                Object.defineProperty(proto, "addEventListener", def);
+            """,
+                "listener_detector",
+            )
+        return Client.ListenerTracker(self.listener_preload_script, type, selector)
+
+    @contextlib.asynccontextmanager
+    async def preload_script(self, text, *args):
+        script = await self.make_preload_script(text, "preload", args=args)
+        yield script
+        await script.stop()
+
+    def back(self):
+        self.session.back()
+
+    def switch_to_frame(self, frame):
+        return self.session.transport.send(
+            "POST",
+            "session/{session_id}/frame".format(**vars(self.session)),
+            {"id": frame},
+            encoder=webdriver.protocol.Encoder,
+            decoder=webdriver.protocol.Decoder,
+            session=self.session,
+        )
+
+    def switch_frame(self, frame):
+        self.session.switch_frame(frame)
+
+    async def load_page_and_wait_for_iframe(
+        self, url, finder, loads=1, timeout=None, **kwargs
+    ):
+        while loads > 0:
+            await self.navigate(url, **kwargs)
+            frame = self.await_element(finder, timeout=timeout)
+            loads -= 1
+        self.switch_frame(frame)
+        return frame
+
+    def execute_script(self, script, *args):
+        return self.session.execute_script(script, args=args)
+
+    def execute_async_script(self, script, *args, **kwargs):
+        return self.session.execute_async_script(script, args, **kwargs)
+
+    def clear_all_cookies(self):
+        self.session.transport.send(
+            "DELETE", "session/%s/cookie" % self.session.session_id
+        )
+
+    def send_element_command(self, element, method, uri, body=None):
+        url = "element/%s/%s" % (element.id, uri)
+        return self.session.send_session_command(method, url, body)
+
+    def get_element_attribute(self, element, name):
+        return self.send_element_command(element, "GET", "attribute/%s" % name)
+
+    def _do_is_displayed_check(self, ele, is_displayed):
+        if ele is None:
+            return None
+
+        if type(ele) in [list, tuple]:
+            return [x for x in ele if self._do_is_displayed_check(x, is_displayed)]
+
+        if is_displayed is False and ele and self.is_displayed(ele):
+            return None
+        if is_displayed is True and ele and not self.is_displayed(ele):
+            return None
+        return ele
+
+    def find_css(self, *args, all=False, is_displayed=None, **kwargs):
+        try:
+            ele = self.session.find.css(*args, all=all, **kwargs)
+            return self._do_is_displayed_check(ele, is_displayed)
+        except webdriver.error.NoSuchElementException:
+            return None
+
+    def find_xpath(self, xpath, all=False, is_displayed=None):
+        route = "elements" if all else "element"
+        body = {"using": "xpath", "value": xpath}
+        try:
+            ele = self.session.send_session_command("POST", route, body)
+            return self._do_is_displayed_check(ele, is_displayed)
+        except webdriver.error.NoSuchElementException:
+            return None
+
+    def find_text(self, text, is_displayed=None, **kwargs):
+        try:
+            ele = self.find_xpath(f"//*[contains(text(),'{text}')]", **kwargs)
+            return self._do_is_displayed_check(ele, is_displayed)
+        except webdriver.error.NoSuchElementException:
+            return None
+
+    def find_element(self, finder, is_displayed=None, **kwargs):
+        ele = finder.find(self, **kwargs)
+        return self._do_is_displayed_check(ele, is_displayed)
+
+    def await_css(self, selector, **kwargs):
+        return self.await_element(self.css(selector), **kwargs)
+
+    def await_xpath(self, selector, **kwargs):
+        return self.await_element(self.xpath(selector), **kwargs)
+
+    def await_text(self, selector, *args, **kwargs):
+        return self.await_element(self.text(selector), **kwargs)
+
+    def await_element(self, finder, **kwargs):
+        return self.await_first_element_of([finder], **kwargs)[0]
+
+    class css:
+        def __init__(self, selector):
+            self.selector = selector
+
+        def find(self, client, **kwargs):
+            return client.find_css(self.selector, **kwargs)
+
+    class xpath:
+        def __init__(self, selector):
+            self.selector = selector
+
+        def find(self, client, **kwargs):
+            return client.find_xpath(self.selector, **kwargs)
+
+    class text:
+        def __init__(self, selector):
+            self.selector = selector
+
+        def find(self, client, **kwargs):
+            return client.find_text(self.selector, **kwargs)
+
+    def await_first_element_of(
+        self, finders, timeout=None, delay=0.25, condition=False, **kwargs
+    ):
+        t0 = time.time()
+        condition = f"var elem=arguments[0]; return {condition}" if condition else False
+
+        if timeout is None:
+            timeout = 10
+
+        found = [None for finder in finders]
+
+        exc = None
+        while time.time() < t0 + timeout:
+            for i, finder in enumerate(finders):
+                try:
+                    result = finder.find(self, **kwargs)
+                    if result and (
+                        not condition
+                        or self.session.execute_script(condition, [result])
+                    ):
+                        found[i] = result
+                        return found
+                except webdriver.error.NoSuchElementException as e:
+                    exc = e
+            time.sleep(delay)
+        raise exc if exc is not None else webdriver.error.NoSuchElementException
+        return found
+
+    async def dom_ready(self, timeout=None):
+        if timeout is None:
+            timeout = 20
+
+        async def wait():
+            return self.session.execute_async_script(
+                """
+                const cb = arguments[0];
+                setInterval(() => {
+                    if (document.readyState === "complete") {
+                        cb();
+                    }
+                }, 500);
+            """
+            )
+
+        task = asyncio.create_task(wait())
+        return await asyncio.wait_for(task, timeout)
+
+    def is_float_cleared(self, elem1, elem2):
+        return self.session.execute_script(
+            """return (function(a, b) {
+            // Ensure that a is placed under b (and not to its right)
+            return a?.offsetTop >= b?.offsetTop + b?.offsetHeight &&
+                   a?.offsetLeft < b?.offsetLeft + b?.offsetWidth;
+            }(arguments[0], arguments[1]));""",
+            elem1,
+            elem2,
+        )
+
+    @contextlib.contextmanager
+    def assert_getUserMedia_called(self):
+        self.execute_script(
+            """
+            navigator.mediaDevices.getUserMedia =
+                navigator.mozGetUserMedia =
+                navigator.getUserMedia =
+                () => { window.__gumCalled = true; };
+        """
+        )
+        yield
+        assert self.execute_script("return window.__gumCalled === true;")
+
+    def await_element_hidden(self, finder, timeout=None, delay=0.25):
+        t0 = time.time()
+
+        if timeout is None:
+            timeout = 20
+
+        elem = finder.find(self)
+        while time.time() < t0 + timeout:
+            try:
+                if not self.is_displayed(elem):
+                    return
+                time.sleep(delay)
+            except webdriver.error.StaleElementReferenceException:
+                return
+
+    def soft_click(self, element):
+        self.execute_script("arguments[0].click()", element)
+
+    def remove_element(self, element):
+        self.execute_script("arguments[0].remove()", element)
+
+    def scroll_into_view(self, element):
+        self.execute_script(
+            "arguments[0].scrollIntoView({block:'center', inline:'center', behavior: 'instant'})",
+            element,
+        )
+
+    @contextlib.asynccontextmanager
+    async def ensure_fastclick_activates(self):
+        fastclick_preload_script = await self.make_preload_script(
+            """
+                var _ = document.createElement("webcompat_test");
+                _.style = "position:absolute;right:-1px;width:1px;height:1px";
+                document.documentElement.appendChild(_);
+            """,
+            "fastclick_forcer",
+        )
+        yield
+        fastclick_preload_script.stop()
+
+    def test_for_fastclick(self, element):
+        # FastClick cancels touchend, breaking default actions on Fenix.
+        # It instead fires a mousedown or click, which we can detect.
+        self.execute_script(
+            """
+                const sel = arguments[0];
+                window.fastclicked = false;
+                const evt = sel.nodeName === "SELECT" ? "mousedown" : "click";
+                document.addEventListener(evt, e => {
+                    if (e.target === sel && !e.isTrusted) {
+                        window.fastclicked = true;
+                    }
+                }, true);
+            """,
+            element,
+        )
+        self.scroll_into_view(element)
+        # tap a few times in case the site's other code interferes
+        self.touch.click(element=element).perform()
+        self.touch.click(element=element).perform()
+        self.touch.click(element=element).perform()
+        return self.execute_script("return window.fastclicked")
+
+    def is_displayed(self, element):
+        if element is None:
+            return False
+
+        return self.session.execute_script(
+            """
+              const e = arguments[0],
+                    s = window.getComputedStyle(e),
+                    v = s.visibility === "visible",
+                    o = Math.abs(parseFloat(s.opacity));
+              return e.getClientRects().length > 0 && v && (isNaN(o) || o === 1.0);
+          """,
+            args=[element],
+        )
-- 
cgit v1.2.3