diff --git a/src/govoplan_campaign/backend/persistence/campaigns.py b/src/govoplan_campaign/backend/persistence/campaigns.py index 92612ad..817a9fa 100644 --- a/src/govoplan_campaign/backend/persistence/campaigns.py +++ b/src/govoplan_campaign/backend/persistence/campaigns.py @@ -157,6 +157,7 @@ def create_campaign_version_from_json( raw_json: dict[str, Any], source_filename: str | None = None, source_base_path: str | None = None, + commit: bool = True, ) -> tuple[Campaign, CampaignVersion]: assert_server_safe_campaign_paths( raw_json, @@ -211,8 +212,11 @@ def create_campaign_version_from_json( session.flush() campaign.current_version_id = version.id session.add(campaign) - _write_campaign_snapshot(version) - session.commit() + if commit: + _write_campaign_snapshot(version) + session.commit() + else: + session.flush() return campaign, version diff --git a/src/govoplan_campaign/backend/persistence/versions.py b/src/govoplan_campaign/backend/persistence/versions.py index 3edaa55..e708c83 100644 --- a/src/govoplan_campaign/backend/persistence/versions.py +++ b/src/govoplan_campaign/backend/persistence/versions.py @@ -208,6 +208,7 @@ def create_minimal_campaign( description: str | None = None, current_flow: str = CampaignVersionFlow.CREATE.value, current_step: str = "basics", + commit: bool = True, ) -> tuple[Campaign, CampaignVersion]: existing = session.query(Campaign).filter(Campaign.tenant_id == tenant_id, Campaign.external_id == external_id).one_or_none() if existing: @@ -241,8 +242,11 @@ def create_minimal_campaign( session.flush() campaign.current_version_id = version.id session.add(campaign) - _write_campaign_snapshot(version) - session.commit() + if commit: + _write_campaign_snapshot(version) + session.commit() + else: + session.flush() return campaign, version @@ -338,6 +342,7 @@ def fork_campaign_version_for_edit( source_base_path: str | None = None, autosave: bool = True, migrate_legacy_mail_settings: bool = False, + commit: bool = True, ) -> CampaignVersion: """Create the next sole working version from immutable campaign history. @@ -407,8 +412,11 @@ def fork_campaign_version_for_edit( campaign.current_version_id = new_version.id campaign.status = CampaignStatus.DRAFT.value session.add(campaign) - _write_campaign_snapshot(new_version) - session.commit() + if commit: + _write_campaign_snapshot(new_version) + session.commit() + else: + session.flush() return new_version @@ -466,6 +474,7 @@ def unlock_validated_campaign_version( tenant_id: str, campaign_id: str, version_id: str, + commit: bool = True, ) -> CampaignVersion: """Unlock a validation snapshot so it can be edited again. @@ -515,7 +524,10 @@ def unlock_validated_campaign_version( campaign.status = CampaignStatus.DRAFT.value session.add(version) session.add(campaign) - session.commit() + if commit: + session.commit() + else: + session.flush() return version def update_campaign_version( @@ -534,6 +546,7 @@ def update_campaign_version( source_base_path: str | None = None, autosave: bool = False, migrate_legacy_mail_settings: bool = False, + commit: bool = True, ) -> CampaignVersion: if raw_json is not None or source_filename is not None or source_base_path is not None: assert_server_safe_campaign_paths( @@ -602,8 +615,11 @@ def update_campaign_version( session.add(version) session.add(campaign) session.flush() - _write_campaign_snapshot(version) - session.commit() + if commit: + _write_campaign_snapshot(version) + session.commit() + else: + session.flush() return version @@ -616,6 +632,7 @@ def update_campaign_review_state( inspection_complete: bool, reviewed_message_keys: list[str], user_id: str | None, + commit: bool = True, ) -> CampaignVersion: """Persist review acknowledgement without mutating the locked campaign data. @@ -646,7 +663,10 @@ def update_campaign_review_state( user_id=user_id, ) session.add(version) - session.commit() + if commit: + session.commit() + else: + session.flush() return version @@ -728,6 +748,7 @@ def lock_campaign_version_temporarily( campaign_id: str, version_id: str, user_id: str | None, + commit: bool = True, ) -> CampaignVersion: """Apply a reversible user-requested lock without changing workflow state.""" @@ -752,7 +773,10 @@ def lock_campaign_version_temporarily( version.user_locked_at = datetime.now(UTC) version.user_locked_by_user_id = user_id session.add(version) - session.commit() + if commit: + session.commit() + else: + session.flush() return version @@ -762,6 +786,7 @@ def unlock_user_locked_campaign_version( tenant_id: str, campaign_id: str, version_id: str, + commit: bool = True, ) -> CampaignVersion: """Remove a reversible user lock without invalidating campaign data.""" @@ -785,7 +810,10 @@ def unlock_user_locked_campaign_version( version.user_locked_at = None version.user_locked_by_user_id = None session.add(version) - session.commit() + if commit: + session.commit() + else: + session.flush() return version @@ -796,6 +824,7 @@ def permanently_lock_campaign_version( campaign_id: str, version_id: str, user_id: str | None, + commit: bool = True, ) -> CampaignVersion: """Apply an irreversible user lock. @@ -823,7 +852,10 @@ def permanently_lock_campaign_version( # Retain published_at as a compatibility marker for existing integrations. version.published_at = version.published_at or now session.add(version) - session.commit() + if commit: + session.commit() + else: + session.flush() return version @@ -834,6 +866,7 @@ def publish_campaign_version( campaign_id: str, version_id: str, user_id: str | None = None, + commit: bool = True, ) -> CampaignVersion: """Backwards-compatible alias for the permanent user lock.""" @@ -843,6 +876,7 @@ def publish_campaign_version( campaign_id=campaign_id, version_id=version_id, user_id=user_id, + commit=commit, ) diff --git a/src/govoplan_campaign/backend/router.py b/src/govoplan_campaign/backend/router.py index 8253427..a2df8d6 100644 --- a/src/govoplan_campaign/backend/router.py +++ b/src/govoplan_campaign/backend/router.py @@ -390,14 +390,22 @@ def _campaign_version_detail_response( details=audit_details, commit=True, ) - return CampaignVersionDetailResponse.model_validate(version) + _write_current_version_snapshot_if_available(version) + return CampaignVersionDetailResponse.model_validate( + version, + context=_campaign_response_context(principal), + ) except LockedCampaignVersionError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc except CampaignPathSecurityError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except CampaignPersistenceError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: + session.rollback() if validation_error_status is None: raise raise HTTPException(status_code=validation_error_status, detail=str(exc)) from exc @@ -437,6 +445,7 @@ def _update_campaign_version_detail_response( source_base_path=payload.source_base_path, autosave=autosave, migrate_legacy_mail_settings=payload.migrate_legacy_mail_settings, + commit=False, ), audit_action=audit_action, details=lambda version: { @@ -611,6 +620,7 @@ def create_campaign( raw_json=payload.config, source_filename=payload.source_filename, source_base_path=payload.source_base_path, + commit=False, ) audit_from_principal( session, @@ -621,11 +631,20 @@ def create_campaign( details={"version_id": version.id, "external_id": campaign.external_id}, commit=True, ) + _write_current_version_snapshot_if_available(version) except HTTPException: + session.rollback() raise except Exception as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc - return CampaignCreateResponse(campaign=CampaignResponse.model_validate(campaign), version=CampaignVersionResponse.model_validate(version)) + return CampaignCreateResponse( + campaign=CampaignResponse.model_validate(campaign), + version=CampaignVersionResponse.model_validate( + version, + context=_campaign_response_context(principal), + ), + ) @router.post("/new", response_model=CampaignCreateResponse) @@ -651,6 +670,7 @@ def create_minimal_campaign_endpoint( description=payload.description, current_flow=payload.current_flow, current_step=payload.current_step, + commit=False, ) audit_from_principal( session, @@ -661,9 +681,20 @@ def create_minimal_campaign_endpoint( details={"version_id": version.id, "external_id": campaign.external_id}, commit=True, ) - return CampaignCreateResponse(campaign=CampaignResponse.model_validate(campaign), version=CampaignVersionResponse.model_validate(version)) + _write_current_version_snapshot_if_available(version) + return CampaignCreateResponse( + campaign=CampaignResponse.model_validate(campaign), + version=CampaignVersionResponse.model_validate( + version, + context=_campaign_response_context(principal), + ), + ) except CampaignPersistenceError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc + except Exception: + session.rollback() + raise @@ -1045,14 +1076,30 @@ def _campaign_workspace_response( campaign_id=campaign_id, version_id=selected_version_id, include_jobs=False, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), ) except CampaignReportError: summary_payload = None return CampaignWorkspaceResponse( campaign=CampaignResponse.model_validate(campaign), - versions=[CampaignVersionResponse.model_validate(item) for item in versions] if include_versions else [], - current_version=CampaignVersionDetailResponse.model_validate(current_version) if current_version is not None else None, + versions=[ + CampaignVersionResponse.model_validate( + item, + context=_campaign_response_context(principal), + ) + for item in versions + ] + if include_versions + else [], + current_version=( + CampaignVersionDetailResponse.model_validate( + current_version, + context=_campaign_response_context(principal), + ) + if current_version is not None + else None + ), summary=summary_payload, selected_version_id=selected_version_id, ) @@ -1167,7 +1214,13 @@ def _campaign_workspace_delta_response( ) return CampaignWorkspaceDeltaResponse( campaign=CampaignResponse.model_validate(campaign) if delta.changed_campaign else None, - versions=_campaign_workspace_delta_versions(session, campaign, delta, include_versions=include_versions), + versions=_campaign_workspace_delta_versions( + session, + principal, + campaign, + delta, + include_versions=include_versions, + ), current_version=_campaign_workspace_delta_current_version( session, principal, @@ -1240,6 +1293,7 @@ def _campaign_workspace_delta_deleted(entries: list[object]) -> list[dict[str, o def _campaign_workspace_delta_versions( session: Session, + principal: ApiPrincipal, campaign: Campaign, delta: _WorkspaceDeltaState, *, @@ -1253,7 +1307,13 @@ def _campaign_workspace_delta_versions( .order_by(CampaignVersion.version_number.desc()) .all() ) - return [CampaignVersionResponse.model_validate(item) for item in versions] + return [ + CampaignVersionResponse.model_validate( + item, + context=_campaign_response_context(principal), + ) + for item in versions + ] def _campaign_workspace_delta_current_version( @@ -1274,7 +1334,14 @@ def _campaign_workspace_delta_current_version( .filter(CampaignVersion.id == delta.selected_version_id, CampaignVersion.campaign_id == campaign.id) .one_or_none() ) - return CampaignVersionDetailResponse.model_validate(current_version) if current_version is not None else None + return ( + CampaignVersionDetailResponse.model_validate( + current_version, + context=_campaign_response_context(principal), + ) + if current_version is not None + else None + ) def _campaign_workspace_delta_summary( @@ -1296,6 +1363,7 @@ def _campaign_workspace_delta_summary( campaign_id=campaign_id, version_id=delta.selected_version_id, include_jobs=False, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), ) except CampaignReportError: return None @@ -1378,17 +1446,25 @@ def update_campaign_metadata_endpoint( _sync_campaign_metadata_to_current_version(session, campaign) session.add(campaign) - session.commit() + session.flush() + try: + audit_from_principal( + session, + principal, + action="campaign.metadata_updated", + object_type="campaign", + object_id=campaign.id, + details={"external_id": campaign.external_id, "name": campaign.name}, + commit=True, + ) + except Exception: + session.rollback() + raise session.refresh(campaign) - audit_from_principal( - session, - principal, - action="campaign.metadata_updated", - object_type="campaign", - object_id=campaign.id, - details={"external_id": campaign.external_id, "name": campaign.name}, - commit=True, - ) + if campaign.current_version_id: + current_version = session.get(CampaignVersion, campaign.current_version_id) + if current_version is not None: + _write_current_version_snapshot_if_available(current_version) return CampaignResponse.model_validate(campaign) @@ -1458,7 +1534,13 @@ def list_versions( .order_by(CampaignVersion.version_number.desc()) .all() ) - return [CampaignVersionResponse.model_validate(item) for item in versions] + return [ + CampaignVersionResponse.model_validate( + item, + context=_campaign_response_context(principal), + ) + for item in versions + ] @router.get("/{campaign_id}/versions/{version_id}", response_model=CampaignVersionDetailResponse) @@ -1474,7 +1556,10 @@ def get_version_detail( version = get_campaign_version_for_tenant( session, tenant_id=principal.tenant_id, campaign_id=campaign_id, version_id=version_id ) - return CampaignVersionDetailResponse.model_validate(version) + return CampaignVersionDetailResponse.model_validate( + version, + context=_campaign_response_context(principal), + ) except CampaignPersistenceError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @@ -1523,6 +1608,7 @@ def fork_version_for_edit( source_base_path=payload.source_base_path, autosave=True, migrate_legacy_mail_settings=payload.migrate_legacy_mail_settings, + commit=False, ) campaign = _get_campaign_for_tenant(session, campaign_id, principal.tenant_id) audit_from_principal( @@ -1539,16 +1625,26 @@ def fork_version_for_edit( }, commit=True, ) + _write_current_version_snapshot_if_available(version) return CampaignCreateResponse( campaign=CampaignResponse.model_validate(campaign), - version=CampaignVersionResponse.model_validate(version), + version=CampaignVersionResponse.model_validate( + version, + context=_campaign_response_context(principal), + ), ) except LockedCampaignVersionError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc except CampaignPathSecurityError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc except CampaignPersistenceError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + except Exception: + session.rollback() + raise @router.post("/{campaign_id}/versions/{version_id}/unlock-validation", response_model=CampaignVersionDetailResponse) @@ -1574,6 +1670,7 @@ def unlock_version_validation( tenant_id=principal.tenant_id, campaign_id=campaign_id, version_id=version_id, + commit=False, ), audit_action="campaign.version_validation_unlocked", ) @@ -1597,6 +1694,7 @@ def lock_version_temporarily( campaign_id=campaign_id, version_id=version_id, user_id=principal.user.id, + commit=False, ), audit_action="campaign.version_user_locked_temporarily", ) @@ -1619,6 +1717,7 @@ def unlock_version_user_lock( tenant_id=principal.tenant_id, campaign_id=campaign_id, version_id=version_id, + commit=False, ), audit_action="campaign.version_user_lock_removed", ) @@ -1642,6 +1741,7 @@ def lock_version_permanently( campaign_id=campaign_id, version_id=version_id, user_id=principal.user.id, + commit=False, ), audit_action="campaign.version_user_locked_permanently", ) @@ -1694,8 +1794,11 @@ def set_version_step( principal: ApiPrincipal = Depends(require_scope("campaigns:campaign:update")), ): _get_campaign_for_principal(session, campaign_id, principal, write=True) - try: - version = update_campaign_version( + return _campaign_version_detail_response( + session, + principal, + campaign_id, + lambda: update_campaign_version( session, tenant_id=principal.tenant_id, campaign_id=campaign_id, @@ -1703,12 +1806,15 @@ def set_version_step( current_flow=payload.current_flow, current_step=payload.current_step, autosave=True, - ) - return CampaignVersionDetailResponse.model_validate(version) - except LockedCampaignVersionError as exc: - raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc - except CampaignPersistenceError as exc: - raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc + commit=False, + ), + audit_action="campaign.version_step_updated", + details={ + "campaign_id": campaign_id, + "current_flow": payload.current_flow, + "current_step": payload.current_step, + }, + ) @router.post("/{campaign_id}/versions/{version_id}/review-state", response_model=CampaignVersionDetailResponse) @@ -1729,6 +1835,7 @@ def set_version_review_state( inspection_complete=payload.inspection_complete, reviewed_message_keys=payload.reviewed_message_keys, user_id=principal.user.id, + commit=False, ) audit_from_principal( session, @@ -1743,11 +1850,19 @@ def set_version_review_state( }, commit=True, ) - return CampaignVersionDetailResponse.model_validate(version) + return CampaignVersionDetailResponse.model_validate( + version, + context=_campaign_response_context(principal), + ) except LockedCampaignVersionError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(exc)) from exc except CampaignPersistenceError as exc: + session.rollback() raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_CONTENT, detail=str(exc)) from exc + except Exception: + session.rollback() + raise @router.post("/{campaign_id}/versions/{version_id}/validate-partial", response_model=CampaignPartialValidationResponse) @@ -1797,6 +1912,7 @@ def publish_version( campaign_id=campaign_id, version_id=version_id, user_id=principal.user.id, + commit=False, ), audit_action="campaign.version_user_locked_permanently", ) @@ -1867,7 +1983,10 @@ def validate_version( }, commit=True, ) - return public_campaign_payload(result) + return public_campaign_payload( + result, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), + ) except HTTPException: raise except CampaignPersistenceError as exc: @@ -1902,7 +2021,10 @@ def build_version( details={"write_eml": payload.write_eml if payload else True, "built_count": result.get("built_count")}, commit=True, ) - return public_campaign_payload(result) + return public_campaign_payload( + result, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), + ) except CampaignPersistenceError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc except Exception as exc: @@ -1941,8 +2063,6 @@ def _job_summary_payload( imap_status=job.imap_status, ), "queued_at": job.queued_at, - "claimed_at": job.claimed_at, - "smtp_started_at": job.smtp_started_at, "outcome_unknown_at": job.outcome_unknown_at, "sent_at": job.sent_at, "created_at": job.created_at, @@ -2646,6 +2766,7 @@ def campaign_summary( version_id=version_id, include_jobs=include_jobs, include_recent_failures=include_jobs, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), ) except CampaignReportError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @@ -2672,6 +2793,7 @@ def campaign_report( version_id=version_id, include_jobs=include_jobs, include_recent_failures=include_jobs, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), ) except CampaignReportError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc @@ -2690,7 +2812,11 @@ def campaign_jobs_csv( try: csv_text = generate_jobs_csv( - session, tenant_id=principal.tenant_id, campaign_id=campaign_id, version_id=version_id + session, + tenant_id=principal.tenant_id, + campaign_id=campaign_id, + version_id=version_id, + include_diagnostics=has_scope(principal, "campaigns:diagnostic:read"), ) except CampaignReportError as exc: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail=str(exc)) from exc diff --git a/tests/test_campaign_audit_atomicity.py b/tests/test_campaign_audit_atomicity.py new file mode 100644 index 0000000..f8ae40f --- /dev/null +++ b/tests/test_campaign_audit_atomicity.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import MagicMock, patch + +import pytest +from fastapi import HTTPException + +from govoplan_campaign.backend import router +from govoplan_campaign.backend.schemas import CampaignUpdateRequest, CampaignVersionUpdateRequest + + +def _principal() -> SimpleNamespace: + return SimpleNamespace( + tenant_id="tenant-1", + user=SimpleNamespace(id="user-1"), + api_key=None, + ) + + +def test_version_update_rolls_back_when_its_audit_record_cannot_be_written() -> None: + session = MagicMock() + principal = _principal() + version = SimpleNamespace( + id="version-1", + raw_json={}, + current_flow="manual", + current_step="recipients", + ) + + def mutate(*_args, **kwargs): + assert kwargs["commit"] is False + session.flush() + return version + + with ( + patch.object(router, "_get_campaign_for_principal"), + patch.object(router, "_get_version_for_tenant", return_value=version), + patch.object(router, "update_campaign_version", side_effect=mutate), + patch.object(router, "audit_from_principal", side_effect=RuntimeError("audit unavailable")), + ): + with pytest.raises(HTTPException, match="audit unavailable") as captured: + router._update_campaign_version_detail_response( # noqa: SLF001 - transaction regression test + session, + principal, # type: ignore[arg-type] + "campaign-1", + "version-1", + CampaignVersionUpdateRequest(current_step="recipients"), + autosave=True, + audit_action="campaign.version_autosaved", + ) + + assert captured.value.status_code == 422 + session.commit.assert_not_called() + session.rollback.assert_called_once_with() + + +def test_version_fork_rolls_back_when_its_audit_record_cannot_be_written() -> None: + session = MagicMock() + principal = _principal() + campaign = SimpleNamespace(id="campaign-1") + source = SimpleNamespace(id="version-1", campaign_id="campaign-1", raw_json={}) + forked = SimpleNamespace(id="version-2", campaign_id="campaign-1", version_number=2) + + def mutate(*_args, **kwargs): + assert kwargs["commit"] is False + session.flush() + return forked + + with ( + patch.object(router, "_get_campaign_for_principal", return_value=campaign), + patch.object(router, "_require_permission"), + patch.object(router, "_get_version_for_tenant", return_value=source), + patch.object(router, "_get_campaign_for_tenant", return_value=campaign), + patch.object(router, "fork_campaign_version_for_edit", side_effect=mutate), + patch.object(router, "audit_from_principal", side_effect=RuntimeError("audit unavailable")), + ): + with pytest.raises(RuntimeError, match="audit unavailable"): + router.fork_version_for_edit( + "campaign-1", + "version-1", + CampaignVersionUpdateRequest(), + session=session, + principal=principal, # type: ignore[arg-type] + ) + + session.commit.assert_not_called() + session.rollback.assert_called_once_with() + + +def test_metadata_update_rolls_back_when_its_audit_record_cannot_be_written() -> None: + session = MagicMock() + principal = _principal() + campaign = SimpleNamespace( + id="campaign-1", + tenant_id="tenant-1", + external_id="C-1", + name="Old name", + description=None, + status="draft", + current_version_id=None, + ) + + with ( + patch.object(router, "_get_campaign_for_principal", return_value=campaign), + patch.object(router, "_sync_campaign_metadata_to_current_version"), + patch.object(router, "audit_from_principal", side_effect=RuntimeError("audit unavailable")), + ): + with pytest.raises(RuntimeError, match="audit unavailable"): + router.update_campaign_metadata_endpoint( + "campaign-1", + CampaignUpdateRequest(name="New name"), + session=session, + principal=principal, # type: ignore[arg-type] + ) + + session.flush.assert_called_once_with() + session.commit.assert_not_called() + session.rollback.assert_called_once_with()