dodil/erp-brick-projects
tests/test_item_v3.py
# ruff: noqa: F811 (the `planned` fixture is imported from test_doors and used by name)
"""Item process v3: whoever submitted the work records its acceptance only with pj:item:accept_own; the item's
work_ref; what a screen reads (readiness of an item and of a project, the project's stage)."""
import json
from datetime import datetime
import pytest
from conftest import LEAD, LEAD2, LEAD_PERMS, MANAGER, NOBODY, SCHEDULER, WORKER, WORKER_PERMS, api, person, tool
from test_doors import I1, I2, P, done_by_will, item, path, planned, step # noqa: F401
from brick_projects.application import use_cases
from brick_projects.domain.errors import RuleBroken
from brick_projects.domain.item import Item
from brick_projects.domain.values import Settings
MON = datetime(2026, 11, 2, 9, 0)
SENTENCE = "You submitted this work, so someone else records its acceptance"
SOLO = person("sol@agency.example", LEAD_PERMS + WORKER_PERMS + ["pj:item:accept_own"], ["firm_of_one"])
WILD = person("wil@agency.example", ["pj:*:*"], []) # a wildcard does NOT reach pj:item:accept_own
def ident(sub, email, permissions):
"""The same caller as the gateway may present them: a sub with or without an email."""
out = person(email or sub, permissions)
out["X-Dodil-User"] = json.dumps({"sub": sub, **({"email": email} if email else {})})
return out
def submitted_by(who, item_id=I1):
assert step("start", item_id, who)[0] == 200
assert step("submit", item_id, who, {"evidence": "https://x.example/proof"})[0] == 200
@pytest.fixture
def mine(planned):
"""Lena leads the project and submits item 1 herself."""
submitted_by(LEAD)
def test_the_default_roles_do_not_hold_accept_own():
from conftest import ROLES
assert not any("pj:item:accept_own" in perms for perms in ROLES.values())
def test_the_submitter_cannot_record_acceptance_without_accept_own(mine):
status, out = step("accept", I1, LEAD)
assert (status, out["detail"]) == (409, SENTENCE)
assert item(I1)[1]["state"] == "in_review"
def test_a_manager_who_did_not_submit_records_it(mine):
assert step("accept", I1, MANAGER)[1]["state"] == "done"
def test_accept_own_lets_the_submitter_record_it():
who = SOLO
sale = {"source_kind": "contract", "source_id": "C-200", "bp_id": "1",
"lines": [{"line_id": "L1", "title": "T", "quantity": 1, "billing_kind": "one_off",
"billing_plan": "single"}]}
assert api("POST", "/plans", who, sale)[0] == 200
mine_id = "pj:contract:C-200|L1|1"
submitted_by(who, mine_id)
assert step("accept", mine_id, who)[1]["state"] == "done"
def test_the_domain_gate_covers_both_acceptances():
it = Item("i", project_id="p", billing_kind="one_off", billing_plan="single", evidence="e", evidence_kind="link")
it.start()
it.submit(now=MON, by="sol@agency.example")
with pytest.raises(RuleBroken, match=SENTENCE):
it.accept(now=MON, settings=Settings(), by="sol@agency.example")
it.accept(now=MON, settings=Settings(), by="sol@agency.example", may_accept_own=True)
assert it.state == "done"
deemed = Item("d", evidence="e", evidence_kind="link")
deemed.start()
deemed.submit(now=MON, by="sol@agency.example")
from erp_platform.work import WorkCalendar
deemed.warned_at = datetime(2026, 9, 1)
cal = WorkCalendar(working_weekdays=frozenset(range(5)), holidays=frozenset())
with pytest.raises(RuleBroken, match=SENTENCE):
deemed.accept_deemed(now=MON, cal=cal, settings=Settings(), by="sol@agency.example")
deemed.accept_deemed(now=MON, cal=cal, settings=Settings(), by="sol@agency.example", may_accept_own=True)
assert deemed.approved_how == "deemed"
# ── work_ref ──────────────────────────────────────────────────────────────────────────────────
def test_work_ref_is_https_only_optional_and_not_evidence(planned):
ok = step("assign", I1, LEAD, {"work_ref": "https://git.example/team/issues/12"})
assert ok[0] == 200 and ok[1]["work_ref"] == "https://git.example/team/issues/12"
for bad in ("http://git.example/x", "ftp://git.example/x", "git.example/x", "https://", "https://a b.example",
"javascript:alert(1)"):
status, out = step("assign", I1, LEAD, {"work_ref": bad})
assert status == 409 and "https link" in out["detail"], bad
assert item(I1)[1]["work_ref"] == "https://git.example/team/issues/12"
assert item(I1)[1]["evidence"] == "" # separate from evidence
assert step("assign", I1, WORKER, {"work_ref": "https://git.example/x"})[0] == 403 # pj:item:write
assert step("assign", I1, LEAD, {"work_ref": ""})[1]["work_ref"] == ""
# ── readiness ─────────────────────────────────────────────────────────────────────────────────
def readiness(item_id, who):
return api("GET", path("/items/{i}/readiness", i=item_id), who)
def by_key(out):
return {t["transition"]: t for t in out["transitions"]}
def test_item_readiness_names_each_gate_in_words(planned):
done_by_will(I1)
status, out = readiness(I1, LEAD)
assert status == 200
accept = by_key(out)["accept"]
assert accept["ready"] and all(set(g) == {"gate", "met", "reason", "fix"} for g in accept["gates"])
deemed = by_key(out)["accept_deemed"]
assert not deemed["ready"]
unmet = {g["gate"]: g for g in deemed["gates"] if not g["met"]}
assert "The client was warned" in unmet and unmet["The client was warned"]["fix"]
def test_the_submitter_sees_the_unmet_gate_and_accept_own_meets_it(mine):
gate = {g["gate"]: g for g in by_key(readiness(I1, LEAD)[1])["accept"]["gates"]}["Someone else records it"]
assert gate["met"] is False and gate["reason"] == SENTENCE and "accept_own" in gate["fix"]
def test_readiness_omits_what_the_caller_may_never_do(planned):
done_by_will(I1)
assert "accept" not in by_key(readiness(I1, WORKER)[1]) # no pj:item:accept
assert "descope" not in by_key(readiness(I1, WORKER)[1])
assert by_key(readiness(I1, MANAGER)[1]).keys() >= {"accept", "descope", "request_changes"}
planned_view = readiness(I2, LEAD)[1]
assert set(by_key(planned_view)) == {"start", "descope"}
def test_readiness_follows_the_record_scope(planned):
assert readiness(I1, LEAD2)[0] == 404
assert readiness("nope", LEAD)[0] == 404
assert api("GET", path("/projects/{p}/readiness", p=P), LEAD2)[0] == 404
assert readiness(I1, NOBODY)[0] == 403
def test_project_readiness_and_stage(planned):
def project(who=LEAD):
return api("GET", path("/projects/{p}/readiness", p=P), who)[1]
assert project()["stage"] == "plan"
assert set(by_key(project())) == {"cancel_project"} # the lead never reconciles
assert project(SCHEDULER)["transitions"] == [] # planned: nothing to reconcile yet
assert step("start", I1, LEAD)[0] == 200
assert by_key(project(SCHEDULER)).keys() == {"reconcile"} # the scheduler never cancels
window = {g["gate"]: g for g in by_key(project(SCHEDULER))["reconcile"]["gates"]}["The window has closed"]
assert window["met"] is False and "2026-12-31" in window["reason"]
assert api("GET", path("/projects/{p}", p=P), LEAD)[1]["stage"] == "work"
assert step("submit", I1, LEAD, {"evidence": "https://x.example/p"})[0] == 200
assert api("GET", path("/projects/{p}", p=P), LEAD)[1]["stage"] == "review"
assert step("accept", I1, MANAGER)[0] == 200
assert api("GET", path("/projects/{p}", p=P), LEAD)[1]["stage"] == "work"
from conftest import clock
clock("2027-01-05T09:00")
assert api("GET", path("/projects/{p}", p=P), LEAD)[1]["stage"] == "reconcile"
closed = {g["gate"]: g for g in by_key(project(SCHEDULER))["reconcile"]["gates"]}["The window has closed"]
assert closed["met"] is True
def test_both_readiness_answers_are_tools_too(planned):
done_by_will(I1)
assert tool(MANAGER, "item_readiness", {"item_id": I1})[1] == readiness(I1, MANAGER)[1]
assert tool(LEAD, "project_readiness", {"project_id": P})[1] == api(
"GET", path("/projects/{p}/readiness", p=P), LEAD)[1]
assert tool(LEAD2, "item_readiness", {"item_id": I1})[0] == "refused"
assert use_cases.PJ.name == "pj"
def test_a_lead_without_the_permission_is_not_offered_the_transition():
"""Leading the project is not enough: the permission decides, so a lead who may only plan and work sees no
descope, and an in-review item shows them no acceptance."""
limited = person("lia@agency.example", ["pj:project:read", "pj:project:plan", "pj:item:read", "pj:item:work"], [])
sale = {"source_kind": "contract", "source_id": "C-300", "bp_id": "1",
"lines": [{"line_id": "L1", "title": "T", "quantity": 1, "billing_kind": "one_off",
"billing_plan": "single"}]}
assert api("POST", "/plans", limited, sale)[0] == 200
mine_id = "pj:contract:C-300|L1|1"
assert set(by_key(readiness(mine_id, limited)[1])) == {"start"}
submitted_by(limited, mine_id)
assert set(by_key(readiness(mine_id, limited)[1])) == set()
def sale(source_id):
return {"source_kind": "contract", "source_id": source_id, "bp_id": "1",
"lines": [{"line_id": "L1", "title": "T", "quantity": 1, "billing_kind": "one_off",
"billing_plan": "single"}]}
def test_a_wildcard_does_not_reach_accept_own():
assert api("POST", "/plans", WILD, sale("C-400"))[0] == 200
mine_id = "pj:contract:C-400|L1|1"
submitted_by(WILD, mine_id)
status, out = step("accept", mine_id, WILD)
assert (status, out["detail"]) == (409, SENTENCE)
@pytest.mark.parametrize("accepting", [("u-max", None), ("u-max", "other@agency.example"), ("u-other", "max@agency.example")],
ids=["same sub, no email", "same sub, other email", "same email, other sub"])
def test_the_submitter_is_the_same_person_under_either_identity(accepting):
perms = LEAD_PERMS + WORKER_PERMS + ["pj:project:manage", "pj:project:read_all"]
first = ident("u-max", "max@agency.example", perms)
assert api("POST", "/plans", first, sale("C-500"))[0] == 200
mine_id = "pj:contract:C-500|L1|1"
submitted_by(first, mine_id)
status, out = step("accept", mine_id, ident(accepting[0], accepting[1], perms))
assert (status, out["detail"]) == (409, SENTENCE)
gate = {g["gate"]: g for g in by_key(readiness(mine_id, ident(*accepting, perms))[1])["accept"]["gates"]}
assert gate["Someone else records it"]["met"] is False
stranger = ident("u-else", "else@agency.example", perms)
assert step("accept", mine_id, stranger)[1]["state"] == "done"
@pytest.mark.parametrize("bad", ["https://user:pw@git.example/x", "https://git.example\\@evil.example/x",
"https://git.example/\u202ex", "https://git.example/a\u200bb",
"https://git.example:99999/x", "https://git.example:abc/x",
"https://git.example/\\evil", "https://git.example/" + "a" * 500])
def test_work_ref_refuses_tricks(bad, planned):
assert step("assign", I1, LEAD, {"work_ref": bad})[0] in (409, 422)
assert step("assign", I1, LEAD, {"work_ref": "https://git.example:8443/t/issues/1"})[0] == 200
def test_stage_is_what_the_caller_sees_and_reads_sensibly_when_cancelled_or_all_descoped(planned):
from conftest import clock
for i in (I1, I2):
assert step("descope", i, LEAD, {"reason": "not needed"})[0] == 200
stage = lambda who: api("GET", path("/projects/{p}", p=P), who)[1]["stage"] # noqa: E731
assert stage(LEAD) == "plan"
clock("2027-01-05T09:00")
assert stage(LEAD) == "reconcile"
assert api("GET", path("/projects/{p}/readiness", p=P), LEAD)[1]["stage"] == "reconcile"
clock("2026-11-02T09:00")
assert project_cancel(LEAD)[0] == 200
assert stage(LEAD) == "cancelled"
def project_cancel(who):
return api("POST", path("/projects/{p}/cancel", p=P), who, {"reason": "no longer wanted"})
def test_stage_ignores_items_the_caller_does_not_see(planned):
done_by_will(I1)
stage = lambda who: api("GET", path("/projects/{p}", p=P), who)[1]["stage"] # noqa: E731
assert stage(MANAGER) == "review"
other = person("zed@agency.example", WORKER_PERMS + ["pj:project:read"])
assert step("assign", I2, LEAD, {"owner": "zed@agency.example"})[0] == 200
assert stage(other) == "plan" # zed sees only their planned item
def test_a_wildcard_does_not_reach_reconcile_but_the_scheduler_does(planned):
assert step("start", I1, LEAD)[0] == 200
status, out = api("POST", path("/projects/{p}/reconcile", p=P), WILD)
assert status == 403 and "pj:project:reconcile" in out["detail"]
assert "reconcile" not in by_key(api("GET", path("/projects/{p}/readiness", p=P), WILD)[1])
assert api("POST", path("/projects/{p}/reconcile", p=P), SCHEDULER)[0] == 200