devplacepy/tests/unit/services/skills/install.py

# retoor <retoor@molodetz.nl>

import asyncio

import httpx
import pytest

from devplacepy.net_guard import BlockedAddressError
from devplacepy.services.skills import fetcher, limits
from devplacepy.services.skills.fetcher import SkillFetchError, fetch_skill
from tests.conftest import run_async as run

SKILL = """---
name: demo-skill
description: Demonstrates an install.
license: MIT
---
# Demo

Run `scripts/run.py`, read [the guide](references/guide.md), see `./assets/logo.png`,
[outside](https://other.example/scripts/x.py), [missing](references/missing.md)
and [escape](scripts/../../etc/passwd).
"""


def serve(routes):
    seen = []

    def handler(request: httpx.Request) -> httpx.Response:
        seen.append(str(request.url))
        entry = routes.get(request.url.path)
        if entry is None:
            return httpx.Response(404, text="missing")
        if callable(entry):
            return entry(request)
        status, body, content_type = entry
        return httpx.Response(status, content=body, headers={"content-type": content_type})

    return httpx.MockTransport(handler), seen


@pytest.fixture
def mock_origin(monkeypatch):
    def install(routes):
        transport, seen = serve(routes)

        def client_factory(**kwargs):
            return httpx.AsyncClient(
                transport=transport,
                follow_redirects=kwargs.get("follow_redirects", True),
                max_redirects=kwargs.get("max_redirects", 20),
                headers=kwargs.get("headers"),
            )

        monkeypatch.setattr(fetcher, "guarded_async_client", client_factory)
        return seen

    return install


def text(body, content_type="text/markdown"):
    return (200, body.encode("utf-8"), content_type)


def test_direct_skill_url_installs_with_bundled_files_and_reasons(mock_origin):
    seen = mock_origin(
        {
            "/pack/SKILL.md": text(SKILL),
            "/pack/scripts/run.py": text("print('hi')", "text/x-python"),
            "/pack/references/guide.md": text("# Guide"),
        }
    )
    result = run(fetch_skill("https://skills.example/pack/SKILL.md"))
    assert result.parsed.name == "demo-skill"
    assert result.parsed.license == "MIT"
    assert result.source_url == "https://skills.example/pack/SKILL.md"
    assert result.files == {"scripts/run.py": "print('hi')", "references/guide.md": "# Guide"}
    reasons = {item.path: item.reason for item in result.skipped}
    assert "binary" in reasons["./assets/logo.png"]
    assert "references/missing.md" in reasons and "404" in reasons["references/missing.md"]
    assert not any("other.example" in url for url in seen)
    assert not any("etc/passwd" in url for url in seen)


def test_directory_url_resolves_to_skill_md(mock_origin):
    mock_origin(
        {
            "/pack": (200, b"<html>index</html>", "text/html"),
            "/pack/SKILL.md": text(SKILL),
        }
    )
    result = run(fetch_skill("https://skills.example/pack"))
    assert result.source_url == "https://skills.example/pack/SKILL.md"
    assert result.parsed.name == "demo-skill"


def test_index_page_redirect_resolves_to_its_directory(mock_origin):
    def redirect(request):
        return httpx.Response(302, headers={"location": "/pack/index.html"})

    mock_origin(
        {
            "/start": redirect,
            "/pack/index.html": (200, b"<html></html>", "text/html"),
            "/pack/SKILL.md": text(SKILL),
        }
    )
    result = run(fetch_skill("https://skills.example/start"))
    assert result.source_url == "https://skills.example/pack/SKILL.md"


def test_html_where_skill_md_was_expected_is_an_error(mock_origin):
    mock_origin({"/SKILL.md": (200, b"<html>login</html>", "text/html; charset=utf-8")})
    with pytest.raises(SkillFetchError, match="HTML page"):
        run(fetch_skill("https://skills.example/SKILL.md"))


def test_http_error_and_invalid_skill_are_reported(mock_origin):
    mock_origin({"/gone/SKILL.md": (500, b"boom", "text/plain"), "/bad/SKILL.md": text("no frontmatter")})
    with pytest.raises(SkillFetchError, match="500"):
        run(fetch_skill("https://skills.example/gone/SKILL.md"))
    with pytest.raises(SkillFetchError, match="frontmatter"):
        run(fetch_skill("https://skills.example/bad/SKILL.md"))


def test_oversize_skill_md_is_refused(mock_origin):
    mock_origin({"/big/SKILL.md": text("x" * (limits.SKILL_MD_MAX_BYTES + 10))})
    with pytest.raises(SkillFetchError, match="exceeds"):
        run(fetch_skill("https://skills.example/big/SKILL.md"))


def test_bundled_file_count_and_size_limits(mock_origin):
    links = " ".join(f"`scripts/f{index}.py`" for index in range(limits.BUNDLED_MAX_FILES + 3))
    routes = {"/many/SKILL.md": text(f"---\nname: many\ndescription: d\n---\n{links}")}
    for index in range(limits.BUNDLED_MAX_FILES + 3):
        routes[f"/many/scripts/f{index}.py"] = text("x = 1")
    mock_origin(routes)
    result = run(fetch_skill("https://skills.example/many/SKILL.md"))
    assert len(result.files) == limits.BUNDLED_MAX_FILES
    assert sum(1 for item in result.skipped if "count limit" in item.reason) == 3

    big = "y" * limits.BUNDLED_MAX_FILE_BYTES
    routes = {"/size/SKILL.md": text("---\nname: size\ndescription: d\n---\n`scripts/a.py` `scripts/b.py` `scripts/c.py` `scripts/d.py` `scripts/e.py`")}
    for letter in "abcde":
        routes[f"/size/scripts/{letter}.py"] = text(big)
    mock_origin(routes)
    result = run(fetch_skill("https://skills.example/size/SKILL.md"))
    assert sum(len(content) for content in result.files.values()) <= limits.BUNDLED_MAX_TOTAL_BYTES
    assert any("total size limit" in item.reason for item in result.skipped)


def test_a_binary_bundled_file_with_a_text_extension_is_skipped(mock_origin):
    mock_origin(
        {
            "/bin/SKILL.md": text("---\nname: bin\ndescription: d\n---\n`scripts/blob.py`"),
            "/bin/scripts/blob.py": (200, b"abc\x00def", "text/x-python"),
        }
    )
    result = run(fetch_skill("https://skills.example/bin/SKILL.md"))
    assert result.files == {}
    assert result.skipped[0].path == "scripts/blob.py"


def test_a_blocked_address_is_a_fetch_error(monkeypatch):
    def refuse(**kwargs):
        raise BlockedAddressError("Refusing to reach a private address.")

    monkeypatch.setattr(fetcher, "guarded_async_client", refuse)
    with pytest.raises(SkillFetchError, match="private address"):
        run(fetch_skill("http://10.0.0.1/SKILL.md"))


def test_the_overall_deadline_is_enforced(monkeypatch):
    async def slow_fetch(url):
        await asyncio.sleep(5)

    monkeypatch.setattr(fetcher, "_fetch", slow_fetch)
    with pytest.raises(SkillFetchError, match="deadline"):
        run(fetch_skill("https://skills.example/slow/SKILL.md", deadline_seconds=0.05))