feat(e1): AuditMiddleware als systematisches Safety-Net — alle erfolgreichen POST/PATCH/DELETE erzeugen Audit-Eintrag (Session-basierte user/tenant-Attribuierung, entity_type aus Pfad, source=middleware in changes); schließt Lücke von 349 mutierenden Endpoints in 59 Dateien ohne Audit; Skip-Liste auth/health/errors/audit/external; best-effort; Beweistest test_audit_middleware.py grün (POST ohne explizites log_audit → Audit-Zeile); Regressionssmoke 23/23 grün
This commit is contained in:
@@ -0,0 +1,48 @@
|
||||
"""Tests for AuditMiddleware (Block E / E1).
|
||||
|
||||
Proves the AGENTS.md requirement "every mutation produces an audit entry"
|
||||
for routes that lack an explicit ``log_audit`` call.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from app.models.audit import AuditLog
|
||||
from tests.conftest import ORIGIN_HEADER, login_client, seed_tenant_and_users
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_post_without_explicit_audit_creates_middleware_entry(
|
||||
client, db_session: AsyncSession
|
||||
):
|
||||
"""POST to a route without explicit log_audit still yields an audit row."""
|
||||
seed = await seed_tenant_and_users(db_session)
|
||||
await login_client(client, "admin@tenanta.com")
|
||||
|
||||
# Mutating request on a route WITHOUT explicit log_audit:
|
||||
# create a saved view via core route (no log_audit in saved_views.py)
|
||||
view_name = f"audit-mw-{uuid.uuid4().hex[:8]}"
|
||||
resp = await client.post(
|
||||
"/api/v1/saved-views",
|
||||
json={"name": view_name, "entity_type": "contact", "config": {}},
|
||||
headers=ORIGIN_HEADER,
|
||||
)
|
||||
assert resp.status_code in (200, 201), resp.text
|
||||
|
||||
# A middleware-written entry must exist with source=middleware in changes
|
||||
rows = (
|
||||
await db_session.execute(select(AuditLog).where(AuditLog.tenant_id == seed["tenant_a"].id))
|
||||
).scalars().all()
|
||||
mw_rows = [
|
||||
r for r in rows if r.changes and isinstance(r.changes, dict) and r.changes.get("source") == "middleware"
|
||||
]
|
||||
assert len(mw_rows) > 0, "expected at least one middleware-sourced audit entry"
|
||||
latest = max(mw_rows, key=lambda r: r.timestamp)
|
||||
assert latest.changes["method"] == "POST"
|
||||
assert latest.changes["path"].startswith("/api/v1/")
|
||||
assert latest.action in ("create", "update", "delete")
|
||||
Reference in New Issue
Block a user