test_chromewebstoreprovider.py

from .conftest import (
    assert_extension_binary_loaded,
    restore_signed_store_metadata_from_real_crx,
)
import asyncio
import hashlib
import json
import os
import subprocess
import sys
import tempfile
import threading
import zipfile
from functools import partial
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

import abxbus

from abxpkg import Binary, ChromeWebstoreProvider
from abxpkg.binary_service import BinaryRequestEvent, BinaryService


PACKAGED_CHROMEWEBSTORE_UTILS_PATH = (
    Path(__file__).resolve().parent.parent / "abxpkg" / "chromewebstore_utils.js"
)
UBLOCK_WEBSTORE_ID = "ddkjiahejlhfcafbddmgiahcphecmpfh"


class TestChromeWebstoreProvider:
    def test_install_accepts_pinned_extension_archive_url(self, test_machine):
        test_machine.require_tool("node")

        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            archive_path = root / "extension.zip"
            manifest = json.dumps(
                {"manifest_version": 3, "name": "Local Extension", "version": "1.2.3"},
            )
            with zipfile.ZipFile(archive_path, "w") as archive:
                archive.writestr("manifest.json", manifest)
            sha256 = hashlib.sha256(archive_path.read_bytes()).hexdigest()

            server = ThreadingHTTPServer(
                ("127.0.0.1", 0),
                partial(SimpleHTTPRequestHandler, directory=str(root)),
            )
            thread = threading.Thread(target=server.serve_forever, daemon=True)
            thread.start()
            try:
                provider = ChromeWebstoreProvider(
                    install_root=root / "chromewebstore-root",
                    bin_dir=root / "chromewebstore-root/extensions",
                ).get_provider_with_overrides(
                    overrides={
                        "local-extension": {
                            "install_args": [
                                "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa",
                                "--name=local-extension",
                                f"--url=http://127.0.0.1:{server.server_port}/extension.zip",
                                f"--sha256={sha256}",
                            ],
                        },
                    },
                )

                installed = provider.install("local-extension")

                assert_extension_binary_loaded(installed)
                assert installed is not None
                assert str(installed.loaded_version) == "1.2.3"
                metadata = json.loads(installed.loaded_abspath.read_text())
                assert metadata["crx_sha256"] == sha256
            finally:
                server.shutdown()
                thread.join()

    def test_unzip_dependency_uses_native_system_package_providers(self):
        provider = ChromeWebstoreProvider()

        unzip = provider.unzip_binary()

        assert [binprovider.name for binprovider in unzip.binproviders] == [
            "env",
            "brew",
            "apt",
        ]

    def test_install_root_alias_without_explicit_bin_dir_uses_root_extensions(
        self,
        test_machine,
    ):
        test_machine.require_tool("node")
        assert PACKAGED_CHROMEWEBSTORE_UTILS_PATH.exists(), (
            PACKAGED_CHROMEWEBSTORE_UTILS_PATH
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            install_root = (Path(temp_dir) / "chromewebstore-root").resolve()
            provider = ChromeWebstoreProvider.model_validate(
                {
                    "install_root": install_root,
                    "postinstall_scripts": True,
                    "min_release_age": 0,
                },
            ).get_provider_with_overrides(
                overrides={
                    "ublock": {
                        "install_args": [UBLOCK_WEBSTORE_ID, "--name=ublock"],
                    },
                },
            )

            installed = provider.install("ublock")

            assert_extension_binary_loaded(installed)
            assert installed is not None
            bin_dir = provider.bin_dir
            assert bin_dir is not None
            assert provider.install_root == install_root
            assert bin_dir == install_root / "extensions"
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.is_relative_to(bin_dir)
            assert not os.access(installed.loaded_abspath, os.X_OK)

            lib_dir = Path(temp_dir) / "lib"
            projection_overrides = {
                "chromewebstore": {
                    "install_root": str(install_root),
                    "install_args": [UBLOCK_WEBSTORE_ID, "--name=ublock"],
                },
            }

            async def cache_script_dependencies() -> None:
                bus = abxbus.EventBus(name="chromewebstore_script_dependency")
                BinaryService(bus, auto_install=False, lib_dir=lib_dir)
                await bus.emit(
                    BinaryRequestEvent(name="python3", binproviders="env"),
                ).now()
                await bus.emit(
                    BinaryRequestEvent(
                        name="ublock",
                        binproviders="env,chromewebstore",
                        overrides=projection_overrides,
                    ),
                ).now()
                await bus.wait_until_idle()
                await bus.destroy(clear=False)

            asyncio.run(cache_script_dependencies())

            config_path = Path(temp_dir) / "config.json"
            config_path.write_text(
                json.dumps(
                    {
                        "required_binaries": [
                            {
                                "name": "ublock",
                                "binproviders": "env,chromewebstore",
                                "overrides": projection_overrides,
                            },
                        ],
                    },
                ),
            )
            script_path = Path(temp_dir) / "hook.py"
            script_path.write_text(
                '# /// script\n# ///\nprint("projected-extension")\n',
            )
            abxpkg = Path(sys.executable).parent / "abxpkg"
            assert abxpkg.is_file()
            projected = subprocess.run(
                [
                    str(abxpkg),
                    f"--lib={lib_dir}",
                    "--binproviders=env",
                    "run",
                    "--script",
                    f"--deps-from={config_path}:required_binaries",
                    "python3",
                    str(script_path),
                ],
                capture_output=True,
                text=True,
                check=False,
                env={**os.environ, "PYTHONPROFILEIMPORTTIME": "1"},
            )

            assert projected.returncode == 0, projected.stderr
            assert projected.stdout.strip() == "projected-extension"
            assert "rich_click" not in projected.stderr
            assert "pydantic" not in projected.stderr

    def test_install_root_and_bin_dir_aliases_install_into_the_requested_paths(
        self,
        test_machine,
    ):
        test_machine.require_tool("node")
        assert PACKAGED_CHROMEWEBSTORE_UTILS_PATH.exists(), (
            PACKAGED_CHROMEWEBSTORE_UTILS_PATH
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            install_root = (Path(temp_dir) / "chromewebstore-root").resolve()
            bin_dir = (Path(temp_dir) / "custom-extensions").resolve()
            provider = ChromeWebstoreProvider.model_validate(
                {
                    "install_root": install_root,
                    "bin_dir": bin_dir,
                    "postinstall_scripts": True,
                    "min_release_age": 0,
                },
            ).get_provider_with_overrides(
                overrides={
                    "ublock": {
                        "install_args": [UBLOCK_WEBSTORE_ID, "--name=ublock"],
                    },
                },
            )

            installed = provider.install("ublock")

            assert_extension_binary_loaded(installed)
            assert installed is not None
            assert bin_dir is not None
            assert provider.install_root == install_root
            assert provider.bin_dir == bin_dir
            assert installed.loaded_abspath is not None
            assert installed.loaded_abspath.is_relative_to(bin_dir)

    def test_provider_direct_methods_exercise_real_lifecycle(self, test_machine):
        test_machine.require_tool("node")
        unzip = test_machine.require_tool("unzip")
        assert PACKAGED_CHROMEWEBSTORE_UTILS_PATH.exists(), (
            PACKAGED_CHROMEWEBSTORE_UTILS_PATH
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            provider = ChromeWebstoreProvider(
                install_root=Path(temp_dir) / "chromewebstore-root",
                bin_dir=Path(temp_dir) / "chromewebstore-root/extensions",
                postinstall_scripts=True,
                min_release_age=3,
            ).get_provider_with_overrides(
                overrides={
                    "ublock": {
                        "install_args": [UBLOCK_WEBSTORE_ID, "--name=ublock"],
                    },
                },
            )

            provider.setup(
                postinstall_scripts=True,
                min_release_age=3,
                min_version=None,
            )
            assert provider.load("ublock", quiet=True, no_cache=True) is None

            installed = provider.install("ublock", no_cache=True)
            assert_extension_binary_loaded(installed)
            assert installed is not None
            assert installed.loaded_abspath is not None
            crx_path = provider.bin_dir / f"{UBLOCK_WEBSTORE_ID}__ublock.crx"
            metadata = json.loads(installed.loaded_abspath.read_text(encoding="utf-8"))
            restore_signed_store_metadata_from_real_crx(
                unzip,
                crx_path,
                Path(metadata["unpacked_path"]),
            )

            loaded = provider.load("ublock", no_cache=True)
            assert_extension_binary_loaded(loaded)

            loaded_or_installed = provider.install("ublock", no_cache=True)
            assert_extension_binary_loaded(loaded_or_installed)

            updated = provider.update("ublock", no_cache=True)
            assert_extension_binary_loaded(updated)

            assert provider.uninstall("ublock", no_cache=True) is True
            assert provider.load("ublock", quiet=True, no_cache=True) is None

    def test_binary_direct_methods_exercise_real_lifecycle(self, test_machine):
        test_machine.require_tool("node")
        assert PACKAGED_CHROMEWEBSTORE_UTILS_PATH.exists(), (
            PACKAGED_CHROMEWEBSTORE_UTILS_PATH
        )

        with tempfile.TemporaryDirectory() as temp_dir:
            binary = Binary(
                name="ublock",
                binproviders=[
                    ChromeWebstoreProvider(
                        install_root=Path(temp_dir) / "chromewebstore-root",
                        bin_dir=Path(temp_dir) / "chromewebstore-root/extensions",
                        postinstall_scripts=True,
                        min_release_age=3,
                    ),
                ],
                overrides={
                    "chromewebstore": {
                        "install_args": [UBLOCK_WEBSTORE_ID, "--name=ublock"],
                    },
                },
                postinstall_scripts=True,
                min_release_age=3,
            )

            fresh = test_machine.unloaded_binary(binary)
            test_machine.assert_binary_missing(fresh)

            installed = fresh.install()
            assert_extension_binary_loaded(installed)

            loaded = test_machine.unloaded_binary(binary).load(no_cache=True)
            assert_extension_binary_loaded(loaded)

            loaded_or_installed = test_machine.unloaded_binary(binary).install(
                no_cache=True,
            )
            assert_extension_binary_loaded(loaded_or_installed)

            updated = installed.update()
            assert_extension_binary_loaded(updated)

            removed = updated.uninstall()
            assert not removed.is_valid
            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_search_finds_real_chromewebstore_extension_by_id(self):
        # Search resolves a 32-char Chrome Web Store extension ID into
        # its canonical name by scraping the public detail page's
        # ``<title>`` tag. The Binary's ``.name`` is the stable webstore
        # ID (BinName-safe); the human extension name lives in
        # ``description`` and the ``--name=`` install_arg. Non-ID
        # queries return [] because the Web Store doesn't expose a JSON
        # search API for keyword lookups.
        results = ChromeWebstoreProvider().search(UBLOCK_WEBSTORE_ID)
        assert results, "chromewebstore search by id should resolve uBlock Origin Lite"
        assert len(results) == 1
        match = results[0]
        assert match.name == UBLOCK_WEBSTORE_ID
        assert "ublock" in match.description.lower()
        assert match.overrides == {
            "chromewebstore": {
                "install_args": [
                    UBLOCK_WEBSTORE_ID,
                    f"--name={match.description.split(' (')[0]}",
                ],
            },
        }
        assert ChromeWebstoreProvider().search("not-an-id") == []