Restructure Engagement Forgeron + fix GenesisBlock + InertiaSlider
- Seed: restructure Engagement Forgeron (51→59 items) avec 3 nouvelles sections: Engagements fondamentaux (EF1-EF3), Engagements techniques (ET1-ET3), Qualification (Q0-Q1) liée au protocole Embarquement - Seed: ajout protocole Embarquement Forgeron (5 jalons: candidature, miroir, évaluation, certification Smith, mise en ligne) - GenesisBlock: fix lisibilité — fond mood-surface teinté accent au lieu de mood-text inversé, texte mood-aware au lieu de rgba blanc hardcodé - InertiaSlider: mini affiche "Inertie" sous le curseur, compact en width:fit-content pour s'adapter au label - Frontend: ajout section qualification dans SECTION_META/SECTION_ORDER - Pages, composants et tests des sprints précédents Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
856
backend/app/tests/test_doc_protocol_integration.py
Normal file
856
backend/app/tests/test_doc_protocol_integration.py
Normal file
@@ -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
|
||||
@@ -43,6 +43,7 @@ def _make_document_mock(
|
||||
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 = []
|
||||
@@ -68,6 +69,9 @@ def _make_item_mock(
|
||||
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
|
||||
@@ -135,8 +139,8 @@ class TestDocumentOutSchema:
|
||||
|
||||
expected_fields = {
|
||||
"id", "slug", "title", "doc_type", "version", "status",
|
||||
"description", "ipfs_cid", "chain_anchor", "created_at",
|
||||
"updated_at", "items_count",
|
||||
"description", "ipfs_cid", "chain_anchor", "genesis_json",
|
||||
"created_at", "updated_at", "items_count",
|
||||
}
|
||||
assert expected_fields.issubset(set(data.keys()))
|
||||
|
||||
|
||||
1033
backend/seed.py
1033
backend/seed.py
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user