dodil/erp-brick-lead
tests/test_rules.py
"""The lead rules, with no database — named for the CRM spec's stories (modules/crm/module.yaml)."""
from datetime import datetime, timedelta, timezone
import pytest
from brick_lead.domain.lead import (DEFAULT_POLICY, DIM, Email, Lead, LeadIncomplete, LeadKey, LeadRuleBroken,
ProfileVector, Rule, Score, ScoringPolicy, Signal, company_domain,
cosine_distance, lead_id_for, normalise_domain, normalise_size,
plan_import, profile, score, signal_id_for)
NOW = datetime(2026, 10, 17, 12, 0, tzinfo=timezone.utc)
GATE = ScoringPolicy(60, 30, ())
def points(n):
return Score(n, 0, n, ())
def person(email="ana@acme.com", **kw):
return Lead.new("tom@co.example", email=email, name=kw.pop("name", "Ana Berg"),
company=kw.pop("company", "Acme"), **kw)
# ── the key: derived, so a re-import lands on the same row ─────────────────────────────────
def test_crm13_a_persons_lead_id_is_derived_from_their_email_however_it_is_typed():
assert lead_id_for(" Ana@ACME.com ") == lead_id_for("ana@acme.com") == person().lead_id
assert lead_id_for("ana@acme.com") != lead_id_for(domain="acme.com") # a person is not their company
assert lead_id_for(domain="https://www.Acme.com/about") == lead_id_for(domain="acme.com")
def test_a_free_mail_address_says_nothing_about_the_company():
assert company_domain("ana@gmail.com") == ""
assert company_domain("ana@acme.com") == "acme.com"
assert company_domain("ana@gmail.com", "acme.com") == "acme.com"
assert person("ana@gmail.com").domain == ""
@pytest.mark.parametrize("bad", ["", "ana", "ana@", "ana@acme", "a b@acme.com", "ana@@acme.com"])
def test_an_address_that_is_not_one_is_refused(bad):
with pytest.raises(LeadIncomplete):
lead_id_for(bad)
def test_size_takes_a_band_or_a_head_count():
assert [normalise_size(v) for v in ("51-200", "250", "1,200 staff", "5000+", "12000", "3")] == \
["51-200", "201-1000", "1001-5000", "5000+", "5000+", "1-10"]
with pytest.raises(LeadIncomplete):
normalise_size("lots")
# ── CRM-13: an import plan ─────────────────────────────────────────────────────────────────
CSV = """Name,Email,Company,Website,Country,Size,Badge colour
Ana Berg,ana@acme.com,Acme,acme.com,se,250,blue
No Email,,Nobody Ltd,,,,
Bad Address,not-an-address,Bad Co,,,,
Ana Again,ANA@acme.com,Acme,,,,
Ben Ode,ben@beta.io,Beta,,uk,lots,
"""
def test_crm13_rows_with_no_email_are_reported_by_row_number_never_dropped():
plan = plan_import(CSV, {}, "mia@co.example")
assert [l.email for l in plan.create] == ["ana@acme.com", "ben@beta.io"]
assert plan.skipped == [{"row": 3, "reason": "no email"},
{"row": 4, "reason": "'not-an-address' is not an email address"},
{"row": 5, "reason": "same email as row 2"}]
assert plan.warnings[0]["row"] == 6 and "size_band not imported" in plan.warnings[0]["warning"]
assert plan.ignored_columns == ["Badge colour"]
ana = plan.create[0]
assert (ana.domain, ana.country, ana.size_band, ana.owner, ana.source) == \
("acme.com", "SE", "201-1000", "mia@co.example", "import")
assert plan.create[1].country == "GB"
def test_crm13_importing_twice_each_person_exists_once_and_a_blank_cell_erases_nothing():
first = plan_import(CSV, {}, "mia@co.example")
existing = {l.lead_id: l for l in first.create}
existing[lead_id_for("ana@acme.com")].start_working()
again = plan_import(CSV, existing, "someone@else.example")
assert (len(again.create), len(again.update), len(again.unchanged)) == (0, 0, 2)
sparser = "email,company,industry\nana@acme.com,,SaaS\n"
third = plan_import(sparser, existing, "x@y.z")
(lead, changed), = third.update
assert changed == ["industry"] and lead.industry == "software"
assert (lead.company, lead.status, lead.owner) == ("Acme", "working", "mia@co.example")
def test_a_semicolon_export_and_an_owner_column_are_understood():
plan = plan_import("e-mail;Full name;Owner\nana@acme.com;Ana;sofia@co.example\n", {}, "mia@co.example")
assert plan.create[0].owner == "sofia@co.example" and plan.create[0].name == "Ana"
def test_a_file_without_an_email_column_is_refused_whole():
with pytest.raises(LeadIncomplete, match="email column"):
plan_import("name,company\nAna,Acme\n", {}, "m@co")
# ── the workflow ───────────────────────────────────────────────────────────────────────────
def test_the_lead_workflow_new_working_qualified_and_its_gate():
lead = person()
with pytest.raises(LeadRuleBroken, match="start working it first"):
lead.qualify(points(90), GATE, 'tom')
assert lead.start_working() and not lead.start_working()
with pytest.raises(LeadRuleBroken, match="score is 59, below the 60"):
lead.qualify(points(59), GATE, 'tom')
assert lead.qualify(points(60), GATE, 'tom') and lead.status == "qualified"
with pytest.raises(LeadRuleBroken, match="cannot be disqualified"):
lead.disqualify("no budget")
def test_disqualifying_needs_a_reason():
lead = person()
with pytest.raises(LeadIncomplete, match="needs a reason"):
lead.disqualify(" ")
assert lead.disqualify("no budget") and not lead.disqualify("no budget")
with pytest.raises(LeadRuleBroken):
lead.start_working()
def test_crm03_conversion_is_kept_step_by_step_and_completes_once():
lead = person()
lead.start_working()
lead.qualify(points(70), GATE, 'tom')
lead.ready_to_convert()
assert not lead.record_step(bp_id=54848654143250435)
assert not lead.record_step(contact_id="C-1")
assert lead.status == "qualified" # part-converted is still qualified
assert lead.record_step(opportunity_id="O-1", at="2026-10-17T12:00:00+00:00", by="tom@co.example")
assert (lead.status, lead.bp_id) == ("converted", 54848654143250435)
assert not lead.record_step(opportunity_id="O-1") # again: nothing changes
with pytest.raises(LeadRuleBroken):
lead.describe(company="Acme AB")
def test_crm03_converting_needs_the_company_domain_and_the_persons_email():
lead = person("ana@gmail.com")
lead.start_working()
lead.qualify(points(70), GATE, 'tom')
with pytest.raises(LeadIncomplete, match="web domain"):
lead.ready_to_convert()
company = Lead.new("tom@co.example", domain="acme.com", company="Acme")
company.start_working()
company.qualify(points(70), GATE, 'tom')
with pytest.raises(LeadIncomplete, match="email"):
company.ready_to_convert()
def test_a_step_cannot_silently_replace_what_a_previous_step_produced():
lead = person()
lead.start_working()
lead.qualify(points(70), GATE, 'tom')
lead.record_step(bp_id=1)
with pytest.raises(LeadRuleBroken, match="already 1"):
lead.record_step(bp_id=2)
# ── CRM-10: the score is derived ───────────────────────────────────────────────────────────
POLICY = ScoringPolicy(60, 30, (Rule("fit", "industry", "software", 20), Rule("fit", "region", "nordics", 10),
Rule("signal", "email_opened_pricing", "", 25, 2),
Rule("signal", "unsubscribed", "", -30, 1)))
def test_crm10_a_pricing_email_opened_an_hour_ago_counts_and_says_so():
lead = person(industry="SaaS", country="SE")
cold = score(lead, [], POLICY, NOW)
hot = score(lead, [("email_opened_pricing", NOW - timedelta(hours=1))], POLICY, NOW)
assert (cold.score, hot.score) == (30, 55)
assert hot.reasons == ("industry=software +20", "region=nordics +10", "email_opened_pricing ×1 +25")
def test_crm10_the_same_inputs_always_give_the_same_score():
sig = [("email_opened_pricing", NOW - timedelta(days=1))] * 5
assert score(person(), sig, POLICY, NOW) == score(person(), sig, POLICY, NOW)
assert score(person(), sig, POLICY, NOW).engagement_points == 50 # capped at max_count 2
def test_signals_outside_the_window_or_in_the_future_do_not_count_and_scores_stay_0_to_100():
lead = person()
old = [("email_opened_pricing", NOW - timedelta(days=31)), ("email_opened_pricing", NOW + timedelta(hours=1))]
assert score(lead, old, POLICY, NOW).score == 0
assert score(lead, [("unsubscribed", NOW)], POLICY, NOW).score == 0 # floored, not negative
rich = ScoringPolicy(60, 30, (Rule("signal", "demo_requested", "", 100, 1), Rule("fit", "source", "manual", 50)))
assert score(Lead.new("t@co", email="a@b.co", source="manual"), [("demo_requested", NOW)], rich, NOW).score == 100
def test_a_policy_is_validated_as_a_whole():
with pytest.raises(LeadIncomplete):
ScoringPolicy(101, 30, ())
with pytest.raises(LeadIncomplete, match="same thing"):
ScoringPolicy(60, 30, (Rule("signal", "x_y", "", 1), Rule("signal", "x_y", "", 2)))
assert DEFAULT_POLICY.qualify_threshold == 60
def test_the_same_event_reported_twice_is_one_signal():
assert signal_id_for("email_opened", NOW) == signal_id_for("email_opened", NOW)
assert signal_id_for("email_opened", NOW) != signal_id_for("email_opened", NOW + timedelta(seconds=1))
# ── CRM-14: the profile ────────────────────────────────────────────────────────────────────
def test_crm14_a_profile_is_a_deterministic_unit_vector_of_fit():
p = profile("software", "51-200", "SE")
assert len(p) == DIM == 31 and abs(sum(x * x for x in p) - 1) < 1e-5
assert p == profile("SaaS", "120", "se")
assert cosine_distance(p, p) < 1e-9
assert profile("", "", "") != [0.0] * DIM # never a zero vector
def test_crm14_same_industry_and_a_neighbouring_size_is_nearer_than_another_industry():
won = profile("software", "51-200", "SE")
near = cosine_distance(won, profile("software", "201-1000", "NO"))
far = cosine_distance(won, profile("retail", "51-200", "SE"))
farther = cosine_distance(won, profile("retail", "5000+", "US"))
assert near < far < farther
def test_normalise_domain_takes_what_people_paste():
assert normalise_domain("HTTPS://www.Acme.co.uk:443/contact?x=1") == "acme.co.uk"
with pytest.raises(LeadIncomplete):
normalise_domain("not a domain")
# ── value objects and domain events ────────────────────────────────────────────────────────
def test_value_objects_are_equal_by_value_in_their_one_normal_form():
assert Email(" Ana@ACME.com ") == Email("ana@acme.com") and Email("ana@acme.com").domain == "acme.com"
assert LeadKey.of("ana@acme.com") == LeadKey("person", "ana@acme.com")
assert LeadKey.of(domain="www.acme.com").kind == "company"
assert Signal("Email-Opened", NOW).signal_id == Signal("email_opened", NOW).signal_id
assert ProfileVector.of("software", "51-200", "SE").distance(ProfileVector.of("SaaS", "99", "se")) < 1e-9
with pytest.raises(LeadIncomplete):
Signal("email_opened", NOW.replace(tzinfo=None))
def test_the_aggregate_raises_past_tense_events_for_what_happened():
lead = person()
lead.start_working()
lead.qualify(points(75), GATE, "tom")
lead.qualify(points(75), GATE, "tom") # no change, no event
lead.record_step(bp_id=1, detail="partner 1")
lead.record_step(contact_id="C-1", detail="contact C-1")
lead.record_step(opportunity_id="O-1", detail="deal O-1")
assert [e.name for e in lead.events] == ["added", "work started", "qualified", "conversion step recorded",
"conversion step recorded", "converted"]
assert "score 75 ≥ 60" in lead.events[2].detail
def test_the_policy_decides_qualification_not_the_caller():
assert GATE.qualifies(points(60)) and not GATE.qualifies(points(59))
assert DEFAULT_POLICY.weighs("email_opened_pricing") and not DEFAULT_POLICY.weighs("shoe_size")