forked from yvv/decision
Mandats : origin→FK identité + nomination auto + boutons + tests intégration
ci/woodpecker/push/woodpecker Pipeline was successful
ci/woodpecker/push/woodpecker Pipeline was successful
- origin TEXT → origin_id UUID FK duniter_identities (migration e3f4a5b6c7d8) - GET /auth/identities?q= : recherche d'identités par nom/adresse - MandateCreate.nomination_mode : auto (auto-assign auteur), collective, postpone - Wizard new.vue : champ origine = picker identité, checkbox "Démarrer maintenant" - [id].vue : modal "Assigner" = search-picker (résout UUID vs adresse SS58), affiche origin_display_name + mandatee_display_name, inputs natifs (<input>/<textarea>) - Erreurs API visibles dans l'UI (plus de catch silencieux) - test_mandate_flows.py : 17 tests intégration SQLite réels (origin, nomination, assign, lifecycle, revocation, interactions croisées) — 241 tests total OK Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Yvv
@@ -0,0 +1,405 @@
|
||||
"""Integration tests for mandate flows: nomination, lifecycle, assignment, revocation.
|
||||
|
||||
Uses a real in-memory SQLite database — no mocks of the DB layer.
|
||||
Tests the service functions directly to verify interconnected business logic.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
from sqlalchemy.orm import selectinload
|
||||
from sqlalchemy import select
|
||||
|
||||
import app.models # noqa: F401 — registers all models with Base.metadata
|
||||
from app.database import Base
|
||||
from app.models.mandate import Mandate, MandateStep
|
||||
from app.models.user import DuniterIdentity
|
||||
from app.services.mandate_service import (
|
||||
advance_mandate,
|
||||
assign_mandatee,
|
||||
revoke_mandate,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def engine():
|
||||
eng = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
||||
async with eng.begin() as conn:
|
||||
await conn.run_sync(Base.metadata.create_all)
|
||||
yield eng
|
||||
await eng.dispose()
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db(engine):
|
||||
factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
|
||||
async with factory() as session:
|
||||
yield session
|
||||
|
||||
|
||||
async def _mk_identity(db: AsyncSession, display_name: str = "Alice") -> DuniterIdentity:
|
||||
ident = DuniterIdentity(
|
||||
id=uuid.uuid4(),
|
||||
address=f"5{uuid.uuid4().hex[:46]}",
|
||||
display_name=display_name,
|
||||
wot_status="member",
|
||||
is_smith=False,
|
||||
is_techcomm=False,
|
||||
)
|
||||
db.add(ident)
|
||||
await db.commit()
|
||||
await db.refresh(ident)
|
||||
return ident
|
||||
|
||||
|
||||
async def _mk_mandate(
|
||||
db: AsyncSession,
|
||||
*,
|
||||
mandatee_id: uuid.UUID | None = None,
|
||||
origin_id: uuid.UUID | None = None,
|
||||
status: str = "draft",
|
||||
steps: list[dict] | None = None,
|
||||
) -> Mandate:
|
||||
mandate = Mandate(
|
||||
id=uuid.uuid4(),
|
||||
title="Mandat test",
|
||||
mandate_type="functional",
|
||||
status=status,
|
||||
mandatee_id=mandatee_id,
|
||||
origin_id=origin_id,
|
||||
)
|
||||
db.add(mandate)
|
||||
await db.flush()
|
||||
|
||||
for i, s in enumerate(steps or []):
|
||||
step = MandateStep(
|
||||
id=uuid.uuid4(),
|
||||
mandate_id=mandate.id,
|
||||
step_order=i,
|
||||
step_type=s.get("step_type", "formulation"),
|
||||
title=s.get("title"),
|
||||
status=s.get("status", "pending"),
|
||||
)
|
||||
db.add(step)
|
||||
|
||||
await db.commit()
|
||||
await db.refresh(mandate)
|
||||
return mandate
|
||||
|
||||
|
||||
async def _reload(db: AsyncSession, mandate_id: uuid.UUID) -> Mandate:
|
||||
result = await db.execute(
|
||||
select(Mandate)
|
||||
.options(
|
||||
selectinload(Mandate.steps),
|
||||
selectinload(Mandate.origin_identity),
|
||||
selectinload(Mandate.mandatee),
|
||||
)
|
||||
.where(Mandate.id == mandate_id)
|
||||
)
|
||||
return result.scalar_one()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestMandateOrigin
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMandateOrigin:
|
||||
"""origin_id must link to a real DuniterIdentity and expose display_name."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_origin_id_linked(self, db: AsyncSession):
|
||||
author = await _mk_identity(db, "Baptiste")
|
||||
mandate = await _mk_mandate(db, origin_id=author.id)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
|
||||
assert loaded.origin_id == author.id
|
||||
assert loaded.origin_display_name == "Baptiste"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_origin_id_optional(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
|
||||
assert loaded.origin_id is None
|
||||
assert loaded.origin_display_name is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestAutoNomination
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestAutoNomination:
|
||||
"""Auto-désignation: mandatee = author, no candidacy steps needed."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_assign_author(self, db: AsyncSession):
|
||||
author = await _mk_identity(db, "Constance")
|
||||
mandate = await _mk_mandate(db, mandatee_id=author.id)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
|
||||
assert loaded.mandatee_id == author.id
|
||||
assert loaded.mandatee_display_name == "Constance"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_auto_assign_then_advance_to_active(self, db: AsyncSession):
|
||||
author = await _mk_identity(db, "David")
|
||||
mandate = await _mk_mandate(
|
||||
db,
|
||||
mandatee_id=author.id,
|
||||
steps=[
|
||||
{"step_type": "formulation", "status": "pending"},
|
||||
{"step_type": "assignment", "status": "pending"},
|
||||
{"step_type": "reporting", "status": "pending"},
|
||||
],
|
||||
)
|
||||
|
||||
# First advance: draft → candidacy, activates step 0
|
||||
result = await advance_mandate(mandate.id, db)
|
||||
assert result.status == "candidacy"
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.steps[0].status == "active"
|
||||
assert loaded.steps[1].status == "pending"
|
||||
|
||||
# Second advance: step 0 → completed, step 1 → active
|
||||
await advance_mandate(mandate.id, db)
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.steps[0].status == "completed"
|
||||
assert loaded.steps[1].status == "active"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestMandateAssign
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMandateAssign:
|
||||
"""assign_mandatee service: proper UUID lookup, display_name populated."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_assign_sets_mandatee_and_starts_at(self, db: AsyncSession):
|
||||
identity = await _mk_identity(db, "Elodie")
|
||||
mandate = await _mk_mandate(db, status="active")
|
||||
|
||||
await assign_mandatee(mandate.id, identity.id, db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.mandatee_id == identity.id
|
||||
assert loaded.starts_at is not None
|
||||
assert loaded.mandatee_display_name == "Elodie"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_assign_unknown_identity_raises(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(db, status="active")
|
||||
|
||||
with pytest.raises(ValueError, match="Identite Duniter introuvable"):
|
||||
await assign_mandatee(mandate.id, uuid.uuid4(), db)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_assign_completed_mandate_raises(self, db: AsyncSession):
|
||||
identity = await _mk_identity(db, "Fabien")
|
||||
mandate = await _mk_mandate(db, status="completed")
|
||||
|
||||
with pytest.raises(ValueError, match="statut terminal"):
|
||||
await assign_mandatee(mandate.id, identity.id, db)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_reassign_replaces_mandatee(self, db: AsyncSession):
|
||||
first = await _mk_identity(db, "Gilles")
|
||||
second = await _mk_identity(db, "Hélène")
|
||||
mandate = await _mk_mandate(db, mandatee_id=first.id, status="active")
|
||||
|
||||
await assign_mandatee(mandate.id, second.id, db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.mandatee_id == second.id
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestMandateLifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMandateLifecycle:
|
||||
"""advance_mandate: full lifecycle with and without steps."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_full_lifecycle_no_steps(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(db)
|
||||
statuses = [mandate.status]
|
||||
|
||||
for _ in range(10):
|
||||
m = await advance_mandate(mandate.id, db)
|
||||
statuses.append(m.status)
|
||||
if m.status == "completed":
|
||||
break
|
||||
|
||||
assert statuses == ["draft", "candidacy", "voting", "active", "reporting", "completed"]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_steps_activate_in_order(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(
|
||||
db,
|
||||
steps=[
|
||||
{"step_type": "formulation", "status": "pending"},
|
||||
{"step_type": "candidacy", "status": "pending"},
|
||||
{"step_type": "vote", "status": "pending"},
|
||||
],
|
||||
)
|
||||
|
||||
# Advance 1: activates step 0, moves to candidacy
|
||||
await advance_mandate(mandate.id, db)
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.status == "candidacy"
|
||||
assert loaded.steps[0].status == "active"
|
||||
assert loaded.steps[1].status == "pending"
|
||||
|
||||
# Advance 2: step 0 → completed, step 1 → active
|
||||
await advance_mandate(mandate.id, db)
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.steps[0].status == "completed"
|
||||
assert loaded.steps[1].status == "active"
|
||||
|
||||
# Advance 3: step 1 → completed, step 2 → active
|
||||
await advance_mandate(mandate.id, db)
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.steps[1].status == "completed"
|
||||
assert loaded.steps[2].status == "active"
|
||||
|
||||
# Advance 4: step 2 → completed, no more pending → advance mandate status
|
||||
await advance_mandate(mandate.id, db)
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.steps[2].status == "completed"
|
||||
assert loaded.status == "voting"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_advance_terminal_raises(self, db: AsyncSession):
|
||||
for terminal in ("completed", "revoked"):
|
||||
mandate = await _mk_mandate(db, status=terminal)
|
||||
with pytest.raises(ValueError, match="statut terminal"):
|
||||
await advance_mandate(mandate.id, db)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestMandateRevocation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestMandateRevocation:
|
||||
"""revoke_mandate: active/pending steps cancelled, completed steps preserved."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_cancels_active_and_pending(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(
|
||||
db,
|
||||
status="active",
|
||||
steps=[
|
||||
{"step_type": "formulation", "status": "completed"},
|
||||
{"step_type": "assignment", "status": "active"},
|
||||
{"step_type": "reporting", "status": "pending"},
|
||||
],
|
||||
)
|
||||
|
||||
await revoke_mandate(mandate.id, db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.status == "revoked"
|
||||
assert loaded.ends_at is not None
|
||||
assert loaded.steps[0].status == "completed"
|
||||
assert loaded.steps[1].status == "cancelled"
|
||||
assert loaded.steps[2].status == "cancelled"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_sets_ends_at(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(db, status="draft")
|
||||
|
||||
await revoke_mandate(mandate.id, db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.ends_at is not None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_already_revoked_raises(self, db: AsyncSession):
|
||||
mandate = await _mk_mandate(db, status="revoked")
|
||||
|
||||
with pytest.raises(ValueError, match="statut terminal"):
|
||||
await revoke_mandate(mandate.id, db)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TestNominationInteractions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestNominationInteractions:
|
||||
"""Cross-process: nomination + assignment + lifecycle are consistent."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_assign_then_advance_full_cycle(self, db: AsyncSession):
|
||||
"""Assigning a mandatee then running the full lifecycle completes cleanly."""
|
||||
mandatee = await _mk_identity(db, "Isabelle")
|
||||
mandate = await _mk_mandate(
|
||||
db,
|
||||
steps=[
|
||||
{"step_type": "formulation", "status": "pending"},
|
||||
{"step_type": "assignment", "status": "pending"},
|
||||
{"step_type": "completion", "status": "pending"},
|
||||
],
|
||||
)
|
||||
|
||||
# Assign before starting
|
||||
await assign_mandatee(mandate.id, mandatee.id, db)
|
||||
|
||||
# Run through all steps
|
||||
for _ in range(5):
|
||||
m = await advance_mandate(mandate.id, db)
|
||||
if m.status in ("completed", "revoked"):
|
||||
break
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
# All steps completed, mandate advanced beyond steps
|
||||
assert all(s.status == "completed" for s in loaded.steps)
|
||||
assert loaded.mandatee_id == mandatee.id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_revoke_after_assign_preserves_mandatee_id(self, db: AsyncSession):
|
||||
"""Revoking a mandate keeps mandatee_id (for audit trail)."""
|
||||
mandatee = await _mk_identity(db, "Jacques")
|
||||
mandate = await _mk_mandate(db, mandatee_id=mandatee.id, status="active")
|
||||
|
||||
await revoke_mandate(mandate.id, db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.status == "revoked"
|
||||
assert loaded.mandatee_id == mandatee.id
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_origin_and_mandatee_can_differ(self, db: AsyncSession):
|
||||
"""The person who proposed the mandate (origin) is different from the mandatee."""
|
||||
proposer = await _mk_identity(db, "Kim")
|
||||
mandatee = await _mk_identity(db, "Laurent")
|
||||
|
||||
mandate = await _mk_mandate(db, origin_id=proposer.id)
|
||||
await assign_mandatee(mandate.id, mandatee.id, db)
|
||||
|
||||
loaded = await _reload(db, mandate.id)
|
||||
assert loaded.origin_id == proposer.id
|
||||
assert loaded.mandatee_id == mandatee.id
|
||||
assert loaded.origin_display_name == "Kim"
|
||||
assert loaded.mandatee_display_name == "Laurent"
|
||||
Reference in New Issue
Block a user