camoufox/tests/conftest.py

# Copyright (c) Microsoft Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import asyncio
import inspect
import io
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any, AsyncGenerator, Callable, Dict, Generator, List, Optional, cast

import playwright
import playwright._impl._path_utils
import pytest
from PIL import Image
from pixelmatch import pixelmatch
from pixelmatch.contrib.PIL import from_PIL_to_raw_data
from playwright._impl._path_utils import get_file_dirname

from .server import Server, test_server

_dirname = get_file_dirname()


"""
Patch playwright to not rely on module path for assets.
"""

original_get_file_dirname = playwright._impl._path_utils.get_file_dirname


def _run_in_the_pages_own_world() -> None:
    """Run this suite in the page's world rather than Camoufox's isolated one.

    This is upstream Playwright's conformance suite, so it asserts upstream
    semantics: tests read globals their own page scripts defined, and pass
    element handles into evaluate(). Camoufox evaluates in an isolated world by
    default -- the reason this fork exists -- and about 37 of these tests fail
    on "X is not defined" for a global the page really did set.

    The `mw:` prefix cannot stand in for this. It refuses handles by design
    (Runtime.js), and much of this suite needs them. So isolation is turned off
    for this suite alone. Camoufox's isolated-world behaviour keeps its own
    coverage in tests/patches/isolated-evaluate.py, which must go on passing
    without this flag -- that is the file to check if isolation regresses, not
    this one.
    """
    raw = os.environ.get("CAMOU_CONFIG")
    camou_config = json.loads(raw) if raw else {}
    camou_config["disableWorldIsolation"] = True
    os.environ["CAMOU_CONFIG"] = json.dumps(camou_config)


@pytest.hookimpl(tryfirst=True)
def pytest_configure(config):
    def patched_get_file_dirname():
        return _dirname

    playwright._impl._path_utils.get_file_dirname = patched_get_file_dirname

    _run_in_the_pages_own_world()


@pytest.hookimpl(trylast=True)
def pytest_unconfigure(config):
    playwright._impl._path_utils.get_file_dirname = original_get_file_dirname


def pytest_generate_tests(metafunc: pytest.Metafunc) -> None:
    if "browser_name" in metafunc.fixturenames:
        browsers = ["firefox"]
        metafunc.parametrize("browser_name", browsers, scope="session")


"""
Playwright fixtures.
"""


@pytest.fixture(scope="session")
def event_loop() -> Generator[asyncio.AbstractEventLoop, None, None]:
    # Not asyncio.get_event_loop(): with no running loop that is deprecated on
    # 3.12 and raises RuntimeError on 3.14, which takes down every async test in
    # the suite at fixture setup ("There is no current event loop in thread
    # 'MainThread'") -- 1151 errors that look like browser failures but are not.
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        yield loop
    finally:
        asyncio.set_event_loop(None)
        loop.close()


@pytest.fixture(scope="session")
def assetdir() -> Path:
    return _dirname / "assets"


@pytest.fixture(scope="session")
def headless(pytestconfig: pytest.Config) -> bool:
    return pytestconfig.getoption("--headless") or os.getenv("HEADLESS", False)


@pytest.fixture(scope="session")
def launch_arguments(pytestconfig: pytest.Config, headless: bool) -> Dict:
    args = {
        "headless": headless,
        "channel": pytestconfig.getoption("--browser-channel"),
    }
    executable_path = os.getenv("CAMOUFOX_EXECUTABLE_PATH", None)
    if executable_path:
        args["executable_path"] = os.path.abspath(executable_path)
    return args


@pytest.fixture
def server() -> Generator[Server, None, None]:
    yield test_server.server


@pytest.fixture
def https_server() -> Generator[Server, None, None]:
    yield test_server.https_server


@pytest.fixture(autouse=True, scope="session")
async def start_server() -> AsyncGenerator[None, None]:
    test_server.start()
    yield
    test_server.stop()


@pytest.fixture(autouse=True)
def after_each_hook() -> Generator[None, None, None]:
    yield
    test_server.reset()


@pytest.fixture(scope="session")
def browser_name(pytestconfig: pytest.Config) -> str:
    # Always use Firefox
    return 'firefox'


@pytest.fixture(scope="session")
def browser_channel(pytestconfig: pytest.Config) -> Optional[str]:
    return cast(Optional[str], pytestconfig.getoption("--browser-channel"))


@pytest.fixture(scope="session")
def is_webkit(browser_name: str) -> bool:
    return browser_name == "webkit"


@pytest.fixture(scope="session")
def is_firefox(browser_name: str) -> bool:
    return browser_name == "firefox"


@pytest.fixture(scope="session")
def is_chromium(browser_name: str) -> bool:
    return browser_name == "chromium"


@pytest.fixture(scope="session")
def is_win() -> bool:
    return sys.platform == "win32"


@pytest.fixture(scope="session")
def is_linux() -> bool:
    return sys.platform == "linux"


@pytest.fixture(scope="session")
def is_mac() -> bool:
    return sys.platform == "darwin"


"""
Helper to skip tests by browser or platform.
"""


