564 lines
20 KiB
Python
564 lines
20 KiB
Python
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()
|