from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from types import SimpleNamespace from fastapi import HTTPException from pydantic import ValidationError from sqlalchemy import create_engine from sqlalchemy.orm import Session, sessionmaker from govoplan_core.auth import ApiPrincipal from govoplan_core.core.access import PrincipalRef from govoplan_core.db.base import Base from govoplan_poll.backend.db.models import ( Poll, PollInvitation, PollLifecycleTransition, PollOption, PollResponse, ) from govoplan_poll.backend.router import api_poll_lifecycle, api_transition_poll from govoplan_poll.backend.schemas import ( PollAnswerInput, PollCreateRequest, PollOptionInput, PollSubmitResponseRequest, PollTransitionRequest, ) from govoplan_poll.backend.service import ( PollError, create_poll, list_poll_lifecycle_transitions, poll_response, poll_results_are_visible, response_datetime, submit_poll_response, transition_poll, ) from govoplan_poll.backend.transitions import ( DEFAULT_POLL_TRANSITION_POLICY, PollTransitionEngine, PollTransitionPolicy, PollTransitionRule, ) WRITE_SCOPE = "poll:poll:write" class PollLifecycleTests(unittest.TestCase): def setUp(self) -> None: self.engine = create_engine("sqlite:///:memory:") Base.metadata.create_all( self.engine, tables=[ Poll.__table__, PollOption.__table__, PollResponse.__table__, PollInvitation.__table__, PollLifecycleTransition.__table__, ], ) self.Session = sessionmaker(bind=self.engine) self.session: Session = self.Session() def tearDown(self) -> None: self.session.close() Base.metadata.drop_all( self.engine, tables=[ PollLifecycleTransition.__table__, PollInvitation.__table__, PollResponse.__table__, PollOption.__table__, Poll.__table__, ], ) self.engine.dispose() def _poll(self, *, status: str = "draft", title: str = "Decision") -> Poll: return create_poll( self.session, tenant_id="tenant-1", user_id="owner", payload=PollCreateRequest( title=title, kind="single_choice", status=status, visibility="tenant", options=[ PollOptionInput(key="yes", label="Yes"), PollOptionInput(key="no", label="No"), ], ), ) @staticmethod def _principal() -> ApiPrincipal: return ApiPrincipal( principal=PrincipalRef( account_id="owner", membership_id="owner-membership", tenant_id="tenant-1", scopes=frozenset({WRITE_SCOPE}), display_name="Owner", ), account=SimpleNamespace(id="owner"), user=SimpleNamespace(id="owner-membership"), ) def test_default_policy_exposes_every_allowed_and_rejected_transition(self) -> None: expected = { "draft": {"open", "archive"}, "open": {"draft", "close", "archive"}, "closed": {"open", "decide", "archive"}, "decided": {"open", "decide", "archive"}, "archived": {"unarchive"}, } engine = PollTransitionEngine() for status, allowed in expected.items(): archived_from_status = "closed" if status == "archived" else None availability = engine.available_actions( current_status=status, archived_from_status=archived_from_status, ) self.assertEqual({item.action for item in availability if item.available}, allowed) for item in availability: if item.available: self.assertIsNone(item.reason) else: self.assertTrue(item.reason) with self.assertRaisesRegex(ValueError, "not allowed"): engine.plan(current_status="draft", action="close") def test_injected_policy_governs_service_transition_and_projection(self) -> None: rules = dict(DEFAULT_POLL_TRANSITION_POLICY.rules) rules["close"] = PollTransitionRule( action="close", source_statuses=frozenset({"draft", "open"}), target_status="closed", ) custom_engine = PollTransitionEngine(PollTransitionPolicy(rules)) poll = self._poll() projected = poll_response(poll, transition_engine=custom_engine) close_action = next(item for item in projected["lifecycle_actions"] if item["action"] == "close") result = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="close", transition_engine=custom_engine, ) self.assertTrue(close_action["available"]) self.assertEqual(result.poll.status, "closed") self.assertEqual(result.transition.from_status, "draft") def test_only_draft_or_open_are_valid_initial_states(self) -> None: for initial_status in ("closed", "decided", "archived"): with self.subTest(initial_status=initial_status): with self.assertRaises(ValidationError): PollCreateRequest(title="Invalid initial state", kind="yes_no", status=initial_status) bypassed_schema = PollCreateRequest(title="Invalid initial state", kind="yes_no").model_copy( update={"status": "closed"} ) with self.assertRaisesRegex(PollError, "initial status must be draft or open"): create_poll( self.session, tenant_id="tenant-1", user_id="owner", payload=bypassed_schema, ) def test_reopening_preserves_responses_close_history_and_previous_decision(self) -> None: poll = self._poll() transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") response = submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( respondent_id="participant", answers=[PollAnswerInput(option_key="yes")], ), ) transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close") first_closed_at = poll.closed_at transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="yes", ) decided_at = poll.decided_at decided_option_id = poll.decided_option_id transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") self.assertEqual(poll.status, "open") self.assertEqual(response_datetime(poll.closed_at), response_datetime(first_closed_at)) self.assertEqual(response_datetime(poll.decided_at), response_datetime(decided_at)) self.assertEqual(poll.decided_option_id, decided_option_id) self.assertEqual(response.deleted_at, None) self.assertEqual(response.answers[0]["option_key"], "yes") history = list_poll_lifecycle_transitions(self.session, tenant_id="tenant-1", poll_id=poll.id) self.assertEqual([item.action for item in history], ["open", "close", "decide", "open"]) def test_direct_redecision_is_audited_and_exact_retry_is_idempotent(self) -> None: poll = self._poll() transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close") first = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="yes", idempotency_key="decision-1", ) replay = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="yes", idempotency_key="decision-1", ) second = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="no", idempotency_key="decision-2", ) self.assertFalse(first.replayed) self.assertTrue(replay.replayed) self.assertEqual(replay.transition.id, first.transition.id) self.assertFalse(second.replayed) self.assertEqual(second.transition.previous_decision_option_id, first.transition.decision_option_id) self.assertEqual(poll.decided_option_id, second.transition.decision_option_id) self.assertEqual( self.session.query(PollLifecycleTransition) .filter( PollLifecycleTransition.poll_id == poll.id, PollLifecycleTransition.action == "decide", ) .count(), 2, ) with self.assertRaisesRegex(PollError, "different poll transition"): transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="yes", idempotency_key="decision-2", ) def test_repeated_exact_transition_without_an_identity_is_an_audit_free_noop(self) -> None: poll = self._poll() applied = transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") with self.assertLogs("govoplan.poll.lifecycle", level="INFO") as logs: repeated = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="open", actor_user_id="owner", ) self.assertFalse(applied.replayed) self.assertTrue(repeated.replayed) self.assertIsNone(repeated.transition) self.assertEqual(poll.status, "open") self.assertEqual(len(poll.lifecycle_transitions), 1) self.assertIn("Ignored repeated exact Poll lifecycle transition", logs.output[0]) def test_same_decision_is_a_noop_but_a_changed_decision_is_audited(self) -> None: poll = self._poll() transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close") first = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="yes", ) repeated = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="yes", ) changed = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="decide", option_key="no", ) self.assertTrue(repeated.replayed) self.assertIsNone(repeated.transition) self.assertIsNotNone(first.transition) self.assertIsNotNone(changed.transition) self.assertEqual(changed.transition.previous_decision_option_id, first.transition.decision_option_id) self.assertEqual( [item.action for item in poll.lifecycle_transitions], ["open", "close", "decide", "decide"], ) def test_exact_unarchive_retry_is_a_noop_but_unarchive_without_history_is_invalid(self) -> None: poll = self._poll() transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="archive") transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="unarchive") repeated = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="unarchive", ) self.assertTrue(repeated.replayed) self.assertIsNone(repeated.transition) self.assertEqual([item.action for item in poll.lifecycle_transitions], ["archive", "unarchive"]) never_archived = self._poll(title="Never archived") with self.assertRaisesRegex(PollError, "not allowed"): transition_poll( self.session, tenant_id="tenant-1", poll_id=never_archived.id, action="unarchive", ) def test_archive_and_unarchive_restore_status_and_append_audit_history(self) -> None: poll = self._poll() transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close") response = PollResponse( tenant_id="tenant-1", poll_id=poll.id, respondent_id="participant", answers=[{"option_key": "yes"}], submitted_at=poll.created_at, ) self.session.add(response) self.session.flush() archived = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="archive", idempotency_key="archive-1", ) restored = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="unarchive", idempotency_key="unarchive-1", ) self.assertEqual(archived.transition.to_status, "archived") self.assertEqual(restored.transition.to_status, "closed") self.assertEqual(poll.status, "closed") self.assertIsNone(poll.archived_from_status) self.assertIsNone(response.deleted_at) self.assertEqual( [item.action for item in poll.lifecycle_transitions], ["open", "close", "archive", "unarchive"], ) def test_legacy_archive_without_origin_restores_to_draft_with_audit_marker(self) -> None: poll = Poll( tenant_id="tenant-1", slug="legacy-archive", title="Legacy archive", kind="yes_no", status="archived", visibility="tenant", result_visibility="after_close", workflow_steps=[], allow_anonymous=False, allow_response_update=True, min_choices=1, ) self.session.add(poll) self.session.flush() restored = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="unarchive", ) self.assertEqual(restored.poll.status, "draft") self.assertEqual(restored.transition.to_status, "draft") self.assertEqual( restored.transition.metadata_["legacy_restore_fallback"], { "reason": "missing_archived_from_status", "restored_status": "draft", }, ) def test_reopening_clears_only_an_expired_deadline_and_preserves_close_history(self) -> None: poll = self._poll() transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="open") transition_poll(self.session, tenant_id="tenant-1", poll_id=poll.id, action="close") closed_at = poll.closed_at expired_deadline = datetime.now(timezone.utc) - timedelta(days=1) poll.closes_at = expired_deadline reopened = transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="open", ) self.assertEqual(reopened.poll.status, "open") self.assertIsNone(reopened.poll.closes_at) self.assertEqual(response_datetime(reopened.poll.closed_at), response_datetime(closed_at)) self.assertEqual( reopened.transition.metadata_["cleared_expired_closes_at"], expired_deadline.isoformat(), ) submit_poll_response( self.session, tenant_id="tenant-1", poll_id=poll.id, payload=PollSubmitResponseRequest( respondent_id="participant-after-reopen", answers=[PollAnswerInput(option_key="yes")], ), ) future_deadline = datetime.now(timezone.utc) + timedelta(days=1) future_poll = self._poll(title="Future deadline") future_poll.closes_at = future_deadline opened = transition_poll( self.session, tenant_id="tenant-1", poll_id=future_poll.id, action="open", ) self.assertEqual(response_datetime(opened.poll.closes_at), future_deadline) self.assertNotIn("cleared_expired_closes_at", opened.transition.metadata_) def test_archiving_draft_or_open_poll_does_not_grant_after_close_visibility(self) -> None: for initial_status in ("draft", "open"): with self.subTest(initial_status=initial_status): poll = self._poll(status=initial_status, title=f"Archive from {initial_status}") transition_poll( self.session, tenant_id="tenant-1", poll_id=poll.id, action="archive", ) self.assertFalse( poll_results_are_visible( self.session, poll=poll, actor_ids=("participant",), ) ) closed_poll = self._poll(title="Archive from closed") transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="open") transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="close") transition_poll(self.session, tenant_id="tenant-1", poll_id=closed_poll.id, action="archive") self.assertTrue( poll_results_are_visible( self.session, poll=closed_poll, actor_ids=("participant",), ) ) def test_generic_api_returns_action_availability_history_and_actor(self) -> None: poll = self._poll() principal = self._principal() opened = api_transition_poll( poll.id, PollTransitionRequest(action="open"), idempotency_key="open-api-1", session=self.session, principal=principal, ) replay = api_transition_poll( poll.id, PollTransitionRequest(action="open"), idempotency_key="open-api-1", session=self.session, principal=principal, ) lifecycle = api_poll_lifecycle(poll.id, session=self.session, principal=principal) self.assertEqual(opened.poll.status, "open") self.assertFalse(opened.replayed) self.assertTrue(replay.replayed) self.assertIsNotNone(replay.transition) self.assertEqual(len(lifecycle.history), 1) self.assertEqual(lifecycle.history[0].actor_user_id, "owner-membership") self.assertEqual( {item.action for item in lifecycle.actions if item.available}, {"draft", "close", "archive"}, ) draft = self._poll(title="Draft cannot close") with self.assertRaises(HTTPException) as rejected: api_transition_poll( draft.id, PollTransitionRequest(action="close"), idempotency_key=None, session=self.session, principal=principal, ) self.assertEqual(rejected.exception.status_code, 400) def test_generic_api_returns_null_transition_for_domain_noop(self) -> None: poll = self._poll(status="open", title="Already open") repeated = api_transition_poll( poll.id, PollTransitionRequest(action="open"), idempotency_key=None, session=self.session, principal=self._principal(), ) self.assertEqual(repeated.poll.status, "open") self.assertTrue(repeated.replayed) self.assertIsNone(repeated.transition) self.assertEqual( self.session.query(PollLifecycleTransition) .filter(PollLifecycleTransition.poll_id == poll.id) .count(), 0, ) if __name__ == "__main__": unittest.main()