v3 : socle Rust (moteur pur, accès trustinWallet, serveur) ; maquette Nuxt rangée dans maquette/

- crates/ld-engine : formule inertielle (parité avec la maquette, garde C),
  coquille / changement de fond, assistant d'édition à règles, composition
  d'un document sous vote permanent (adoptions, aperçus, rivales).
- crates/trustin-access : qui voit, qui ouvre, qui gère les accès.
- crates/ld-server : API axum + SQLite, connexion Ğ1 (client OAuth2 de
  connect.monnaie-libre.fr), invitations, entrée en vigueur automatique,
  signalements de contradiction, traductions éditoriales, publication git,
  taille de la toile lue sur la chaîne.
- seeds/acte-engagement-certification.json : 33 articles, fr + en + es ;
  en vigueur : préambule, E1, E2 (seed a minima).
- maquette/ : frontend, backend et docker v2 déplacés tels quels.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Yvv
2026-10-21 02:04:42 +02:00
co-authored by Claude Opus 5.5
parent 4b1732916d
commit 67ddee28b2
275 changed files with 8979 additions and 0 deletions
@@ -0,0 +1,215 @@
"""Tests for cache_service: get, set, invalidate, and cleanup.
Uses mock database sessions to test cache logic in isolation
without requiring a real PostgreSQL connection.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timedelta, timezone
from unittest.mock import AsyncMock, MagicMock
import pytest
sqlalchemy = pytest.importorskip("sqlalchemy", reason="sqlalchemy required for cache service tests")
from app.services.cache_service import ( # noqa: E402
cleanup_expired,
get_cached,
invalidate,
set_cached,
)
# ---------------------------------------------------------------------------
# Helpers: mock objects for BlockchainCache entries
# ---------------------------------------------------------------------------
def _make_cache_entry(
key: str = "test:key",
value: dict | None = None,
expired: bool = False,
) -> MagicMock:
"""Create a mock BlockchainCache entry."""
entry = MagicMock()
entry.id = uuid.uuid4()
entry.cache_key = key
entry.cache_value = value or {"data": 42}
entry.fetched_at = datetime.now(timezone.utc)
if expired:
entry.expires_at = datetime.now(timezone.utc) - timedelta(hours=1)
else:
entry.expires_at = datetime.now(timezone.utc) + timedelta(hours=1)
return entry
def _make_async_db(entry: MagicMock | None = None) -> AsyncMock:
"""Create a mock async database session.
The session's execute() returns a result with scalar_one_or_none()
that returns the given entry (or None).
"""
db = AsyncMock()
result = MagicMock()
result.scalar_one_or_none.return_value = entry
db.execute = AsyncMock(return_value=result)
db.commit = AsyncMock()
db.add = MagicMock()
return db
# ---------------------------------------------------------------------------
# Tests: get_cached
# ---------------------------------------------------------------------------
class TestGetCached:
"""Test cache_service.get_cached."""
@pytest.mark.asyncio
async def test_returns_none_for_missing_key(self):
"""get_cached returns None when no entry exists for the key."""
db = _make_async_db(entry=None)
result = await get_cached("nonexistent:key", db)
assert result is None
@pytest.mark.asyncio
async def test_returns_value_for_valid_entry(self):
"""get_cached returns the cached value when entry exists and is not expired."""
entry = _make_cache_entry(key="blockchain:wot_size", value={"value": 7224})
db = _make_async_db(entry=entry)
result = await get_cached("blockchain:wot_size", db)
assert result == {"value": 7224}
@pytest.mark.asyncio
async def test_returns_none_for_expired_entry(self):
"""get_cached returns None when the entry is expired.
Note: In the real implementation, the SQL query filters by expires_at.
Here we test that when the DB returns None (as it would for expired),
get_cached returns None.
"""
db = _make_async_db(entry=None)
result = await get_cached("expired:key", db)
assert result is None
# ---------------------------------------------------------------------------
# Tests: set_cached + get_cached roundtrip
# ---------------------------------------------------------------------------
class TestSetCached:
"""Test cache_service.set_cached."""
@pytest.mark.asyncio
async def test_set_cached_creates_new_entry(self):
"""set_cached creates a new entry when key does not exist."""
db = _make_async_db(entry=None)
await set_cached("new:key", {"value": 100}, db, ttl_seconds=3600)
db.add.assert_called_once()
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_set_cached_updates_existing_entry(self):
"""set_cached updates the value when key already exists."""
existing = _make_cache_entry(key="existing:key", value={"value": 50})
db = _make_async_db(entry=existing)
await set_cached("existing:key", {"value": 200}, db, ttl_seconds=7200)
# Should update in-place, not add new
db.add.assert_not_called()
assert existing.cache_value == {"value": 200}
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_set_cached_roundtrip(self):
"""set_cached followed by get_cached returns the stored value."""
# First call: set (no existing entry)
db_set = _make_async_db(entry=None)
await set_cached("roundtrip:key", {"data": "test"}, db_set, ttl_seconds=3600)
db_set.add.assert_called_once()
# Extract the added entry from the mock
added_entry = db_set.add.call_args[0][0]
# Second call: get (returns the added entry)
db_get = _make_async_db(entry=added_entry)
result = await get_cached("roundtrip:key", db_get)
assert result == {"data": "test"}
# ---------------------------------------------------------------------------
# Tests: invalidate
# ---------------------------------------------------------------------------
class TestInvalidate:
"""Test cache_service.invalidate."""
@pytest.mark.asyncio
async def test_invalidate_removes_entry(self):
"""invalidate executes a delete and commits."""
db = AsyncMock()
db.execute = AsyncMock()
db.commit = AsyncMock()
await invalidate("some:key", db)
db.execute.assert_awaited_once()
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_invalidate_then_get_returns_none(self):
"""After invalidation, get_cached returns None."""
db = _make_async_db(entry=None)
result = await get_cached("invalidated:key", db)
assert result is None
# ---------------------------------------------------------------------------
# Tests: cleanup_expired
# ---------------------------------------------------------------------------
class TestCleanupExpired:
"""Test cache_service.cleanup_expired."""
@pytest.mark.asyncio
async def test_cleanup_removes_expired_entries(self):
"""cleanup_expired executes a delete for expired entries and returns count."""
db = AsyncMock()
# Mock the execute result with rowcount
exec_result = MagicMock()
exec_result.rowcount = 3
db.execute = AsyncMock(return_value=exec_result)
db.commit = AsyncMock()
count = await cleanup_expired(db)
assert count == 3
db.execute.assert_awaited_once()
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_cleanup_with_no_expired_entries(self):
"""cleanup_expired returns 0 when no expired entries exist."""
db = AsyncMock()
exec_result = MagicMock()
exec_result.rowcount = 0
db.execute = AsyncMock(return_value=exec_result)
db.commit = AsyncMock()
count = await cleanup_expired(db)
assert count == 0
@@ -0,0 +1,386 @@
"""Tests for decision service: advance_decision, 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, patch
import pytest
sqlalchemy = pytest.importorskip("sqlalchemy", reason="sqlalchemy required for decision service tests")
from app.services.decision_service import ( # noqa: E402
advance_decision,
create_vote_session_for_step,
)
# ---------------------------------------------------------------------------
# Helpers: mock objects that behave like SQLAlchemy models
# ---------------------------------------------------------------------------
def _make_step(
step_id: uuid.UUID | None = None,
decision_id: uuid.UUID | None = None,
step_order: int = 0,
step_type: str = "qualification",
status: str = "pending",
vote_session_id: uuid.UUID | None = None,
) -> MagicMock:
"""Create a mock DecisionStep."""
step = MagicMock()
step.id = step_id or uuid.uuid4()
step.decision_id = decision_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_decision(
decision_id: uuid.UUID | None = None,
title: str = "Decision de test",
decision_type: str = "document_change",
status: str = "draft",
voting_protocol_id: uuid.UUID | None = None,
steps: list | None = None,
) -> MagicMock:
"""Create a mock Decision."""
decision = MagicMock()
decision.id = decision_id or uuid.uuid4()
decision.title = title
decision.description = None
decision.context = None
decision.decision_type = decision_type
decision.status = status
decision.voting_protocol_id = voting_protocol_id
decision.created_by_id = None
decision.created_at = datetime.now(timezone.utc)
decision.updated_at = datetime.now(timezone.utc)
decision.steps = steps or []
return decision
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_db_for_advance(decision: MagicMock | None) -> AsyncMock:
"""Create a mock async database session for advance_decision."""
db = AsyncMock()
result = MagicMock()
result.scalar_one_or_none.return_value = decision
db.execute = AsyncMock(return_value=result)
db.commit = AsyncMock()
db.refresh = AsyncMock()
return db
def _make_db_for_vote_session(
decision: MagicMock | None,
protocol: MagicMock | None = None,
) -> AsyncMock:
"""Create a mock async database session for create_vote_session_for_step.
Sequential execute calls:
1st -> decision lookup
2nd -> protocol lookup
"""
db = AsyncMock()
call_results = []
# Decision result
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_decision
# ---------------------------------------------------------------------------
class TestAdvanceDecision:
"""Test decision_service.advance_decision."""
@pytest.mark.asyncio
async def test_advance_from_draft_activates_first_step(self):
"""Advancing a draft decision with pending steps activates the first one
and moves decision to qualification."""
decision_id = uuid.uuid4()
step1 = _make_step(decision_id=decision_id, step_order=0, status="pending")
step2 = _make_step(decision_id=decision_id, step_order=1, status="pending")
decision = _make_decision(
decision_id=decision_id,
status="draft",
steps=[step1, step2],
)
db = _make_db_for_advance(decision)
result = await advance_decision(decision_id, db)
assert step1.status == "active"
assert step2.status == "pending"
assert decision.status == "qualification"
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."""
decision_id = uuid.uuid4()
step1 = _make_step(decision_id=decision_id, step_order=0, status="active")
step2 = _make_step(decision_id=decision_id, step_order=1, status="pending")
decision = _make_decision(
decision_id=decision_id,
status="qualification",
steps=[step1, step2],
)
db = _make_db_for_advance(decision)
result = await advance_decision(decision_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 decision status advances."""
decision_id = uuid.uuid4()
step1 = _make_step(decision_id=decision_id, step_order=0, status="completed")
step2 = _make_step(decision_id=decision_id, step_order=1, status="active")
decision = _make_decision(
decision_id=decision_id,
status="qualification",
steps=[step1, step2],
)
db = _make_db_for_advance(decision)
result = await advance_decision(decision_id, db)
assert step2.status == "completed"
assert decision.status == "review"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_advance_no_steps_advances_status(self):
"""A decision with no steps advances its status directly."""
decision_id = uuid.uuid4()
decision = _make_decision(
decision_id=decision_id,
status="draft",
steps=[],
)
db = _make_db_for_advance(decision)
result = await advance_decision(decision_id, db)
assert decision.status == "qualification"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_advance_full_progression(self):
"""A decision with no steps can progress through all statuses."""
decision_id = uuid.uuid4()
decision = _make_decision(
decision_id=decision_id,
status="draft",
steps=[],
)
statuses_seen = [decision.status]
db = _make_db_for_advance(decision)
# Advance through all states until closed
for _ in range(10): # safety limit
# Recreate db mock for each call (side_effect is consumed)
db = _make_db_for_advance(decision)
await advance_decision(decision_id, db)
statuses_seen.append(decision.status)
if decision.status == "closed":
break
assert statuses_seen == [
"draft",
"qualification",
"review",
"voting",
"executed",
"closed",
]
@pytest.mark.asyncio
async def test_advance_closed_raises_error(self):
"""Advancing a closed decision raises ValueError."""
decision_id = uuid.uuid4()
decision = _make_decision(
decision_id=decision_id,
status="closed",
steps=[],
)
db = _make_db_for_advance(decision)
with pytest.raises(ValueError, match="deja cloturee"):
await advance_decision(decision_id, db)
@pytest.mark.asyncio
async def test_advance_not_found_raises_error(self):
"""Advancing a non-existent decision raises ValueError."""
db = _make_db_for_advance(None)
with pytest.raises(ValueError, match="Decision introuvable"):
await advance_decision(uuid.uuid4(), db)
# ---------------------------------------------------------------------------
# Tests: create_vote_session_for_step
# ---------------------------------------------------------------------------
class TestCreateVoteSessionForStep:
"""Test decision_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 step."""
decision_id = uuid.uuid4()
step_id = uuid.uuid4()
protocol_id = uuid.uuid4()
step = _make_step(
step_id=step_id,
decision_id=decision_id,
step_type="vote",
status="active",
vote_session_id=None,
)
decision = _make_decision(
decision_id=decision_id,
status="voting",
voting_protocol_id=protocol_id,
steps=[step],
)
protocol = _make_protocol(protocol_id=protocol_id, duration_days=14)
db = _make_db_for_vote_session(decision, protocol)
result = await create_vote_session_for_step(decision_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_decision_not_found(self):
"""ValueError when decision does not exist."""
db = _make_db_for_vote_session(None)
with pytest.raises(ValueError, match="Decision 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 decision."""
decision_id = uuid.uuid4()
decision = _make_decision(decision_id=decision_id, steps=[])
db = _make_db_for_vote_session(decision)
with pytest.raises(ValueError, match="Etape introuvable"):
await create_vote_session_for_step(decision_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."""
decision_id = uuid.uuid4()
step_id = uuid.uuid4()
existing_session_id = uuid.uuid4()
step = _make_step(
step_id=step_id,
decision_id=decision_id,
vote_session_id=existing_session_id,
)
decision = _make_decision(
decision_id=decision_id,
voting_protocol_id=uuid.uuid4(),
steps=[step],
)
db = _make_db_for_vote_session(decision)
with pytest.raises(ValueError, match="deja une session de vote"):
await create_vote_session_for_step(decision_id, step_id, db)
@pytest.mark.asyncio
async def test_create_vote_session_no_protocol(self):
"""ValueError when decision has no voting protocol configured."""
decision_id = uuid.uuid4()
step_id = uuid.uuid4()
step = _make_step(
step_id=step_id,
decision_id=decision_id,
vote_session_id=None,
)
decision = _make_decision(
decision_id=decision_id,
voting_protocol_id=None,
steps=[step],
)
db = _make_db_for_vote_session(decision)
with pytest.raises(ValueError, match="Aucun protocole de vote"):
await create_vote_session_for_step(decision_id, step_id, db)
@@ -0,0 +1,856 @@
"""TDD: Document ↔ Protocol ↔ Vote integration tests.
Tests the interrelation between:
- DocumentItem ←→ VotingProtocol (via voting_protocol_id)
- VotingProtocol ←→ FormulaConfig (formula parameters)
- VoteSession creation from DocumentItem context
- Threshold computation using item's protocol (inertia presets)
- Smith vs WoT standard protocol behavior
- ItemVersion lifecycle: propose → vote → accept/reject
- Multi-criteria adoption (WoT + Smith + TechComm)
All tests are pure unit tests exercising engine functions + service logic
without a real database (mocks only).
"""
from __future__ import annotations
import math
import uuid
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.engine.mode_params import parse_mode_params
from app.engine.nuanced_vote import evaluate_nuanced
from app.engine.smith_threshold import smith_threshold
from app.engine.techcomm_threshold import techcomm_threshold
from app.engine.threshold import wot_threshold
# ===========================================================================
# 1. DOCUMENT-PROTOCOL INTERRELATION
# ===========================================================================
class TestInertiaPresetsThresholds:
"""Verify that different inertia presets produce different thresholds.
Inertia presets map to gradient_exponent values:
low → G=0.1 (easy replacement)
standard → G=0.2 (balanced)
high → G=0.4 (hard replacement)
very_high → G=0.6 (very hard replacement)
"""
WOT_SIZE = 7224
TOTAL_VOTES = 120 # ~1.66% participation
INERTIA_MAP = {
"low": {"gradient_exponent": 0.1, "majority_pct": 50},
"standard": {"gradient_exponent": 0.2, "majority_pct": 50},
"high": {"gradient_exponent": 0.4, "majority_pct": 60},
"very_high": {"gradient_exponent": 0.6, "majority_pct": 66},
}
def _threshold_for_preset(self, preset: str) -> int:
params = self.INERTIA_MAP[preset]
return wot_threshold(
wot_size=self.WOT_SIZE,
total_votes=self.TOTAL_VOTES,
majority_pct=params["majority_pct"],
gradient_exponent=params["gradient_exponent"],
)
def test_low_inertia_easiest(self):
"""Low inertia should produce the lowest threshold."""
t_low = self._threshold_for_preset("low")
t_std = self._threshold_for_preset("standard")
assert t_low < t_std, f"Low ({t_low}) should be < standard ({t_std})"
def test_standard_below_high(self):
"""Standard inertia should be below high."""
t_std = self._threshold_for_preset("standard")
t_high = self._threshold_for_preset("high")
assert t_std < t_high, f"Standard ({t_std}) should be < high ({t_high})"
def test_high_below_very_high(self):
"""High inertia should be below very_high."""
t_high = self._threshold_for_preset("high")
t_vh = self._threshold_for_preset("very_high")
assert t_high < t_vh, f"High ({t_high}) should be < very_high ({t_vh})"
def test_monotonic_ordering(self):
"""All 4 presets must be strictly ordered: low < standard < high < very_high."""
thresholds = {p: self._threshold_for_preset(p) for p in self.INERTIA_MAP}
assert thresholds["low"] < thresholds["standard"]
assert thresholds["standard"] < thresholds["high"]
assert thresholds["high"] < thresholds["very_high"]
def test_low_inertia_near_majority(self):
"""With low inertia (G=0.1), even at 1.66% participation,
threshold shouldn't be too far from total votes."""
t = self._threshold_for_preset("low")
# With G=0.1, inertia is mild even at low participation
assert t <= self.TOTAL_VOTES, f"Low inertia threshold ({t}) should be <= total ({self.TOTAL_VOTES})"
def test_very_high_inertia_near_unanimity(self):
"""With very_high inertia (G=0.6, M=66%), at 1.66% participation,
threshold should be very close to total votes (near unanimity)."""
t = self._threshold_for_preset("very_high")
# At very low participation with high G and high M, threshold ≈ total
ratio = t / self.TOTAL_VOTES
assert ratio > 0.85, f"Very high inertia ratio ({ratio:.2f}) should demand near-unanimity"
class TestSmithProtocolVsStandard:
"""Compare the behavior of Smith protocol (D30M50B.1G.2S.1) vs
WoT standard (D30M50B.1G.2) — the only difference is the Smith criterion."""
WOT_SIZE = 7224
SMITH_SIZE = 20
TOTAL = 120
VOTES_FOR = 97
def test_same_wot_threshold(self):
"""Both protocols have M50B.1G.2 → same WoT threshold."""
smith_params = parse_mode_params("D30M50B.1G.2S.1")
std_params = parse_mode_params("D30M50B.1G.2")
t_smith = wot_threshold(
wot_size=self.WOT_SIZE, total_votes=self.TOTAL,
majority_pct=smith_params["majority_pct"],
gradient_exponent=smith_params["gradient_exponent"],
)
t_std = wot_threshold(
wot_size=self.WOT_SIZE, total_votes=self.TOTAL,
majority_pct=std_params["majority_pct"],
gradient_exponent=std_params["gradient_exponent"],
)
assert t_smith == t_std
def test_smith_criterion_present_vs_absent(self):
"""Smith protocol has smith_exponent=0.1, standard has None."""
smith_params = parse_mode_params("D30M50B.1G.2S.1")
std_params = parse_mode_params("D30M50B.1G.2")
assert smith_params["smith_exponent"] == 0.1
assert std_params["smith_exponent"] is None
def test_smith_protocol_can_be_blocked_by_smiths(self):
"""Smith protocol adoption requires both WoT AND Smith criteria."""
wot_thresh = wot_threshold(
wot_size=self.WOT_SIZE, total_votes=self.TOTAL,
majority_pct=50, gradient_exponent=0.2,
)
wot_pass = self.VOTES_FOR >= wot_thresh
smith_thresh = smith_threshold(self.SMITH_SIZE, 0.1) # = 2
smith_votes_for = 1 # Only 1 smith voted → fails
adopted = wot_pass and (smith_votes_for >= smith_thresh)
assert wot_pass is True
assert smith_votes_for < smith_thresh
assert adopted is False
def test_standard_protocol_ignores_smith(self):
"""Standard protocol with no smith_exponent always passes smith criterion."""
params = parse_mode_params("D30M50B.1G.2")
smith_ok = True # Default when smith_exponent is None
if params["smith_exponent"] is not None:
smith_ok = False # Would need smith votes
assert smith_ok is True
def test_smith_threshold_scales_with_smith_size(self):
"""Smith threshold ceil(N^0.1) grows slowly with smith WoT size."""
sizes_and_expected = [
(1, 1), # ceil(1^0.1) = 1
(5, 2), # ceil(5^0.1) = ceil(1.175) = 2
(20, 2), # ceil(20^0.1) = ceil(1.35) = 2
(100, 2), # ceil(100^0.1) = ceil(1.585) = 2
(1000, 2), # ceil(1000^0.1) = ceil(1.995) = 2
(1024, 2), # ceil(1024^0.1) = ceil(2.0) = 2
(1025, 3), # ceil(1025^0.1) > 2
]
for size, expected in sizes_and_expected:
result = smith_threshold(size, 0.1)
assert result == expected, f"smith_threshold({size}, 0.1) = {result}, expected {expected}"
class TestModeParamsRoundtrip:
"""Verify mode_params parsing produces correct formula parameters."""
def test_smith_protocol_params(self):
"""D30M50B.1G.2S.1 — the Forgeron Smith protocol."""
params = parse_mode_params("D30M50B.1G.2S.1")
assert params["duration_days"] == 30
assert params["majority_pct"] == 50
assert params["base_exponent"] == 0.1
assert params["gradient_exponent"] == 0.2
assert params["smith_exponent"] == 0.1
assert params["techcomm_exponent"] is None
def test_techcomm_protocol_params(self):
"""D30M50B.1G.2T.1 — a TechComm protocol."""
params = parse_mode_params("D30M50B.1G.2T.1")
assert params["techcomm_exponent"] == 0.1
assert params["smith_exponent"] is None
def test_full_protocol_params(self):
"""D30M50B.1G.2S.1T.1 — both Smith AND TechComm."""
params = parse_mode_params("D30M50B.1G.2S.1T.1")
assert params["smith_exponent"] == 0.1
assert params["techcomm_exponent"] == 0.1
def test_high_inertia_params(self):
"""D30M60B.1G.4 — high inertia preset."""
params = parse_mode_params("D30M60B.1G.4")
assert params["majority_pct"] == 60
assert params["gradient_exponent"] == 0.4
def test_very_high_inertia_params(self):
"""D30M66B.1G.6 — very high inertia preset."""
params = parse_mode_params("D30M66B.1G.6")
assert params["majority_pct"] == 66
assert params["gradient_exponent"] == 0.6
def test_params_used_in_threshold_match(self):
"""Threshold computed from parsed params must match direct computation."""
params = parse_mode_params("D30M50B.1G.2S.1")
computed = wot_threshold(
wot_size=7224, total_votes=120,
majority_pct=params["majority_pct"],
base_exponent=params["base_exponent"],
gradient_exponent=params["gradient_exponent"],
)
direct = wot_threshold(
wot_size=7224, total_votes=120,
majority_pct=50, base_exponent=0.1, gradient_exponent=0.2,
)
assert computed == direct
# ===========================================================================
# 2. VOTE BEHAVIOR — ADVANCED SCENARIOS
# ===========================================================================
class TestInertiaFormulaBehavior:
"""Deep tests on the inertia formula behavior across participation levels."""
def test_participation_curve_is_monotonically_decreasing(self):
"""As participation increases, required threshold ratio decreases.
This is the fundamental property of inertia-based democracy."""
W = 7224
M = 0.5
G = 0.2
prev_ratio = float("inf")
for t in range(10, W + 1, 100):
participation = t / W
inertia = 1.0 - participation ** G
ratio = M + (1.0 - M) * inertia
assert ratio <= prev_ratio, (
f"Ratio must decrease: at T={t}, ratio={ratio:.4f} > prev={prev_ratio:.4f}"
)
prev_ratio = ratio
def test_at_full_participation_ratio_equals_majority(self):
"""At T=W (100% participation), ratio should equal M exactly."""
W = 7224
M_pct = 50
M = M_pct / 100
threshold = wot_threshold(wot_size=W, total_votes=W, majority_pct=M_pct)
expected = math.ceil(0.1 ** W + M * W)
assert threshold == expected
def test_at_1_percent_participation_near_unanimity(self):
"""At ~1% participation, threshold should be near total votes."""
W = 7224
T = 72 # ~1%
threshold = wot_threshold(wot_size=W, total_votes=T, majority_pct=50)
ratio = threshold / T
assert ratio > 0.75, f"At 1% participation, ratio={ratio:.2f} should be > 0.75"
def test_at_50_percent_participation(self):
"""At 50% participation with G=0.2, threshold is well above simple majority."""
W = 1000
T = 500
threshold = wot_threshold(wot_size=W, total_votes=T, majority_pct=50,
gradient_exponent=0.2)
# (500/1000)^0.2 ≈ 0.87, inertia ≈ 0.13, ratio ≈ 0.565
assert threshold > 250, "Should be above simple majority"
assert threshold < 400, "Should not be near unanimity"
def test_gradient_zero_means_always_majority(self):
"""With G=0, (T/W)^0 = 1, inertia = 0, ratio = M always.
This effectively disables inertia."""
W = 7224
M_pct = 50
for T in [10, 100, 1000, 7224]:
threshold = wot_threshold(wot_size=W, total_votes=T,
majority_pct=M_pct, gradient_exponent=0.0001)
expected_approx = M_pct / 100 * T
# With very small G, threshold should be close to M*T
assert abs(threshold - math.ceil(expected_approx)) <= 2, (
f"At T={T}, threshold={threshold}, expected≈{expected_approx:.0f}"
)
class TestMultiCriteriaAdoption:
"""Test that adoption requires ALL applicable criteria to pass."""
WOT = 7224
TOTAL = 120
VOTES_FOR = 97
SMITH_SIZE = 20
TECHCOMM_SIZE = 5
def _wot_threshold(self):
return wot_threshold(wot_size=self.WOT, total_votes=self.TOTAL,
majority_pct=50, gradient_exponent=0.2)
def test_all_pass(self):
"""WoT pass + Smith pass + TechComm pass → adopted."""
wot_ok = self.VOTES_FOR >= self._wot_threshold()
smith_ok = 3 >= smith_threshold(self.SMITH_SIZE, 0.1) # 3 >= 2
tc_ok = 2 >= techcomm_threshold(self.TECHCOMM_SIZE, 0.1) # 2 >= 2
assert wot_ok and smith_ok and tc_ok
def test_wot_fails(self):
"""WoT fail + Smith pass + TechComm pass → rejected."""
wot_ok = 50 >= self._wot_threshold() # 50 < 94
smith_ok = 3 >= smith_threshold(self.SMITH_SIZE, 0.1)
tc_ok = 2 >= techcomm_threshold(self.TECHCOMM_SIZE, 0.1)
assert not wot_ok
assert not (wot_ok and smith_ok and tc_ok)
def test_smith_fails(self):
"""WoT pass + Smith fail + TechComm pass → rejected."""
wot_ok = self.VOTES_FOR >= self._wot_threshold()
smith_ok = 1 >= smith_threshold(self.SMITH_SIZE, 0.1) # 1 < 2
tc_ok = 2 >= techcomm_threshold(self.TECHCOMM_SIZE, 0.1)
assert wot_ok and tc_ok
assert not smith_ok
assert not (wot_ok and smith_ok and tc_ok)
def test_techcomm_fails(self):
"""WoT pass + Smith pass + TechComm fail → rejected."""
wot_ok = self.VOTES_FOR >= self._wot_threshold()
smith_ok = 3 >= smith_threshold(self.SMITH_SIZE, 0.1)
tc_ok = 1 >= techcomm_threshold(self.TECHCOMM_SIZE, 0.1) # 1 < 2
assert wot_ok and smith_ok
assert not tc_ok
assert not (wot_ok and smith_ok and tc_ok)
def test_all_fail(self):
"""All three fail → rejected."""
wot_ok = 10 >= self._wot_threshold()
smith_ok = 0 >= smith_threshold(self.SMITH_SIZE, 0.1)
tc_ok = 0 >= techcomm_threshold(self.TECHCOMM_SIZE, 0.1)
assert not (wot_ok or smith_ok or tc_ok)
def test_no_smith_no_techcomm(self):
"""When protocol has no Smith/TechComm, only WoT matters."""
params = parse_mode_params("D30M50B.1G.2")
wot_ok = self.VOTES_FOR >= self._wot_threshold()
# smith_exponent and techcomm_exponent are None
smith_ok = True # default when not configured
tc_ok = True
if params["smith_exponent"] is not None:
smith_ok = False
if params["techcomm_exponent"] is not None:
tc_ok = False
assert wot_ok and smith_ok and tc_ok
class TestEdgeCasesVotes:
"""Edge cases in vote behavior."""
def test_single_vote_small_wot(self):
"""1 vote out of 5 WoT members → threshold near 1 (almost unanimity)."""
threshold = wot_threshold(wot_size=5, total_votes=1, majority_pct=50)
# With 1/5 = 20% participation, inertia is high → threshold ≈ 1
assert threshold == 1
def test_single_vote_large_wot(self):
"""1 vote out of 7224 WoT → threshold = 1 (need that 1 vote to be for)."""
threshold = wot_threshold(wot_size=7224, total_votes=1, majority_pct=50)
assert threshold == 1
def test_two_votes_disagree(self):
"""2 votes: 1 for + 1 against. At low participation → need near-unanimity."""
threshold = wot_threshold(wot_size=7224, total_votes=2, majority_pct=50)
# Threshold should be close to 2 (near unanimity)
assert threshold == 2
def test_exact_threshold_boundary(self):
"""votes_for == threshold exactly → adopted (>= comparison)."""
threshold = wot_threshold(wot_size=7224, total_votes=120, majority_pct=50)
assert threshold >= threshold # votes_for == threshold → adopted
def test_one_below_threshold_boundary(self):
"""votes_for == threshold - 1 → rejected."""
threshold = wot_threshold(wot_size=7224, total_votes=120, majority_pct=50)
assert (threshold - 1) < threshold
def test_constant_base_raises_minimum(self):
"""With C=5, threshold is at least 5 even with no participation effect."""
threshold = wot_threshold(wot_size=100, total_votes=0,
majority_pct=50, constant_base=5.0)
assert threshold >= 5
def test_wot_size_1_minimal(self):
"""WoT of 1 member, 1 vote → threshold = 1."""
threshold = wot_threshold(wot_size=1, total_votes=1, majority_pct=50)
assert threshold == 1
class TestNuancedVoteAdvanced:
"""Advanced nuanced vote scenarios."""
def test_all_level_5_adopted(self):
"""60 voters all at level 5 (TOUT A FAIT) → adopted."""
result = evaluate_nuanced([5] * 60)
assert result["adopted"] is True
assert result["positive_pct"] == 100.0
def test_all_level_0_rejected(self):
"""60 voters all at level 0 (CONTRE) → rejected."""
result = evaluate_nuanced([0] * 60)
assert result["adopted"] is False
assert result["positive_pct"] == 0.0
def test_exactly_80_pct_positive(self):
"""Exactly 80% positive (48/60) → threshold_met = True."""
# 48 positive (levels 3-5) + 12 negative (levels 0-2) = 60
votes = [4] * 48 + [1] * 12
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["positive_pct"] == 80.0
assert result["threshold_met"] is True
assert result["adopted"] is True
def test_just_below_80_pct(self):
"""79.67% positive (47.8/60 ≈ 47/59) → threshold_met = False."""
# 47 positive + 13 negative = 60
votes = [4] * 47 + [1] * 13
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["positive_pct"] < 80.0
assert result["threshold_met"] is False
assert result["adopted"] is False
def test_min_participants_exactly_met(self):
"""59 participants exactly → min_participants_met = True."""
votes = [5] * 59
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["min_participants_met"] is True
assert result["adopted"] is True
def test_min_participants_not_met(self):
"""58 participants → min_participants_met = False, even if 100% positive."""
votes = [5] * 58
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["min_participants_met"] is False
assert result["threshold_met"] is True # 100% > 80%
assert result["adopted"] is False # but quorum not met
def test_neutre_counts_as_positive(self):
"""Level 3 (NEUTRE) counts as positive in the formula."""
votes = [3] * 60
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["positive_count"] == 60
assert result["positive_pct"] == 100.0
assert result["adopted"] is True
def test_invalid_level_raises(self):
"""Vote level 6 should raise ValueError."""
with pytest.raises(ValueError, match="invalide"):
evaluate_nuanced([6])
def test_negative_level_raises(self):
"""Vote level -1 should raise ValueError."""
with pytest.raises(ValueError, match="invalide"):
evaluate_nuanced([-1])
# ===========================================================================
# 3. ITEM MODIFICATION / DELETION / ADDITION WORKFLOW
# ===========================================================================
class TestItemVersionWorkflow:
"""Test the ItemVersion status lifecycle and apply/reject logic
using the document_service functions with mock database."""
def _make_item(self, item_id=None, text="Original text"):
item = MagicMock()
item.id = item_id or uuid.uuid4()
item.current_text = text
return item
def _make_version(self, version_id=None, item_id=None, proposed_text="New text",
status="proposed"):
version = MagicMock()
version.id = version_id or uuid.uuid4()
version.item_id = item_id or uuid.uuid4()
version.proposed_text = proposed_text
version.status = status
version.rationale = "Test rationale"
return version
@pytest.mark.asyncio
async def test_apply_version_updates_current_text(self):
"""When a version is accepted, item.current_text is updated."""
from app.services.document_service import apply_version
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = self._make_item(item_id, "Old text")
version = self._make_version(version_id, item_id, "New improved text")
db = AsyncMock()
# Mock item query
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = item
# Mock version query
version_result = MagicMock()
version_result.scalar_one_or_none.return_value = version
# Mock other versions query (no other pending versions)
other_result = MagicMock()
other_result.scalars.return_value = iter([])
db.execute = AsyncMock(side_effect=[item_result, version_result, other_result])
db.commit = AsyncMock()
db.refresh = AsyncMock()
result = await apply_version(item_id, version_id, db)
assert item.current_text == "New improved text"
assert version.status == "accepted"
@pytest.mark.asyncio
async def test_apply_version_rejects_competing_versions(self):
"""When a version is accepted, all other pending versions are rejected."""
from app.services.document_service import apply_version
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = self._make_item(item_id)
version = self._make_version(version_id, item_id, "Winning text")
other1 = self._make_version(status="proposed")
other2 = self._make_version(status="voting")
db = AsyncMock()
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = item
version_result = MagicMock()
version_result.scalar_one_or_none.return_value = version
other_result = MagicMock()
other_result.scalars.return_value = iter([other1, other2])
db.execute = AsyncMock(side_effect=[item_result, version_result, other_result])
db.commit = AsyncMock()
db.refresh = AsyncMock()
await apply_version(item_id, version_id, db)
assert other1.status == "rejected"
assert other2.status == "rejected"
@pytest.mark.asyncio
async def test_apply_version_wrong_item_raises(self):
"""Applying a version that belongs to a different item should raise."""
from app.services.document_service import apply_version
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = self._make_item(item_id)
version = self._make_version(version_id, uuid.uuid4(), "Text") # different item_id
db = AsyncMock()
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = item
version_result = MagicMock()
version_result.scalar_one_or_none.return_value = version
db.execute = AsyncMock(side_effect=[item_result, version_result])
with pytest.raises(ValueError, match="n'appartient pas"):
await apply_version(item_id, version_id, db)
@pytest.mark.asyncio
async def test_reject_version_sets_status(self):
"""Rejecting a version sets its status to 'rejected'."""
from app.services.document_service import reject_version
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = self._make_item(item_id)
version = self._make_version(version_id, item_id, status="proposed")
db = AsyncMock()
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = item
version_result = MagicMock()
version_result.scalar_one_or_none.return_value = version
db.execute = AsyncMock(side_effect=[item_result, version_result])
db.commit = AsyncMock()
db.refresh = AsyncMock()
result = await reject_version(item_id, version_id, db)
assert version.status == "rejected"
@pytest.mark.asyncio
async def test_apply_nonexistent_item_raises(self):
"""Applying a version to a nonexistent item should raise."""
from app.services.document_service import apply_version
db = AsyncMock()
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = None
db.execute = AsyncMock(return_value=item_result)
with pytest.raises(ValueError, match="introuvable"):
await apply_version(uuid.uuid4(), uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_apply_nonexistent_version_raises(self):
"""Applying a nonexistent version should raise."""
from app.services.document_service import apply_version
item_id = uuid.uuid4()
item = self._make_item(item_id)
db = AsyncMock()
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = item
version_result = MagicMock()
version_result.scalar_one_or_none.return_value = None
db.execute = AsyncMock(side_effect=[item_result, version_result])
with pytest.raises(ValueError, match="introuvable"):
await apply_version(item_id, uuid.uuid4(), db)
class TestDocumentSerialization:
"""Test document serialization for IPFS archival with new fields."""
def _make_doc(self, items=None):
from app.services.document_service import serialize_document_to_text
doc = MagicMock()
doc.title = "Acte d'engagement Certification"
doc.version = "1.0.0"
doc.doc_type = "engagement"
doc.status = "active"
doc.description = "Test document"
doc.items = items or []
return doc
def _make_item(self, position, sort_order, title, text, section_tag=None, item_type="clause"):
item = MagicMock()
item.position = position
item.sort_order = sort_order
item.title = title
item.current_text = text
item.item_type = item_type
item.section_tag = section_tag
return item
def test_serialization_includes_section_items(self):
"""Serialization should include items from all sections."""
from app.services.document_service import serialize_document_to_text
items = [
self._make_item("I1", 0, "Preambule", "Texte preambule", "introduction"),
self._make_item("E1", 1, "Clause 1", "Texte clause", "fondamental"),
self._make_item("X1", 2, "Annexe 1", "Texte annexe", "annexe"),
]
doc = self._make_doc(items)
result = serialize_document_to_text(doc)
assert "Preambule" in result
assert "Clause 1" in result
assert "Annexe 1" in result
assert "Texte preambule" in result
def test_serialization_preserves_order(self):
"""Items should appear in sort_order, not insertion order."""
from app.services.document_service import serialize_document_to_text
items = [
self._make_item("X1", 2, "Third", "Text 3"),
self._make_item("I1", 0, "First", "Text 1"),
self._make_item("E1", 1, "Second", "Text 2"),
]
doc = self._make_doc(items)
result = serialize_document_to_text(doc)
first_pos = result.index("First")
second_pos = result.index("Second")
third_pos = result.index("Third")
assert first_pos < second_pos < third_pos
def test_serialization_deterministic(self):
"""Same document serialized twice must produce identical output."""
from app.services.document_service import serialize_document_to_text
items = [self._make_item("E1", 0, "Clause 1", "Text 1")]
doc = self._make_doc(items)
result1 = serialize_document_to_text(doc)
result2 = serialize_document_to_text(doc)
assert result1 == result2
class TestVoteSessionCreationFromItem:
"""Test the logic for creating a VoteSession from a DocumentItem's protocol context.
This simulates what the votes router does when creating a session."""
def test_session_inherits_protocol_params(self):
"""A vote session created for an item should use the item's protocol."""
# Simulate: DocumentItem has voting_protocol_id → VotingProtocol has mode_params
protocol_params = parse_mode_params("D30M50B.1G.2S.1")
# These params should be used in threshold computation
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=protocol_params["majority_pct"],
gradient_exponent=protocol_params["gradient_exponent"],
base_exponent=protocol_params["base_exponent"],
)
assert threshold == 94
def test_different_protocols_different_thresholds(self):
"""Items with different protocols should produce different thresholds."""
# Certification item (standard WoT)
std_params = parse_mode_params("D30M50B.1G.2")
# Forgeron item (Smith protocol)
smith_params = parse_mode_params("D30M50B.1G.2S.1")
# TechComm item
tc_params = parse_mode_params("D30M50B.1G.2T.1")
# WoT thresholds are the same (same M/B/G)
t_std = wot_threshold(wot_size=7224, total_votes=120,
majority_pct=std_params["majority_pct"],
gradient_exponent=std_params["gradient_exponent"])
t_smith = wot_threshold(wot_size=7224, total_votes=120,
majority_pct=smith_params["majority_pct"],
gradient_exponent=smith_params["gradient_exponent"])
assert t_std == t_smith
# But Smith protocol requires additional smith votes
smith_req = smith_threshold(20, smith_params["smith_exponent"])
assert smith_req == 2
# TechComm protocol requires additional techcomm votes
tc_req = techcomm_threshold(5, tc_params["techcomm_exponent"])
assert tc_req == 2
def test_session_duration_from_protocol(self):
"""Session duration should come from protocol's duration_days."""
params = parse_mode_params("D30M50B.1G.2")
assert params["duration_days"] == 30
params_short = parse_mode_params("D7M50B.1G.2")
assert params_short["duration_days"] == 7
class TestRealWorldScenarios:
"""Real-world voting scenarios from Duniter community."""
def test_forgeron_vote_feb_2026(self):
"""Engagement Forgeron v2.0.0 — Feb 2026.
97 pour / 23 contre, WoT 7224, Smith 20.
Mode: D30M50B.1G.2S.1 → threshold=94 → adopted."""
params = parse_mode_params("D30M50B.1G.2S.1")
threshold = wot_threshold(
wot_size=7224, total_votes=120,
majority_pct=params["majority_pct"],
gradient_exponent=params["gradient_exponent"],
)
assert threshold == 94
adopted = 97 >= threshold
assert adopted is True
# Smith criterion
smith_req = smith_threshold(20, params["smith_exponent"])
assert smith_req == 2
# Assume at least 2 smiths voted for → passes
smith_ok = 5 >= smith_req
assert smith_ok is True
def test_forgeron_vote_barely_passes(self):
"""Same scenario but with exactly 94 votes for → still passes."""
threshold = wot_threshold(wot_size=7224, total_votes=117,
majority_pct=50, gradient_exponent=0.2)
# Slightly different total (94+23=117)
assert 94 >= threshold
def test_forgeron_vote_would_fail_at_93(self):
"""93 votes for out of 116 → fails (threshold likely ~93-94)."""
threshold = wot_threshold(wot_size=7224, total_votes=116,
majority_pct=50, gradient_exponent=0.2)
# With 116 total, threshold is still high
# 93 may or may not pass depending on exact computation
if threshold > 93:
assert 93 < threshold
def test_certification_item_low_inertia(self):
"""A certification document item with low inertia (G=0.1) is easier to replace."""
threshold_low = wot_threshold(wot_size=7224, total_votes=120,
majority_pct=50, gradient_exponent=0.1)
threshold_std = wot_threshold(wot_size=7224, total_votes=120,
majority_pct=50, gradient_exponent=0.2)
assert threshold_low < threshold_std
def test_very_high_inertia_item(self):
"""A formule/ordonnancement item with very_high inertia (G=0.6, M=66%)."""
threshold = wot_threshold(wot_size=7224, total_votes=120,
majority_pct=66, gradient_exponent=0.6)
# Should be very close to 120 (near unanimity at such low participation)
assert threshold >= 110, f"Very high inertia should demand near-unanimity, got {threshold}"
def test_full_wot_participation_simple_majority_wins(self):
"""If entire WoT of 7224 votes, simple majority (3613) suffices with standard params."""
threshold = wot_threshold(wot_size=7224, total_votes=7224,
majority_pct=50, gradient_exponent=0.2)
# At full participation: threshold ≈ 0.5 * 7224 = 3612
assert threshold == math.ceil(0.1 ** 7224 + 0.5 * 7224)
assert threshold == 3612
def test_techcomm_vote_cotec_5_members(self):
"""TechComm criterion with 5 members, exponent 0.1 → need 2 votes."""
tc_threshold = techcomm_threshold(5, 0.1)
assert tc_threshold == 2
# 1 TC vote → fails
assert 1 < tc_threshold
# 2 TC votes → passes
assert 2 >= tc_threshold
@@ -0,0 +1,418 @@
"""Tests for document service: apply_version, reject_version, and serialization.
These are pure unit tests that mock the database layer to test
the service logic in isolation.
"""
from __future__ import annotations
import hashlib
import uuid
from datetime import datetime, timezone
from unittest.mock import AsyncMock, MagicMock
import pytest
sqlalchemy = pytest.importorskip("sqlalchemy", reason="sqlalchemy required for document service tests")
from app.services.document_service import ( # noqa: E402
apply_version,
reject_version,
serialize_document_to_text,
)
# ---------------------------------------------------------------------------
# Helpers: mock objects that behave like SQLAlchemy models
# ---------------------------------------------------------------------------
def _make_item(
item_id: uuid.UUID | None = None,
document_id: uuid.UUID | None = None,
current_text: str = "Texte original",
position: str = "1",
item_type: str = "clause",
title: str | None = None,
sort_order: int = 0,
) -> MagicMock:
"""Create a mock DocumentItem."""
item = MagicMock()
item.id = item_id or uuid.uuid4()
item.document_id = document_id or uuid.uuid4()
item.current_text = current_text
item.position = position
item.item_type = item_type
item.title = title
item.sort_order = sort_order
item.created_at = datetime.now(timezone.utc)
item.updated_at = datetime.now(timezone.utc)
return item
def _make_version(
version_id: uuid.UUID | None = None,
item_id: uuid.UUID | None = None,
proposed_text: str = "Texte propose",
status: str = "proposed",
) -> MagicMock:
"""Create a mock ItemVersion."""
version = MagicMock()
version.id = version_id or uuid.uuid4()
version.item_id = item_id or uuid.uuid4()
version.proposed_text = proposed_text
version.status = status
version.diff_text = None
version.rationale = None
version.decision_id = None
version.proposed_by_id = None
version.created_at = datetime.now(timezone.utc)
return version
def _make_document(
doc_id: uuid.UUID | None = None,
slug: str = "test-doc",
title: str = "Document de test",
doc_type: str = "licence",
version: str = "1.0.0",
status: str = "active",
description: str | None = "Description de test",
items: list | None = None,
) -> MagicMock:
"""Create a mock Document."""
doc = MagicMock()
doc.id = doc_id or uuid.uuid4()
doc.slug = slug
doc.title = title
doc.doc_type = doc_type
doc.version = version
doc.status = status
doc.description = description
doc.ipfs_cid = None
doc.chain_anchor = None
doc.items = items or []
doc.created_at = datetime.now(timezone.utc)
doc.updated_at = datetime.now(timezone.utc)
return doc
def _make_async_db(
item: MagicMock | None = None,
version: MagicMock | None = None,
other_versions: list[MagicMock] | None = None,
) -> AsyncMock:
"""Create a mock async database session.
The mock session's execute() returns appropriate results based on
the query being run. It supports multiple sequential calls:
1st call -> item lookup
2nd call -> version lookup
3rd call (optional) -> other versions lookup (for apply_version)
"""
db = AsyncMock()
call_results = []
# Item result
item_result = MagicMock()
item_result.scalar_one_or_none.return_value = item
call_results.append(item_result)
# Version result
version_result = MagicMock()
version_result.scalar_one_or_none.return_value = version
call_results.append(version_result)
# Other versions result (for apply_version)
if other_versions is not None:
other_result = MagicMock()
other_scalars = MagicMock()
other_scalars.__iter__ = MagicMock(return_value=iter(other_versions))
other_result.scalars.return_value = other_scalars
call_results.append(other_result)
db.execute = AsyncMock(side_effect=call_results)
db.commit = AsyncMock()
db.refresh = AsyncMock()
return db
# ---------------------------------------------------------------------------
# Tests: apply_version
# ---------------------------------------------------------------------------
class TestApplyVersion:
"""Test document_service.apply_version."""
@pytest.mark.asyncio
async def test_apply_version_updates_text(self):
"""Applying a version replaces item's current_text with proposed_text."""
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = _make_item(item_id=item_id, current_text="Ancien texte")
version = _make_version(
version_id=version_id,
item_id=item_id,
proposed_text="Nouveau texte",
)
db = _make_async_db(item=item, version=version, other_versions=[])
result = await apply_version(item_id, version_id, db)
assert result.current_text == "Nouveau texte"
assert version.status == "accepted"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_apply_version_rejects_other_pending(self):
"""Applying a version rejects other pending/voting versions."""
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = _make_item(item_id=item_id)
version = _make_version(version_id=version_id, item_id=item_id)
other_v1 = _make_version(item_id=item_id, status="proposed")
other_v2 = _make_version(item_id=item_id, status="voting")
db = _make_async_db(
item=item,
version=version,
other_versions=[other_v1, other_v2],
)
await apply_version(item_id, version_id, db)
assert other_v1.status == "rejected"
assert other_v2.status == "rejected"
@pytest.mark.asyncio
async def test_apply_version_item_not_found(self):
"""ValueError is raised when item does not exist."""
db = _make_async_db(item=None, version=None)
with pytest.raises(ValueError, match="Element de document introuvable"):
await apply_version(uuid.uuid4(), uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_apply_version_version_not_found(self):
"""ValueError is raised when version does not exist."""
item = _make_item()
db = _make_async_db(item=item, version=None)
with pytest.raises(ValueError, match="Version introuvable"):
await apply_version(item.id, uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_apply_version_wrong_item(self):
"""ValueError is raised when version belongs to a different item."""
item_id = uuid.uuid4()
other_item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = _make_item(item_id=item_id)
version = _make_version(version_id=version_id, item_id=other_item_id)
db = _make_async_db(item=item, version=version)
with pytest.raises(ValueError, match="n'appartient pas"):
await apply_version(item_id, version_id, db)
# ---------------------------------------------------------------------------
# Tests: reject_version
# ---------------------------------------------------------------------------
class TestRejectVersion:
"""Test document_service.reject_version."""
@pytest.mark.asyncio
async def test_reject_version_sets_status(self):
"""Rejecting a version sets its status to 'rejected'."""
item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = _make_item(item_id=item_id)
version = _make_version(
version_id=version_id,
item_id=item_id,
status="proposed",
)
db = _make_async_db(item=item, version=version)
result = await reject_version(item_id, version_id, db)
assert result.status == "rejected"
db.commit.assert_awaited_once()
@pytest.mark.asyncio
async def test_reject_version_item_not_found(self):
"""ValueError is raised when item does not exist."""
db = _make_async_db(item=None, version=None)
with pytest.raises(ValueError, match="Element de document introuvable"):
await reject_version(uuid.uuid4(), uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_reject_version_version_not_found(self):
"""ValueError is raised when version does not exist."""
item = _make_item()
db = _make_async_db(item=item, version=None)
with pytest.raises(ValueError, match="Version introuvable"):
await reject_version(item.id, uuid.uuid4(), db)
@pytest.mark.asyncio
async def test_reject_version_wrong_item(self):
"""ValueError is raised when version belongs to a different item."""
item_id = uuid.uuid4()
other_item_id = uuid.uuid4()
version_id = uuid.uuid4()
item = _make_item(item_id=item_id)
version = _make_version(version_id=version_id, item_id=other_item_id)
db = _make_async_db(item=item, version=version)
with pytest.raises(ValueError, match="n'appartient pas"):
await reject_version(item_id, version_id, db)
# ---------------------------------------------------------------------------
# Tests: serialize_document_to_text
# ---------------------------------------------------------------------------
class TestSerializeDocumentToText:
"""Test document serialization for archival."""
def test_basic_serialization(self):
"""A document with items serializes to the expected text format."""
doc_id = uuid.uuid4()
item1 = _make_item(
document_id=doc_id,
position="1",
title="Preambule",
item_type="preamble",
current_text="Le texte du preambule.",
sort_order=0,
)
item2 = _make_item(
document_id=doc_id,
position="2",
title="Article premier",
item_type="clause",
current_text="Le texte de l'article premier.",
sort_order=1,
)
item3 = _make_item(
document_id=doc_id,
position="2.1",
title=None,
item_type="rule",
current_text="Sous-article sans titre.",
sort_order=2,
)
doc = _make_document(
doc_id=doc_id,
title="Licence G1",
version="2.0.0",
doc_type="licence",
status="active",
description="La licence monetaire de la G1",
items=[item1, item2, item3],
)
text = serialize_document_to_text(doc)
assert "# Licence G1" in text
assert "Version: 2.0.0" in text
assert "Type: licence" in text
assert "Statut: active" in text
assert "Description: La licence monetaire de la G1" in text
# Items
assert "## 1 - Preambule [preamble]" in text
assert "Le texte du preambule." in text
assert "## 2 - Article premier [clause]" in text
assert "Le texte de l'article premier." in text
assert "## 2.1 [rule]" in text
assert "Sous-article sans titre." in text
def test_serialization_ordering(self):
"""Items are serialized in sort_order, not insertion order."""
doc_id = uuid.uuid4()
item_b = _make_item(
document_id=doc_id,
position="2",
title="Second",
current_text="Texte B",
sort_order=1,
)
item_a = _make_item(
document_id=doc_id,
position="1",
title="Premier",
current_text="Texte A",
sort_order=0,
)
# Insert in reverse order
doc = _make_document(doc_id=doc_id, items=[item_b, item_a])
text = serialize_document_to_text(doc)
# "Premier" should appear before "Second"
idx_a = text.index("Texte A")
idx_b = text.index("Texte B")
assert idx_a < idx_b, "Items should be ordered by sort_order"
def test_serialization_without_description(self):
"""Document without description omits that line."""
doc = _make_document(description=None, items=[])
text = serialize_document_to_text(doc)
assert "Description:" not in text
def test_serialization_hash_is_deterministic(self):
"""Same document content produces the same SHA-256 hash."""
doc_id = uuid.uuid4()
item = _make_item(
document_id=doc_id,
position="1",
title="Test",
current_text="Contenu identique",
sort_order=0,
)
doc1 = _make_document(doc_id=doc_id, title="Doc", version="1.0", items=[item])
doc2 = _make_document(doc_id=doc_id, title="Doc", version="1.0", items=[item])
text1 = serialize_document_to_text(doc1)
text2 = serialize_document_to_text(doc2)
hash1 = hashlib.sha256(text1.encode("utf-8")).hexdigest()
hash2 = hashlib.sha256(text2.encode("utf-8")).hexdigest()
assert hash1 == hash2
def test_empty_document(self):
"""A document with no items serializes header only."""
doc = _make_document(items=[])
text = serialize_document_to_text(doc)
assert "# Document de test" in text
assert "Version: 1.0.0" in text
# Should end with just a newline after the header block
lines = text.strip().split("\n")
assert len(lines) >= 4 # title, version, type, status
@@ -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"
+655
View File
@@ -0,0 +1,655 @@
"""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)
@@ -0,0 +1,185 @@
"""Tests for middleware stack: CORS headers, rate limiting, dev auth flow.
Critical invariants:
- ALL responses (including 429) must carry CORS headers when origin is allowed
- Dev login flow must survive repeated logins without hitting rate limit
- OPTIONS preflight must never be rate-limited
Note: each test uses a unique X-Forwarded-For IP to isolate rate limit counters,
since the rate limiter is in-memory and shared across the app instance.
"""
from __future__ import annotations
import pytest
from httpx import ASGITransport, AsyncClient
import app.models # noqa: F401 — registers all models with Base.metadata before create_all
from app.database import init_db
from app.main import app
@pytest.fixture(scope="module", autouse=True)
async def _create_tables():
"""Create DB tables once for this module.
ASGITransport does not trigger the FastAPI lifespan, so init_db() would
never run. Tests that hit endpoints backed by the DB need the tables to
exist beforehand.
"""
await init_db()
ORIGIN = "http://localhost:3002"
CHALLENGE_URL = "/api/v1/auth/challenge"
VERIFY_URL = "/api/v1/auth/verify"
ME_URL = "/api/v1/auth/me"
DEV_ADDRESS = "5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"
# ---------------------------------------------------------------------------
# INVARIANT 1: 429 responses must include CORS headers
# Without this, the browser sees "Failed to fetch" instead of "Too Many Requests"
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_rate_limited_response_has_cors_headers():
"""A 429 from the rate limiter must still carry Access-Control-Allow-Origin.
Root cause of the "no response / Failed to fetch" bug: the rate limiter
sits outside CORS in the middleware stack, so its 429 responses have no
CORS headers and the browser discards them as network errors.
"""
# dev auth limit = 60/min (RATE_LIMIT_DEFAULT), prod = 10/min (RATE_LIMIT_AUTH)
# Send 65 requests to guarantee 429 regardless of environment.
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
last_response = None
for _ in range(65):
resp = await client.post(
CHALLENGE_URL,
json={"address": DEV_ADDRESS},
headers={"Origin": ORIGIN, "X-Forwarded-For": "10.0.1.1"},
)
last_response = resp
if resp.status_code == 429:
break
assert last_response is not None
assert last_response.status_code == 429, (
"Expected 429 after exceeding auth rate limit"
)
assert "access-control-allow-origin" in last_response.headers, (
"429 response must include CORS headers so the browser can read the error"
)
assert last_response.headers["access-control-allow-origin"] == ORIGIN
# ---------------------------------------------------------------------------
# INVARIANT 2: OPTIONS preflight must never be rate-limited
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_options_preflight_never_rate_limited():
"""OPTIONS requests must pass through regardless of request count.
Browsers send a preflight before every cross-origin POST with custom headers.
A 429 on OPTIONS prevents the real request from ever being sent.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
for i in range(20):
resp = await client.options(
CHALLENGE_URL,
headers={
"Origin": ORIGIN,
"Access-Control-Request-Method": "POST",
"Access-Control-Request-Headers": "content-type",
"X-Forwarded-For": "10.0.2.1",
},
)
assert resp.status_code != 429, (
f"OPTIONS request #{i + 1} was rate-limited (429) — preflights must never be blocked"
)
# ---------------------------------------------------------------------------
# INVARIANT 3: Dev login flow must survive ≥ 10 consecutive logins
# (challenge + verify cycle, dev profile bypass)
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_dev_login_survives_repeated_cycles():
"""Complete login cycle (challenge → verify) must work ≥ 10 times in a row.
In dev mode, the developer disconnects and reconnects frequently.
With auth rate limit = 10/min, the 6th challenge request would be blocked.
Dev mode must use a higher limit (≥ 60/min) to prevent login lockout.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
tokens = []
for i in range(10):
# Step 1: get challenge
ch_resp = await client.post(
CHALLENGE_URL,
json={"address": DEV_ADDRESS},
headers={"Origin": ORIGIN, "X-Forwarded-For": "10.0.3.1"},
)
assert ch_resp.status_code == 200, (
f"Login cycle #{i + 1}: challenge returned {ch_resp.status_code} — "
f"rate limit likely hit. Dev mode requires RATE_LIMIT_AUTH ≥ 60/min."
)
challenge = ch_resp.json()["challenge"]
# Step 2: verify (dev bypass — any signature accepted for dev addresses)
v_resp = await client.post(
VERIFY_URL,
json={
"address": DEV_ADDRESS,
"challenge": challenge,
"signature": "0x" + "ab" * 64,
},
headers={"Origin": ORIGIN, "X-Forwarded-For": "10.0.3.1"},
)
assert v_resp.status_code == 200, (
f"Login cycle #{i + 1}: verify returned {v_resp.status_code}"
)
tokens.append(v_resp.json()["token"])
assert len(tokens) == 10, "Expected 10 successful login cycles"
assert len(set(tokens)) == 10, "Each login cycle must produce a unique token"
# ---------------------------------------------------------------------------
# INVARIANT 4: /auth/me OPTIONS preflight must return CORS headers
# ---------------------------------------------------------------------------
@pytest.mark.anyio
async def test_auth_me_options_preflight_returns_cors():
"""OPTIONS preflight for /auth/me must return 200 with CORS headers.
This was the root cause of the session-lost-on-reload bug:
repeated /auth/me calls would exhaust the auth rate limit,
the 429 OPTIONS response had no CORS headers,
and the browser threw 'Failed to fetch'.
"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as client:
resp = await client.options(
ME_URL,
headers={
"Origin": ORIGIN,
"Access-Control-Request-Method": "GET",
"Access-Control-Request-Headers": "authorization",
"X-Forwarded-For": "10.0.4.1",
},
)
assert resp.status_code in (200, 204), (
f"OPTIONS /auth/me returned {resp.status_code}"
)
assert "access-control-allow-origin" in resp.headers
@@ -0,0 +1,75 @@
"""Tests for mode-params string parser."""
from app.engine.mode_params import parse_mode_params
class TestParseModeParams:
"""Parse compact parameter strings into structured dicts."""
def test_standard_params(self):
"""D30M50B.1G.2 => standard Licence G1 params."""
result = parse_mode_params("D30M50B.1G.2")
assert result["duration_days"] == 30
assert result["majority_pct"] == 50
assert result["base_exponent"] == 0.1
assert result["gradient_exponent"] == 0.2
# Optional criteria absent
assert result["smith_exponent"] is None
assert result["techcomm_exponent"] is None
def test_with_smith_exponent(self):
"""D30M50B.1G.2S.1 => standard + smith_exponent=0.1."""
result = parse_mode_params("D30M50B.1G.2S.1")
assert result["duration_days"] == 30
assert result["majority_pct"] == 50
assert result["base_exponent"] == 0.1
assert result["gradient_exponent"] == 0.2
assert result["smith_exponent"] == 0.1
assert result["techcomm_exponent"] is None
def test_with_techcomm_exponent(self):
"""D30M50B.1G.2T.1 => standard + techcomm_exponent=0.1."""
result = parse_mode_params("D30M50B.1G.2T.1")
assert result["duration_days"] == 30
assert result["majority_pct"] == 50
assert result["base_exponent"] == 0.1
assert result["gradient_exponent"] == 0.2
assert result["smith_exponent"] is None
assert result["techcomm_exponent"] == 0.1
def test_full_params_with_constant(self):
"""D30M50B1G.5C10 => integer base, gradient=0.5, constant=10."""
result = parse_mode_params("D30M50B1G.5C10")
assert result["duration_days"] == 30
assert result["majority_pct"] == 50
assert result["base_exponent"] == 1.0
assert result["gradient_exponent"] == 0.5
assert result["constant_base"] == 10.0
def test_empty_string_defaults(self):
"""Empty string returns all defaults."""
result = parse_mode_params("")
assert result["duration_days"] == 30
assert result["majority_pct"] == 50
assert result["base_exponent"] == 0.1
assert result["gradient_exponent"] == 0.2
assert result["constant_base"] == 0.0
assert result["smith_exponent"] is None
assert result["techcomm_exponent"] is None
assert result["ratio_multiplier"] is None
assert result["is_ratio_mode"] is False
def test_whitespace_only_returns_defaults(self):
"""Whitespace-only string treated as empty."""
result = parse_mode_params(" ")
assert result["duration_days"] == 30
def test_roundtrip_consistency(self):
"""Parsing a standard string then re-checking all keys."""
result = parse_mode_params("D30M50B.1G.2")
expected_keys = {
"duration_days", "majority_pct", "base_exponent",
"gradient_exponent", "constant_base", "smith_exponent",
"techcomm_exponent", "ratio_multiplier", "is_ratio_mode",
}
assert set(result.keys()) == expected_keys
+120
View File
@@ -0,0 +1,120 @@
"""Tests for six-level nuanced vote evaluation.
Levels: 0-CONTRE, 1-PAS DU TOUT, 2-PAS D'ACCORD, 3-NEUTRE, 4-D'ACCORD, 5-TOUT A FAIT
Positive = levels 3 + 4 + 5
Adoption requires: positive_pct >= threshold (80%) AND total >= min_participants (59).
"""
import pytest
from app.engine.nuanced_vote import evaluate_nuanced
class TestNuancedVoteAdoption:
"""Cases where the vote should be adopted."""
def test_59_positive_10_negative_adopted(self):
"""59 positive (levels 3-5) + 10 negative = 69 total.
positive_pct = 59/69 ~ 85.5% >= 80% and 69 >= 59 => adopted.
"""
votes = [5] * 20 + [4] * 20 + [3] * 19 + [2] * 5 + [1] * 3 + [0] * 2
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 69
assert result["positive_count"] == 59
assert result["positive_pct"] == pytest.approx(85.51, abs=0.1)
assert result["threshold_met"] is True
assert result["min_participants_met"] is True
assert result["adopted"] is True
def test_all_tout_a_fait_adopted(self):
"""All 59 voters at level 5 => 100% positive, adopted."""
votes = [5] * 59
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 59
assert result["positive_count"] == 59
assert result["positive_pct"] == 100.0
assert result["adopted"] is True
class TestNuancedVoteRejection:
"""Cases where the vote should be rejected."""
def test_40_positive_30_negative_rejected(self):
"""40 positive + 30 negative = 70 total.
positive_pct = 40/70 ~ 57.14% < 80% => threshold not met.
"""
votes = [5] * 15 + [4] * 15 + [3] * 10 + [2] * 10 + [1] * 10 + [0] * 10
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 70
assert result["positive_count"] == 40
assert result["positive_pct"] == pytest.approx(57.14, abs=0.1)
assert result["threshold_met"] is False
assert result["min_participants_met"] is True # 70 >= 59
assert result["adopted"] is False
def test_min_participants_not_met(self):
"""50 positive + 5 negative = 55 total < 59 min_participants.
Even though 50/55 ~ 90.9% > 80%, adoption fails on min_participants.
"""
votes = [5] * 30 + [4] * 10 + [3] * 10 + [1] * 3 + [0] * 2
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 55
assert result["positive_count"] == 50
assert result["positive_pct"] > 80
assert result["threshold_met"] is True
assert result["min_participants_met"] is False
assert result["adopted"] is False
class TestNuancedVoteEdgeCases:
"""Edge cases and exact boundary conditions."""
def test_exact_threshold_80_percent(self):
"""Exactly 80% positive votes should pass the threshold."""
# 80 positive out of 100 = exactly 80%
votes = [5] * 40 + [4] * 20 + [3] * 20 + [2] * 10 + [1] * 5 + [0] * 5
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 100
assert result["positive_count"] == 80
assert result["positive_pct"] == 80.0
assert result["threshold_met"] is True
assert result["min_participants_met"] is True
assert result["adopted"] is True
def test_just_below_threshold(self):
"""79 positive out of 100 = 79% < 80% => rejected."""
votes = [5] * 39 + [4] * 20 + [3] * 20 + [2] * 11 + [1] * 5 + [0] * 5
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 100
assert result["positive_count"] == 79
assert result["positive_pct"] == 79.0
assert result["threshold_met"] is False
assert result["adopted"] is False
def test_empty_votes(self):
"""Zero votes => not adopted."""
result = evaluate_nuanced([], threshold_pct=80, min_participants=59)
assert result["total"] == 0
assert result["positive_count"] == 0
assert result["positive_pct"] == 0.0
assert result["adopted"] is False
def test_invalid_vote_level(self):
"""Vote level outside 0-5 raises ValueError."""
with pytest.raises(ValueError, match="invalide"):
evaluate_nuanced([5, 3, 6])
def test_per_level_counts(self):
"""Verify per-level breakdown is correct."""
votes = [0, 1, 2, 3, 4, 5, 5, 4, 3]
result = evaluate_nuanced(votes, threshold_pct=50, min_participants=1)
assert result["per_level_counts"] == {0: 1, 1: 1, 2: 1, 3: 2, 4: 2, 5: 2}
assert result["positive_count"] == 6 # 2+2+2
assert result["total"] == 9
@@ -0,0 +1,226 @@
"""Tests for public API: basic schema validation and serialization.
Uses mock database sessions to test the public router logic
without requiring a real PostgreSQL connection.
"""
from __future__ import annotations
import uuid
from datetime import datetime, timezone
from unittest.mock import MagicMock
import pytest
sqlalchemy = pytest.importorskip("sqlalchemy", reason="sqlalchemy required for public API tests")
from app.schemas.document import DocumentFullOut, DocumentOut # noqa: E402
from app.schemas.sanctuary import SanctuaryEntryOut # noqa: E402
# ---------------------------------------------------------------------------
# Helpers: mock objects
# ---------------------------------------------------------------------------
def _make_document_mock(
doc_id: uuid.UUID | None = None,
slug: str = "licence-g1",
title: str = "Licence G1",
doc_type: str = "licence",
version: str = "2.0.0",
doc_status: str = "active",
description: str | None = "La licence monetaire",
) -> MagicMock:
"""Create a mock Document for schema validation."""
doc = MagicMock()
doc.id = doc_id or uuid.uuid4()
doc.slug = slug
doc.title = title
doc.doc_type = doc_type
doc.version = version
doc.status = doc_status
doc.description = description
doc.ipfs_cid = None
doc.chain_anchor = None
doc.genesis_json = None
doc.created_at = datetime.now(timezone.utc)
doc.updated_at = datetime.now(timezone.utc)
doc.items = []
return doc
def _make_item_mock(
item_id: uuid.UUID | None = None,
document_id: uuid.UUID | None = None,
position: str = "1",
item_type: str = "clause",
title: str | None = "Article 1",
current_text: str = "Texte de l'article",
sort_order: int = 0,
) -> MagicMock:
"""Create a mock DocumentItem for schema validation."""
item = MagicMock()
item.id = item_id or uuid.uuid4()
item.document_id = document_id or uuid.uuid4()
item.position = position
item.item_type = item_type
item.title = title
item.current_text = current_text
item.voting_protocol_id = None
item.sort_order = sort_order
item.section_tag = None
item.inertia_preset = "standard"
item.is_permanent_vote = True
item.created_at = datetime.now(timezone.utc)
item.updated_at = datetime.now(timezone.utc)
return item
def _make_sanctuary_entry_mock(
entry_id: uuid.UUID | None = None,
entry_type: str = "document",
reference_id: uuid.UUID | None = None,
title: str | None = "Licence G1 v2.0.0",
content_hash: str = "abc123def456",
ipfs_cid: str | None = "QmTestCid123",
chain_tx_hash: str | None = "0xdeadbeef",
) -> MagicMock:
"""Create a mock SanctuaryEntry for schema validation."""
entry = MagicMock()
entry.id = entry_id or uuid.uuid4()
entry.entry_type = entry_type
entry.reference_id = reference_id or uuid.uuid4()
entry.title = title
entry.content_hash = content_hash
entry.ipfs_cid = ipfs_cid
entry.chain_tx_hash = chain_tx_hash
entry.chain_block = 12345 if chain_tx_hash else None
entry.metadata_json = None
entry.created_at = datetime.now(timezone.utc)
return entry
# ---------------------------------------------------------------------------
# Tests: DocumentOut schema serialization
# ---------------------------------------------------------------------------
class TestDocumentOutSchema:
"""Test DocumentOut schema validation from mock objects."""
def test_document_out_basic(self):
"""DocumentOut validates from a mock document object."""
doc = _make_document_mock()
out = DocumentOut.model_validate(doc)
assert out.slug == "licence-g1"
assert out.title == "Licence G1"
assert out.doc_type == "licence"
assert out.version == "2.0.0"
assert out.status == "active"
# items_count is set explicitly after validation (not from model)
out.items_count = 0
assert out.items_count == 0
def test_document_out_with_items_count(self):
"""DocumentOut can have items_count set after validation."""
doc = _make_document_mock()
out = DocumentOut.model_validate(doc)
out.items_count = 42
assert out.items_count == 42
def test_document_out_all_fields_present(self):
"""All expected fields are present in the DocumentOut serialization."""
doc = _make_document_mock()
out = DocumentOut.model_validate(doc)
data = out.model_dump()
expected_fields = {
"id", "slug", "title", "doc_type", "version", "status",
"description", "ipfs_cid", "chain_anchor", "genesis_json",
"created_at", "updated_at", "items_count",
}
assert expected_fields.issubset(set(data.keys()))
# ---------------------------------------------------------------------------
# Tests: DocumentFullOut schema serialization
# ---------------------------------------------------------------------------
class TestDocumentFullOutSchema:
"""Test DocumentFullOut schema validation (document with items)."""
def test_document_full_out_empty_items(self):
"""DocumentFullOut works with an empty items list."""
doc = _make_document_mock()
doc.items = []
out = DocumentFullOut.model_validate(doc)
assert out.slug == "licence-g1"
assert out.items == []
def test_document_full_out_with_items(self):
"""DocumentFullOut includes items when present."""
doc_id = uuid.uuid4()
doc = _make_document_mock(doc_id=doc_id)
item1 = _make_item_mock(document_id=doc_id, position="1", sort_order=0)
item2 = _make_item_mock(document_id=doc_id, position="2", sort_order=1)
doc.items = [item1, item2]
out = DocumentFullOut.model_validate(doc)
assert len(out.items) == 2
assert out.items[0].position == "1"
assert out.items[1].position == "2"
# ---------------------------------------------------------------------------
# Tests: SanctuaryEntryOut schema serialization
# ---------------------------------------------------------------------------
class TestSanctuaryEntryOutSchema:
"""Test SanctuaryEntryOut schema validation."""
def test_sanctuary_entry_out_basic(self):
"""SanctuaryEntryOut validates from a mock entry."""
entry = _make_sanctuary_entry_mock()
out = SanctuaryEntryOut.model_validate(entry)
assert out.entry_type == "document"
assert out.content_hash == "abc123def456"
assert out.ipfs_cid == "QmTestCid123"
assert out.chain_tx_hash == "0xdeadbeef"
assert out.chain_block == 12345
def test_sanctuary_entry_out_without_ipfs(self):
"""SanctuaryEntryOut works when IPFS CID is None."""
entry = _make_sanctuary_entry_mock(ipfs_cid=None, chain_tx_hash=None)
out = SanctuaryEntryOut.model_validate(entry)
assert out.ipfs_cid is None
assert out.chain_tx_hash is None
assert out.chain_block is None
def test_sanctuary_entry_out_all_fields(self):
"""All expected fields are present in SanctuaryEntryOut."""
entry = _make_sanctuary_entry_mock()
out = SanctuaryEntryOut.model_validate(entry)
data = out.model_dump()
expected_fields = {
"id", "entry_type", "reference_id", "title",
"content_hash", "ipfs_cid", "chain_tx_hash",
"chain_block", "metadata_json", "created_at",
}
assert expected_fields.issubset(set(data.keys()))
def test_sanctuary_entry_types(self):
"""Different entry_type values are accepted."""
for entry_type in ("document", "decision", "vote_result"):
entry = _make_sanctuary_entry_mock(entry_type=entry_type)
out = SanctuaryEntryOut.model_validate(entry)
assert out.entry_type == entry_type
@@ -0,0 +1,270 @@
"""TDD — Moteur de qualification des décisions.
Source de vérité exécutable des règles métier du tunnel "Décider".
Règles testées :
R1 within_mandate → individual + consultation_avis
R2 within_mandate → aucune modalité de vote + consignation Observatoire
R4 2 ≤ affected_count ≤ small_group_max → individual recommandé, collectif disponible
R5 small_group_max < affected_count ≤ collective_wot_min → collective recommandé
R6 affected_count > collective_wot_min → collective recommandé (WoT applicable, non obligatoire)
R7 is_structural → recommend_onchain + raison explicite
R8 is_structural=False → recommend_onchain=False
GARDE-FOUS (invariants internes qui ne doivent jamais régresser) :
G1 decision_type est toujours dans l'enum autorisé
G2 individual n'expose jamais de modalités de vote
G3 collective expose au moins une modalité
G4 les seuils sont lus depuis QualificationConfig (configurables)
"""
from __future__ import annotations
import pytest
from app.engine.qualifier import (
DecisionType,
QualificationConfig,
QualificationInput,
QualificationResult,
qualify,
)
DEFAULT_CONFIG = QualificationConfig()
# ---------------------------------------------------------------------------
# R1 — within_mandate → individual + consultation_avis
# ---------------------------------------------------------------------------
def test_r1_within_mandate_gives_individual():
result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG)
assert result.decision_type == DecisionType.INDIVIDUAL
def test_r1_within_mandate_gives_consultation_avis():
result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG)
assert result.process == "consultation_avis"
def test_r1_within_mandate_overrides_large_affected_count():
"""Même si de nombreuses personnes sont concernées, un mandat impose individual."""
result = qualify(
QualificationInput(within_mandate=True, affected_count=500),
DEFAULT_CONFIG,
)
assert result.decision_type == DecisionType.INDIVIDUAL
assert result.process == "consultation_avis"
# ---------------------------------------------------------------------------
# R2 — within_mandate → aucune modalité de vote + consignation Observatoire
# ---------------------------------------------------------------------------
def test_r2_within_mandate_no_vote_modalities():
"""Le mandataire décide seul après consultation — pas de vote collectif."""
result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG)
assert result.recommended_modalities == []
def test_r2_within_mandate_records_in_observatory():
"""Une décision dans un mandat doit être consignée dans l'Observatoire."""
result = qualify(QualificationInput(within_mandate=True), DEFAULT_CONFIG)
assert result.record_in_observatory is True
def test_r2_out_of_mandate_does_not_force_observatory():
"""Hors mandat, la consignation dans l'Observatoire n'est pas imposée."""
result = qualify(
QualificationInput(within_mandate=False, affected_count=10),
DEFAULT_CONFIG,
)
assert result.record_in_observatory is False
# ---------------------------------------------------------------------------
# R4 — 2 ≤ affected_count ≤ small_group_max → individual recommandé
# ---------------------------------------------------------------------------
def test_r4_small_group_recommends_individual():
for count in range(2, DEFAULT_CONFIG.small_group_max + 1):
result = qualify(
QualificationInput(within_mandate=False, affected_count=count),
DEFAULT_CONFIG,
)
assert result.decision_type == DecisionType.INDIVIDUAL, (
f"affected_count={count} devrait recommander individual"
)
def test_r4_small_group_collective_is_available():
result = qualify(
QualificationInput(within_mandate=False, affected_count=3),
DEFAULT_CONFIG,
)
assert result.collective_available is True
def test_r4_small_group_confidence_is_recommended():
result = qualify(
QualificationInput(within_mandate=False, affected_count=3),
DEFAULT_CONFIG,
)
assert result.confidence == "recommended"
# ---------------------------------------------------------------------------
# R5 — small_group_max < affected_count ≤ collective_wot_min → collective recommandé
# ---------------------------------------------------------------------------
def test_r5_medium_group_recommends_collective():
mid = DEFAULT_CONFIG.small_group_max + 1
result = qualify(
QualificationInput(within_mandate=False, affected_count=mid),
DEFAULT_CONFIG,
)
assert result.decision_type == DecisionType.COLLECTIVE
def test_r5_medium_group_confidence_is_recommended():
mid = DEFAULT_CONFIG.small_group_max + 1
result = qualify(
QualificationInput(within_mandate=False, affected_count=mid),
DEFAULT_CONFIG,
)
assert result.confidence == "recommended"
# ---------------------------------------------------------------------------
# R6 — affected_count > collective_wot_min → collective recommandé (pas obligatoire)
# ---------------------------------------------------------------------------
def test_r6_large_group_recommends_collective():
result = qualify(
QualificationInput(within_mandate=False, affected_count=DEFAULT_CONFIG.collective_wot_min + 1),
DEFAULT_CONFIG,
)
assert result.decision_type == DecisionType.COLLECTIVE
def test_r6_large_group_confidence_is_recommended_not_required():
"""Au-delà du seuil WoT, le vote collectif est recommandé — pas imposé."""
result = qualify(
QualificationInput(within_mandate=False, affected_count=DEFAULT_CONFIG.collective_wot_min + 1),
DEFAULT_CONFIG,
)
assert result.confidence == "recommended"
def test_r6_large_group_includes_vote_wot_modality():
result = qualify(
QualificationInput(within_mandate=False, affected_count=DEFAULT_CONFIG.collective_wot_min + 1),
DEFAULT_CONFIG,
)
assert "vote_wot" in result.recommended_modalities
# ---------------------------------------------------------------------------
# R7 — is_structural=True → recommend_onchain + raison
# ---------------------------------------------------------------------------
def test_r7_structural_recommends_onchain():
result = qualify(
QualificationInput(within_mandate=False, affected_count=100, is_structural=True),
DEFAULT_CONFIG,
)
assert result.recommend_onchain is True
def test_r7_structural_provides_onchain_reason():
result = qualify(
QualificationInput(within_mandate=False, affected_count=100, is_structural=True),
DEFAULT_CONFIG,
)
assert result.onchain_reason is not None and len(result.onchain_reason) > 0
def test_r7_structural_within_mandate_can_also_recommend_onchain():
"""Même une décision dans un mandat peut être gravée si structurante."""
result = qualify(
QualificationInput(within_mandate=True, is_structural=True),
DEFAULT_CONFIG,
)
assert result.recommend_onchain is True
# ---------------------------------------------------------------------------
# R8 — is_structural=False → recommend_onchain=False
# ---------------------------------------------------------------------------
def test_r8_non_structural_never_recommends_onchain():
for count in [2, 3, 10, 100]:
result = qualify(
QualificationInput(within_mandate=False, affected_count=count, is_structural=False),
DEFAULT_CONFIG,
)
assert result.recommend_onchain is False, (
f"affected_count={count} non structurant : on-chain ne doit pas être proposé"
)
# ---------------------------------------------------------------------------
# GARDE-FOUS internes (régressions silencieuses)
# ---------------------------------------------------------------------------
def test_g1_decision_type_always_valid():
valid_types = set(DecisionType)
for inp in [
QualificationInput(within_mandate=True),
QualificationInput(within_mandate=False, affected_count=2),
QualificationInput(within_mandate=False, affected_count=10),
QualificationInput(within_mandate=False, affected_count=100),
]:
result = qualify(inp, DEFAULT_CONFIG)
assert result.decision_type in valid_types
def test_g2_individual_never_has_vote_modalities():
for inp in [
QualificationInput(within_mandate=True),
QualificationInput(within_mandate=False, affected_count=2),
QualificationInput(within_mandate=False, affected_count=3),
]:
result = qualify(inp, DEFAULT_CONFIG)
if result.decision_type == DecisionType.INDIVIDUAL:
assert result.recommended_modalities == [], (
f"Individual ne doit pas exposer de modalités : {inp}"
)
def test_g3_collective_has_at_least_one_modality():
result = qualify(
QualificationInput(within_mandate=False, affected_count=20),
DEFAULT_CONFIG,
)
assert result.decision_type == DecisionType.COLLECTIVE
assert len(result.recommended_modalities) >= 1
def test_g4_custom_config_overrides_thresholds():
"""Les seuils viennent de QualificationConfig — pas de constantes hardcodées."""
custom = QualificationConfig(small_group_max=2, collective_wot_min=10)
result = qualify(
QualificationInput(within_mandate=False, affected_count=3),
custom,
)
assert result.decision_type == DecisionType.COLLECTIVE
def test_g4_default_thresholds_are_stable():
cfg = QualificationConfig()
assert cfg.small_group_max == 5
assert cfg.collective_wot_min == 50
@@ -0,0 +1,172 @@
"""TDD — Service AI de cadrage des décisions (qualify/ai-chat).
Invariants testés :
A1 Premier appel (messages=[]) → retourne toujours 2 questions, done=False
A2 Les 2 questions couvrent réversibilité et urgence (ids stables)
A3 Deuxième appel (messages=[q+réponse]) → done=True, résultat qualifié
A4 Réponse "irréversible" → recommend_onchain conservé si is_structural
A5 Réponse "urgente" → raison "urgence" présente dans le résultat
A6 La qualification finale respecte les règles du moteur (R1/R2/R4/R5/R6)
A7 Sans contexte, les questions restent les mêmes (stub ne dépend pas du LLM)
"""
from __future__ import annotations
import pytest
from app.services.qualify_ai_service import (
AIFrameRequest,
AIMessage,
ai_frame,
)
DEFAULT_REQUEST = AIFrameRequest(
context="Révision du règlement intérieur de l'association",
within_mandate=False,
affected_count=20,
is_structural=False,
messages=[],
)
# ---------------------------------------------------------------------------
# A1 — Premier appel → 2 questions, done=False
# ---------------------------------------------------------------------------
def test_a1_first_call_returns_questions():
resp = ai_frame(DEFAULT_REQUEST)
assert resp.done is False
assert len(resp.questions) == 2
def test_a1_first_call_result_is_none():
resp = ai_frame(DEFAULT_REQUEST)
assert resp.result is None
# ---------------------------------------------------------------------------
# A2 — Questions couvrent réversibilité et urgence
# ---------------------------------------------------------------------------
def test_a2_questions_have_stable_ids():
resp = ai_frame(DEFAULT_REQUEST)
ids = {q.id for q in resp.questions}
assert "reversibility" in ids
assert "urgency" in ids
def test_a2_questions_have_options():
resp = ai_frame(DEFAULT_REQUEST)
for q in resp.questions:
assert len(q.options) >= 2, f"Question '{q.id}' doit avoir au moins 2 options"
# ---------------------------------------------------------------------------
# A3 — Deuxième appel (avec réponses) → done=True + résultat
# ---------------------------------------------------------------------------
def _make_second_request(reversibility_ans: str, urgency_ans: str, **kwargs) -> AIFrameRequest:
questions = ai_frame(DEFAULT_REQUEST).questions
messages = []
for q in questions:
messages.append(AIMessage(role="assistant", content=q.text))
# One user message bundling all answers
messages.append(AIMessage(
role="user",
content=f"reversibility:{reversibility_ans}|urgency:{urgency_ans}",
))
return AIFrameRequest(
**{**vars(DEFAULT_REQUEST), "messages": messages, **kwargs}
)
def test_a3_second_call_is_done():
req = _make_second_request("Difficilement", "Pas d'urgence")
resp = ai_frame(req)
assert resp.done is True
def test_a3_second_call_has_result():
req = _make_second_request("Difficilement", "Pas d'urgence")
resp = ai_frame(req)
assert resp.result is not None
assert resp.result.decision_type in ("individual", "collective")
# ---------------------------------------------------------------------------
# A4 — Irréversible + structurant → recommend_onchain
# ---------------------------------------------------------------------------
def test_a4_irreversible_structural_recommends_onchain():
req = _make_second_request(
"Non, c'est irréversible",
"Pas d'urgence",
is_structural=True,
)
resp = ai_frame(req)
assert resp.result is not None
assert resp.result.recommend_onchain is True
# ---------------------------------------------------------------------------
# A5 — Urgence → raison présente
# ---------------------------------------------------------------------------
def test_a5_urgent_adds_urgency_reason():
req = _make_second_request("Oui, facilement", "Urgente (< 1 semaine)")
resp = ai_frame(req)
assert resp.result is not None
reasons_text = " ".join(resp.result.reasons).lower()
assert "urgence" in reasons_text or "urgent" in reasons_text
# ---------------------------------------------------------------------------
# A6 — Résultat respecte les règles du moteur
# ---------------------------------------------------------------------------
def test_a6_within_mandate_gives_individual():
req = AIFrameRequest(
within_mandate=True,
affected_count=None,
messages=[
AIMessage(role="assistant", content="q"),
AIMessage(role="user", content="reversibility:Facilement|urgency:Pas d'urgence"),
],
)
resp = ai_frame(req)
assert resp.done is True
assert resp.result is not None
assert resp.result.decision_type == "individual"
assert resp.result.process == "consultation_avis"
def test_a6_large_group_gives_collective():
req = _make_second_request("Difficilement", "Pas d'urgence", affected_count=100)
resp = ai_frame(req)
assert resp.result is not None
assert resp.result.decision_type == "collective"
# ---------------------------------------------------------------------------
# A7 — Sans contexte, mêmes questions (stub ne dépend pas du LLM)
# ---------------------------------------------------------------------------
def test_a7_no_context_same_question_ids():
req_with = DEFAULT_REQUEST
req_without = AIFrameRequest(
context=None,
within_mandate=False,
affected_count=20,
messages=[],
)
ids_with = {q.id for q in ai_frame(req_with).questions}
ids_without = {q.id for q in ai_frame(req_without).questions}
assert ids_with == ids_without
@@ -0,0 +1,180 @@
"""Tests for WoT threshold formula, Smith threshold, and TechComm threshold.
Real-world reference case:
Vote Engagement Forgeron v2.0.0 (Feb 2026)
wot_size=7224, votes_for=97, votes_against=23, total=120
params M=50, B=0.1, G=0.2 => threshold=94 => adopted (97 >= 94)
"""
import math
import pytest
from app.engine.threshold import wot_threshold
from app.engine.smith_threshold import smith_threshold
from app.engine.techcomm_threshold import techcomm_threshold
# ---------------------------------------------------------------------------
# WoT threshold: real-world vote Forgeron
# ---------------------------------------------------------------------------
class TestWotThresholdForgeron:
"""Test with the actual Engagement Forgeron v2.0.0 vote numbers."""
def test_forgeron_vote_passes(self):
"""97 votes_for out of 120 total (wot=7224) must pass."""
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# With low participation (120/7224 ~ 1.66%), near-unanimity is required.
# The historical threshold was 94, and 97 >= 94.
assert 97 >= threshold
# The threshold should be high relative to total votes (inertia effect)
assert threshold > 60, f"Threshold {threshold} should be well above simple majority"
def test_forgeron_vote_threshold_value(self):
"""Verify the computed threshold is in a reasonable range."""
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# At ~1.66% participation, inertia should push threshold close to 78-95%
# of total votes. The exact value depends on the formula.
assert 80 <= threshold <= 120
# ---------------------------------------------------------------------------
# WoT threshold: low participation
# ---------------------------------------------------------------------------
class TestWotThresholdLowParticipation:
"""With very low participation, near-unanimity should be required."""
def test_ten_votes_out_of_7224(self):
"""10 voters out of 7224 => nearly all must vote 'for'."""
threshold = wot_threshold(
wot_size=7224,
total_votes=10,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# With participation ratio 10/7224 ~ 0.14%, threshold should be
# very close to total_votes (near-unanimity).
assert threshold >= 9, f"Expected near-unanimity but got threshold={threshold}"
assert threshold <= 10
# ---------------------------------------------------------------------------
# WoT threshold: high participation
# ---------------------------------------------------------------------------
class TestWotThresholdHighParticipation:
"""With high participation, threshold should approach simple majority M."""
def test_3000_votes_out_of_7224(self):
"""3000/7224 ~ 41.5% participation => threshold closer to 50%."""
threshold = wot_threshold(
wot_size=7224,
total_votes=3000,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# With ~42% participation, the inertia factor diminishes.
# threshold should be well below 90% of votes but above simple majority.
simple_majority = math.ceil(3000 * 0.5)
assert threshold >= simple_majority, (
f"Threshold {threshold} should be at least simple majority {simple_majority}"
)
# Should be noticeably less than near-unanimity
assert threshold < 2700, (
f"Threshold {threshold} should be much less than near-unanimity at high participation"
)
# ---------------------------------------------------------------------------
# WoT threshold: edge cases
# ---------------------------------------------------------------------------
class TestWotThresholdEdgeCases:
"""Edge-case behaviour."""
def test_zero_total_votes(self):
"""With zero votes, threshold is ceil(C + B^W)."""
threshold = wot_threshold(
wot_size=7224,
total_votes=0,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# B^W = 0.1^7224 is effectively 0
expected = math.ceil(0.0 + 0.1 ** 7224)
assert threshold == expected
def test_invalid_wot_size_zero(self):
with pytest.raises(ValueError, match="wot_size"):
wot_threshold(wot_size=0, total_votes=10)
def test_invalid_negative_votes(self):
with pytest.raises(ValueError, match="total_votes"):
wot_threshold(wot_size=100, total_votes=-1)
def test_invalid_majority_pct(self):
with pytest.raises(ValueError, match="majority_pct"):
wot_threshold(wot_size=100, total_votes=10, majority_pct=150)
# ---------------------------------------------------------------------------
# Smith threshold
# ---------------------------------------------------------------------------
class TestSmithThreshold:
"""Test Smith sub-WoT threshold: ceil(smith_size ^ S)."""
def test_smith_size_20_exponent_01(self):
"""smith_size=20, exponent=0.1 => ceil(20^0.1)."""
result = smith_threshold(smith_wot_size=20, exponent=0.1)
expected = math.ceil(20 ** 0.1)
assert result == expected
# 20^0.1 ~ 1.35, ceil => 2
assert result == 2
def test_smith_size_1(self):
"""smith_size=1 => ceil(1^0.1) = 1."""
assert smith_threshold(smith_wot_size=1, exponent=0.1) == 1
def test_smith_invalid(self):
with pytest.raises(ValueError):
smith_threshold(smith_wot_size=0)
# ---------------------------------------------------------------------------
# TechComm threshold
# ---------------------------------------------------------------------------
class TestTechcommThreshold:
"""Test TechComm threshold: ceil(cotec_size ^ T)."""
def test_cotec_size_5_exponent_01(self):
"""cotec_size=5, exponent=0.1 => ceil(5^0.1)."""
result = techcomm_threshold(cotec_size=5, exponent=0.1)
expected = math.ceil(5 ** 0.1)
assert result == expected
# 5^0.1 ~ 1.175, ceil => 2
assert result == 2
def test_cotec_size_1(self):
assert techcomm_threshold(cotec_size=1, exponent=0.1) == 1
def test_cotec_invalid(self):
with pytest.raises(ValueError):
techcomm_threshold(cotec_size=0)
+478
View File
@@ -0,0 +1,478 @@
"""Tests for vote service engine functions: threshold details, session logic, nuanced evaluation, and simulation.
All tests are pure unit tests that exercise the engine functions directly
without any database dependency.
Real-world reference case:
Vote Engagement Forgeron v2.0.0 (Feb 2026)
wot_size=7224, votes_for=97, votes_against=23, total=120
params M=50, B=0.1, G=0.2 => threshold=94 => adopted (97 >= 94)
"""
from __future__ import annotations
import math
import pytest
from app.engine.nuanced_vote import evaluate_nuanced
from app.engine.smith_threshold import smith_threshold
from app.engine.techcomm_threshold import techcomm_threshold
from app.engine.threshold import wot_threshold
# ---------------------------------------------------------------------------
# Threshold details computation: real Forgeron case (97/23/7224)
# ---------------------------------------------------------------------------
class TestThresholdDetailsForgeron:
"""Simulate the threshold details computation that the service would return
for the real Engagement Forgeron v2.0.0 vote."""
WOT_SIZE = 7224
VOTES_FOR = 97
VOTES_AGAINST = 23
TOTAL = 120
SMITH_SIZE = 20
SMITH_EXPONENT = 0.1
TECHCOMM_SIZE = 5
TECHCOMM_EXPONENT = 0.1
def _compute_details(self) -> dict:
"""Reproduce the threshold details logic from vote_service.get_threshold_details."""
wot_thresh = wot_threshold(
wot_size=self.WOT_SIZE,
total_votes=self.TOTAL,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
constant_base=0.0,
)
wot_passed = self.VOTES_FOR >= wot_thresh
smith_thresh = smith_threshold(self.SMITH_SIZE, self.SMITH_EXPONENT)
# Assume all smith members voted for
smith_votes_for = 5
smith_passed = smith_votes_for >= smith_thresh
techcomm_thresh = techcomm_threshold(self.TECHCOMM_SIZE, self.TECHCOMM_EXPONENT)
# Assume 2 techcomm members voted for
techcomm_votes_for = 2
techcomm_passed = techcomm_votes_for >= techcomm_thresh
overall_passed = wot_passed and smith_passed and techcomm_passed
participation_rate = self.TOTAL / self.WOT_SIZE
return {
"wot_threshold": wot_thresh,
"smith_threshold": smith_thresh,
"techcomm_threshold": techcomm_thresh,
"votes_for": self.VOTES_FOR,
"votes_against": self.VOTES_AGAINST,
"votes_total": self.TOTAL,
"wot_passed": wot_passed,
"smith_passed": smith_passed,
"techcomm_passed": techcomm_passed,
"overall_passed": overall_passed,
"participation_rate": round(participation_rate, 6),
"formula_params": {
"M": 50, "B": 0.1, "G": 0.2, "C": 0.0,
"S": self.SMITH_EXPONENT, "T": self.TECHCOMM_EXPONENT,
},
}
def test_wot_threshold_value(self):
"""WoT threshold for Forgeron vote should be in the expected range."""
details = self._compute_details()
assert 80 <= details["wot_threshold"] <= 120
# 97 votes_for must pass
assert details["wot_passed"] is True
def test_smith_threshold_value(self):
"""Smith threshold: ceil(20^0.1) = ceil(1.35) = 2."""
details = self._compute_details()
assert details["smith_threshold"] == 2
assert details["smith_passed"] is True
def test_techcomm_threshold_value(self):
"""TechComm threshold: ceil(5^0.1) = ceil(1.175) = 2."""
details = self._compute_details()
assert details["techcomm_threshold"] == 2
assert details["techcomm_passed"] is True
def test_overall_pass(self):
"""All three criteria pass => overall adopted."""
details = self._compute_details()
assert details["overall_passed"] is True
def test_participation_rate(self):
"""Participation rate for 120/7224 ~ 1.66%."""
details = self._compute_details()
expected = round(120 / 7224, 6)
assert details["participation_rate"] == expected
assert details["participation_rate"] < 0.02 # less than 2%
def test_formula_params_present(self):
"""All formula params must be present and correct."""
details = self._compute_details()
params = details["formula_params"]
assert params["M"] == 50
assert params["B"] == 0.1
assert params["G"] == 0.2
assert params["C"] == 0.0
assert params["S"] == 0.1
assert params["T"] == 0.1
# ---------------------------------------------------------------------------
# Close session behavior (engine-level logic)
# ---------------------------------------------------------------------------
class TestCloseSessionLogic:
"""Test the logic that would execute when a session is closed:
computing the final tally and determining adopted/rejected."""
def test_binary_vote_adopted(self):
"""97 for / 23 against out of 7224 WoT => adopted."""
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
adopted = 97 >= threshold
assert adopted is True
result = "adopted" if adopted else "rejected"
assert result == "adopted"
def test_binary_vote_rejected(self):
"""50 for / 70 against out of 7224 WoT => rejected."""
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
adopted = 50 >= threshold
assert adopted is False
result = "adopted" if adopted else "rejected"
assert result == "rejected"
def test_close_with_zero_votes(self):
"""Session with 0 votes => threshold is ~0 (B^W), effectively rejected
because 0 votes_for cannot meet even a tiny threshold."""
threshold = wot_threshold(
wot_size=7224,
total_votes=0,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# B^W = 0.1^7224 -> effectively 0, ceil(0) = 0
# But with 0 votes_for, 0 >= 0 is True
# This is a degenerate case; in practice sessions with 0 votes
# would be marked invalid
assert threshold == 0 or threshold == math.ceil(0.1 ** 7224)
def test_close_high_participation(self):
"""3000/7224 participating, 2500 for / 500 against => should pass.
At ~41.5% participation with G=0.2, the inertia factor is low,
so a strong supermajority should pass.
"""
threshold = wot_threshold(
wot_size=7224,
total_votes=3000,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# At ~41.5% participation, threshold is lower than near-unanimity
# but still above simple majority (1500)
assert threshold > 1500, f"Threshold {threshold} should be above simple majority"
assert threshold < 3000, f"Threshold {threshold} should be below total votes"
adopted = 2500 >= threshold
assert adopted is True
def test_close_barely_fails(self):
"""Use exact threshold to verify borderline rejection."""
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# votes_for = threshold - 1 => should fail
barely_fail = (threshold - 1) >= threshold
assert barely_fail is False
def test_close_smith_criterion_blocks(self):
"""Even if WoT passes, failing Smith criterion blocks adoption."""
threshold = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
wot_pass = 97 >= threshold
smith_required = smith_threshold(20, 0.1) # ceil(20^0.1) = 2
smith_ok = 1 >= smith_required # Only 1 smith voted -> fails
adopted = wot_pass and smith_ok
assert wot_pass is True
assert smith_ok is False
assert adopted is False
# ---------------------------------------------------------------------------
# Nuanced vote result evaluation
# ---------------------------------------------------------------------------
class TestNuancedVoteResult:
"""Test nuanced vote evaluation with per-level breakdown,
as would be returned by compute_result for nuanced votes."""
def test_nuanced_adopted_with_breakdown(self):
"""Standard nuanced vote that passes threshold and min_participants."""
votes = [5] * 30 + [4] * 20 + [3] * 15 + [2] * 5 + [1] * 3 + [0] * 2
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
# Total = 75, positive = 30+20+15 = 65
assert result["total"] == 75
assert result["positive_count"] == 65
assert result["positive_pct"] == pytest.approx(86.67, abs=0.1)
assert result["adopted"] is True
# Verify per-level breakdown
assert result["per_level_counts"][5] == 30
assert result["per_level_counts"][4] == 20
assert result["per_level_counts"][3] == 15
assert result["per_level_counts"][2] == 5
assert result["per_level_counts"][1] == 3
assert result["per_level_counts"][0] == 2
def test_nuanced_rejected_threshold_not_met(self):
"""Nuanced vote where positive percentage is below threshold."""
votes = [5] * 10 + [4] * 10 + [3] * 10 + [2] * 15 + [1] * 15 + [0] * 10
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
# Total = 70, positive = 10+10+10 = 30
assert result["total"] == 70
assert result["positive_count"] == 30
assert result["positive_pct"] == pytest.approx(42.86, abs=0.1)
assert result["threshold_met"] is False
assert result["min_participants_met"] is True
assert result["adopted"] is False
def test_nuanced_all_neutre(self):
"""All voters at level 3 (NEUTRE) still counts as positive."""
votes = [3] * 60
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
assert result["total"] == 60
assert result["positive_count"] == 60
assert result["positive_pct"] == 100.0
assert result["adopted"] is True
def test_nuanced_mixed_heavy_negative(self):
"""Majority at levels 0-2 => rejected."""
votes = [0] * 20 + [1] * 20 + [2] * 15 + [3] * 2 + [4] * 1 + [5] * 1
result = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
# Total = 59, positive = 2+1+1 = 4
assert result["total"] == 59
assert result["positive_count"] == 4
assert result["positive_pct"] < 10
assert result["adopted"] is False
def test_nuanced_breakdown_dict_structure(self):
"""Verify the breakdown structure matches what the service builds."""
votes = [5] * 20 + [4] * 20 + [3] * 19 + [2] * 5 + [1] * 3 + [0] * 2
evaluation = evaluate_nuanced(votes, threshold_pct=80, min_participants=59)
# Simulate the nuanced_breakdown dict the service builds
nuanced_breakdown = {
"per_level_counts": evaluation["per_level_counts"],
"positive_count": evaluation["positive_count"],
"positive_pct": evaluation["positive_pct"],
"threshold_met": evaluation["threshold_met"],
"min_participants_met": evaluation["min_participants_met"],
"threshold_pct": 80,
"min_participants": 59,
}
assert "per_level_counts" in nuanced_breakdown
assert nuanced_breakdown["threshold_pct"] == 80
assert nuanced_breakdown["min_participants"] == 59
assert nuanced_breakdown["positive_count"] == 59
assert nuanced_breakdown["threshold_met"] is True
# ---------------------------------------------------------------------------
# Simulation endpoint logic (pure engine functions)
# ---------------------------------------------------------------------------
class TestSimulationLogic:
"""Test the formula simulation that the /simulate endpoint performs,
exercising all three threshold engine functions together."""
def test_simulation_forgeron_case(self):
"""Simulate the Forgeron vote: wot=7224, total=120, M50 B.1 G.2."""
wot_thresh = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# Derived values (matching the simulate endpoint logic)
w = 7224
t = 120
m = 0.5
g = 0.2
participation_rate = t / w
turnout_ratio = min(t / w, 1.0)
inertia_factor = 1.0 - turnout_ratio ** g
required_ratio = m + (1.0 - m) * inertia_factor
assert 80 <= wot_thresh <= 120
assert participation_rate == pytest.approx(0.016611, abs=0.001)
assert 0 < inertia_factor < 1
assert required_ratio > m # Inertia pushes above simple majority
def test_simulation_full_participation(self):
"""At 100% participation, threshold approaches simple majority."""
wot_size = 100
total = 100
wot_thresh = wot_threshold(
wot_size=wot_size,
total_votes=total,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# At full participation, turnout_ratio = 1.0
# inertia_factor = 1.0 - 1.0^0.2 = 0
# required_ratio = 0.5 + 0.5*0 = 0.5
# threshold = 0 + 0.1^100 + 0.5 * 100 = ~50
assert wot_thresh == math.ceil(0.1 ** 100 + 0.5 * 100)
assert wot_thresh == 50
def test_simulation_with_smith_and_techcomm(self):
"""Simulate with all three criteria."""
wot_thresh = wot_threshold(
wot_size=7224,
total_votes=120,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
smith_thresh = smith_threshold(smith_wot_size=20, exponent=0.1)
techcomm_thresh = techcomm_threshold(cotec_size=5, exponent=0.1)
assert wot_thresh > 0
assert smith_thresh == 2 # ceil(20^0.1) = ceil(1.35) = 2
assert techcomm_thresh == 2 # ceil(5^0.1) = ceil(1.175) = 2
def test_simulation_varying_majority(self):
"""Higher majority_pct increases the threshold at full participation."""
thresh_50 = wot_threshold(wot_size=100, total_votes=100, majority_pct=50)
thresh_66 = wot_threshold(wot_size=100, total_votes=100, majority_pct=66)
thresh_80 = wot_threshold(wot_size=100, total_votes=100, majority_pct=80)
assert thresh_50 < thresh_66 < thresh_80
def test_simulation_varying_gradient(self):
"""Higher gradient exponent means more inertia at partial participation.
With T/W < 1, a higher G makes (T/W)^G smaller, so
inertia_factor = 1 - (T/W)^G becomes larger, raising the threshold.
At 50% participation: (0.5)^0.2 ~ 0.87, (0.5)^1.0 = 0.5.
"""
thresh_g02 = wot_threshold(
wot_size=200, total_votes=100,
majority_pct=50, gradient_exponent=0.2,
)
thresh_g10 = wot_threshold(
wot_size=200, total_votes=100,
majority_pct=50, gradient_exponent=1.0,
)
# Higher G = more inertia at partial participation = higher threshold
assert thresh_g10 > thresh_g02
def test_simulation_zero_votes(self):
"""Zero votes produces a minimal threshold."""
thresh = wot_threshold(
wot_size=7224,
total_votes=0,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# B^W = 0.1^7224 ~ 0, ceil(0) = 0
assert thresh == 0
def test_simulation_small_wot(self):
"""Small WoT size (e.g. 5 members)."""
thresh = wot_threshold(
wot_size=5,
total_votes=3,
majority_pct=50,
base_exponent=0.1,
gradient_exponent=0.2,
)
# With 3/5 participation, should require more than simple majority of 3
assert thresh >= 2
assert thresh <= 3
def test_simulation_result_structure(self):
"""Verify the simulation produces all expected values for the API response."""
w = 7224
t = 120
m_pct = 50
m = m_pct / 100.0
b = 0.1
g = 0.2
c = 0.0
wot_thresh = wot_threshold(
wot_size=w, total_votes=t,
majority_pct=m_pct, base_exponent=b,
gradient_exponent=g, constant_base=c,
)
smith_thresh = smith_threshold(smith_wot_size=20, exponent=0.1)
techcomm_thresh = techcomm_threshold(cotec_size=5, exponent=0.1)
participation_rate = t / w
turnout_ratio = min(t / w, 1.0)
inertia_factor = 1.0 - turnout_ratio ** g
required_ratio = m + (1.0 - m) * inertia_factor
# Build the simulation result dict (matching FormulaSimulationResult schema)
result = {
"wot_threshold": wot_thresh,
"smith_threshold": smith_thresh,
"techcomm_threshold": techcomm_thresh,
"participation_rate": round(participation_rate, 6),
"required_ratio": round(required_ratio, 6),
"inertia_factor": round(inertia_factor, 6),
}
assert isinstance(result["wot_threshold"], int)
assert isinstance(result["smith_threshold"], int)
assert isinstance(result["techcomm_threshold"], int)
assert 0 < result["participation_rate"] < 1
assert 0 < result["required_ratio"] <= 1
assert 0 < result["inertia_factor"] < 1