From 220dd6b8b13e194f3000c443d98533889839eeca Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Thu, 29 Oct 2026 13:17:37 +0200 Subject: [PATCH 1/2] tests: apply dependency overrides to versioned_app The unit tests send requests to versioned_app, which gets the API routes through include_router() and resolves dependencies against its own dependency_overrides. The auth mocks were only registered on app, so every authenticated endpoint test got 401 Unauthorized. Register the same overrides on versioned_app. Signed-off-by: Denys Fedoryshchenko --- tests/unit_tests/conftest.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/unit_tests/conftest.py b/tests/unit_tests/conftest.py index e73e6d90..20506b5a 100644 --- a/tests/unit_tests/conftest.py +++ b/tests/unit_tests/conftest.py @@ -89,6 +89,7 @@ def mock_get_current_admin_user(request: Request): # Mock dependency callables for getting current user app.dependency_overrides[get_current_user] = mock_get_current_user app.dependency_overrides[get_current_superuser] = mock_get_current_admin_user +versioned_app.dependency_overrides.update(app.dependency_overrides) @pytest.fixture(scope="session") From 6a0126b67908c26a2605634bbf9b81e173f4137d Mon Sep 17 00:00:00 2001 From: Denys Fedoryshchenko Date: Thu, 29 Oct 2026 13:17:37 +0200 Subject: [PATCH 2/2] api: reject state/result changes to done nodes A lab can report a job more than once. The later report silently replaced the first result of a node that was already done (e.g. pass, then fail 15 minutes later), while the KCIDB bridge had already sent the first result to the dashboard, leaving Maestro and the dashboard out of sync with nobody noticing. Make the first final result win: PUT/PATCH /node and PUT /nodes now return 409 Conflict when they would change the state or result of a done node. The hierarchy is checked as a whole before anything is written. Resubmitting the same result is still allowed, and so is the late result for a node closed without one (incomplete, e.g. by node_timeout), which the LAVA callback relies on. Admins can override with force=true. Every rejected or forced change is counted in the metrics and published on the node-integrity channel, so it can be monitored through /events?channel=node-integrity. Signed-off-by: Denys Fedoryshchenko --- api/main.py | 108 ++++++++++++ tests/unit_tests/test_node_done_update.py | 206 ++++++++++++++++++++++ 2 files changed, 314 insertions(+) create mode 100644 tests/unit_tests/test_node_done_update.py diff --git a/api/main.py b/api/main.py index be3a1adc..7e161b86 100644 --- a/api/main.py +++ b/api/main.py @@ -1827,12 +1827,100 @@ def is_same_flags(old_node, new_node): return False +# Rejected (or admin-forced) changes to completed nodes are published on this +# channel, so they can be monitored via /events?channel=node-integrity. +NODE_INTEGRITY_CHANNEL = "node-integrity" + + +def _done_node_update_allowed(existing: Node, new_state, new_result) -> bool: + """Check if an update may be applied to a node that is already done""" + if existing.state != "done": + return True + if new_state != "done": + return False + if new_result == existing.result: + return True + # A node closed without a final result (e.g. by node_timeout) can still + # receive the late result from its lab. + return existing.result in (None, "incomplete") + + +async def _check_done_node_update( + existing: Node, new_state, new_result, user: User, force: Optional[bool] +): + """Reject state/result changes to done nodes and report every attempt + + The first final result wins: once a node is done, its state and result + can only be changed by an admin with force=true. Both rejected and + forced changes are published on NODE_INTEGRITY_CHANNEL. + """ + # PATCH passes StateValues/ResultValues enums, PUT plain strings + new_state = getattr(new_state, "value", new_state) + new_result = getattr(new_result, "value", new_result) + if _done_node_update_allowed(existing, new_state, new_result): + return + forced = bool(force and user.is_superuser) + op = "done_update_forced" if forced else "done_update_rejected" + metrics.add(f"node_{op}_total", 1) + data = { + "op": op, + "id": str(existing.id), + "kind": existing.kind, + "name": existing.name, + "path": existing.path, + "group": existing.group, + "owner": existing.owner, + "runtime": (existing.data or {}).get("runtime"), + "platform": (existing.data or {}).get("platform"), + "state": existing.state, + "result": existing.result, + "attempted_state": new_state, + "attempted_result": new_result, + "user": user.username, + } + try: + await pubsub.publish_cloudevent( + NODE_INTEGRITY_CHANNEL, data, {"owner": existing.owner} + ) + except Exception as exc: + # Reporting must not turn the rejection into a server error + print(f"Failed to publish {op} event for {existing.id}: {exc}") + if not forced: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + f"Node {existing.id} is already done with result " + f"'{existing.result}', refusing to change it to " + f"state '{new_state}' result '{new_result}'" + ), + ) + + +async def _check_done_hierarchy_update( + hierarchy: Hierarchy, root: Node, user: User, force: Optional[bool] +): + """Run _check_done_node_update on every existing node of a hierarchy""" + node = hierarchy.node + existing = root + if existing is None and node.id: + existing = await db.find_by_id(Node, node.id) + if existing is not None: + await _check_done_node_update( + existing, node.state, node.result, user, force + ) + for child in hierarchy.child_nodes: + await _check_done_hierarchy_update(child, None, user, force) + + @app.put("/node/{node_id}", response_model=Node, response_model_by_alias=False) async def put_node( node_id: str, node: Node, user: str = Depends(authorize_user), noevent: Optional[bool] = Query(None), + force: Optional[bool] = Query( + None, description="Admin only: allow changing a done node" + ), ): """Update an already added node""" metrics.add("http_requests_total", 1) @@ -1847,6 +1935,10 @@ async def put_node( # [TODO] Remove translation below once we can use it in the pipeline node = _translate_version_fields(node) + await _check_done_node_update( + node_from_id, node.state, node.result, user, force + ) + # Sanity checks # Note: do not update node ownership fields, don't update 'state' # until we've checked the state transition is valid. @@ -1913,6 +2005,9 @@ async def patch_node( patch: NodePatchRequest, user: str = Depends(authorize_user), noevent: Optional[bool] = Query(None), + force: Optional[bool] = Query( + None, description="Admin only: allow changing a done node" + ), ): """Partial update of an existing node""" metrics.add("http_requests_total", 1) @@ -1930,6 +2025,14 @@ async def patch_node( detail="No fields to update", ) + await _check_done_node_update( + node_from_id, + update_data.get("state", node_from_id.state), + update_data.get("result", node_from_id.result), + user, + force, + ) + # Handle state transition separately new_state = update_data.pop("state", None) @@ -2041,6 +2144,9 @@ async def put_nodes( nodes: Hierarchy, authorization: str | None = Header(default=None), user: str = Depends(authorize_user), + force: Optional[bool] = Query( + None, description="Admin only: allow changing done nodes" + ), ): """Add a hierarchy of nodes to an existing root node""" metrics.add("http_requests_total", 1) @@ -2052,6 +2158,8 @@ async def put_nodes( status_code=status.HTTP_404_NOT_FOUND, detail=f"Node not found with id: {node_id}", ) + # Check the whole hierarchy before writing any of it + await _check_done_hierarchy_update(nodes, node_from_id, user, force) submitter = node_from_id.submitter treeid = node_from_id.treeid diff --git a/tests/unit_tests/test_node_done_update.py b/tests/unit_tests/test_node_done_update.py new file mode 100644 index 00000000..dc3108f6 --- /dev/null +++ b/tests/unit_tests/test_node_done_update.py @@ -0,0 +1,206 @@ +# SPDX-License-Identifier: LGPL-2.1-or-later + +"""Unit tests for rejecting state/result changes to done nodes""" + +import json + +import pytest +from fastapi import HTTPException +from kernelci.api.models import Node + +from api.main import NODE_INTEGRITY_CHANNEL, _check_done_node_update +from api.models import User +from tests.unit_tests.conftest import BEARER_TOKEN + +NODE_ID = "6ae228345c0836a8c2f041a8" +HEADERS = {"Accept": "application/json", "Authorization": BEARER_TOKEN} + + +def _make_node(state="done", result="pass", node_id=NODE_ID): + return Node( + id=node_id, + kind="job", + name="baseline-arm64", + path=["checkout", "kbuild-gcc-14-arm64", "baseline-arm64"], + group="baseline-arm64", + parent="6ae2138e5c0836a8c2eeeac6", + state=state, + result=result, + owner="bob", + data={"runtime": "sasha-lab", "platform": "jetson-tx1-devkit"}, + ) + + +def _make_user(is_superuser=False): + return User( + id="65265305c74695807499037f", + username="bob", + hashed_password="hash", + email="bob@kernelci.org", + is_active=True, + is_superuser=is_superuser, + is_verified=True, + ) + + +def _node_payload(node, **update): + payload = json.loads(node.model_dump_json()) + payload.update(update) + return payload + + +@pytest.fixture +def mock_integrity_publish(mocker): + """Capture events published by the done-node guard""" + pubsub = mocker.AsyncMock() + mocker.patch("api.main.pubsub", pubsub) + return pubsub.publish_cloudevent + + +def test_put_node_rejects_result_flip( + mock_db_find_by_id, mock_db_update, mock_integrity_publish, test_client +): + """PUT changing pass to fail on a done node is rejected and reported""" + node = _make_node() + mock_db_find_by_id.return_value = node + + response = test_client.put( + f"node/{NODE_ID}", + headers=HEADERS, + json=_node_payload(node, result="fail"), + ) + + assert response.status_code == 409 + mock_db_update.assert_not_called() + mock_integrity_publish.assert_awaited_once() + channel, data, _ = mock_integrity_publish.await_args.args + assert channel == NODE_INTEGRITY_CHANNEL + assert data["op"] == "done_update_rejected" + assert data["id"] == NODE_ID + assert data["result"] == "pass" + assert data["attempted_result"] == "fail" + assert data["runtime"] == "sasha-lab" + + +def test_put_node_allows_same_result( + mock_db_find_by_id, + mock_db_update, + mock_publish_cloudevent, + mock_integrity_publish, + test_client, +): + """PUT keeping the result of a done node (e.g. adding data) is allowed""" + node = _make_node() + mock_db_find_by_id.return_value = node + mock_db_update.return_value = node + + response = test_client.put( + f"node/{NODE_ID}", + headers=HEADERS, + json=_node_payload(node), + ) + + assert response.status_code == 200 + channels = [c.args[0] for c in mock_integrity_publish.await_args_list] + assert NODE_INTEGRITY_CHANNEL not in channels + + +def test_put_node_allows_late_result_after_timeout( + mock_db_find_by_id, mock_db_update, mock_integrity_publish, test_client +): + """A node closed as incomplete can still get its final result""" + node = _make_node(result="incomplete") + mock_db_find_by_id.return_value = node + mock_db_update.return_value = _make_node(result="fail") + + response = test_client.put( + f"node/{NODE_ID}", + headers=HEADERS, + json=_node_payload(node, result="fail"), + ) + + assert response.status_code == 200 + channels = [c.args[0] for c in mock_integrity_publish.await_args_list] + assert NODE_INTEGRITY_CHANNEL not in channels + + +def test_patch_node_rejects_result_flip( + mock_db_find_by_id, mock_db_update, mock_integrity_publish, test_client +): + """PATCH changing the result of a done node is rejected""" + mock_db_find_by_id.return_value = _make_node() + + response = test_client.patch( + f"node/{NODE_ID}", + headers=HEADERS, + json={"result": "fail"}, + ) + + assert response.status_code == 409 + mock_db_update.assert_not_called() + mock_integrity_publish.assert_awaited_once() + + +def test_put_nodes_rejects_hierarchy_overwrite( + mocker, mock_db_find_by_id, mock_integrity_publish, test_client +): + """A hierarchy re-submitted with a different root result is rejected""" + node = _make_node() + mock_db_find_by_id.return_value = node + create_hierarchy = mocker.patch( + "api.db.Database.create_hierarchy", side_effect=mocker.AsyncMock() + ) + + response = test_client.put( + f"nodes/{NODE_ID}", + headers=HEADERS, + json={ + "node": _node_payload(node, result="fail"), + "child_nodes": [], + }, + ) + + assert response.status_code == 409 + create_hierarchy.assert_not_called() + mock_integrity_publish.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_force_by_admin_is_allowed_and_reported(mock_integrity_publish): + """An admin can force the change; it is still reported""" + await _check_done_node_update( + _make_node(), "done", "fail", _make_user(is_superuser=True), True + ) + + mock_integrity_publish.assert_awaited_once() + assert ( + mock_integrity_publish.await_args.args[1]["op"] == "done_update_forced" + ) + + +@pytest.mark.asyncio +async def test_force_by_regular_user_is_rejected(mock_integrity_publish): + """force=true is ignored for non-admin users""" + with pytest.raises(HTTPException) as exc: + await _check_done_node_update( + _make_node(), "done", "fail", _make_user(), True + ) + + assert exc.value.status_code == 409 + assert ( + mock_integrity_publish.await_args.args[1]["op"] + == "done_update_rejected" + ) + + +@pytest.mark.asyncio +async def test_publish_failure_still_rejects(mock_integrity_publish): + """A failing event publish must not turn the 409 into a 500""" + mock_integrity_publish.side_effect = RuntimeError("redis down") + + with pytest.raises(HTTPException) as exc: + await _check_done_node_update( + _make_node(), "done", "fail", _make_user(), None + ) + + assert exc.value.status_code == 409