FabledSteward/tests/core/test_plugin_install.py

"""Plugin installs fetch only what the catalog names, verify it, and unpack it
only inside the plugin's own folder."""
import hashlib
import io
import zipfile
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch

import pytest

from steward.core import plugin_manager as pm


def make_zip(files: dict[str, str]) -> bytes:
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, "w") as zf:
        for name, text in files.items():
            zf.writestr(name, text)
    return buf.getvalue()


def sha(content: bytes) -> str:
    return hashlib.sha256(content).hexdigest()


@pytest.mark.parametrize("name", ["..", "a/b", "evil-name", "", "x.y", "ümlaut"])
def test_name_must_be_an_identifier(name):
    with pytest.raises(pm.PluginInstallError):
        pm.check_plugin_name(name)


def test_checksum_is_required_and_compared():
    content = b"zip"
    with pytest.raises(pm.PluginInstallError, match="no checksum"):
        pm.verify_checksum(content, "")
    with pytest.raises(pm.PluginInstallError, match="mismatch"):
        pm.verify_checksum(content, "0" * 64)
    pm.verify_checksum(content, sha(content).upper())


def test_extract_strips_the_plugin_folder(tmp_path):
    content = make_zip({"demo-v1.2/__init__.py": "x = 1", "demo-v1.2/sub/a.txt": "a"})
    pm.extract_plugin(content, "demo", tmp_path / "demo")
    assert (tmp_path / "demo" / "__init__.py").read_text() == "x = 1"
    assert (tmp_path / "demo" / "sub" / "a.txt").read_text() == "a"


@pytest.mark.parametrize("member", ["../escape.py", "demo/../../escape.py", "/abs.py"])
def test_extract_refuses_members_outside_the_folder(tmp_path, member):
    content = make_zip({"__init__.py": "", member: "boom"})
    with pytest.raises(pm.PluginInstallError, match="outside"):
        pm.extract_plugin(content, "demo", tmp_path / "plugins" / "demo")
    assert not (tmp_path / "escape.py").exists()
    assert not (tmp_path / "plugins" / "escape.py").exists()
    assert not (tmp_path / "plugins" / "demo").exists()  # nothing half-written


def test_extract_refuses_an_oversized_unpack(tmp_path, monkeypatch):
    monkeypatch.setattr(pm, "MAX_PLUGIN_UNPACKED_BYTES", 10)
    with pytest.raises(pm.PluginInstallError, match="Unpacks to more than"):
        pm.extract_plugin(make_zip({"__init__.py": "x" * 11}), "demo", tmp_path / "demo")


def _app(tmp_path):
    return SimpleNamespace(config={"PLUGIN_INSTALL_DIR": str(tmp_path), "PLUGIN_DIRS": [str(tmp_path)],
                                   "DATABASE_URL": ""})


@pytest.mark.asyncio
async def test_update_replaces_the_folder_whole(tmp_path):
    old = tmp_path / "demo"
    old.mkdir()
    (old / "gone.py").write_text("old")
    content = make_zip({"demo/__init__.py": "new"})
    with patch.object(pm, "_download", new=AsyncMock(return_value=content)):
        ok, msg = await pm.download_and_install_plugin(_app(tmp_path), "demo", "https://x/demo.zip", sha(content))
    assert ok, msg
    assert (old / "__init__.py").read_text() == "new"
    assert not (old / "gone.py").exists()
    assert sorted(p.name for p in tmp_path.iterdir()) == ["demo"]


@pytest.mark.asyncio
async def test_a_refused_download_leaves_the_installed_copy(tmp_path):
    old = tmp_path / "demo"
    old.mkdir()
    (old / "__init__.py").write_text("old")
    content = make_zip({"demo/__init__.py": "new"})
    with patch.object(pm, "_download", new=AsyncMock(return_value=content)):
        ok, msg = await pm.download_and_install_plugin(_app(tmp_path), "demo", "https://x/demo.zip", "")
    assert not ok and "no checksum" in msg
    assert (old / "__init__.py").read_text() == "old"
    assert sorted(p.name for p in tmp_path.iterdir()) == ["demo"]


