TST Add a fixture for testing wheels from external URL

tests/conftest.py

@@ -3,11 +3,15 @@
 import io
 import sys
 import zipfile
+from dataclasses import dataclass
 from importlib.metadata import Distribution, PackageNotFoundError
 from pathlib import Path
 from tempfile import TemporaryDirectory
+from typing import Any
 
 import pytest
+from packaging.utils import parse_wheel_filename
+from pytest_httpserver import HTTPServer
 from pytest_pyodide import spawn_web_server
 
 
@@ -56,16 +60,6 @@ def wheel_path(tmp_path_factory):
     yield output_dir
 
 
-@pytest.fixture(scope="session")
-def test_wheel_path(tmp_path_factory):
-    # Build a test wheel for testing
-    output_dir = tmp_path_factory.mktemp("wheel")
-
-    _build(Path(__file__).parent / "test_data" / "test_wheel_uninstall", output_dir)
-
-    yield output_dir
-
-
 @pytest.fixture
 def selenium_standalone_micropip(selenium_standalone, wheel_path):
     """Import micropip before entering test so that global initialization of
@@ -93,6 +87,68 @@ def selenium_standalone_micropip(selenium_standalone, wheel_path):
     yield selenium_standalone
 
 
+class WheelCatalog:
+    """
+    A catalog of wheels for testing.
+    """
+
+    @dataclass
+    class Wheel:
+        _path: Path
+
+        name: str
+        filename: str
+        top_level: str
+        url: str
+
+        @property
+        def content(self) -> bytes:
+            return self._path.read_bytes()
+
+    def __init__(self):
+        self._wheels = {}
+
+        self._httpserver = HTTPServer()
+        self._httpserver.no_handler_status_code = 404
+
+    def __enter__(self):
+        self._httpserver.__enter__()
+        return self
+
+    def __exit__(self, *args: Any):
+        self._httpserver.__exit__(*args)
+
+    def _register_handler(self, path: Path) -> str:
+        self._httpserver.expect_request(f"/{path.name}").respond_with_data(
+            path.read_bytes(),
+            content_type="application/zip",
+            headers={"Access-Control-Allow-Origin": "*"},
+        )
+
+        return self._httpserver.url_for(f"/{path.name}")
+
+    def add_wheel(self, path: Path):
+        name = parse_wheel_filename(path.name)[0]
+        url = self._register_handler(path)
+
+        self._wheels[name] = self.Wheel(
+            path, name, path.name, name.replace("-", "_"), url
+        )
+
+    def get(self, name: str) -> Wheel:
+        return self._wheels[name]
+
+
+@pytest.fixture(scope="session")
+def test_wheel_catalog():
+    """Run a mock server that serves pre-built wheels"""
+    with WheelCatalog() as catalog:
+        for wheel in TEST_WHEEL_DIR.glob("*.whl"):
+            catalog.add_wheel(wheel)
+
+        yield catalog
+
+
 @pytest.fixture
 def mock_platform(monkeypatch):
     monkeypatch.setenv("_PYTHON_HOST_PLATFORM", PLATFORM)

tests/test_data/test_wheel_uninstall/README.md

@@ -0,0 +1,7 @@
+This directory contains a wheel that is used to test the uninstallation functionality of the micropip.
+If something is changed in the wheel, create a new wheel and copy it to the `wheel` directory.
+
+```sh
+python -m build
+cp dist/test_wheel_uninstall-1.0.0-py3-none-any.whl ../wheel/
+```

tests/test_install.py

@@ -1,7 +1,7 @@
 import pytest
-from conftest import SNOWBALL_WHEEL, TEST_WHEEL_DIR, mock_fetch_cls
+from conftest import mock_fetch_cls
 from packaging.utils import parse_wheel_filename
-from pytest_pyodide import run_in_pyodide, spawn_web_server
+from pytest_pyodide import run_in_pyodide
 
 import micropip
 
@@ -29,23 +29,20 @@ def test_install_simple(selenium_standalone_micropip):
 
 
 @pytest.mark.parametrize("base_url", ["'{base_url}'", "'.'"])
-def test_install_custom_url(selenium_standalone_micropip, base_url):
+def test_install_custom_url(selenium_standalone_micropip, base_url, test_wheel_catalog):
     selenium = selenium_standalone_micropip
+    snowball_wheel = test_wheel_catalog.get("snowballstemmer")
+    url = snowball_wheel.url
 
-    with spawn_web_server(TEST_WHEEL_DIR) as server:
-        server_hostname, server_port, _ = server
-        base_url = f"http://{server_hostname}:{server_port}/"
-        url = base_url + SNOWBALL_WHEEL
-
-        selenium.run_js(
-            f"""
-            await pyodide.runPythonAsync(`
-                import micropip
-                await micropip.install('{url}')
-                import snowballstemmer
-            `);
-            """
-        )
+    selenium.run_js(
+        f"""
+        await pyodide.runPythonAsync(`
+            import micropip
+            await micropip.install('{url}')
+            import snowballstemmer
+        `);
+        """
+    )
 
 
 @pytest.mark.xfail_browsers(chrome="node only", firefox="node only")
@@ -313,59 +310,54 @@ async def test_load_binary_wheel2(selenium):
     import regex  # noqa: F401
 
 
-def test_emfs(selenium_standalone_micropip):
-    with spawn_web_server(TEST_WHEEL_DIR) as server:
-        server_hostname, server_port, _ = server
-        url = f"http://{server_hostname}:{server_port}/"
+def test_emfs(selenium_standalone_micropip, test_wheel_catalog):
+    snowball_wheel = test_wheel_catalog.get("snowballstemmer")
 
-        @run_in_pyodide(packages=["micropip"])
-        async def run_test(selenium, url, wheel_name):
-            from pyodide.http import pyfetch
+    @run_in_pyodide()
+    async def run_test(selenium, url, wheel_name):
+        from pyodide.http import pyfetch
 
-            import micropip
+        import micropip
 
-            resp = await pyfetch(url + wheel_name)
-            await resp._into_file(open(wheel_name, "wb"))
-            await micropip.install("emfs:" + wheel_name)
-            import snowballstemmer
+        resp = await pyfetch(url)
+        await resp._into_file(open(wheel_name, "wb"))
+        await micropip.install("emfs:" + wheel_name)
+        import snowballstemmer
 
-            stemmer = snowballstemmer.stemmer("english")
-            assert stemmer.stemWords("go going goes gone".split()) == [
-                "go",
-                "go",
-                "goe",
-                "gone",
-            ]
+        stemmer = snowballstemmer.stemmer("english")
+        assert stemmer.stemWords("go going goes gone".split()) == [
+            "go",
+            "go",
+            "goe",
+            "gone",
+        ]
 
-        run_test(selenium_standalone_micropip, url, SNOWBALL_WHEEL)
+    run_test(selenium_standalone_micropip, snowball_wheel.url, snowball_wheel.filename)
 
 
-def test_logging(selenium_standalone_micropip):
-    # TODO: make a fixture for this, it's used in a few places
-    with spawn_web_server(TEST_WHEEL_DIR) as server:
-        server_hostname, server_port, _ = server
-        url = f"http://{server_hostname}:{server_port}/"
-        wheel_url = url + SNOWBALL_WHEEL
-        name, version, _, _ = parse_wheel_filename(SNOWBALL_WHEEL)
+def test_logging(selenium_standalone_micropip, test_wheel_catalog):
+    @run_in_pyodide(packages=["micropip"])
+    async def run_test(selenium, url, name, version):
+        import contextlib
+        import io
 
-        @run_in_pyodide(packages=["micropip"])
-        async def run_test(selenium, url, name, version):
-            import contextlib
-            import io
+        import micropip
 
-            import micropip
+        with io.StringIO() as buf, contextlib.redirect_stdout(buf):
+            await micropip.install(url, verbose=True)
 
-            with io.StringIO() as buf, contextlib.redirect_stdout(buf):
-                await micropip.install(url, verbose=True)
+            captured = buf.getvalue()
 
-                captured = buf.getvalue()
+            assert f"Collecting {name}" in captured
+            assert f"  Downloading {name}" in captured
+            assert f"Installing collected packages: {name}" in captured
+            assert f"Successfully installed {name}-{version}" in captured
 
-                assert f"Collecting {name}" in captured
-                assert f"  Downloading {name}" in captured
-                assert f"Installing collected packages: {name}" in captured
-                assert f"Successfully installed {name}-{version}" in captured
+    snowball_wheel = test_wheel_catalog.get("snowballstemmer")
+    wheel_url = snowball_wheel.url
+    name, version, _, _ = parse_wheel_filename(snowball_wheel.filename)
 
-        run_test(selenium_standalone_micropip, wheel_url, name, version)
+    run_test(selenium_standalone_micropip, wheel_url, name, version)
 
 
 @pytest.mark.asyncio

tests/test_list.py

@@ -1,6 +1,5 @@
 import pytest
-from conftest import SNOWBALL_WHEEL, TEST_WHEEL_DIR, mock_fetch_cls
-from pytest_pyodide import spawn_web_server
+from conftest import mock_fetch_cls
 
 import micropip
 
@@ -42,22 +41,20 @@ async def test_list_wheel_name_mismatch(mock_fetch: mock_fetch_cls) -> None:
     assert pkg_list[dummy_pkg_name].source.lower() == dummy_url
 
 
-def test_list_load_package_from_url(selenium_standalone_micropip):
-    with spawn_web_server(TEST_WHEEL_DIR) as server:
-        server_hostname, server_port, _ = server
-        base_url = f"http://{server_hostname}:{server_port}/"
-        url = base_url + SNOWBALL_WHEEL
-
-        selenium = selenium_standalone_micropip
-        selenium.run_js(
-            f"""
-            await pyodide.loadPackage({url!r});
-            await pyodide.runPythonAsync(`
-                import micropip
-                assert "snowballstemmer" in micropip.list()
-            `);
-            """
-        )
+def test_list_load_package_from_url(selenium_standalone_micropip, test_wheel_catalog):
+    snowball_wheel = test_wheel_catalog.get("snowballstemmer")
+    url = snowball_wheel.url
+
+    selenium = selenium_standalone_micropip
+    selenium.run_js(
+        f"""
+        await pyodide.loadPackage({url!r});
+        await pyodide.runPythonAsync(`
+            import micropip
+            assert "snowballstemmer" in micropip.list()
+        `);
+        """
+    )
 
 
 def test_list_pyodide_package(selenium_standalone_micropip):

tests/test_transaction.py

@@ -1,7 +1,6 @@
 import pytest
-from conftest import SNOWBALL_WHEEL, TEST_WHEEL_DIR
+from conftest import SNOWBALL_WHEEL
 from packaging.tags import Tag
-from pytest_pyodide import spawn_web_server
 
 
 @pytest.mark.parametrize(
@@ -65,23 +64,20 @@ def create_transaction(Transaction):
 
 
 @pytest.mark.asyncio
-async def test_add_requirement():
+async def test_add_requirement(test_wheel_catalog):
     pytest.importorskip("packaging")
     from micropip.transaction import Transaction
 
-    with spawn_web_server(TEST_WHEEL_DIR) as server:
-        server_hostname, server_port, _ = server
-        base_url = f"http://{server_hostname}:{server_port}/"
-        url = base_url + SNOWBALL_WHEEL
+    snowballstemmer_wheel = test_wheel_catalog.get("snowballstemmer")
 
-        transaction = create_transaction(Transaction)
-        await transaction.add_requirement(url)
+    transaction = create_transaction(Transaction)
+    await transaction.add_requirement(snowballstemmer_wheel.url)
 
     wheel = transaction.wheels[0]
     assert wheel.name == "snowballstemmer"
     assert str(wheel.version) == "2.0.0"
     assert wheel.filename == SNOWBALL_WHEEL
-    assert wheel.url == url
+    assert wheel.url == snowballstemmer_wheel.url
     assert wheel.tags == frozenset(
         {Tag("py2", "none", "any"), Tag("py3", "none", "any")}
     )