feat(campaigns): add get_campaign_timeline service function

This commit is contained in:
kitos
2026-07-03 11:23:37 +02:00
parent fd47db7bea
commit 6749daa7cb
2 changed files with 58 additions and 0 deletions
@@ -26,6 +26,9 @@ from app.domain.errors import (
# Import Campaign, CampaignModificationRequest, CampaignTest from app.models.campaign
from app.models.campaign import Campaign, CampaignModificationRequest, CampaignTest
# Import AuditLog from app.models.audit
from app.models.audit import AuditLog
# Import Technique from app.models.technique
from app.models.technique import Technique
@@ -1133,3 +1136,31 @@ def list_modification_requests(
query = query.filter(CampaignModificationRequest.status == status)
requests = query.order_by(CampaignModificationRequest.created_at.desc()).all()
return [serialize_modification_request(db, r) for r in requests]
def get_campaign_timeline(db: Session, campaign_id: str) -> list[dict]:
"""Return the chronological audit-log history for a campaign.
Raises EntityNotFoundError if the campaign does not exist.
"""
campaign = db.query(Campaign).filter(Campaign.id == uuid.UUID(campaign_id)).first()
if not campaign:
raise EntityNotFoundError("Campaign", campaign_id)
logs = (
db.query(AuditLog)
.filter(AuditLog.entity_type == "campaign", AuditLog.entity_id == str(campaign_id))
.order_by(AuditLog.timestamp.asc())
.all()
)
return [
{
"id": str(log.id),
"action": log.action,
"user_id": str(log.user_id) if log.user_id else None,
"timestamp": log.timestamp.isoformat() if log.timestamp else None,
"details": log.details or {},
}
for log in logs
]
+27
View File
@@ -377,3 +377,30 @@ class TestModificationRequests:
pending = list_modification_requests(db, status="pending")
assert len(pending) == 1
assert pending[0]["action"] == "remove_test"
from app.services.audit_service import log_action
from app.services.campaign_crud_service import get_campaign_timeline
class TestCampaignTimeline:
def test_timeline_returns_entries_in_chronological_order(self, db, draft_campaign_with_test, admin_user):
log_action(
db, user_id=admin_user.id, action="create_campaign",
entity_type="campaign", entity_id=str(draft_campaign_with_test.id), details={},
)
log_action(
db, user_id=admin_user.id, action="submit_campaign_for_approval",
entity_type="campaign", entity_id=str(draft_campaign_with_test.id), details={},
)
db.commit()
timeline = get_campaign_timeline(db, str(draft_campaign_with_test.id))
assert len(timeline) == 2
assert timeline[0]["action"] == "create_campaign"
assert timeline[1]["action"] == "submit_campaign_for_approval"
def test_timeline_missing_campaign_raises(self, db):
import uuid as uuid_mod
with pytest.raises(EntityNotFoundError):
get_campaign_timeline(db, str(uuid_mod.uuid4()))