import pytest # ---- LOCKED BASELINE: paths + methods ---- BASELINE_ROUTES = { ("GET", "/workloads"), ("GET", "/workloads/read/{filename:path}"), ("POST", "/workloads/save-file"), ("POST", "/workloads/deploy/{filename:path}"), ("DELETE", "/files/rmdir"), ("GET", "/pods"), ("POST", "/actions/{action}/{name}"), ("GET", "/pods-dashboard"), ("POST", "/pods/actions/{action}/{podname}"), ("GET", "/containers-dashboard"), ("GET", "/containers"), ("POST", "/containers/{action}/{name}"), ("GET", "/debug/defined-containers"), ("GET", "/dashboard"), ("GET", "/test-hybrid"), ("GET", "/files/tree"), ("GET", "/files/read"), ("POST", "/files/save"), ("DELETE", "/files/delete"), ("POST", "/files/mkdir"), ("GET", "/containers/logs/{name}"), ("GET", "/containers/inspect/{name}"), ("GET", "/systemd/allowlist"), ("GET", "/"), ("POST", "/daemon-reload"), ("POST", "/{action}/{unit}"), ("POST", "/action"), ("POST", "/api//"), } def _route_tuples_from_app(app): found = set() for r in app.routes: methods = getattr(r, "methods", None) path = getattr(r, "path", None) if not methods or not path: continue for m in methods: # ignore implicit HEAD/OPTIONS etc; baseline is explicit if m in {"HEAD", "OPTIONS"}: continue found.add((m, path)) return found def _assert_keys_exact(obj: dict, expected_keys: list[str]): assert isinstance(obj, dict), f"Expected dict, got {type(obj)}" assert sorted(list(obj.keys())) == sorted(expected_keys), f"Keys mismatch: {sorted(obj.keys())} != {sorted(expected_keys)}" # ---- 1) Route existence: all 28 paths/methods ---- def test_baseline_routes_exist(app_module): found = _route_tuples_from_app(app_module.app) missing = sorted(BASELINE_ROUTES - found) assert not missing, f"Missing baseline routes: {missing}" # ---- 2) FIXED/VARIANT keyset verification ---- def test_files_tree_fixed_shape(client): r = client.get("/files/tree") assert r.status_code == 200 data = r.json() assert isinstance(data, list) if data: _assert_keys_exact(data[0], ["dirs", "files", "path"]) def test_files_read_fixed_shape(client): r = client.get("/files/read", params={"path": "systemd/test.txt"}) assert r.status_code == 200 _assert_keys_exact(r.json(), ["content"]) def test_files_save_fixed_shape(client): r = client.post("/files/save", params={"path": "systemd/test.txt"}, json={"content": "updated\n"}) assert r.status_code == 200 _assert_keys_exact(r.json(), ["path", "status"]) def test_files_delete_fixed_shape(client): # create a file first (via save) client.post("/files/save", params={"path": "systemd/delete-me.txt"}, json={"content": "x"}) r = client.delete("/files/delete", params={"path": "systemd/delete-me.txt"}) assert r.status_code == 200 _assert_keys_exact(r.json(), ["status", "type"]) def test_files_mkdir_fixed_shape(client): r = client.post("/files/mkdir", params={"path": "systemd/newdir"}) assert r.status_code == 200 _assert_keys_exact(r.json(), ["path", "status"]) def test_files_rmdir_variant_shapes(client): # 409: directory not empty -> FastAPI HTTPException detail object r = client.delete("/files/rmdir", params={"path": "systemd/nonempty"}) assert r.status_code == 409 body = r.json() _assert_keys_exact(body, ["detail"]) assert isinstance(body["detail"], dict) _assert_keys_exact(body["detail"], ["dirs", "error", "files"]) # 200: empty directory -> success keys client.post("/files/mkdir", params={"path": "systemd/emptydir"}) r2 = client.delete("/files/rmdir", params={"path": "systemd/emptydir"}) assert r2.status_code == 200 _assert_keys_exact(r2.json(), ["deleted", "path"]) def test_systemd_allowlist_fixed_shape(client): r = client.get("/systemd/allowlist") assert r.status_code == 200 _assert_keys_exact(r.json(), ["allow_mode", "units"]) def test_daemon_reload_fixed_shape(client): r = client.post("/daemon-reload") assert r.status_code == 200 _assert_keys_exact(r.json(), ["cmd", "exit", "output"]) def test_systemd_action_fixed_shape_and_allow_enforcement(client): # allowed unit r = client.post("/status/sonarr.service") assert r.status_code == 200 _assert_keys_exact(r.json(), ["cmd", "exit", "output"]) # not allowed unit -> 403 with {"detail": ...} r2 = client.post("/status/notallowed.service") assert r2.status_code == 403 _assert_keys_exact(r2.json(), ["detail"]) def test_pods_dashboard_fixed_shape(client): r = client.get("/pods-dashboard") assert r.status_code == 200 data = r.json() assert isinstance(data, list) # Expect at least the defined pod from mediaserver.yaml -> "podmediaserver" # Keys per item must be fixed if data: _assert_keys_exact(data[0], ["Containers", "Name", "Source", "Status", "Unit"]) def test_containers_dashboard_variant_shape(client): r = client.get("/containers-dashboard") assert r.status_code == 200 data = r.json() assert isinstance(data, list) # Because we created systemd/sonarr.container, we expect a defined entry # in fixed defined-shape. defined_shape = ["Image", "Names", "PodName", "Ports", "State", "Status", "_dashboard_def_path", "_dashboard_source", "_dashboard_unit"] found_defined = False for item in data: if isinstance(item, dict) and item.get("_dashboard_source") == "systemd": _assert_keys_exact(item, defined_shape) found_defined = True break assert found_defined, "Expected at least one defined container entry with fixed defined-shape" def test_containers_logs_fixed_shape(client): r = client.get("/containers/logs/testcontainer") assert r.status_code == 200 _assert_keys_exact(r.json(), ["logs"]) # NOTE: tuple-return endpoints (invalid action) behave as HTTP 200 with JSON list [dict, status] def test_tuple_return_invalid_action_shapes(client): r = client.post("/containers/foo/bar") assert r.status_code == 200 body = r.json() assert isinstance(body, list) and len(body) == 2 assert isinstance(body[0], dict) and "error" in body[0] r2 = client.post("/pods/actions/foo/podx") assert r2.status_code == 200 body2 = r2.json() assert isinstance(body2, list) and len(body2) == 2 assert isinstance(body2[0], dict) and "error" in body2[0] # ---- 3) DYNAMIC endpoints: statuscode + is JSON only ---- @pytest.mark.parametrize( "method,path,kwargs", [ ("GET", "/pods", {}), ("GET", "/containers", {}), ("GET", "/containers/inspect/testcontainer", {}), ("POST", "/workloads/deploy/mediaserver.yaml", {}), ("GET", "/debug/defined-containers", {}), ], ) def test_dynamic_endpoints_only_json_and_200(client, method, path, kwargs): req = getattr(client, method.lower()) r = req(path, **kwargs) assert r.status_code == 200 data = r.json() assert isinstance(data, (dict, list)), f"Expected JSON (dict/list), got {type(data)}"