230 lines
8 KiB
Python
230 lines
8 KiB
Python
import json
|
|
|
|
from furtka.cli import main
|
|
|
|
|
|
def _set_env(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("FURTKA_APPS_DIR", str(tmp_path))
|
|
|
|
|
|
def test_app_list_empty(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
rc = main(["app", "list"])
|
|
assert rc == 0
|
|
assert "no apps installed" in capsys.readouterr().out
|
|
|
|
|
|
def test_app_list_json_empty(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
rc = main(["app", "list", "--json"])
|
|
assert rc == 0
|
|
assert json.loads(capsys.readouterr().out) == []
|
|
|
|
|
|
def test_app_list_json_with_one_app(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
app = tmp_path / "fileshare"
|
|
app.mkdir()
|
|
(app / "manifest.json").write_text(
|
|
json.dumps(
|
|
{
|
|
"name": "fileshare",
|
|
"display_name": "Network Files",
|
|
"version": "0.1.0",
|
|
"description": "SMB",
|
|
"description_long": "Long description here.",
|
|
"volumes": ["files"],
|
|
"ports": [445],
|
|
"icon": "icon.svg",
|
|
"open_url": "smb://{host}/files",
|
|
"settings": [
|
|
{
|
|
"name": "SMB_USER",
|
|
"label": "User",
|
|
"description": "SMB user",
|
|
"type": "text",
|
|
"default": "furtka",
|
|
"required": True,
|
|
}
|
|
],
|
|
}
|
|
)
|
|
)
|
|
rc = main(["app", "list", "--json"])
|
|
assert rc == 0
|
|
data = json.loads(capsys.readouterr().out)
|
|
assert len(data) == 1
|
|
assert data[0]["ok"] is True
|
|
m = data[0]["manifest"]
|
|
assert m["name"] == "fileshare"
|
|
assert m["description_long"] == "Long description here."
|
|
assert m["open_url"] == "smb://{host}/files"
|
|
assert len(m["settings"]) == 1
|
|
assert m["settings"][0]["name"] == "SMB_USER"
|
|
assert m["settings"][0]["required"] is True
|
|
assert m["settings"][0]["default"] == "furtka"
|
|
assert m["internet"] == {"viable": False, "port": None}
|
|
|
|
|
|
def test_gateway_connect_bg_dispatches_to_run_connect(tmp_path, monkeypatch):
|
|
"""CLI `gateway connect-bg <url> <token> <name>` must call
|
|
gateway.run_connect(url, token, name) — the entry point the HTTP API
|
|
fires via systemd-run for POST /api/gateway/connect."""
|
|
_set_env(monkeypatch, tmp_path)
|
|
from furtka import gateway
|
|
|
|
called = []
|
|
monkeypatch.setattr(
|
|
gateway, "run_connect", lambda url, token, name: called.append((url, token, name))
|
|
)
|
|
rc = main(["gateway", "connect-bg", "https://gw.example.com", "reg-token", "my-box"])
|
|
assert rc == 0
|
|
assert called == [("https://gw.example.com", "reg-token", "my-box")]
|
|
|
|
|
|
def test_gateway_connect_bg_returns_1_on_failure(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
from furtka import gateway
|
|
|
|
def boom(url, token, name):
|
|
raise RuntimeError("gateway unreachable")
|
|
|
|
monkeypatch.setattr(gateway, "run_connect", boom)
|
|
rc = main(["gateway", "connect-bg", "https://gw.example.com", "reg-token", "my-box"])
|
|
assert rc == 1
|
|
assert "gateway connect-bg failed" in capsys.readouterr().err
|
|
|
|
|
|
def test_reconcile_dry_run_empty(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
rc = main(["reconcile", "--dry-run"])
|
|
assert rc == 0
|
|
out = capsys.readouterr().out
|
|
assert "0 actions" in out
|
|
|
|
|
|
def test_app_install_bg_dispatches_to_runner(tmp_path, monkeypatch):
|
|
"""CLI `app install-bg <name>` must call install_runner.run_install(name).
|
|
|
|
This is the entry point the HTTP API fires via systemd-run; regression
|
|
here would leave the UI hanging at "pulling_image…" forever because
|
|
the background never transitions state.
|
|
"""
|
|
_set_env(monkeypatch, tmp_path)
|
|
from furtka import install_runner
|
|
|
|
called = []
|
|
monkeypatch.setattr(install_runner, "run_install", lambda name: called.append(name))
|
|
rc = main(["app", "install-bg", "fileshare"])
|
|
assert rc == 0
|
|
assert called == ["fileshare"]
|
|
|
|
|
|
def test_app_install_bg_returns_1_on_failure(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
from furtka import install_runner
|
|
|
|
def boom(name):
|
|
raise RuntimeError("compose pull failed")
|
|
|
|
monkeypatch.setattr(install_runner, "run_install", boom)
|
|
rc = main(["app", "install-bg", "fileshare"])
|
|
assert rc == 1
|
|
err = capsys.readouterr().err
|
|
assert "install-bg failed" in err
|
|
assert "compose pull failed" in err
|
|
|
|
|
|
# --- Dependency-aware install + remove ---------------------------------------
|
|
|
|
|
|
def _write_manifest(root, name, **overrides):
|
|
app = root / name
|
|
app.mkdir(parents=True, exist_ok=True)
|
|
payload = {
|
|
"name": name,
|
|
"display_name": name,
|
|
"version": "0.1.0",
|
|
"description": "x",
|
|
"volumes": [],
|
|
"ports": [],
|
|
"icon": "icon.svg",
|
|
**overrides,
|
|
}
|
|
(app / "manifest.json").write_text(json.dumps(payload))
|
|
(app / "docker-compose.yaml").write_text("services: {}\n")
|
|
return app
|
|
|
|
|
|
def test_app_remove_blocked_by_dependent(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
_write_manifest(tmp_path, "mosquitto")
|
|
_write_manifest(tmp_path, "zigbee2mqtt", requires=[{"app": "mosquitto"}])
|
|
rc = main(["app", "remove", "mosquitto"])
|
|
assert rc == 2
|
|
err = capsys.readouterr().err
|
|
assert "required by: zigbee2mqtt" in err
|
|
|
|
|
|
def test_app_remove_unblocked_when_no_dependents(tmp_path, monkeypatch):
|
|
_set_env(monkeypatch, tmp_path)
|
|
_write_manifest(tmp_path, "mosquitto")
|
|
from furtka import dockerops
|
|
|
|
monkeypatch.setattr(dockerops, "compose_down", lambda *a, **k: None)
|
|
rc = main(["app", "remove", "mosquitto"])
|
|
assert rc == 0
|
|
assert not (tmp_path / "mosquitto").exists()
|
|
|
|
|
|
def test_app_install_uses_plan_for_named_install(tmp_path, monkeypatch, capsys):
|
|
"""Named install pulls in dependencies via plan_install."""
|
|
_set_env(monkeypatch, tmp_path)
|
|
bundled = tmp_path / "bundled"
|
|
bundled.mkdir()
|
|
monkeypatch.setenv("FURTKA_BUNDLED_APPS_DIR", str(bundled))
|
|
# No catalog dir — bundled-only.
|
|
monkeypatch.setenv("FURTKA_CATALOG_DIR", str(tmp_path / "catalog"))
|
|
_write_manifest(bundled, "mosquitto")
|
|
_write_manifest(bundled, "zigbee2mqtt", requires=[{"app": "mosquitto"}])
|
|
|
|
from furtka import install_runner, installer
|
|
|
|
# Stub install_from so we don't actually copy files / mess with placeholders.
|
|
install_calls: list[str] = []
|
|
|
|
def fake_install_from(src, settings=None):
|
|
install_calls.append(src.name)
|
|
return tmp_path / src.name
|
|
|
|
monkeypatch.setattr(installer, "install_from", fake_install_from)
|
|
monkeypatch.setattr(install_runner, "_INSTALL_PLAN", tmp_path / "install-plan.json")
|
|
# Named installs drive the docker phase through run_install (so on_install
|
|
# hooks fire) — stub it and assert it's invoked with the chosen target.
|
|
run_calls: list[str] = []
|
|
monkeypatch.setattr(install_runner, "run_install", lambda name: run_calls.append(name))
|
|
|
|
rc = main(["app", "install", "zigbee2mqtt"])
|
|
assert rc == 0
|
|
# Provider installed before consumer.
|
|
assert install_calls == ["mosquitto", "zigbee2mqtt"]
|
|
# The docker phase (which fires on_install hooks) ran for the target.
|
|
assert run_calls == ["zigbee2mqtt"]
|
|
# And the plan file was staged for run_install to consume.
|
|
plan = json.loads(install_runner.plan_path().read_text())
|
|
assert plan["target"] == "zigbee2mqtt"
|
|
assert plan["to_install"] == ["mosquitto", "zigbee2mqtt"]
|
|
|
|
|
|
def test_app_install_named_with_cycle_exits_2(tmp_path, monkeypatch, capsys):
|
|
_set_env(monkeypatch, tmp_path)
|
|
bundled = tmp_path / "bundled"
|
|
bundled.mkdir()
|
|
monkeypatch.setenv("FURTKA_BUNDLED_APPS_DIR", str(bundled))
|
|
monkeypatch.setenv("FURTKA_CATALOG_DIR", str(tmp_path / "catalog"))
|
|
_write_manifest(bundled, "a", requires=[{"app": "b"}])
|
|
_write_manifest(bundled, "b", requires=[{"app": "a"}])
|
|
rc = main(["app", "install", "a"])
|
|
assert rc == 2
|
|
err = capsys.readouterr().err
|
|
assert "circular" in err.lower()
|