@pytest.mark.asyncio
async def test_route_takes_url_and_checksum_from_the_catalog(client):
    async with client.session_transaction() as sess:
        sess.update(user_id="u", user_role="admin", username="admin", _csrf="t")
    entry = SimpleNamespace(name="demo", version="1.0", download_url="https://catalog/demo.zip",
                            checksum_sha256="abc")
    install = AsyncMock(return_value=(False, "stopped here"))
    with patch("steward.settings.routes.get_all_settings", new=AsyncMock(return_value={})), \
         patch("steward.settings.routes.repo_urls", return_value=["https://catalog/"]), \
         patch("steward.core.plugin_index.fetch_catalog", new=AsyncMock(return_value=[entry])), \
         patch("steward.core.plugin_manager.download_and_install_plugin", new=install), \
         patch("steward.settings.routes.log_audit", new=AsyncMock()):
        resp = await client.post("/settings/plugins/install/demo", headers={"X-CSRF-Token": "t"}, form={
            "download_url": "https://attacker/evil.zip", "checksum_sha256": "dead"})
        assert resp.status_code == 200
        assert install.await_args.args[1:] == ("demo", "https://catalog/demo.zip", "abc")

        install.reset_mock()
        resp = await client.post("/settings/plugins/install/other", headers={"X-CSRF-Token": "t"})
        assert "No configured plugin repository lists this plugin" in await resp.get_data(as_text=True)
        install.assert_not_awaited()


# ── The catalog is fetched by a scheduled task, never by a page load ─────────
# Background Jobs ruling (operator, 2026-10-08): a Steward page reads stored
# records; the sources are contacted by scheduled collection, or because the
# operator asked (Refresh, install).

from steward.core import plugin_index as pi  # noqa: E402


@pytest.fixture
def fresh_catalog():
    pi.clear_catalog_cache()
    yield
    pi.clear_catalog_cache()


def _fetcher(fetched, fail=()):
    async def fetch(url):
        fetched.append(url)
        if url in fail:
            pi._failed_at[url] = 0.0
            return False
        pi._url_cache[url] = ({"plugins": [{"name": f"p-{url[-1]}", "version": "1.0"}]}, 0.0)
        return True
    return fetch


@pytest.mark.asyncio
async def test_the_tick_fetches_what_is_due_and_waits_out_a_failure(fresh_catalog):
    fetched = []
    with patch.object(pi, "_fetch_one", new=_fetcher(fetched, fail={"https://r/b"})):
        assert await pi.refresh_due(["https://r/a", "https://r/b"], now=10.0) == 1
        assert fetched == ["https://r/a", "https://r/b"]
        fetched.clear()
        # a fresh copy is left alone; the failure is not retried before RETRY_AFTER
        assert await pi.refresh_due(["https://r/a", "https://r/b"], now=pi.RETRY_AFTER - 1.0) == 0
        assert fetched == []
        assert await pi.refresh_due(["https://r/a", "https://r/b"], now=pi.RETRY_AFTER + 1.0) == 0
        assert fetched == ["https://r/b"]
        fetched.clear()
        assert await pi.refresh_due(["https://r/a"], now=pi.CATALOG_MAX_AGE + 1.0) == 1
        assert fetched == ["https://r/a"]


@pytest.mark.asyncio
async def test_the_catalog_page_reads_the_cache_and_only_refresh_fetches(client, fresh_catalog):
    async with client.session_transaction() as sess:
        sess.update(user_id="u", user_role="admin", username="admin", _csrf="t")
    fetched = []
    with patch("steward.settings.routes.get_all_settings", new=AsyncMock(return_value={})), \
         patch("steward.settings.routes.repo_urls", return_value=["https://r/a"]), \
         patch.object(pi, "_fetch_one", new=_fetcher(fetched)):
        page = await (await client.get("/settings/plugins/catalog/")).get_data(as_text=True)
        assert fetched == [] and "Not fetched yet: https://r/a" in page
        page = await (await client.post("/settings/plugins/catalog/refresh",
                                        headers={"X-CSRF-Token": "t"})).get_data(as_text=True)
        assert fetched == ["https://r/a"] and "p-a" in page and "Not fetched yet" not in page
        # the next page load shows the stored copy without fetching again
        page = await (await client.get("/settings/plugins/catalog/")).get_data(as_text=True)
        assert fetched == ["https://r/a"] and "p-a" in page