246 lines
13 KiB
Python
246 lines
13 KiB
Python
"""Real database/route checks: a review acceptance is durable before completion."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
from types import SimpleNamespace
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import Column, String, Table, create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
from govoplan_campaign.backend.db.models import Campaign, CampaignJob, CampaignVersion
|
|
from govoplan_campaign.backend.routes import versions as routes
|
|
from govoplan_campaign.backend.schemas import CampaignReviewStateRequest, CampaignVersionDetailResponse
|
|
from govoplan_campaign.backend.services.job_queries import _review_metadata, _review_metadata_counts
|
|
from govoplan_campaign.backend.services.review_decisions import review_decision_metadata
|
|
from govoplan_campaign.backend.sending.jobs import _reviewed_needs_review_keys
|
|
from govoplan_core.auth import ApiPrincipal
|
|
from govoplan_core.core.access import PrincipalRef
|
|
from govoplan_core.core.change_sequence import ChangeSequenceEntry
|
|
from govoplan_core.db.base import Base
|
|
|
|
|
|
@pytest.fixture
|
|
def review(tmp_path):
|
|
engine = create_engine(f"sqlite+pysqlite:///{tmp_path / 'review.db'}")
|
|
for name in ("access_users", "access_groups"):
|
|
if name not in Base.metadata.tables:
|
|
Table(name, Base.metadata, Column("id", String(36), primary_key=True))
|
|
Base.metadata.create_all(engine, tables=[
|
|
Base.metadata.tables["access_users"], Base.metadata.tables["access_groups"],
|
|
ChangeSequenceEntry.__table__, Campaign.__table__, CampaignVersion.__table__, CampaignJob.__table__,
|
|
])
|
|
with Session(engine) as session:
|
|
campaign = Campaign(id="campaign-1", tenant_id="tenant-1", external_id="C1", name="Campaign", current_version_id="version-1")
|
|
version = CampaignVersion(id="version-1", campaign_id=campaign.id, version_number=1,
|
|
raw_json={"version": "1.0", "campaign": {"id": "C1", "name": "Campaign"}},
|
|
build_summary={"build_token": "private-build-1", "built_count": 2}, editor_state={"created_from": "minimal_campaign"})
|
|
session.add_all([campaign, version, _job(1), _job(2)])
|
|
session.commit()
|
|
audits = []
|
|
def audit(current_session, _principal, **kwargs):
|
|
audits.append(kwargs)
|
|
if kwargs.get("commit"):
|
|
current_session.commit()
|
|
with patch.object(routes, "_get_campaign_for_principal", return_value=campaign), patch.object(routes, "audit_from_principal", side_effect=audit):
|
|
yield SimpleNamespace(engine=engine, session=session, version=version, campaign=campaign, audits=audits)
|
|
engine.dispose()
|
|
|
|
|
|
def _job(index, *, validation="needs_review", build="built", code="missing_optional_attachment", behavior="ask"):
|
|
return CampaignJob(id=f"job-{index}", tenant_id="tenant-1", campaign_id="campaign-1", campaign_version_id="version-1",
|
|
entry_index=index, entry_id=f"entry-{index}", validation_status=validation, build_status=build,
|
|
eml_sha256=str(index % 10) * 64, issues_snapshot=[{"code": code, "behavior": behavior, "source": "attachments", "details": {"rule_id": f"rule-{index}"}}])
|
|
|
|
|
|
def _principal(actor="reviewer-1"):
|
|
return ApiPrincipal(principal=PrincipalRef(account_id=actor, membership_id=actor, tenant_id="tenant-1", scopes=frozenset({"campaigns:campaign:review"})), account=SimpleNamespace(id=actor), user=SimpleNamespace(id=actor))
|
|
|
|
|
|
def _save(review, *, ids=(1,), complete=False, actor="reviewer-1", session=None, **overrides):
|
|
payload = {
|
|
"inspection_complete": complete, "merge_progress": True,
|
|
"build_token": review.version.review_build_token, "base_revision": review.version.edit_revision,
|
|
"reviewed_message_keys": [f"entry-{index}" for index in ids],
|
|
"issue_decisions": [{"job_id": f"job-{index}", "decision": "accept", "reason": f"Reason {index}"} for index in ids],
|
|
**overrides,
|
|
}
|
|
return routes.set_version_review_state("campaign-1", "version-1", CampaignReviewStateRequest(**payload), session=session or review.session, principal=_principal(actor))
|
|
|
|
|
|
def test_partial_reason_and_reviewed_state_survive_fresh_reload_without_completion(review):
|
|
response = _save(review)
|
|
review.session.expire_all()
|
|
stored = review.session.get(CampaignVersion, "version-1")
|
|
state = stored.editor_state["review_send"]
|
|
assert state["inspection_complete"] is False
|
|
assert state["reviewed_message_keys"] == ["entry-1"]
|
|
assert state["issue_decisions"][0]["reason"] == "Reason 1"
|
|
assert state["issue_decisions"][0]["actor_user_id"] == "reviewer-1"
|
|
assert state["issue_decisions"][0]["message_sha256"] == "1" * 64
|
|
assert response.edit_revision == 2
|
|
assert response.review_build_token and "private-build-1" not in repr(response)
|
|
public_state = response.editor_state["review_send"]
|
|
assert public_state["review_build_token"] == response.review_build_token
|
|
assert "build_token" not in public_state and "build_token" not in (response.build_summary or {})
|
|
metadata, reviewed_keys = _review_metadata(review.session, stored, [CampaignJob.campaign_version_id == stored.id])
|
|
assert metadata["reviewed_required_count"] == 1 and reviewed_keys == {"entry-1"}
|
|
assert _reviewed_needs_review_keys(stored) == set(), "partial progress never authorizes delivery"
|
|
assert review.audits[-1]["action"] == "campaign.message_review_updated"
|
|
|
|
|
|
def test_second_reviewer_and_final_completion_preserve_first_decision_evidence(review):
|
|
_save(review)
|
|
first = copy.deepcopy(review.version.editor_state["review_send"]["issue_decisions"][0])
|
|
_save(review, ids=(2,), actor="reviewer-2")
|
|
state = review.version.editor_state["review_send"]
|
|
assert state["reviewed_message_keys"] == ["entry-1", "entry-2"]
|
|
assert state["issue_decisions"][0] == first
|
|
second = copy.deepcopy(state["issue_decisions"][1])
|
|
assert review.audits[-1]["details"]["issue_decisions"]["count"] == 1
|
|
assert review.audits[-1]["details"]["issue_decisions"] == routes._review_decision_audit_evidence(review.version, job_ids={"job-2"})
|
|
_save(review, ids=(), complete=True, actor="reviewer-3")
|
|
review.session.expire_all()
|
|
state = review.version.editor_state["review_send"]
|
|
assert state["inspection_complete"] is True
|
|
assert state["issue_decisions"] == [first, second]
|
|
assert review.audits[-1]["details"]["issue_decisions"]["count"] == 2
|
|
assert _reviewed_needs_review_keys(review.version) == {"entry-1", "entry-2"}
|
|
|
|
|
|
def test_incremental_save_loads_only_selected_job_and_never_materializes_files(review):
|
|
review.session.add_all([_job(index, validation="ready", behavior="continue") for index in range(3, 503)])
|
|
review.session.commit()
|
|
review.session.refresh(review.version)
|
|
review.session.refresh(review.campaign)
|
|
review.session.expunge_all()
|
|
loaded_jobs = []
|
|
def loaded(_session, instance):
|
|
if isinstance(instance, CampaignJob):
|
|
loaded_jobs.append(instance.id)
|
|
event.listen(review.session, "loaded_as_persistent", loaded)
|
|
try:
|
|
with patch("govoplan_campaign.backend.persistence.campaigns.load_version_config", side_effect=AssertionError("Review progress must not rebuild or resolve Files")):
|
|
_save(review)
|
|
assert loaded_jobs == ["job-1"]
|
|
finally:
|
|
event.remove(review.session, "loaded_as_persistent", loaded)
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["unknown_job", "unknown_key", "missing_reason", "duplicate", "blocked", "hard_issue", "excluded", "category"])
|
|
def test_invalid_incremental_acceptance_is_atomic(review, mutation):
|
|
options = {}
|
|
if mutation == "unknown_job":
|
|
options["issue_decisions"] = [{"job_id": "outside-build", "reason": "Not allowed"}]
|
|
elif mutation == "unknown_key":
|
|
options["reviewed_message_keys"] = ["outside-build"]
|
|
elif mutation == "missing_reason":
|
|
options["issue_decisions"] = [{"job_id": "job-1", "reason": " "}]
|
|
elif mutation == "duplicate":
|
|
options["issue_decisions"] = [{"job_id": "job-1", "reason": "A"}, {"job_id": "job-1", "reason": "B"}]
|
|
elif mutation == "category":
|
|
options["decision_category_key"] = "wrong-category"
|
|
else:
|
|
job = review.session.get(CampaignJob, "job-1")
|
|
if mutation == "hard_issue":
|
|
job.issues_snapshot = [{"code": "required", "source": "attachments", "behavior": "block"}]
|
|
else:
|
|
job.validation_status = mutation
|
|
review.session.commit()
|
|
original = copy.deepcopy(review.version.editor_state)
|
|
with pytest.raises(HTTPException) as error:
|
|
_save(review, **options)
|
|
assert error.value.status_code == 422
|
|
review.session.expire_all()
|
|
assert review.version.editor_state == original and review.version.edit_revision == 1
|
|
assert review.audits == []
|
|
|
|
|
|
@pytest.mark.parametrize("precondition", ["build", "revision"])
|
|
def test_stale_progress_is_conflict_and_preserves_acknowledged_progress(review, precondition):
|
|
_save(review)
|
|
options = {"build_token": "old-build"} if precondition == "build" else {"base_revision": 1}
|
|
with pytest.raises(HTTPException) as error:
|
|
_save(review, ids=(2,), **options)
|
|
assert error.value.status_code == 409
|
|
assert review.version.editor_state["review_send"]["reviewed_message_keys"] == ["entry-1"]
|
|
|
|
|
|
def test_same_category_bulk_is_bound_to_exact_selected_jobs(review):
|
|
category = review_decision_metadata(review.session.get(CampaignJob, "job-1"))["category_key"]
|
|
result = _save(review, ids=(1, 2), decision_category_key=category)
|
|
assert result.editor_state["review_send"]["reviewed_message_keys"] == ["entry-1", "entry-2"]
|
|
assert len(result.editor_state["review_send"]["issue_decisions"]) == 2
|
|
|
|
|
|
def test_deliberately_excluded_and_inactive_rows_do_not_block_completion_or_need_bulk_acceptance(review):
|
|
review.session.add_all([_job(3, validation="excluded", build="skipped", behavior="drop"), _job(4, validation="inactive", build="skipped", behavior="continue")])
|
|
review.session.commit()
|
|
_save(review, ids=(1, 2), complete=True)
|
|
counts = _review_metadata_counts([(None, 3, "skipped", "excluded"), (None, 4, "skipped", "inactive")], set())
|
|
assert counts == {"blocking_count": 0, "required_count": 0, "reviewed_required_count": 0, "bulk_acceptable_count": 0}
|
|
assert review.version.editor_state["review_send"]["reviewed_message_keys"] == ["entry-1", "entry-2"]
|
|
assert not review_decision_metadata(review.session.get(CampaignJob, "job-3"))["eligible"]
|
|
|
|
|
|
def test_final_review_cannot_override_remaining_hard_blockers(review):
|
|
review.session.add(_job(3, validation="blocked", build="build_failed", behavior="block"))
|
|
review.session.commit()
|
|
_save(review, ids=(1, 2))
|
|
with pytest.raises(HTTPException, match="Blocked or failed"):
|
|
_save(review, ids=(), complete=True)
|
|
assert review.version.editor_state["review_send"]["inspection_complete"] is False
|
|
|
|
|
|
def test_existing_frozen_blocker_is_not_reclassified_from_current_allowed_empty_settings(review):
|
|
job = review.session.get(CampaignJob, "job-1")
|
|
job.validation_status = "blocked"
|
|
job.build_status = "build_failed"
|
|
job.issues_snapshot = [{"code": "missing_required_attachment", "source": "attachments", "behavior": "block"}]
|
|
review.version.raw_json = {**review.version.raw_json, "attachments": {"missing_behavior": "continue", "send_without_attachments_behavior": "continue"}}
|
|
review.session.commit()
|
|
evidence = copy.deepcopy(job.issues_snapshot)
|
|
with pytest.raises(HTTPException):
|
|
_save(review, ids=(1,))
|
|
assert job.issues_snapshot == evidence and job.validation_status == "blocked"
|
|
|
|
|
|
def test_concurrent_database_write_is_409_without_losing_other_reviewer(review):
|
|
# Keep a genuinely stale ORM identity in a second transaction so the SQL
|
|
# version-column check, rather than just the request comparison, must fire.
|
|
with Session(review.engine) as stale_session:
|
|
stale = stale_session.get(CampaignVersion, "version-1")
|
|
assert stale is not None and stale.edit_revision == 1
|
|
_save(review, ids=(1,))
|
|
with pytest.raises(HTTPException) as error:
|
|
_save(review, ids=(2,), session=stale_session, base_revision=1)
|
|
assert error.value.status_code == 409
|
|
review.session.expire_all()
|
|
assert review.version.editor_state["review_send"]["reviewed_message_keys"] == ["entry-1"]
|
|
assert len(review.audits) == 1
|
|
|
|
|
|
def test_delivery_final_lock_rejects_incremental_acceptance(review):
|
|
review.version.workflow_state = "completed"
|
|
review.session.commit()
|
|
with pytest.raises(HTTPException) as error:
|
|
_save(review)
|
|
assert error.value.status_code == 409
|
|
assert "review_send" not in review.version.editor_state
|
|
|
|
|
|
def test_owner_denial_rejects_before_persisting_review(review):
|
|
with patch.object(routes, "_get_campaign_for_principal", side_effect=HTTPException(status_code=403, detail="Owner access denied")):
|
|
with pytest.raises(HTTPException) as error:
|
|
_save(review)
|
|
assert error.value.status_code == 403
|
|
assert "review_send" not in review.version.editor_state and review.audits == []
|
|
|
|
|
|
def test_progress_contract_requires_both_preconditions():
|
|
with pytest.raises(ValidationError, match="build_token and base_revision"):
|
|
CampaignReviewStateRequest(merge_progress=True, build_token="build-1")
|