76a4e41e3b
Backend: - conftest with async SQLite DB, factory fixtures for all models - pytest-asyncio config in pyproject.toml - Tests: auth (JWT, dev login), RBAC (access service), audit (query, export), brand enforcement (colors, fonts, logos, contrast), retention (cleanup, purge), content intelligence (regex classifiers), slide mapping, review workflow, analytics data queries Frontend: - Cypress E2E config with baseUrl and viewport settings - Custom commands (devLogin, createPresentation) - E2E specs: login flow, wizard navigation, admin panel, review workflow - Test scripts in package.json Infrastructure: - Makefile: test-e2e and test-all targets Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
108 lines
3.4 KiB
Python
108 lines
3.4 KiB
Python
"""Tests for audit_service: query, export_csv, export_json."""
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from models.sql.audit_log import AuditLogModel
|
|
from services import audit_service
|
|
|
|
|
|
class TestAuditQuery:
|
|
async def test_query_returns_entries(self, session):
|
|
entry = AuditLogModel(
|
|
user_id=uuid.uuid4(),
|
|
action="create",
|
|
resource_type="presentation",
|
|
resource_id=uuid.uuid4(),
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
session.add(entry)
|
|
await session.commit()
|
|
|
|
logs = await audit_service.query(session)
|
|
assert len(logs) == 1
|
|
assert logs[0].action == "create"
|
|
|
|
async def test_query_filter_by_action(self, session):
|
|
uid = uuid.uuid4()
|
|
for action in ["create", "update", "delete"]:
|
|
session.add(AuditLogModel(
|
|
user_id=uid,
|
|
action=action,
|
|
resource_type="presentation",
|
|
created_at=datetime.now(timezone.utc),
|
|
))
|
|
await session.commit()
|
|
|
|
logs = await audit_service.query(session, action="delete")
|
|
assert len(logs) == 1
|
|
assert logs[0].action == "delete"
|
|
|
|
async def test_query_filter_by_user_id(self, session):
|
|
uid1 = uuid.uuid4()
|
|
uid2 = uuid.uuid4()
|
|
session.add(AuditLogModel(
|
|
user_id=uid1, action="create", resource_type="deck",
|
|
created_at=datetime.now(timezone.utc),
|
|
))
|
|
session.add(AuditLogModel(
|
|
user_id=uid2, action="create", resource_type="deck",
|
|
created_at=datetime.now(timezone.utc),
|
|
))
|
|
await session.commit()
|
|
|
|
logs = await audit_service.query(session, user_id=uid1)
|
|
assert len(logs) == 1
|
|
|
|
async def test_query_pagination(self, session):
|
|
uid = uuid.uuid4()
|
|
for i in range(5):
|
|
session.add(AuditLogModel(
|
|
user_id=uid, action=f"action_{i}", resource_type="test",
|
|
created_at=datetime.now(timezone.utc),
|
|
))
|
|
await session.commit()
|
|
|
|
page1 = await audit_service.query(session, limit=2, offset=0)
|
|
page2 = await audit_service.query(session, limit=2, offset=2)
|
|
assert len(page1) == 2
|
|
assert len(page2) == 2
|
|
|
|
|
|
class TestAuditExport:
|
|
def test_export_csv(self):
|
|
entries = [
|
|
AuditLogModel(
|
|
id=uuid.uuid4(),
|
|
user_id=uuid.uuid4(),
|
|
action="export",
|
|
resource_type="presentation",
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
]
|
|
csv_str = audit_service.export_csv(entries)
|
|
assert "export" in csv_str
|
|
assert "presentation" in csv_str
|
|
# Header row
|
|
assert "id,user_id,action" in csv_str
|
|
|
|
def test_export_json(self):
|
|
entries = [
|
|
AuditLogModel(
|
|
id=uuid.uuid4(),
|
|
user_id=uuid.uuid4(),
|
|
action="login",
|
|
resource_type="session",
|
|
details={"method": "dev"},
|
|
created_at=datetime.now(timezone.utc),
|
|
)
|
|
]
|
|
json_str = audit_service.export_json(entries)
|
|
assert '"action": "login"' in json_str
|
|
assert '"method": "dev"' in json_str
|
|
|
|
def test_export_csv_empty(self):
|
|
csv_str = audit_service.export_csv([])
|
|
lines = csv_str.strip().split("\n")
|
|
assert len(lines) == 1 # Only header
|