diff --git a/backend/app/services/campaign_crud_service.py b/backend/app/services/campaign_crud_service.py index c1599bf..cc18313 100644 --- a/backend/app/services/campaign_crud_service.py +++ b/backend/app/services/campaign_crud_service.py @@ -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 + ] diff --git a/backend/tests/test_campaign_approval.py b/backend/tests/test_campaign_approval.py index 3d7abf2..39184ab 100644 --- a/backend/tests/test_campaign_approval.py +++ b/backend/tests/test_campaign_approval.py @@ -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()))