""" node-26#14 / #15 — approve must not rotate an existing key, and the host updater's HTTPS channel (request-update + update-check). """ from unittest.mock import AsyncMock, MagicMock, patch from fastapi.testclient import TestClient from app.main import app from app.internal.auth import require_admin_token, require_service_or_firebase_token from app.routers import enrollment, nodes client = TestClient(app) def setup_function(): app.dependency_overrides[require_admin_token] = lambda: {"admin": True} app.dependency_overrides[require_service_or_firebase_token] = lambda: {"admin": True} def teardown_function(): app.dependency_overrides.pop(require_admin_token, None) app.dependency_overrides.pop(require_service_or_firebase_token, None) def _store(node, key_doc): return AsyncMock(side_effect=lambda c, i: {"nodes": node, "node_keys": key_doc}.get(c)) # --------------------------------------------------------------------------- # approve_node # --------------------------------------------------------------------------- def _approve(node, key_doc): with patch.object(nodes.fstore, "doc_get", _store(node, key_doc)), \ patch.object(nodes.fstore, "doc_set", AsyncMock()) as doc_set, \ patch.object(nodes.fstore, "doc_update", AsyncMock()) as doc_update, \ patch.object(nodes.dynsec, "upsert_node_client", AsyncMock()) as upsert, \ patch.object(nodes.mqtt_handler, "publish_node_key", MagicMock()) as publish: resp = client.post("/nodes/n1/approve") return resp, doc_set, doc_update, upsert, publish def test_approve_already_keyed_node_reuses_key_and_never_overwrites_node_keys(): node = {"node_id": "n1", "approval_status": "approved"} resp, doc_set, doc_update, upsert, publish = _approve(node, {"node_id": "n1", "api_key": "live-key"}) assert resp.status_code == 200 assert resp.json()["key_rotated"] is False doc_set.assert_not_awaited() upsert.assert_awaited_once_with("n1", "live-key") publish.assert_called_once_with("n1", "live-key") def test_approve_new_node_mints_a_key(): resp, doc_set, doc_update, upsert, publish = _approve({"node_id": "n1", "approval_status": "pending"}, None) assert resp.status_code == 200 (_, _, written), kwargs = doc_set.await_args assert len(written["api_key"]) == 64 assert kwargs == {"merge": False} upsert.assert_awaited_once_with("n1", written["api_key"]) doc_update.assert_awaited_once_with("nodes", "n1", {"approval_status": "approved"}) def test_approve_twice_keeps_the_same_key(): store = {"nodes": {"node_id": "n1", "approval_status": "pending"}, "node_keys": None} async def doc_get(c, i): return store[c] async def doc_set(c, i, data, merge=False): store[c] = data with patch.object(nodes.fstore, "doc_get", doc_get), \ patch.object(nodes.fstore, "doc_set", doc_set), \ patch.object(nodes.fstore, "doc_update", AsyncMock()), \ patch.object(nodes.dynsec, "upsert_node_client", AsyncMock()), \ patch.object(nodes.mqtt_handler, "publish_node_key", MagicMock()): client.post("/nodes/n1/approve") first = store["node_keys"]["api_key"] client.post("/nodes/n1/approve") assert store["node_keys"]["api_key"] == first # --------------------------------------------------------------------------- # request-update (admin) # --------------------------------------------------------------------------- def test_request_update_sets_flag_even_when_mqtt_is_down(): with patch.object(nodes.fstore, "doc_get", _store({"node_id": "n1"}, None)), \ patch.object(nodes.fstore, "doc_update", AsyncMock()) as doc_update, \ patch.object(nodes.mqtt_handler, "send_command", MagicMock(return_value=False)) as cmd: resp = client.post("/nodes/n1/request-update") assert resp.status_code == 200 body = resp.json() assert body["mqtt_sent"] is False doc_update.assert_awaited_once_with("nodes", "n1", {"update_requested_at": body["update_requested_at"]}) cmd.assert_called_once_with("n1", {"action": "node_update"}) def test_request_update_requires_admin(): app.dependency_overrides.pop(require_admin_token, None) resp = client.post("/nodes/n1/request-update") assert resp.status_code in (401, 403) # --------------------------------------------------------------------------- # update-check (node, HTTPS) # --------------------------------------------------------------------------- NODE = { "node_id": "n1", "pickup_secret_hash": enrollment._hash_secret("pickup-1"), "update_requested_at": "2026-10-04T12:00:00+00:00", } def _check(headers, body=None, node=NODE, key_doc=None): key_doc = key_doc if key_doc is not None else {"api_key": "k" * 64} # A node holds no Firebase/service credential: prove the route needs none. app.dependency_overrides.clear() with patch.object(enrollment.fstore, "doc_get", _store(node, key_doc)), \ patch.object(enrollment.fstore, "doc_update", AsyncMock()) as doc_update: resp = client.post("/nodes/n1/update-check", json=body or {}, headers=headers) return resp, doc_update def test_update_check_with_api_key_returns_request(): resp, doc_update = _check({"Authorization": "Bearer " + "k" * 64}, {"sha": "abc1234"}) assert resp.status_code == 200 assert resp.json() == {"update_requested_at": NODE["update_requested_at"]} (_, _, updates), _ = doc_update.await_args assert updates["updater_sha"] == "abc1234" assert "update_applied_at" not in updates def test_update_check_with_pickup_secret_works_for_a_keyless_node(): resp, _ = _check({"X-Pickup-Secret": "pickup-1"}, key_doc={}) assert resp.status_code == 200 def test_update_check_rejects_wrong_or_missing_credentials(): assert _check({"Authorization": "Bearer wrong"})[0].status_code == 401 assert _check({"X-Pickup-Secret": "wrong"})[0].status_code == 401 assert _check({})[0].status_code == 401 def test_update_check_does_not_reveal_unknown_node_ids(): resp, _ = _check({"X-Pickup-Secret": "pickup-1"}, node=None) assert resp.status_code == 401 def test_update_check_empty_stored_key_never_matches_empty_bearer(): resp, _ = _check({"Authorization": "Bearer "}, key_doc={"api_key": ""}) assert resp.status_code == 401 def test_update_check_records_applied_request(): resp, doc_update = _check( {"Authorization": "Bearer " + "k" * 64}, {"applied_request_at": NODE["update_requested_at"], "result": "ok", "sha": "not-a-sha!"}, ) assert resp.status_code == 200 (_, _, updates), _ = doc_update.await_args assert updates["update_result"] == "ok" assert "update_applied_at" in updates assert "updater_sha" not in updates # junk sha is never written def test_update_check_rejects_unknown_result_value(): resp, _ = _check({"Authorization": "Bearer " + "k" * 64}, {"result": "rm -rf /"}) assert resp.status_code == 422