def _get_skiplist(request: pytest.FixtureRequest, values: List[str], value_name: str) -> List[str]:
    skipped_values = []
    # Allowlist
    only_marker = request.node.get_closest_marker(f"only_{value_name}")
    if only_marker:
        skipped_values = values
        skipped_values.remove(only_marker.args[0])

    # Denylist
    skip_marker = request.node.get_closest_marker(f"skip_{value_name}")
    if skip_marker:
        skipped_values.append(skip_marker.args[0])

    return skipped_values


@pytest.fixture(autouse=True)
def skip_by_browser(request: pytest.FixtureRequest, browser_name: str) -> None:
    skip_browsers_names = _get_skiplist(request, ["chromium", "firefox", "webkit"], "browser")

    if browser_name in skip_browsers_names:
        pytest.skip(f"skipped for this browser: {browser_name}")


@pytest.fixture(autouse=True)
def skip_by_platform(request: pytest.FixtureRequest) -> None:
    skip_platform_names = _get_skiplist(request, ["win32", "linux", "darwin"], "platform")

    if sys.platform in skip_platform_names:
        pytest.skip(f"skipped on this platform: {sys.platform}")


def pytest_addoption(parser: pytest.Parser) -> None:
    group = parser.getgroup("playwright", "Playwright")
    parser.addoption(
        "--headless",
        action="store_true",
        default=False,
        help="Run tests in headless mode.",
    )
    group.addoption(
        "--browser-channel",
        action="store",
        default=None,
        help="Browser channel to be used.",
    )


@pytest.fixture(scope="session")
def assert_to_be_golden(browser_name: str) -> Callable[[bytes, str], None]:
    def compare(received_raw: bytes, golden_name: str) -> None:
        golden_file_path = _dirname / f"golden-{browser_name}" / golden_name
        try:
            golden_file = golden_file_path.read_bytes()
            received_image = Image.open(io.BytesIO(received_raw))
            golden_image = Image.open(io.BytesIO(golden_file))

            if golden_image.size != received_image.size:
                pytest.fail("Image size differs to golden image")
                return
            diff_pixels = pixelmatch(
                from_PIL_to_raw_data(received_image),
                from_PIL_to_raw_data(golden_image),
                golden_image.size[0],
                golden_image.size[1],
                threshold=0.2,
            )
            assert diff_pixels == 0
        except Exception:
            if os.getenv("PW_WRITE_SCREENSHOT"):
                golden_file_path.parent.mkdir(parents=True, exist_ok=True)
                golden_file_path.write_bytes(received_raw)
                print(f"Wrote {golden_file_path}")
            raise

    return compare


def _to_camel_case_keys(options: Dict) -> Dict:
    """snake_case launch options -> the camelCase the Node driver expects.

    Unset options are dropped rather than serialised as null: the driver
    validates types strictly, so a `channel: None` from an unused --browser-channel
    aborts the server with "channel: expected string, got object".
    """

    def camel(key: str) -> str:
        head, *rest = key.split("_")
        return head + "".join(part.title() for part in rest)

    return {camel(key): value for key, value in options.items() if value is not None}


class RemoteServer:
    def __init__(self, browser_name: str, launch_server_options: Dict, tmpfile: Path) -> None:
        driver_dir = Path(inspect.getfile(playwright)).parent / "driver"
        if sys.platform == "win32":
            node_executable = driver_dir / "node.exe"
        else:
            node_executable = driver_dir / "node"
        cli_js = driver_dir / "package" / "cli.js"
        # `launch-server --config` is read by the Node driver as JS launch
        # options, so the keys have to be camelCase. Handing it the Python
        # fixture's snake_case `executable_path` silently dropped it: the server
        # fell back to Playwright's bundled Firefox, failed with "Executable
        # doesn't exist at .../firefox-1522/firefox", printed no endpoint, and
        # every connect test then died on an empty ws_endpoint with the
        # misleading "Port should be >= 0 and < 65536. Received type string ('')".
        tmpfile.write_text(json.dumps(_to_camel_case_keys(launch_server_options)))
        self.process = subprocess.Popen(
            [
                str(node_executable),
                str(cli_js),
                "launch-server",
                "--browser",
                browser_name,
                "--config",
                str(tmpfile),
            ],
            stdout=subprocess.PIPE,
            stderr=sys.stderr,
            cwd=driver_dir,
        )
        assert self.process.stdout
        self.ws_endpoint = self.process.stdout.readline().decode().strip()
        self.process.stdout.close()

    def kill(self) -> None:
        # Send the signal to all the process groups
        if self.process.poll() is not None:
            return
        self.process.kill()
        self.process.wait()


@pytest.fixture
def launch_server(
    browser_name: str, launch_arguments: Dict, tmp_path: Path
) -> Generator[Callable[..., RemoteServer], None, None]:
    remotes: List[RemoteServer] = []

    def _launch_server(**kwargs: Dict[str, Any]) -> RemoteServer:
        remote = RemoteServer(
            browser_name,
            {
                **launch_arguments,
                **kwargs,
            },
            tmp_path / f"settings-{len(remotes)}.json",
        )
        remotes.append(remote)
        return remote

    yield _launch_server

    for remote in remotes:
        remote.kill()
        remote.kill()