furtka/tests/test_api.py

import json
import threading
import urllib.error
import urllib.request

import pytest

from furtka import api, dockerops

VALID_MANIFEST = {
    "name": "fileshare",
    "display_name": "Network Files",
    "version": "0.1.0",
    "description": "SMB share",
    "volumes": ["files"],
    "ports": [445],
    "icon": "icon.svg",
}


@pytest.fixture
def fake_dirs(tmp_path, monkeypatch):
    apps = tmp_path / "apps"
    bundled = tmp_path / "bundled"
    apps.mkdir()
    bundled.mkdir()
    monkeypatch.setenv("FURTKA_APPS_DIR", str(apps))
    monkeypatch.setenv("FURTKA_BUNDLED_APPS_DIR", str(bundled))
    return apps, bundled


@pytest.fixture
def no_docker(monkeypatch):
    """Stub docker calls so install/remove can run without a daemon."""
    monkeypatch.setattr(dockerops, "ensure_volume", lambda name: True)
    monkeypatch.setattr(dockerops, "compose_up", lambda app_dir, project: None)
    monkeypatch.setattr(dockerops, "compose_down", lambda app_dir, project: None)


def _write_bundled(bundled, name, manifest=None, env_example=None):
    app = bundled / name
    app.mkdir()
    (app / "manifest.json").write_text(json.dumps(manifest or VALID_MANIFEST))
    (app / "docker-compose.yaml").write_text("services: {}\n")
    if env_example is not None:
        (app / ".env.example").write_text(env_example)
    return app


def test_list_installed_empty(fake_dirs):
    assert api._list_installed() == []


def test_list_bundled_empty(fake_dirs):
    assert api._list_bundled() == []


def test_list_bundled_shows_uninstalled(fake_dirs):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare")
    out = api._list_bundled()
    assert len(out) == 1
    assert out[0]["name"] == "fileshare"
    assert "display_name" in out[0]


def test_list_bundled_hides_already_installed(fake_dirs, no_docker):
    apps, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", env_example="A=real")
    status, _ = api._do_install("fileshare")
    assert status == 200
    # Now bundled should NOT include fileshare anymore.
    assert api._list_bundled() == []
    # But installed list should.
    installed = api._list_installed()
    assert len(installed) == 1 and installed[0]["name"] == "fileshare"


def test_install_endpoint_rejects_placeholder(fake_dirs):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", env_example="SMB_PASSWORD=changeme")
    status, body = api._do_install("fileshare")
    assert status == 400
    assert "placeholder" in body["error"]


def test_install_endpoint_rejects_unknown_app(fake_dirs):
    status, body = api._do_install("does-not-exist")
    assert status == 400
    assert "not found" in body["error"]


def test_remove_endpoint_unknown(fake_dirs, no_docker):
    status, body = api._do_remove("ghost")
    assert status == 404


def test_remove_endpoint_happy_path(fake_dirs, no_docker):
    apps, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", env_example="A=real")
    api._do_install("fileshare")
    assert (apps / "fileshare").exists()
    status, body = api._do_remove("fileshare")
    assert status == 200
    assert body["removed"] == "fileshare"
    assert not (apps / "fileshare").exists()


def test_http_get_apps_route(fake_dirs, no_docker):
    """Smoke test the actual HTTP server with a real socket, urllib client."""
    server = api.HTTPServer(("127.0.0.1", 0), api._Handler)  # port 0 → ephemeral
    port = server.server_address[1]
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/apps") as r:
            assert r.status == 200
            data = json.loads(r.read())
            assert data == []
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/") as r:
            assert r.status == 200
            assert b"Furtka Apps" in r.read()
        # Unknown route → 404 JSON.
        try:
            urllib.request.urlopen(f"http://127.0.0.1:{port}/api/nope")
            raise AssertionError("expected 404")
        except urllib.error.HTTPError as e:
            assert e.code == 404
    finally:
        server.shutdown()
        server.server_close()


def test_http_post_install_unknown_app(fake_dirs):
    server = api.HTTPServer(("127.0.0.1", 0), api._Handler)
    port = server.server_address[1]
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    try:
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}/api/apps/install",
            data=json.dumps({"name": "ghost"}).encode(),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        try:
            urllib.request.urlopen(req)
            raise AssertionError("expected 400")
        except urllib.error.HTTPError as e:
            assert e.code == 400
            body = json.loads(e.read())
            assert "not found" in body["error"]
    finally:
        server.shutdown()
        server.server_close()


