test_playwrightprovider.py

from .conftest import (
    _resolve_shim_target,
    copy_seeded_playwright_root,
)
import asyncio
import sys
import tempfile
import threading
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
from typing import Any

import abxbus

from abxpkg import Binary, PlaywrightProvider
from abxpkg.binary_service import BinaryEvent, BinaryRequestEvent, BinaryService


class TestPlaywrightProvider:
    def test_concurrent_browser_installs_coordinate_system_dependencies(
        self,
        tmp_path,
        seeded_playwright_root,
    ):
        # Separate browser roots still mutate the same host APT database when
        # Playwright installs Linux libraries. Exercise real installers together,
        # reusing downloaded browser bytes so this tests coordination, not CDN speed.
        providers = []
        for index in range(3):
            root = tmp_path / str(index)
            copy_seeded_playwright_root(seeded_playwright_root, root)
            provider = PlaywrightProvider(install_root=root)
            provider.INSTALLER_BINARY()
            providers.append(provider)
        ready = threading.Barrier(len(providers))

        def install(provider):
            ready.wait()
            provider.default_install_handler(
                "chromium",
                install_args=("--with-deps", "chromium", "--no-shell"),
            )
            loaded = provider.load("chromium", no_cache=True)
            assert loaded is not None
            assert loaded.loaded_abspath.is_file()
            assert loaded.loaded_version

        with ThreadPoolExecutor(max_workers=len(providers)) as pool:
            list(pool.map(install, providers))

    def test_chrome_aliases_install_chromium(self):
        provider = PlaywrightProvider()

        assert provider.install_timeout >= 300
        assert provider.get_install_args("chromium") == ("--with-deps", "chromium")
        assert provider.get_install_args("chrome") == ("--with-deps", "chromium")
        assert provider._normalize_install_args(
            ["chrome@latest", "--no-shell"],
        ) == ["chromium", "--no-shell"]
        assert provider._normalize_install_args(
            ["chromium@latest", "--no-shell"],
        ) == ["chromium", "--no-shell"]
        dry_run_provider = provider.get_provider_with_overrides(dry_run=True)
        assert (
            dry_run_provider.default_install_handler(
                "chrome",
                install_args=("chrome@latest", "--no-shell"),
            )
            == "DRY_RUN would run: playwright install chromium --no-shell"
        )

    def test_chromium_install_puts_real_browser_into_managed_bin_dir(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            playwright_root = Path(temp_dir) / "playwright-root"
            copy_seeded_playwright_root(seeded_playwright_root, playwright_root)
            provider = PlaywrightProvider(install_root=playwright_root)

            installed = provider.load("chromium", no_cache=True)
            assert installed is not None
            test_machine.assert_shallow_binary_loaded(
                installed,
                assert_version_command=False,
            )
            assert installed.name == "chromium"
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.exists()
            assert provider.bin_dir is not None
            assert installed.loaded_abspath.parent == provider.bin_dir
            assert installed.loaded_abspath == provider.bin_dir / "chromium"
            # The shim resolves into ``playwright_root/cache`` (the
            # managed ``PLAYWRIGHT_BROWSERS_PATH`` for this provider).
            real_target = _resolve_shim_target(installed.loaded_abspath)
            assert (playwright_root / "cache").resolve() in real_target.parents
            # Playwright lays out chromium builds as chromium-<build>/.
            assert any(
                child.name.startswith("chromium-")
                for child in (playwright_root / "cache").iterdir()
                if child.is_dir()
            )

            loaded = provider.load("chromium", no_cache=True)
            test_machine.assert_shallow_binary_loaded(
                loaded,
                assert_version_command=False,
            )
            assert loaded is not None
            assert loaded.loaded_abspath is not None
            assert loaded.loaded_abspath.resolve() == installed.loaded_abspath.resolve()

            loaded_or_installed = provider.install("chromium")
            test_machine.assert_shallow_binary_loaded(
                loaded_or_installed,
                assert_version_command=False,
            )
            assert loaded_or_installed is not None
            assert loaded_or_installed.loaded_abspath is not None
            assert (
                loaded_or_installed.loaded_abspath.resolve()
                == installed.loaded_abspath.resolve()
            )

    def test_binary_service_reuses_chromium_request_projection_as_runtime_user(
        self,
        tmp_path,
        seeded_playwright_root,
    ):
        install_root = tmp_path / "playwright"
        copy_seeded_playwright_root(seeded_playwright_root, install_root)

        async def run(run_id: int) -> BinaryEvent:
            bus = abxbus.EventBus(name=f"test_playwright_projection_{run_id}")
            BinaryService(
                bus,
                auto_install=False,
                lib_dir=tmp_path,
            )
            request = BinaryRequestEvent(
                name="chromium",
                binproviders="playwright",
            )
            await bus.emit(request).now()
            await bus.wait_until_idle()
            result = await bus.find(
                BinaryEvent,
                child_of=request,
                past=True,
                future=False,
                name="chromium",
            )
            await bus.destroy(clear=False)
            assert isinstance(result, BinaryEvent)
            return result

        first = asyncio.run(run(1))
        assert Path(first.abspath).is_file()

        binary_load_calls = 0

        def count_binary_loads(frame: Any, event: str, arg: Any) -> None:
            del arg
            nonlocal binary_load_calls
            if event == "call" and frame.f_code is BinaryService._load.__code__:
                binary_load_calls += 1

        sys.setprofile(count_binary_loads)
        threading.setprofile(count_binary_loads)
        try:
            second = asyncio.run(run(2))
        finally:
            sys.setprofile(None)
            threading.setprofile(None)

        assert Path(second.abspath).resolve() == Path(first.abspath).resolve()
        assert binary_load_calls == 0

    def test_install_root_alias_without_explicit_bin_dir_uses_root_bin(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            install_root = Path(temp_dir) / "pw-root"
            copy_seeded_playwright_root(seeded_playwright_root, install_root)
            provider = PlaywrightProvider.model_validate(
                {
                    "install_root": install_root,
                },
            )

            installed = provider.install("chromium")

            test_machine.assert_shallow_binary_loaded(
                installed,
                assert_version_command=False,
            )
            assert installed is not None
            assert provider.install_root == install_root.resolve()
            assert provider.bin_dir == (install_root / "bin").resolve()
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.parent == provider.bin_dir
            # Chromium build landed directly under install_root (which is
            # the effective PLAYWRIGHT_BROWSERS_PATH for this provider).
            assert any(
                child.name.startswith("chromium-")
                for child in (install_root / "cache").iterdir()
                if child.is_dir()
            )

    def test_install_root_and_bin_dir_aliases_install_into_the_requested_paths(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            install_root = Path(temp_dir) / "pw-root"
            bin_dir = Path(temp_dir) / "custom-bin"
            copy_seeded_playwright_root(seeded_playwright_root, install_root)
            provider = PlaywrightProvider.model_validate(
                {
                    "install_root": install_root,
                    "bin_dir": bin_dir,
                },
            )

            installed = provider.install("chromium")

            test_machine.assert_shallow_binary_loaded(
                installed,
                assert_version_command=False,
            )
            assert installed is not None
            assert provider.install_root == install_root.resolve()
            assert provider.bin_dir == bin_dir.resolve()
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.parent == bin_dir.resolve()
            # Browser tree still landed in install_root, not bin_dir.
            assert any(
                child.name.startswith("chromium-")
                for child in (install_root / "cache").iterdir()
                if child.is_dir()
            )

    def test_explicit_bin_dir_takes_precedence_over_existing_PATH_entries(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            temp_dir_path = Path(temp_dir)
            ambient_root = temp_dir_path / "ambient-root"
            copy_seeded_playwright_root(seeded_playwright_root, ambient_root)
            ambient_provider = PlaywrightProvider(
                install_root=ambient_root,
                bin_dir=ambient_root / "bin",
            )
            ambient_installed = ambient_provider.install("chromium")
            assert ambient_installed is not None
            assert ambient_installed.loaded_abspath is not None
            assert ambient_installed.loaded_abspath.parent == ambient_provider.bin_dir

            install_root = temp_dir_path / "playwright-root"
            copy_seeded_playwright_root(seeded_playwright_root, install_root)
            provider = PlaywrightProvider(
                PATH=str(ambient_provider.bin_dir),
                install_root=install_root,
                bin_dir=temp_dir_path / "custom-bin",
            )

            installed = provider.install("chromium")

            test_machine.assert_shallow_binary_loaded(
                installed,
                assert_version_command=False,
            )
            assert installed is not None
            assert provider.bin_dir == (temp_dir_path / "custom-bin").resolve()
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.parent == provider.bin_dir
            # The two providers resolve to different on-disk symlinks.
            assert installed.loaded_abspath != ambient_installed.loaded_abspath

    def test_provider_install_args_are_passed_through_to_playwright_install(
        self,
        test_machine,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            playwright_root = Path(temp_dir) / "playwright-root"
            # Use ``--no-shell`` to skip the headless shell download, which
            # is a real playwright install flag we can verify by checking
            # that ``chromium_headless_shell-*`` does NOT end up on disk.
            provider = PlaywrightProvider(
                install_root=playwright_root,
            ).get_provider_with_overrides(
                overrides={"chromium": {"install_args": ["chromium", "--no-shell"]}},
            )

            installed = provider.install("chromium")

            test_machine.assert_shallow_binary_loaded(
                installed,
                assert_version_command=False,
            )
            assert installed is not None
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.exists()
            cache_dir = playwright_root / "cache"
            chromium_dirs = [
                child
                for child in cache_dir.iterdir()
                if child.is_dir()
                and child.name.startswith("chromium-")
                and not child.name.startswith("chromium_headless_shell")
            ]
            assert chromium_dirs, "chromium-<build> dir should exist on disk"
            # ``--no-shell`` should have skipped the headless shell download.
            headless_shell_dirs = [
                child
                for child in cache_dir.iterdir()
                if child.is_dir() and child.name.startswith("chromium_headless_shell")
            ]
            assert not headless_shell_dirs, (
                f"--no-shell should have skipped chromium_headless_shell, "
                f"but found: {[p.name for p in headless_shell_dirs]}"
            )

    def test_provider_direct_methods_exercise_real_lifecycle(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            install_root = Path(temp_dir) / "playwright-root"
            copy_seeded_playwright_root(seeded_playwright_root, install_root)
            provider = PlaywrightProvider(install_root=install_root)

            loaded = provider.load("chromium", no_cache=True)
            test_machine.assert_shallow_binary_loaded(
                loaded,
                assert_version_command=False,
            )

            loaded_or_installed = provider.install("chromium")
            test_machine.assert_shallow_binary_loaded(
                loaded_or_installed,
                assert_version_command=False,
            )

            updated = provider.update("chromium", no_cache=True)
            test_machine.assert_shallow_binary_loaded(
                updated,
                assert_version_command=False,
            )

            assert provider.uninstall("chromium", no_cache=True) is True
            test_machine.assert_provider_missing(provider, "chromium")

    def test_binary_direct_methods_exercise_real_lifecycle(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            install_root = Path(temp_dir) / "playwright-root"
            copy_seeded_playwright_root(seeded_playwright_root, install_root)
            binary = Binary(
                name="chromium",
                binproviders=[
                    PlaywrightProvider(
                        install_root=install_root,
                    ),
                ],
            )

            loaded = binary.load(no_cache=True)
            test_machine.assert_shallow_binary_loaded(
                loaded,
                assert_version_command=False,
            )

            loaded_or_installed = test_machine.unloaded_binary(binary).install()
            test_machine.assert_shallow_binary_loaded(
                loaded_or_installed,
                assert_version_command=False,
            )

            updated = loaded.update()
            test_machine.assert_shallow_binary_loaded(
                updated,
                assert_version_command=False,
            )

            removed = updated.uninstall()
            assert not removed.is_valid
            assert removed.loaded_binprovider is None
            assert removed.loaded_abspath is None
            assert removed.loaded_version is None
            assert removed.loaded_sha256 is None
            test_machine.assert_binary_missing(binary)

    def test_update_refreshes_chromium_in_place(
        self,
        test_machine,
        seeded_playwright_root,
    ):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            playwright_root = Path(temp_dir) / "playwright-root"
            copy_seeded_playwright_root(seeded_playwright_root, playwright_root)
            provider = PlaywrightProvider(install_root=playwright_root)

            installed = provider.load("chromium", no_cache=True)
            test_machine.assert_shallow_binary_loaded(
                installed,
                assert_version_command=False,
            )
            assert installed is not None
            assert installed.loaded_abspath is not None
            original_target = installed.loaded_abspath.resolve()
            assert original_target.exists()

            updated = provider.update("chromium", no_cache=True)
            test_machine.assert_shallow_binary_loaded(
                updated,
                assert_version_command=False,
            )
            assert updated is not None
            assert updated.loaded_abspath is not None
            # The shim resolves to a chromium build that actually
            # exists on disk after update (whether the build-id moved
            # depends on the current playwright release, but the
            # resolved target must always exist and still live inside
            # ``playwright_root``).
            updated_target = _resolve_shim_target(updated.loaded_abspath)
            assert updated_target.exists()
            assert (playwright_root / "cache").resolve() in updated_target.parents
            assert any(
                child.name.startswith("chromium-")
                for child in (playwright_root / "cache").iterdir()
                if child.is_dir()
            )

    def test_provider_dry_run_does_not_install_chromium(self, test_machine):
        test_machine.require_tool("node")
        test_machine.require_tool("npm")

        with tempfile.TemporaryDirectory() as temp_dir:
            playwright_root = Path(temp_dir) / "playwright-root"
            provider = PlaywrightProvider(install_root=playwright_root)

            test_machine.exercise_provider_dry_run(provider, bin_name="chromium")
            # dry_run must not have actually downloaded any browsers.
            cache_dir = playwright_root / "cache"
            browser_dirs = (
                [
                    p
                    for p in cache_dir.iterdir()
                    if p.is_dir()
                    and p.name.startswith(("chromium-", "firefox-", "webkit-"))
                ]
                if cache_dir.is_dir()
                else []
            )
            assert not browser_dirs, (
                f"dry_run should not have created any browser dirs, got: "
                f"{[p.name for p in browser_dirs]}"
            )

    def test_search_finds_real_playwright_browser_and_install_works(self, test_machine):
        with tempfile.TemporaryDirectory() as temp_dir:
            provider = PlaywrightProvider(
                install_root=Path(temp_dir) / "playwright",
                postinstall_scripts=True,
                min_release_age=3,
            )
            results = provider.search("chromium")
            assert results, (
                "playwright search 'chromium' should return supported browsers"
            )
            names = [r.name for r in results]
            assert "chromium" in names
            match = next(r for r in results if r.name == "chromium")
            assert match.overrides == {"playwright": {"install_args": ["chromium"]}}
            assert match.loaded_abspath is None
            assert match.loaded_version is None
            installed = match.install()
            test_machine.assert_shallow_binary_loaded(installed)
            assert installed.name == "chromium"