Files
decision/backend/app/tests/test_mandates.py
Yvv 3cb1754592 Sprint 4 : decisions et mandats -- workflow complet + vote integration
Backend: 7 nouveaux endpoints (advance, assign, revoke, create-vote-session),
services enrichis avec creation de sessions de vote, assignation de mandataire
et revocation. 35 nouveaux tests (104 total). Frontend: store mandates, page
cadrage decisions, detail mandats, composants DecisionWorkflow, DecisionCadrage,
DecisionCard, MandateTimeline, MandateCard. Documentation mise a jour.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-02-28 14:28:34 +01:00

656 lines
20 KiB
Python

"""Tests for mandate service: advance_mandate, assign_mandatee, revoke_mandate, create_vote_session_for_step.
These are pure unit tests that mock the database layer to test
the service logic in isolation.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock
import pytest
sqlalchemy = pytest.importorskip("sqlalchemy", reason="sqlalchemy required for mandate service tests")
from app.services.mandate_service import ( # noqa: E402
advance_mandate,
assign_mandatee,
create_vote_session_for_step,
revoke_mandate,
)
# ---------------------------------------------------------------------------
# Helpers: mock objects that behave like SQLAlchemy models
# ---------------------------------------------------------------------------
def _make_step(
step_id: uuid.UUID | None = None,
mandate_id: uuid.UUID | None = None,
step_order: int = 0,
step_type: str = "formulation",
status: str = "pending",
vote_session_id: uuid.UUID | None = None,
) -> MagicMock:
"""Create a mock MandateStep."""
step = MagicMock()
step.id = step_id or uuid.uuid4()
step.mandate_id = mandate_id or uuid.uuid4()
step.step_order = step_order
step.step_type = step_type
step.title = None
step.description = None
step.status = status
step.vote_session_id = vote_session_id
step.outcome = None
step.created_at = datetime.now(timezone.utc)
return step
def _make_mandate(
mandate_id: uuid.UUID | None = None,
title: str = "Mandat de test",
mandate_type: str = "techcomm",
status: str = "draft",
mandatee_id: uuid.UUID | None = None,
decision_id: uuid.UUID | None = None,
steps: list | None = None,
) -> MagicMock:
"""Create a mock Mandate."""
mandate = MagicMock()
mandate.id = mandate_id or uuid.uuid4()
mandate.title = title
mandate.description = None
mandate.mandate_type = mandate_type
mandate.status = status
mandate.mandatee_id = mandatee_id
mandate.decision_id = decision_id
mandate.starts_at = None
mandate.ends_at = None
mandate.created_at = datetime.now(timezone.utc)
mandate.updated_at = datetime.now(timezone.utc)
mandate.steps = steps or []
return mandate
def _make_identity(identity_id: uuid.UUID | None = None) -> MagicMock:
"""Create a mock DuniterIdentity."""
identity = MagicMock()
identity.id = identity_id or uuid.uuid4()
identity.address = "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"
identity.username = "Alice"
return identity
def _make_protocol(
protocol_id: uuid.UUID | None = None,
duration_days: int = 30,
) -> MagicMock:
"""Create a mock VotingProtocol with formula_config."""
protocol = MagicMock()
protocol.id = protocol_id or uuid.uuid4()
protocol.name = "Protocole standard"
protocol.vote_type = "binary"
protocol.formula_config = MagicMock()
protocol.formula_config.duration_days = duration_days
return protocol
def _make_decision_mock(
decision_id: uuid.UUID | None = None,
voting_protocol_id: uuid.UUID | None = None,
) -> MagicMock:
"""Create a mock Decision for the vote session lookup."""
decision = MagicMock()
decision.id = decision_id or uuid.uuid4()
decision.voting_protocol_id = voting_protocol_id
return decision
def _make_db_for_advance(mandate: MagicMock | None) -> AsyncMock:
"""Create a mock async database session for advance_mandate."""
db = AsyncMock()
result = MagicMock()
result.scalar_one_or_none.return_value = mandate
db.execute = AsyncMock(return_value=result)
db.commit = AsyncMock()
db.refresh = AsyncMock()
return db
def _make_db_for_assign(
mandate: MagicMock | None,
identity: MagicMock | None = None,
) -> AsyncMock:
"""Create a mock async database session for assign_mandatee.
Sequential execute calls:
1st -> mandate lookup
2nd -> identity lookup
"""
db = AsyncMock()
call_results = []
# Mandate result
mandate_result = MagicMock()
mandate_result.scalar_one_or_none.return_value = mandate
call_results.append(mandate_result)
# Identity result
if identity is not None or mandate is not None:
identity_result = MagicMock()
identity_result.scalar_one_or_none.return_value = identity
call_results.append(identity_result)
db.execute = AsyncMock(side_effect=call_results)
db.commit = AsyncMock()
db.refresh = AsyncMock()
return db
def _make_db_for_revoke(mandate: MagicMock | None) -> AsyncMock:
"""Create a mock async database session for revoke_mandate."""
db = AsyncMock()
result = MagicMock()
result.scalar_one_or_none.return_value = mandate
db.execute = AsyncMock(return_value=result)
db.commit = AsyncMock()
db.refresh = AsyncMock()
return db
def _make_db_for_vote_session(
mandate: MagicMock | None,
decision: MagicMock | None = None,
protocol: MagicMock | None = None,
) -> AsyncMock:
"""Create a mock async database session for create_vote_session_for_step.
Sequential execute calls:
1st -> mandate lookup
2nd -> decision lookup
3rd -> protocol lookup
"""
db = AsyncMock()
call_results = []
# Mandate result
mandate_result = MagicMock()
mandate_result.scalar_one_or_none.return_value = mandate
call_results.append(mandate_result)
# Decision result
if decision is not None:
decision_result = MagicMock()
decision_result.scalar_one_or_none.return_value = decision
call_results.append(decision_result)
# Protocol result
if protocol is not None:
proto_result = MagicMock()
proto_result.scalar_one_or_none.return_value = protocol
call_results.append(proto_result)
db.execute = AsyncMock(side_effect=call_results)
db.commit = AsyncMock()
db.refresh = AsyncMock()
db.flush = AsyncMock()
db.add = MagicMock()
return db
# ---------------------------------------------------------------------------
# Tests: advance_mandate
# ---------------------------------------------------------------------------
class TestAdvanceMandate:
"""Test mandate_service.advance_mandate."""
@pytest.mark.asyncio
async def test_advance_from_draft_activates_first_step(self):
"""Advancing a draft mandate with pending steps activates the first one
and moves mandate to candidacy."""
mandate_id = uuid.uuid4()
step1 = _make_step(mandate_id=mandate_id, step_order=0, status="pending")
step2 = _make_step(mandate_id=mandate_id, step_order=1, status="pending")
mandate = _make_mandate(
mandate_id=mandate_id,
status="draft",
steps=[step1, step2],
)
db = _make_db_for_advance(mandate)
result = await advance_mandate(mandate_id, db)
assert step1.status == "active"
assert step2.status == "pending"
assert mandate.status == "candidacy"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_advance_completes_active_and_activates_next(self):
"""Advancing completes the current active step and activates the next pending one."""
mandate_id = uuid.uuid4()
step1 = _make_step(mandate_id=mandate_id, step_order=0, status="active")
step2 = _make_step(mandate_id=mandate_id, step_order=1, status="pending")
mandate = _make_mandate(
mandate_id=mandate_id,
status="candidacy",
steps=[step1, step2],
)
db = _make_db_for_advance(mandate)
result = await advance_mandate(mandate_id, db)
assert step1.status == "completed"
assert step2.status == "active"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_advance_last_step_advances_status(self):
"""When the last active step is completed with no more pending steps,
the mandate status advances."""
mandate_id = uuid.uuid4()
step1 = _make_step(mandate_id=mandate_id, step_order=0, status="completed")
step2 = _make_step(mandate_id=mandate_id, step_order=1, status="active")
mandate = _make_mandate(
mandate_id=mandate_id,
status="candidacy",
steps=[step1, step2],
)
db = _make_db_for_advance(mandate)
result = await advance_mandate(mandate_id, db)
assert step2.status == "completed"
assert mandate.status == "voting"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_advance_full_progression(self):
"""A mandate with no steps can progress through all statuses."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="draft",
steps=[],
)
statuses_seen = [mandate.status]
# Advance through all states until completed
for _ in range(10): # safety limit
db = _make_db_for_advance(mandate)
await advance_mandate(mandate_id, db)
statuses_seen.append(mandate.status)
if mandate.status == "completed":
break
assert statuses_seen == [
"draft",
"candidacy",
"voting",
"active",
"reporting",
"completed",
]
@pytest.mark.asyncio
async def test_advance_completed_raises_error(self):
"""Advancing a completed mandate raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="completed",
steps=[],
)
db = _make_db_for_advance(mandate)
with pytest.raises(ValueError, match="statut terminal"):
await advance_mandate(mandate_id, db)
@pytest.mark.asyncio
async def test_advance_revoked_raises_error(self):
"""Advancing a revoked mandate raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="revoked",
steps=[],
)
db = _make_db_for_advance(mandate)
with pytest.raises(ValueError, match="statut terminal"):
await advance_mandate(mandate_id, db)
@pytest.mark.asyncio
async def test_advance_not_found_raises_error(self):
"""Advancing a non-existent mandate raises ValueError."""
db = _make_db_for_advance(None)
with pytest.raises(ValueError, match="Mandat introuvable"):
await advance_mandate(uuid.uuid4(), db)
# ---------------------------------------------------------------------------
# Tests: assign_mandatee
# ---------------------------------------------------------------------------
class TestAssignMandatee:
"""Test mandate_service.assign_mandatee."""
@pytest.mark.asyncio
async def test_assign_success(self):
"""Assigning a valid identity to an active mandate succeeds."""
mandate_id = uuid.uuid4()
mandatee_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="active",
)
identity = _make_identity(identity_id=mandatee_id)
db = _make_db_for_assign(mandate, identity)
result = await assign_mandatee(mandate_id, mandatee_id, db)
assert mandate.mandatee_id == mandatee_id
assert mandate.starts_at is not None
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_assign_draft_mandate(self):
"""Assigning to a draft mandate is allowed."""
mandate_id = uuid.uuid4()
mandatee_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="draft",
)
identity = _make_identity(identity_id=mandatee_id)
db = _make_db_for_assign(mandate, identity)
result = await assign_mandatee(mandate_id, mandatee_id, db)
assert mandate.mandatee_id == mandatee_id
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_assign_not_found_raises_error(self):
"""Assigning to a non-existent mandate raises ValueError."""
db = _make_db_for_assign(None)
with pytest.raises(ValueError, match="Mandat introuvable"):
await assign_mandatee(uuid.uuid4(), uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_assign_completed_raises_error(self):
"""Assigning to a completed mandate raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="completed",
)
db = _make_db_for_assign(mandate)
with pytest.raises(ValueError, match="statut terminal"):
await assign_mandatee(mandate_id, uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_assign_revoked_raises_error(self):
"""Assigning to a revoked mandate raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="revoked",
)
db = _make_db_for_assign(mandate)
with pytest.raises(ValueError, match="statut terminal"):
await assign_mandatee(mandate_id, uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_assign_identity_not_found_raises_error(self):
"""Assigning a non-existent identity raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="active",
)
db = _make_db_for_assign(mandate, identity=None)
with pytest.raises(ValueError, match="Identite Duniter introuvable"):
await assign_mandatee(mandate_id, uuid.uuid4(), db)
# ---------------------------------------------------------------------------
# Tests: revoke_mandate
# ---------------------------------------------------------------------------
class TestRevokeMandate:
"""Test mandate_service.revoke_mandate."""
@pytest.mark.asyncio
async def test_revoke_active_mandate(self):
"""Revoking an active mandate sets status to revoked and cancels steps."""
mandate_id = uuid.uuid4()
step1 = _make_step(mandate_id=mandate_id, step_order=0, status="completed")
step2 = _make_step(mandate_id=mandate_id, step_order=1, status="active")
step3 = _make_step(mandate_id=mandate_id, step_order=2, status="pending")
mandate = _make_mandate(
mandate_id=mandate_id,
status="active",
steps=[step1, step2, step3],
)
db = _make_db_for_revoke(mandate)
result = await revoke_mandate(mandate_id, db)
assert mandate.status == "revoked"
assert mandate.ends_at is not None
# Completed steps stay completed
assert step1.status == "completed"
# Active and pending steps are cancelled
assert step2.status == "cancelled"
assert step3.status == "cancelled"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_revoke_draft_mandate(self):
"""Revoking a draft mandate is allowed."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="draft",
steps=[],
)
db = _make_db_for_revoke(mandate)
result = await revoke_mandate(mandate_id, db)
assert mandate.status == "revoked"
assert mandate.ends_at is not None
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_revoke_not_found_raises_error(self):
"""Revoking a non-existent mandate raises ValueError."""
db = _make_db_for_revoke(None)
with pytest.raises(ValueError, match="Mandat introuvable"):
await revoke_mandate(uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_revoke_completed_raises_error(self):
"""Revoking an already completed mandate raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="completed",
)
db = _make_db_for_revoke(mandate)
with pytest.raises(ValueError, match="statut terminal"):
await revoke_mandate(mandate_id, db)
@pytest.mark.asyncio
async def test_revoke_already_revoked_raises_error(self):
"""Revoking an already revoked mandate raises ValueError."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(
mandate_id=mandate_id,
status="revoked",
)
db = _make_db_for_revoke(mandate)
with pytest.raises(ValueError, match="statut terminal"):
await revoke_mandate(mandate_id, db)
# ---------------------------------------------------------------------------
# Tests: create_vote_session_for_step
# ---------------------------------------------------------------------------
class TestCreateVoteSessionForStep:
"""Test mandate_service.create_vote_session_for_step."""
@pytest.mark.asyncio
async def test_create_vote_session_success(self):
"""A vote session is created and linked to the mandate step."""
mandate_id = uuid.uuid4()
step_id = uuid.uuid4()
decision_id = uuid.uuid4()
protocol_id = uuid.uuid4()
step = _make_step(
step_id=step_id,
mandate_id=mandate_id,
step_type="vote",
status="active",
vote_session_id=None,
)
mandate = _make_mandate(
mandate_id=mandate_id,
status="voting",
decision_id=decision_id,
steps=[step],
)
decision = _make_decision_mock(
decision_id=decision_id,
voting_protocol_id=protocol_id,
)
protocol = _make_protocol(protocol_id=protocol_id, duration_days=14)
db = _make_db_for_vote_session(mandate, decision, protocol)
result = await create_vote_session_for_step(mandate_id, step_id, db)
# The step should now have a vote_session_id set
assert step.vote_session_id is not None
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_create_vote_session_mandate_not_found(self):
"""ValueError when mandate does not exist."""
db = _make_db_for_vote_session(None)
with pytest.raises(ValueError, match="Mandat introuvable"):
await create_vote_session_for_step(uuid.uuid4(), uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_create_vote_session_step_not_found(self):
"""ValueError when step does not belong to the mandate."""
mandate_id = uuid.uuid4()
mandate = _make_mandate(mandate_id=mandate_id, decision_id=uuid.uuid4(), steps=[])
db = _make_db_for_vote_session(mandate)
with pytest.raises(ValueError, match="Etape introuvable"):
await create_vote_session_for_step(mandate_id, uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_create_vote_session_step_already_has_session(self):
"""ValueError when step already has a vote session."""
mandate_id = uuid.uuid4()
step_id = uuid.uuid4()
existing_session_id = uuid.uuid4()
step = _make_step(
step_id=step_id,
mandate_id=mandate_id,
vote_session_id=existing_session_id,
)
mandate = _make_mandate(
mandate_id=mandate_id,
decision_id=uuid.uuid4(),
steps=[step],
)
db = _make_db_for_vote_session(mandate)
with pytest.raises(ValueError, match="deja une session de vote"):
await create_vote_session_for_step(mandate_id, step_id, db)
@pytest.mark.asyncio
async def test_create_vote_session_no_decision(self):
"""ValueError when mandate has no linked decision."""
mandate_id = uuid.uuid4()
step_id = uuid.uuid4()
step = _make_step(
step_id=step_id,
mandate_id=mandate_id,
vote_session_id=None,
)
mandate = _make_mandate(
mandate_id=mandate_id,
decision_id=None,
steps=[step],
)
db = _make_db_for_vote_session(mandate)
with pytest.raises(ValueError, match="Aucune decision liee"):
await create_vote_session_for_step(mandate_id, step_id, db)