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] # --- Icon inlining ---------------------------------------------------------- _SIMPLE_SVG = ( '' ) def _write_icon(app_dir, contents, name="icon.svg"): (app_dir / name).write_text(contents) def test_read_icon_svg_returns_content(tmp_path): _write_icon(tmp_path, _SIMPLE_SVG) assert api._read_icon_svg(tmp_path, "icon.svg") == _SIMPLE_SVG def test_read_icon_svg_strips_xml_declaration(tmp_path): _write_icon(tmp_path, '\n' + _SIMPLE_SVG) assert api._read_icon_svg(tmp_path, "icon.svg") == _SIMPLE_SVG def test_read_icon_svg_missing_file_returns_none(tmp_path): assert api._read_icon_svg(tmp_path, "ghost.svg") is None def test_read_icon_svg_no_name_returns_none(tmp_path): assert api._read_icon_svg(tmp_path, None) is None assert api._read_icon_svg(tmp_path, "") is None def test_read_icon_svg_rejects_non_svg(tmp_path): _write_icon(tmp_path, "hi") assert api._read_icon_svg(tmp_path, "icon.svg") is None def test_read_icon_svg_rejects_oversized(tmp_path): _write_icon(tmp_path, "") assert api._read_icon_svg(tmp_path, "icon.svg") is None def test_read_icon_svg_rejects_script_tag(tmp_path): _write_icon(tmp_path, "") assert api._read_icon_svg(tmp_path, "icon.svg") is None def test_read_icon_svg_rejects_event_handler(tmp_path): _write_icon(tmp_path, '') assert api._read_icon_svg(tmp_path, "icon.svg") is None def test_read_icon_svg_rejects_javascript_url(tmp_path): _write_icon(tmp_path, '') assert api._read_icon_svg(tmp_path, "icon.svg") is None def test_list_bundled_inlines_icon_svg(fake_dirs): _, bundled = fake_dirs app = _write_bundled(bundled, "fileshare") _write_icon(app, _SIMPLE_SVG) [entry] = api._list_bundled() assert entry["icon_svg"] == _SIMPLE_SVG def test_list_installed_inlines_icon_svg(fake_dirs, no_docker): apps, bundled = fake_dirs app = _write_bundled(bundled, "fileshare", env_example="A=real") _write_icon(app, _SIMPLE_SVG) api._do_install("fileshare") [entry] = api._list_installed() assert entry["icon_svg"] == _SIMPLE_SVG 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()