46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from src.services.mcp_audit_context import (
|
|
build_mcp_audit_context,
|
|
with_mcp_audit_context,
|
|
)
|
|
|
|
|
|
def test_build_mcp_audit_context_keeps_non_empty_fields() -> None:
|
|
context = build_mcp_audit_context(
|
|
session_id="incident:INC-1:pre_decision",
|
|
incident_id="INC-1",
|
|
flywheel_node="sense",
|
|
agent_role="pre_decision_investigator",
|
|
operator_user_id=None,
|
|
)
|
|
|
|
assert context == {
|
|
"gateway_path": "legacy_registry_provider",
|
|
"session_id": "incident:INC-1:pre_decision",
|
|
"incident_id": "INC-1",
|
|
"flywheel_node": "sense",
|
|
"agent_role": "pre_decision_investigator",
|
|
}
|
|
|
|
|
|
def test_with_mcp_audit_context_merges_existing_context_without_mutating_source() -> None:
|
|
params = {
|
|
"namespace": "awoooi-prod",
|
|
"_mcp_audit": {"trace_id": "trace-1", "flywheel_node": "old"},
|
|
}
|
|
|
|
audited = with_mcp_audit_context(
|
|
params,
|
|
incident_id="INC-2",
|
|
flywheel_node="verify",
|
|
agent_role="post_execution_verifier",
|
|
)
|
|
|
|
assert params["_mcp_audit"]["flywheel_node"] == "old"
|
|
assert audited is not params
|
|
assert audited["_mcp_audit"]["trace_id"] == "trace-1"
|
|
assert audited["_mcp_audit"]["incident_id"] == "INC-2"
|
|
assert audited["_mcp_audit"]["flywheel_node"] == "verify"
|
|
assert audited["_mcp_audit"]["agent_role"] == "post_execution_verifier"
|