furtka/tests/test_deps.py

import json

import pytest

from furtka import deps

BASE_MANIFEST = {
    "display_name": "X",
    "version": "0.1.0",
    "description": "x",
    "volumes": [],
    "ports": [],
    "icon": "icon.svg",
}


@pytest.fixture
def apps_root(tmp_path, monkeypatch):
    """Three roots: installed, catalog, bundled. Each set up empty by default."""
    installed = tmp_path / "installed"
    catalog = tmp_path / "catalog" / "apps"
    bundled = tmp_path / "bundled"
    for p in (installed, catalog, bundled):
        p.mkdir(parents=True)
    monkeypatch.setenv("FURTKA_APPS_DIR", str(installed))
    monkeypatch.setenv("FURTKA_CATALOG_DIR", str(tmp_path / "catalog"))
    monkeypatch.setenv("FURTKA_BUNDLED_APPS_DIR", str(bundled))
    return {"installed": installed, "catalog": catalog, "bundled": bundled}


def _write_manifest(root, name, **overrides):
    app = root / name
    app.mkdir(parents=True, exist_ok=True)
    payload = dict(BASE_MANIFEST, name=name, **overrides)
    (app / "manifest.json").write_text(json.dumps(payload))
    return app


def test_plan_install_no_deps(apps_root):
    _write_manifest(apps_root["catalog"], "alone")
    plan = deps.plan_install("alone")
    assert plan.target == "alone"
    assert plan.install_order == ("alone",)
    assert plan.to_install == ("alone",)
    assert plan.already_installed == frozenset()


def test_plan_install_linear_chain(apps_root):
    # A requires B, B requires C — all in catalog, none installed yet.
    _write_manifest(apps_root["catalog"], "c")
    _write_manifest(apps_root["catalog"], "b", requires=[{"app": "c"}])
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "b"}])
    plan = deps.plan_install("a")
    assert plan.install_order == ("c", "b", "a")
    assert plan.to_install == ("c", "b", "a")


def test_plan_install_diamond(apps_root):
    # A requires B and C; B requires D; C requires D. D must appear once,
    # before B and C, which come before A.
    _write_manifest(apps_root["catalog"], "d")
    _write_manifest(apps_root["catalog"], "b", requires=[{"app": "d"}])
    _write_manifest(apps_root["catalog"], "c", requires=[{"app": "d"}])
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "b"}, {"app": "c"}])
    plan = deps.plan_install("a")
    order = plan.install_order
    # D first, A last, B and C in between (deterministically alphabetical).
    assert order[0] == "d"
    assert order[-1] == "a"
    assert set(order[1:-1]) == {"b", "c"}
    assert order.count("d") == 1


def test_plan_install_already_installed_provider(apps_root):
    _write_manifest(apps_root["installed"], "b")  # provider already installed
    _write_manifest(apps_root["catalog"], "b")
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "b"}])
    plan = deps.plan_install("a")
    assert plan.install_order == ("b", "a")
    assert plan.to_install == ("a",)
    assert plan.already_installed == frozenset({"b"})


def test_plan_install_cycle_two_node(apps_root):
    # Manifest validator already rejects self-reference at load time, but
    # mutual references (A -> B -> A) only show up at plan time.
    _write_manifest(apps_root["catalog"], "b", requires=[{"app": "a"}])
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "b"}])
    with pytest.raises(deps.DependencyError, match="circular"):
        deps.plan_install("a")


def test_plan_install_cycle_three_node(apps_root):
    _write_manifest(apps_root["catalog"], "c", requires=[{"app": "a"}])
    _write_manifest(apps_root["catalog"], "b", requires=[{"app": "c"}])
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "b"}])
    with pytest.raises(deps.DependencyError, match="a -> b -> c -> a"):
        deps.plan_install("a")


def test_plan_install_missing_provider(apps_root):
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "ghost"}])
    with pytest.raises(deps.DependencyError, match="ghost"):
        deps.plan_install("a")


def test_plan_install_prefers_installed_over_catalog(apps_root):
    # If a provider exists in both installed and catalog, we resolve via
    # installed (so we read the actual on-disk manifest the user has).
    _write_manifest(apps_root["installed"], "b")
    _write_manifest(apps_root["catalog"], "b", requires=[{"app": "extra"}])
    _write_manifest(apps_root["catalog"], "a", requires=[{"app": "b"}])
    plan = deps.plan_install("a")
    # The installed manifest has no requires, so "extra" is NOT pulled in.
    assert plan.install_order == ("b", "a")


def test_dependents_of_empty(apps_root):
    assert deps.dependents_of("anything") == ()


def test_dependents_of_finds_consumers(apps_root):
    _write_manifest(apps_root["installed"], "x")
    _write_manifest(apps_root["installed"], "a", requires=[{"app": "x"}])
    _write_manifest(apps_root["installed"], "b", requires=[{"app": "x"}])
    _write_manifest(apps_root["installed"], "unrelated")
    assert deps.dependents_of("x") == ("a", "b")
    assert deps.dependents_of("unrelated") == ()


def test_installed_topo_order_preserves_alpha_when_independent(apps_root):
    from furtka.scanner import scan

    _write_manifest(apps_root["installed"], "alpha")
    _write_manifest(apps_root["installed"], "bravo")
    _write_manifest(apps_root["installed"], "charlie")
    ordered = deps.installed_topo_order(scan(apps_root["installed"]))
    assert [r.manifest.name for r in ordered] == ["alpha", "bravo", "charlie"]


def test_installed_topo_order_puts_providers_first(apps_root):
    from furtka.scanner import scan

    # Alphabetically z2m comes before mqtt? No — but let's force the
    # dependency to win. consumer=alpha requires=provider=zulu, so naive
    # alpha order would put alpha first. Topo must flip them.
    _write_manifest(apps_root["installed"], "zulu")
    _write_manifest(apps_root["installed"], "alpha", requires=[{"app": "zulu"}])
    ordered = deps.installed_topo_order(scan(apps_root["installed"]))
    names = [r.manifest.name for r in ordered]
    assert names == ["zulu", "alpha"]


def test_installed_topo_order_missing_provider_tails_app(apps_root):
    from furtka.scanner import scan

    _write_manifest(apps_root["installed"], "good")
    _write_manifest(apps_root["installed"], "needy", requires=[{"app": "ghost"}])
    ordered = deps.installed_topo_order(scan(apps_root["installed"]))
    names = [r.manifest.name for r in ordered]
    # `good` first (no deps), `needy` last (unresolved).
    assert names == ["good", "needy"]


def test_provider_exec_service_picks_first_service(apps_root, monkeypatch):
    from furtka import dockerops

    monkeypatch.setattr(
        dockerops,
        "compose_image_tags",
        lambda app_dir, project: {"server": "img:1", "worker": "img:2"},
    )
    assert deps.provider_exec_service(apps_root["installed"] / "x", "x") == "server"


def test_provider_exec_service_falls_back_to_project_on_docker_error(apps_root, monkeypatch):
    from furtka import dockerops

    def boom(app_dir, project):
        raise dockerops.DockerError("docker not running")

    monkeypatch.setattr(dockerops, "compose_image_tags", boom)
    assert deps.provider_exec_service(apps_root["installed"] / "x", "myproj") == "myproj"