dodil/erp-brick-pipeline
tests/test_rules.py
"""The pipeline rules, with no database — named for the CRM spec's stories (modules/crm/module.yaml)."""
from decimal import Decimal
import pytest
from erp_platform.money import Money
from brick_pipeline.domain.pipeline import (DEFAULT_PIPELINE as P, EvidenceMissing, Opportunity,
Pipeline, PipelineRuleBroken, RiskPolicy, Stage, forecast,
period_of)
def deal(owner="tom@co.example", amount="10000", close="2026-11-15", stage=None):
o = Opportunity.open("O-1", 4242, "Acme renewal", owner, "EUR", P, amount=amount, close_date=close)
if stage:
o.stage = stage
return o
# ── CRM-04: stages are data; probability and category follow the stage ─────────────────────
def test_crm04_moving_a_deal_takes_the_new_stages_probability_and_category():
o = deal()
assert (o.stage, o.probability_for(P), o.category(P)) == ("discovery", Decimal("10.00"), "pipeline")
o.move("proposal", P, has_accepted_quote=None)
assert (o.probability_for(P), o.category(P)) == (Decimal("50.00"), "best_case")
assert o.weighted(P) == Money.of("5000", "EUR")
def test_a_deals_own_probability_survives_a_move_but_not_closing():
o = deal()
o.set_probability(Decimal("35"), P)
o.move("negotiation", P, has_accepted_quote=None)
assert o.probability_for(P) == Decimal("35.00")
o.move("closed_won", P, has_accepted_quote=True)
assert o.probability_for(P) == Decimal("100.00")
def test_the_customers_own_stages_are_used():
mine = Pipeline((Stage("lead", 1, 5, "pipeline"), Stage("won", 2, 100, "closed", True, True),
Stage("lost", 3, 0, "omitted", is_closed=True)))
o = Opportunity.open("O-2", 1, "x", "a@b", "SEK", mine)
assert o.stage == "lead"
with pytest.raises(PipelineRuleBroken, match="no stage 'proposal'"):
o.move("proposal", mine, has_accepted_quote=None)
@pytest.mark.parametrize("stages, error", [
((Stage("a", 1, 5, "pipeline"),), "exactly one won stage"),
((Stage("a", 1, 5, "pipeline"), Stage("w", 2, 100, "closed", True, True)), "lost stage"),
((Stage("w", 2, 100, "closed", True, True), Stage("l", 3, 0, "omitted", is_closed=True)), "open stage"),
])
def test_a_pipeline_must_be_able_to_win_lose_and_be_open(stages, error):
with pytest.raises(PipelineRuleBroken, match=error):
Pipeline(stages)
# ── won needs evidence; lost needs a reason ────────────────────────────────────────────────
def test_won_needs_an_accepted_quote_when_quotes_are_kept():
o = deal()
with pytest.raises(EvidenceMissing, match="accepted a quote"):
o.move("closed_won", P, has_accepted_quote=False)
assert o.stage == "discovery"
def test_h1_won_takes_the_accepted_quotes_total_whatever_was_typed():
"""A won deal's amount is what the customer accepted. A typed estimate (10 000) left in place made the
forecast's 'won' a number nobody signed (the CRM audit, 2026-10-17)."""
blank = Opportunity.open("O-3", 1, "x", "a@b", "EUR", P)
blank.move("closed_won", P, has_accepted_quote=True, accepted_total=Money.of("9000", "EUR"),
won_quotes=("Q-1",))
assert blank.amount == Money.of("9000", "EUR") and blank.won_quote_ids == ("Q-1",)
typed = deal(amount="10000")
typed.move("closed_won", P, has_accepted_quote=True, accepted_total=Money.of("9000", "EUR"))
assert typed.amount == Money.of("9000", "EUR")
with pytest.raises(PipelineRuleBroken, match="currency"):
deal().move("closed_won", P, has_accepted_quote=True, accepted_total=Money.of("9000", "SEK"))
def test_lost_needs_a_reason_from_the_list():
o = deal()
with pytest.raises(EvidenceMissing, match="price, no_budget"):
o.move("closed_lost", P, has_accepted_quote=None, loss_reason="they went quiet")
o.move("closed_lost", P, has_accepted_quote=None, loss_reason="Price")
assert o.loss_reason == "price"
def test_h3_a_closed_deal_is_final_until_it_is_reopened_with_a_reason():
for stage, kw in (("closed_won", {"has_accepted_quote": True}),
("closed_lost", {"has_accepted_quote": None, "loss_reason": "price"})):
o = deal()
o.move(stage, P, **kw)
for to in ("discovery", "closed_won" if stage == "closed_lost" else "closed_lost"):
with pytest.raises(PipelineRuleBroken, match="reopen"):
o.move(to, P, has_accepted_quote=True, loss_reason="price")
assert o.move(stage, P, has_accepted_quote=None) == stage # the stage it is in: nothing changes
with pytest.raises(EvidenceMissing, match="reason"):
o.reopen("proposal", " ", P)
with pytest.raises(PipelineRuleBroken, match="open stage"):
o.reopen("closed_won", "the order was cancelled", P)
o.reopen("proposal", "the order was cancelled", P)
assert (o.stage, o.loss_reason, o.is_open(P)) == ("proposal", None, True)
with pytest.raises(PipelineRuleBroken, match="is open"):
deal().reopen("proposal", "why", P)
def test_h3_reopening_a_won_deal_voids_the_quotes_that_won_it():
o = deal()
o.move("closed_won", P, has_accepted_quote=True, accepted_total=Money.of("3000", "EUR"),
won_quotes=("Q-1", "Q-2"))
o.reopen("negotiation", "the customer cancelled the order", P)
assert o.won_quote_ids == () and o.voided_quote_ids == ("Q-1", "Q-2")
assert o.counts_as_evidence("Q-3") and not o.counts_as_evidence("Q-1")
def test_h3_a_closed_deals_close_date_and_amount_are_frozen():
won, lost = deal(), deal()
won.move("closed_won", P, has_accepted_quote=True)
lost.move("closed_lost", P, has_accepted_quote=None, loss_reason="price")
for o in (won, lost):
with pytest.raises(PipelineRuleBroken, match="closed"):
o.set_close_date("2020-01-01", P)
with pytest.raises(PipelineRuleBroken):
o.set_amount("1", P)
o.rename("still renameable")
blank = Opportunity.open("O-9", 1, "x", "a@b", "EUR", P)
blank.move("closed_lost", P, has_accepted_quote=None, loss_reason="timing")
with pytest.raises(PipelineRuleBroken, match="closed"):
blank.set_amount("5", P)
def test_a_won_deals_amount_is_the_contracts():
o = deal()
o.move("closed_won", P, has_accepted_quote=True)
with pytest.raises(PipelineRuleBroken, match="contract"):
o.set_amount("1", P)
o.rename("Acme renewal 2027") # the name is not the contract's
# ── CRM-09: the forecast, derived ──────────────────────────────────────────────────────────
def test_crm09_open_weighted_and_committed_per_rep_and_period():
a = deal("tom@co.example", "10000", "2026-11-15", "proposal") # 50%
b = deal("tom@co.example", "20000", "2026-12-01", "negotiation") # 75%, commit
c = deal("tom@co.example", "5000", "2026-10-10", "closed_won")
d = deal("sofia@co.example", "8000", "2027-01-20", "discovery") # next quarter
e = deal("tom@co.example", None, "2026-11-01", "proposal") # no amount: unforecastable
rows = {(r["owner"], r["period"]): r for r in forecast([a, b, c, d, e], P)}
tom = rows[("tom@co.example", "2026-Q4")]
assert (tom["open"], tom["weighted"], tom["commit"], tom["won"]) == (
Decimal("30000.00"), Decimal("20000.00"), Decimal("25000.00"), Decimal("5000.00"))
assert tom["unforecastable"] == 1
assert rows[("sofia@co.example", "2027-Q1")]["weighted"] == Decimal("800.00")
def test_crm09_a_manager_overrides_a_category_with_a_reason():
o = deal(stage="proposal")
with pytest.raises(EvidenceMissing, match="reason"):
o.override_category("commit", " ", P)
o.override_category("commit", "signed LOI on Friday", P)
assert o.category(P) == "commit"
o.move("closed_lost", P, has_accepted_quote=None, loss_reason="competitor")
assert o.category(P) == "omitted" and o.category_override is None
def test_currencies_are_never_summed():
eur, sek = deal(amount="100"), deal(amount="100")
sek.currency, sek.amount = "SEK", Money.of("100", "SEK")
assert {r["currency"] for r in forecast([eur, sek], P)} == {"EUR", "SEK"}
def test_periods_are_calendar_quarters():
assert [period_of(d) for d in ("2026-01-01", "2026-03-31", "2026-04-01", "2026-12-31")] == \
["2026-Q1", "2026-Q1", "2026-Q2", "2026-Q4"]
# ── CRM-16: gone quiet, derived ────────────────────────────────────────────────────────────
def test_crm16_a_late_deal_quiet_for_21_days_is_at_risk_with_the_reason():
o = deal(stage="negotiation", close="2027-03-01")
o.last_activity_at = "2026-10-01T09:00:00+00:00"
assert o.risk(P, RiskPolicy(), today="2026-10-21") is None # 20 days: not yet
assert o.risk(P, RiskPolicy(), today="2026-10-22") == "nothing logged for 21 days in negotiation"
def test_an_early_or_closed_deal_is_never_at_risk():
early = deal(stage="discovery")
early.last_activity_at = "2026-01-01"
assert early.risk(P, RiskPolicy(), today="2026-10-22") is None
won = deal(stage="closed_won")
won.last_activity_at = "2026-01-01"
assert won.risk(P, RiskPolicy(), today="2026-10-22") is None
def test_a_passed_close_date_is_a_risk_too():
o = deal(stage="proposal", close="2026-10-01")
o.last_activity_at = "2026-10-20"
assert "has passed" in o.risk(P, RiskPolicy(), today="2026-10-22")