75 lines
2.5 KiB
Python
75 lines
2.5 KiB
Python
"""FU-7: audit-trail + provenance (offline, monkeypatched I/O)."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from legal_mcp.services import audit, db
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.run(coro)
|
|
|
|
|
|
# ── GAP-18: log_action_safe is non-fatal ───────────────────────────────
|
|
def test_log_action_safe_swallows_db_error(monkeypatch):
|
|
async def _boom(*a, **k):
|
|
raise RuntimeError("db down")
|
|
monkeypatch.setattr(audit, "log_action", _boom)
|
|
# must NOT raise
|
|
_run(audit.log_action_safe("write_block", details={"x": 1}))
|
|
|
|
|
|
def test_log_action_safe_forwards_args(monkeypatch):
|
|
seen = {}
|
|
async def _capture(action, case_id=None, document_id=None, details=None, user="system"):
|
|
seen.update(action=action, details=details)
|
|
monkeypatch.setattr(audit, "log_action", _capture)
|
|
_run(audit.log_action_safe("export_docx", details={"path": "/x"}))
|
|
assert seen["action"] == "export_docx" and seen["details"] == {"path": "/x"}
|
|
|
|
|
|
# ── GAP-20: structural citation resolver ────────────────────────────────
|
|
def test_resolve_citation_case_law_ids_splits(monkeypatch):
|
|
good = uuid4()
|
|
bad = uuid4()
|
|
|
|
class _Conn:
|
|
async def fetchval(self, q, cid):
|
|
return cid == good
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *a): return False
|
|
|
|
class _Pool:
|
|
def acquire(self): return _Conn()
|
|
|
|
async def _pool():
|
|
return _Pool()
|
|
monkeypatch.setattr(db, "get_pool", _pool)
|
|
|
|
out = _run(db.resolve_citation_case_law_ids([good, bad]))
|
|
assert good in out["resolved"] and bad in out["unresolved"]
|
|
|
|
|
|
# ── GAP-17: blocks_stale helper ────────────────────────────────────────
|
|
def test_mark_blocks_stale_executes_update(monkeypatch):
|
|
seen = {}
|
|
|
|
class _Conn:
|
|
async def execute(self, q, *a):
|
|
seen["q"] = q; seen["args"] = a
|
|
async def __aenter__(self): return self
|
|
async def __aexit__(self, *a): return False
|
|
|
|
class _Pool:
|
|
def acquire(self): return _Conn()
|
|
|
|
async def _pool(): return _Pool()
|
|
monkeypatch.setattr(db, "get_pool", _pool)
|
|
|
|
cid = uuid4()
|
|
_run(db.mark_blocks_stale(cid, True))
|
|
assert "blocks_stale" in seen["q"] and seen["args"][0] is True and seen["args"][1] == cid
|