56 lines
1.4 KiB
Python
56 lines
1.4 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from src.services import mcp_audit_service
|
|
|
|
|
|
class _FakeDb:
|
|
def __init__(self) -> None:
|
|
self.executed_params: list[dict[str, Any]] = []
|
|
|
|
async def execute(self, _statement: Any, params: dict[str, Any]) -> None:
|
|
self.executed_params.append(params)
|
|
|
|
|
|
class _FakeDbContext:
|
|
def __init__(self, db: _FakeDb) -> None:
|
|
self.db = db
|
|
|
|
async def __aenter__(self) -> _FakeDb:
|
|
return self.db
|
|
|
|
async def __aexit__(self, *_args: Any) -> None:
|
|
return None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_record_mcp_call_normalizes_long_session_id(monkeypatch) -> None:
|
|
db = _FakeDb()
|
|
monkeypatch.setattr(
|
|
mcp_audit_service,
|
|
"get_db_context",
|
|
lambda: _FakeDbContext(db),
|
|
)
|
|
|
|
await mcp_audit_service.record_mcp_call(
|
|
mcp_server="k8s_provider",
|
|
tool_name="get_pods",
|
|
input_params={
|
|
"_mcp_audit": {
|
|
"session_id": "incident:INC-20260505-E8033A:pre_decision",
|
|
"agent_role": "pre_decision_investigator",
|
|
},
|
|
},
|
|
output_result={"ok": True},
|
|
duration_ms=12,
|
|
success=True,
|
|
error_message=None,
|
|
)
|
|
|
|
audit_insert_params = db.executed_params[0]
|
|
assert audit_insert_params["session_id"] == "inc:INC-20260505-E8033A:pre"
|
|
assert len(audit_insert_params["session_id"]) <= 36
|