# --- Settings endpoints ------------------------------------------------------

SETTINGS_MANIFEST = dict(
    VALID_MANIFEST,
    description_long="Long help text.",
    settings=[
        {
            "name": "SMB_USER",
            "label": "User",
            "type": "text",
            "default": "furtka",
            "required": True,
        },
        {"name": "SMB_PASSWORD", "label": "Pass", "type": "password", "required": True},
    ],
)


def test_get_settings_bundled(fake_dirs):
    _, bundled = fake_dirs
    _write_bundled(
        bundled, "fileshare", manifest=SETTINGS_MANIFEST, env_example="SMB_USER=furtka\n"
    )
    status, body = api._do_get_settings("fileshare")
    assert status == 200
    assert body["installed"] is False
    assert body["description_long"] == "Long help text."
    names = [s["name"] for s in body["settings"]]
    assert names == ["SMB_USER", "SMB_PASSWORD"]
    # Password values never leak back.
    pwd = next(s for s in body["settings"] if s["name"] == "SMB_PASSWORD")
    assert pwd["value"] == ""
    # Text value comes from .env.example.
    user = next(s for s in body["settings"] if s["name"] == "SMB_USER")
    assert user["value"] == "furtka"


def test_get_settings_not_found(fake_dirs):
    status, _ = api._do_get_settings("ghost")
    assert status == 404


def test_install_with_settings_writes_env_via_api(fake_dirs, no_docker):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", manifest=SETTINGS_MANIFEST)
    status, body = api._do_install(
        "fileshare", settings={"SMB_USER": "alice", "SMB_PASSWORD": "s3cret"}
    )
    assert status == 200, body
    apps, _ = fake_dirs
    env = (apps / "fileshare" / ".env").read_text()
    assert "SMB_USER=alice" in env
    assert "SMB_PASSWORD=s3cret" in env


def test_install_with_settings_rejects_empty_required_via_api(fake_dirs, no_docker):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", manifest=SETTINGS_MANIFEST)
    status, body = api._do_install("fileshare", settings={"SMB_USER": "a", "SMB_PASSWORD": ""})
    assert status == 400
    assert "SMB_PASSWORD" in body["error"]


def test_update_settings_merges(fake_dirs, no_docker):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", manifest=SETTINGS_MANIFEST)
    api._do_install("fileshare", settings={"SMB_USER": "alice", "SMB_PASSWORD": "original"})
    # Edit flow: submit only the changed password.
    status, body = api._do_update_settings("fileshare", {"SMB_PASSWORD": "newpass"})
    assert status == 200, body
    apps, _ = fake_dirs
    env = (apps / "fileshare" / ".env").read_text()
    assert "SMB_USER=alice" in env
    assert "SMB_PASSWORD=newpass" in env


def test_update_settings_unknown_app(fake_dirs):
    status, _ = api._do_update_settings("ghost", {"SMB_USER": "x"})
    assert status == 404


def test_http_get_settings_route(fake_dirs, no_docker):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", manifest=SETTINGS_MANIFEST)
    server = api.HTTPServer(("127.0.0.1", 0), api._Handler)
    port = server.server_address[1]
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    try:
        with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/apps/fileshare/settings") as r:
            assert r.status == 200
            data = json.loads(r.read())
            assert data["name"] == "fileshare"
            assert len(data["settings"]) == 2
    finally:
        server.shutdown()
        server.server_close()


def test_http_post_install_with_settings(fake_dirs, no_docker):
    _, bundled = fake_dirs
    _write_bundled(bundled, "fileshare", manifest=SETTINGS_MANIFEST)
    server = api.HTTPServer(("127.0.0.1", 0), api._Handler)
    port = server.server_address[1]
    t = threading.Thread(target=server.serve_forever, daemon=True)
    t.start()
    try:
        req = urllib.request.Request(
            f"http://127.0.0.1:{port}/api/apps/install",
            data=json.dumps(
                {
                    "name": "fileshare",
                    "settings": {"SMB_USER": "alice", "SMB_PASSWORD": "s3cret"},
                }
            ).encode(),
            headers={"Content-Type": "application/json"},
            method="POST",
        )
        with urllib.request.urlopen(req) as r:
            assert r.status == 200
        apps, _ = fake_dirs
        assert "SMB_PASSWORD=s3cret" in (apps / "fileshare" / ".env").read_text()
    finally:
        server.shutdown()
        server.server_close()