Add unit tests, smoke test script, fix CI to debian-host + test job

- tests/test_catalog.py: 5 unit tests for PypiCatalog (fetch, cache, sort prefix, completions) - tests/test_proxy.py: 5 unit tests for BicepProxy (framing, injection, list result, passthrough) - tests/conftest.py: pytest asyncio_mode=auto config - scripts/smoke_test.sh: end-to-end TCP + health smoke test script - .gitea/workflows/ci.yml: split into test + build-and-deploy jobs (test blocks deploy) - runs-on: debian-host (was ubuntu-latest = broken) - test job installs deps + runs pytest before building image - pyproject.toml: [project.optional-dependencies] dev = pytest + pytest-asyncio
2026-05-10 12:38:41 +02:00
parent d8536468ab
commit cd17e9bfaa
6 changed files with 339 additions and 43 deletions
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -0,0 +1,5 @@
+import pytest
+
+# Enable asyncio mode for all tests
+def pytest_configure(config):
+    config.addinivalue_line("markers", "asyncio: mark test as async")
--- a/tests/test_catalog.py
+++ b/tests/test_catalog.py
@@ -0,0 +1,105 @@
+"""Unit tests for the pypi-server.i80.dk catalog fetcher."""
+
+import asyncio
+from unittest.mock import patch
+
+import pytest
+
+from ilsp.python_lsp.catalog import PypiCatalog
+
+
+MOCK_HTML = """
+<!DOCTYPE html>
+<html><head><title>Simple Index</title></head>
+<body>
+<a href="/simple/azure-toolbox-database/">azure-toolbox-database</a>
+<a href="/simple/devops-gitea-sdk/">devops-gitea-sdk</a>
+<a href="/simple/toolbox-tests-framework/">toolbox-tests-framework</a>
+</body></html>
+"""
+
+
+@pytest.fixture(autouse=True)
+def reset_catalog():
+    PypiCatalog._packages = []
+    PypiCatalog._last_refresh = 0
+    yield
+
+
+class _MockResponse:
+    def __init__(self, html: str):
+        self._html = html
+        self.status = 200
+
+    async def text(self):
+        return self._html
+
+    def raise_for_status(self):
+        pass
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        pass
+
+
+class _MockSession:
+    def __init__(self, html: str):
+        self._html = html
+
+    def get(self, url):
+        return _MockResponse(self._html)
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *_):
+        pass
+
+
+@pytest.mark.asyncio
+async def test_fetch_parses_packages():
+    with patch("ilsp.python_lsp.catalog.aiohttp.ClientSession", return_value=_MockSession(MOCK_HTML)):
+        packages = await PypiCatalog._fetch()
+
+    assert len(packages) == 3
+    names = [p["name"] for p in packages]
+    assert "azure-toolbox-database" in names
+    assert "devops-gitea-sdk" in names
+
+
+@pytest.mark.asyncio
+async def test_packages_have_sort_prefix():
+    with patch("ilsp.python_lsp.catalog.aiohttp.ClientSession", return_value=_MockSession(MOCK_HTML)):
+        packages = await PypiCatalog._fetch()
+
+    for pkg in packages:
+        assert pkg["sort_prefix"] == "0_i80_"
+
+
+@pytest.mark.asyncio
+async def test_get_packages_triggers_refresh_when_empty():
+    with patch("ilsp.python_lsp.catalog.aiohttp.ClientSession", return_value=_MockSession(MOCK_HTML)):
+        packages = await PypiCatalog.get_packages()
+
+    assert len(packages) == 3
+
+
+@pytest.mark.asyncio
+async def test_get_packages_uses_cache():
+    PypiCatalog._packages = [{"name": "cached-pkg", "sort_prefix": "0_i80_", "detail": "x"}]
+    PypiCatalog._last_refresh = 9_999_999_999
+
+    packages = await PypiCatalog.get_packages()
+    assert packages[0]["name"] == "cached-pkg"
+
+
+def test_as_completion_items():
+    PypiCatalog._packages = [
+        {"name": "my-pkg", "detail": "i80", "sort_prefix": "0_i80_"},
+    ]
+    items = PypiCatalog.as_completion_items()
+    assert len(items) == 1
+    assert items[0].label == "my-pkg"
+    assert items[0].sort_text == "0_i80_my-pkg"
--- a/tests/test_proxy.py
+++ b/tests/test_proxy.py
@@ -0,0 +1,85 @@
+"""Unit tests for Bicep proxy message injection."""
+
+import json
+
+import pytest
+
+from ilsp.bicep_lsp.modules import BicepModuleCatalog
+from ilsp.bicep_lsp.proxy import BicepProxy, _ContentLengthFramer
+
+
+def test_frame_produces_correct_header():
+    body = b'{"jsonrpc":"2.0"}'
+    framed = _ContentLengthFramer.frame(body)
+    assert framed.startswith(b"Content-Length: 17\r\n\r\n")
+    assert framed.endswith(body)
+
+
+@pytest.fixture(autouse=True)
+def reset_modules():
+    BicepModuleCatalog._modules = []
+    yield
+
+
+def _make_proxy() -> BicepProxy:
+    return BicepProxy.__new__(BicepProxy)
+
+
+def _completion_response(items: list) -> dict:
+    return {
+        "jsonrpc": "2.0",
+        "id": 1,
+        "result": {"isIncomplete": False, "items": items},
+    }
+
+
+def test_standard_items_not_downgraded_without_lru():
+    """Without LRU modules, standard items keep their original sortText."""
+    proxy = _make_proxy()
+    msg = _completion_response([{"label": "Microsoft.Storage", "sortText": "az"}])
+    out = json.loads(proxy._maybe_inject_completions(msg))
+    # No LRU modules → no downgrade, original sortText preserved
+    assert out["result"]["items"][0]["sortText"] == "az"
+
+
+def test_lru_modules_injected_at_top():
+    BicepModuleCatalog._modules = [{
+        "name": "appservice",
+        "path": "bicep/modules/appservice",
+        "versions": ["2.3.0", "latest"],
+        "latest": "latest",
+        "registry": "iactemplatereg.azurecr.io",
+    }]
+
+    proxy = _make_proxy()
+    msg = _completion_response([{"label": "Microsoft.Web/sites", "sortText": "az"}])
+    out = json.loads(proxy._maybe_inject_completions(msg))
+    items = out["result"]["items"]
+
+    assert items[0]["label"] == "appservice"
+    assert items[0]["sortText"].startswith("0_lru_")
+    assert items[1]["label"] == "Microsoft.Web/sites"
+    assert items[1]["sortText"].startswith("1_az_")
+
+
+def test_list_result_also_handled():
+    BicepModuleCatalog._modules = [{
+        "name": "roleassignments",
+        "path": "bicep/modules/roleassignments",
+        "versions": ["2.0.0"],
+        "latest": "2.0.0",
+        "registry": "iactemplatereg.azurecr.io",
+    }]
+
+    proxy = _make_proxy()
+    msg = {"jsonrpc": "2.0", "id": 2, "result": [{"label": "az-item", "sortText": "az"}]}
+    out = json.loads(proxy._maybe_inject_completions(msg))
+    assert isinstance(out["result"], list)
+    assert out["result"][0]["label"] == "roleassignments"
+
+
+def test_non_completion_message_passthrough():
+    proxy = _make_proxy()
+    msg = {"jsonrpc": "2.0", "method": "initialized", "params": {}}
+    out = json.loads(proxy._maybe_inject_completions(msg))
+    assert out["method"] == "initialized"