dodil/erp-brick-projects

tests/test_doors.py

"""The use cases through both doors, on SQLite: every use case is a route AND a tool (bar the one only a job
calls), the same answer through the API and MCP, who may (403) and which rows (404, never 403), the review clock
under a moving business clock, the handovers to invoicing and the order, and the client's read-only link.

A contributor (Will) does the work; the project's lead (Lena) records the client's answer — the person who submitted
work never records its acceptance."""
import pytest
from conftest import (
    FINANCE,
    LEAD,
    LEAD2,
    MANAGER,
    MIXED,
    NAMED_MANAGER,
    NAMED_SYSTEM,
    NOBODY,
    PUBLIC_BASE,
    SCHEDULER,
    WILL,
    WORKER,
    api,
    clock,
    plan,
    tool,
    tools,
)
from erp_platform import public

from brick_projects.application import use_cases
from brick_projects.infrastructure import seams

P = "pj:contract:C-100"
I1, I2 = f"{P}|L1|1", f"{P}|L1|2"
ROUTES = [
    ("list_projects", "GET /projects"), ("get_project", "GET /projects/{project_id}"),
    ("get_item", "GET /items/{item_id}"), ("board", "GET /board"), ("acceptance", "GET /acceptance"),
    ("item_readiness", "GET /items/{item_id}/readiness"), ("project_readiness", "GET /projects/{project_id}/readiness"),
    ("list_states", "GET /states"), ("list_settings", "GET /settings"), ("plan_from_contract", "POST /plans"),
    ("assign_item", "POST /items/{item_id}/assign"), ("start", "POST /items/{item_id}/start"),
    ("block", "POST /items/{item_id}/block"), ("unblock", "POST /items/{item_id}/unblock"),
    ("attach_evidence", "POST /items/{item_id}/evidence"), ("submit", "POST /items/{item_id}/submit"),
    ("send_warning", "POST /items/{item_id}/send-warning"),
    ("request_changes", "POST /items/{item_id}/request-changes"), ("descope", "POST /items/{item_id}/descope"),
    ("accept", "POST /items/{item_id}/accept"), ("accept_deemed", "POST /items/{item_id}/accept-deemed"),
    ("retry_handover", "POST /items/{item_id}/retry-handover"), ("raise_makegood", "POST /items/{item_id}/makegood"),
    ("reconcile", "POST /projects/{project_id}/reconcile"), ("cancel_project", "POST /projects/{project_id}/cancel"),
    ("ai_draft_plan", "GET /projects/{project_id}/ai/draft-plan"),
    ("ai_chase_brief", "GET /items/{item_id}/ai/chase-brief"),
    ("ai_flag_shortfall", "GET /projects/{project_id}/ai/flag-shortfall"),
    ("issue_client_link", "POST /projects/{project_id}/client-link"),
    ("list_client_links", "GET /projects/{project_id}/client-links"),
    ("revoke_client_link", "POST /projects/{project_id}/client-links/{link_id}/revoke"),
    ("rename_state", "POST /states/{state_id}"), ("set_setting", "POST /settings/{setting_key}"),
]


def path(template, **ids):
    from urllib.parse import quote
    for k, v in ids.items():
        template = template.replace("{" + k + "}", quote(v, safe=""))
    return template


def item(item_id, who=LEAD):
    return api("GET", path("/items/{i}", i=item_id), who)


def step(name, item_id, who=LEAD, body=None):
    return api("POST", path(f"/items/{{i}}/{name}", i=item_id), who, body or {})


def project_cmd(name, who=LEAD, body=None, pid=P):
    return api("POST", path(f"/projects/{{p}}/{name}", p=pid), who, body or {})


def done_by_will(item_id, evidence="https://staging.client.example/phase-1"):
    """Will is given the item, does it and submits it with its evidence."""
    step("assign", item_id, LEAD, {"owner": WILL})
    assert step("start", item_id, WORKER)[0] == 200
    assert step("submit", item_id, WORKER, {"evidence": evidence})[0] == 200


@pytest.fixture
def planned():
    status, out = plan()
    assert status == 200, out
    return out


class FakeOrders:
    """The sales-order brick as seen through its seam: up or down, and every delivery it was told about."""
    installed = True

    def __init__(self, up=True):
        self.up, self.told = up, []

    def post(self, path, body, actor=""):
        class Reply:
            def __init__(self, ok, status=200, reason=""):
                self.ok, self.status, self.reason = ok, status, reason

            def __bool__(self):
                return self.ok
        if not self.up:
            return Reply(False, 503, "orders are down")
        self.told.append((path, body["line_no"], body["delivery_ref"]))
        return Reply(True)


def test_every_use_case_is_a_route_and_a_tool_except_the_daily_pass():
    status, spec = api("GET", "/openapi.json", LEAD)
    assert status == 200
    for name, route in ROUTES + [("reconcile_due", "POST /projects/reconcile-due")]:
        method, p = route.split(" ", 1)
        assert method.lower() in spec["paths"].get(p, {}), route
    names = set(tools(LEAD))
    assert {n for n, _ in ROUTES} <= names and "reconcile_due" not in names


# ── PJ-01 / PJ-20 through both doors ────────────────────────────────────────────────────────

def test_pj_20_planning_through_both_doors_and_again_creates_nothing(planned):
    assert planned["created"] == 2
    assert [i["item_id"] for i in planned["project"]["items"]] == [I1, I2]
    assert planned["project"]["bp_id"] == "7311"                         # a 64-bit id leaves as a string
    again = tool(LEAD, "plan_from_contract", {"source_kind": "contract", "source_id": "C-100", "bp_id": "7311",
                                              "lines": MIXED, "starts_on": "2026-11-01", "ends_on": "2026-12-31"})
    assert again[0] == "ok" and again[1]["created"] == 0
    assert again[1]["project"] == api("GET", path("/projects/{p}", p=P), LEAD)[1]


def test_pj_20_the_same_sale_for_another_customer_is_refused(planned):
    status, out = plan(bp_id="9999")
    assert status == 409 and "another customer" in out["detail"]


def test_pj_01_a_sale_of_only_monthly_lines_opens_nothing():
    status, out = plan(lines=[MIXED[1]], source_id="C-200")
    assert status == 200 and out == {"project": None, "created": 0,
                                     "reason": "nothing in this sale is delivered as work"}


def test_pj_20_lines_are_bounded_and_named():
    assert plan(lines=[{"line_id": "L1", "quantity": 2, "colour": "red"}])[0] == 422
    assert plan(lines=[{"line_id": "L1", "quantity": 10**9}])[0] == 422
    assert plan(lines=[])[0] == 422


# ── who sees, who may ───────────────────────────────────────────────────────────────────────

def test_no_permission_is_403_and_writes_nothing():
    assert plan(who=NOBODY)[0] == 403
    assert api("GET", "/projects", LEAD)[1] == {"projects": []}


def test_another_leads_project_is_404_never_403(planned):
    assert api("GET", path("/projects/{p}", p=P), LEAD2)[0] == 404
    assert item(I1, LEAD2)[0] == 404
    assert api("GET", "/projects", LEAD2)[1] == {"projects": []}


def test_finance_and_the_manager_see_every_project_and_finance_changes_nothing(planned):
    for who in (FINANCE, MANAGER):
        assert api("GET", path("/projects/{p}", p=P), who)[0] == 200
    assert step("start", I1, FINANCE)[0] == 403


def test_a_role_name_grants_nothing(planned):
    """Only permissions decide (the brick is generic): a person whose pool role is called "manager" or "system" but
    who holds a viewer's permissions sees no other lead's project and can close none."""
    for who in (NAMED_MANAGER, NAMED_SYSTEM):
        assert api("GET", path("/projects/{p}", p=P), who)[0] == 404
        assert api("POST", "/projects/reconcile-due", who, {})[0] == 403


def test_a_contributor_sees_only_their_items_and_their_projects_headline(planned):
    assert item(I1, WORKER)[0] == 404
    step("assign", I1, LEAD, {"owner": WILL})
    assert item(I1, WORKER)[0] == 200 and item(I2, WORKER)[0] == 404
    status, project = api("GET", path("/projects/{p}", p=P), person_with_read(WORKER))
    assert status == 200 and [i["item_id"] for i in project["items"]] == [I1]
    assert step("start", I1, WORKER)[1]["state"] == "in_progress"


def person_with_read(headers):
    import json

    from conftest import WORKER_PERMS, person
    who = json.loads(headers["X-Dodil-User"])
    return person(who["email"], WORKER_PERMS + ["pj:project:read"], who["app_roles"])


def test_owning_one_item_in_another_leads_project_gives_no_say_over_it(planned):
    """AUDIT #2: Leo owns one item in Lena's project. He may do that item's work; he may not cancel or reconcile
    the project, give the client a link, re-plan or descope its work, or record the client's answer."""
    step("assign", I1, LEAD, {"owner": "leo@agency.example"})
    assert step("start", I1, LEAD2)[0] == 200
    for name, body in (("cancel", {"reason": "mine now"}), ("client-link", {"email": "kim@client.example"})):
        status, out = project_cmd(name, LEAD2, body)
        assert status == 403 and "only the project's lead or a manager" in out["detail"], name
    for name, body in (("assign", {"title": "x"}), ("descope", {"reason": "x"}), ("send-warning", {}),
                       ("accept", {"evidence": "https://x.example"})):
        assert step(name, I1, LEAD2, body)[0] == 403, name
    assert api("GET", path("/projects/{p}", p=P), LEAD)[1]["state"] == "active"


def test_whoever_submitted_the_work_does_not_record_its_acceptance(planned):
    """AUDIT #2: separation of duties — Lena submits her own work; she cannot accept it; a manager can."""
    step("start", I1, LEAD)
    step("submit", I1, LEAD, {"evidence": "https://x.example/1"})
    status, out = step("accept", I1, LEAD)
    assert status == 409 and "You submitted this work" in out["detail"]
    assert step("accept", I1, MANAGER)[1]["state"] == "done"


def test_the_daily_pass_is_the_schedulers_or_a_managers(planned):
    assert api("POST", "/projects/reconcile-due", LEAD, {})[0] == 403
    assert api("POST", "/projects/reconcile-due", SCHEDULER, {})[0] == 200


def test_a_lead_does_not_close_their_own_project_early(planned):
    """The signed process gives reconcile to the system, when the window ends (a probe finding, 0.9.1)."""
    status, out = project_cmd("reconcile", LEAD)
    assert status == 403 and "pj:project:reconcile" in out["detail"]


def test_the_scheduler_does_not_take_a_leads_steps(planned):
    """The scheduler plans and reconciles; cancelling or re-planning is the lead's (a probe finding). It sees every
    project (pj:project:read_all), so it is told 403, not 404."""
    assert project_cmd("cancel", SCHEDULER, {"reason": "x"})[0] == 403
    assert step("descope", I1, SCHEDULER, {"reason": "x"})[0] == 403


def test_a_planner_that_is_a_machine_leads_nothing(planned):
    """The scheduler may plan a sale; the project it plans is led by the owner it names, and with no owner named it
    is the scheduler's own record — still without a lead's permissions, so it cannot cancel it."""
    status, out = plan(who=SCHEDULER, source_id="C-900")
    assert status == 200
    assert project_cmd("cancel", SCHEDULER, {"reason": "x"}, pid="pj:contract:C-900")[0] == 403


# ── a stale screen is refused ───────────────────────────────────────────────────────────────

def test_a_stale_version_is_409_and_changes_nothing(planned):
    v = item(I1)[1]["version"]
    step("start", I1, LEAD, {"version": v})
    status, out = step("assign", I1, LEAD, {"title": "Phase one", "version": v})
    assert status == 409 and "changed since you opened it" in out["detail"]
    assert item(I1)[1]["title"] == "Implementation phase (1 of 2)"


# ── the whole walk, with the clock: PJ-02, PJ-03, PJ-04, PJ-11, PJ-22 ───────────────────────

def test_the_walk_from_start_to_deemed_and_billing_pending_without_invoicing(planned):
    assert not seams.INVOICING.installed
    step("assign", I1, LEAD, {"owner": WILL})
    assert step("start", I1, WORKER)[1]["state"] == "in_progress"
    assert api("GET", path("/projects/{p}", p=P), LEAD)[1]["state"] == "active"
    status, out = step("block", I1, WORKER, {"reason": " "})
    assert status == 409 and out["detail"] == "say what you are waiting for"
    step("block", I1, WORKER, {"reason": "waiting for brief"})
    clock("2026-11-06T09:00")
    assert step("unblock", I1, WORKER)[1]["blocked_days"] == 4
    clock("2026-11-09T09:00")
    status, out = step("accept", I1)
    assert status == 409 and "not in review" in out["detail"]
    step("submit", I1, WORKER)
    assert step("accept", I1)[1]["detail"] == "Attach the evidence first: a link, a file or a sign-off"
    step("evidence", I1, WORKER, {"evidence": "https://staging.client.example/phase-1", "kind": "link"})
    clock("2026-11-19T09:00")                                   # 8 working days after 9 Nov
    clock_view = item(I1)[1]["review_clock"]
    assert (clock_view["working_days_left"], clock_view["warning_due"]) == (2, False)
    assert step("send-warning", I1)[0] == 409
    clock("2026-11-24T09:00")
    assert item(I1)[1]["review_clock"]["warning_due"] is True
    step("send-warning", I1)
    assert step("accept-deemed", I1)[0] == 409                  # the notice runs until 1 Dec
    clock("2026-12-01T09:00")
    status, done = step("accept-deemed", I1)
    assert status == 200 and (done["state"], done["approved_how"]) == ("done", "deemed")
    assert done["billing_status"] == "pending"                  # invoicing is not composed here
    kinds = [h["kind"] for h in item(I1)[1]["history"]]
    assert kinds == ["plan_from_contract", "assign", "start", "block", "unblock", "submit", "attach_evidence",
                     "send_warning", "accept_deemed"]


def test_pj_22_with_invoicing_present_it_is_asked_once(planned, monkeypatch):
    asked = []
    monkeypatch.setattr(seams, "raise_for_acceptance",
                        lambda it, actor: asked.append(f"acc:{it.item_id}") or seams.Asked(True))
    done_by_will(I1)
    status, done = step("accept", I1)
    assert status == 200 and done["billing_status"] == "requested"
    assert step("accept", I1)[1]["billing_status"] == "requested"         # a repeat asks nothing
    assert step("retry-handover", I1)[1]["billing_status"] == "requested"
    assert asked == [f"acc:{I1}"]


def test_pj_22_pending_billing_is_finished_by_a_retry(planned, monkeypatch):
    done_by_will(I1)
    assert step("accept", I1)[1]["billing_status"] == "pending"
    monkeypatch.setattr(seams, "raise_for_acceptance", lambda it, actor: seams.Asked(True))
    assert step("retry-handover", I1)[1]["billing_status"] == "requested"


def test_an_accept_already_saved_is_not_failed_by_a_racing_save(planned, monkeypatch, engine):
    """AUDIT #6: invoicing said yes, and just as its "requested" is being saved, another request saves the item.
    The acceptance is already committed, so the status is recorded on the latest version — the caller gets their
    answer, never a 409 or 500 that invites a second acceptance."""
    from brick_projects.infrastructure import tables
    monkeypatch.setattr(seams, "raise_for_acceptance", lambda it, actor: seams.Asked(True))
    real_save, raced = tables.ItemStore.save, []

    def save(self, agg, expected_version):
        if agg.billing_status == "requested" and not raced:
            raced.append(True)                           # someone else's save lands first, on its own connection
            with engine.begin() as conn:
                other = tables.Repository(conn)
                latest, v = other.items.get(agg.item_id)
                latest.events.append(type("Touch", (), {"subject_id": agg.item_id, "kind": "touch",
                                                        "from_state": "done", "to_state": "done", "by": "x",
                                                        "detail": ""})())
                real_save(other.items, latest, v)
        return real_save(self, agg, expected_version)
    monkeypatch.setattr(tables.ItemStore, "save", save)
    done_by_will(I1)
    status, done = step("accept", I1)
    assert raced and status == 200 and done["billing_status"] == "requested"
    assert item(I1)[1]["billing_status"] == "requested"


# ── the order is told about accepted work only, and never loses the news (AUDIT #1, #4) ──────

ORDER_LINES = [{"line_id": "SO-L1", "title": "Installation", "quantity": 2, "billing_kind": "one_off",
                "billing_plan": "single"}]


def order_plan():
    status, out = api("POST", "/plans", LEAD, {"source_kind": "sales_order", "source_id": "SO-7", "bp_id": "7311",
                                               "lines": ORDER_LINES, "ends_on": "2026-12-31"})
    assert status == 200, out
    return "pj:sales_order:SO-7|SO-L1|1", "pj:sales_order:SO-7|SO-L1|2"


def test_a_retry_tells_the_order_nothing_about_work_not_yet_accepted(monkeypatch):
    orders = FakeOrders()
    monkeypatch.setattr(seams, "ORDERS", orders)
    first, _ = order_plan()
    out = step("retry-handover", first)[1]
    assert (out["delivery_status"], orders.told) == ("not_reported", [])


def test_accepted_order_work_is_reported_once_with_its_line(monkeypatch):
    orders = FakeOrders()
    monkeypatch.setattr(seams, "ORDERS", orders)
    first, _ = order_plan()
    done_by_will(first)
    assert step("accept", first)[1]["delivery_status"] == "reported"
    step("retry-handover", first)
    assert orders.told == [("/orders/SO-7/deliveries", "SO-L1", first)]


def test_an_order_that_is_down_keeps_the_delivery_pending_until_a_retry(monkeypatch):
    orders = FakeOrders(up=False)
    monkeypatch.setattr(seams, "ORDERS", orders)
    first, _ = order_plan()
    done_by_will(first)
    assert step("accept", first)[1]["delivery_status"] == "pending"
    orders.up = True
    assert step("retry-handover", first)[1]["delivery_status"] == "reported"
    assert step("retry-handover", first)[1]["delivery_status"] == "reported"
    assert len(orders.told) == 1


def test_a_makegood_is_never_reported_to_the_order(monkeypatch):
    orders = FakeOrders()
    monkeypatch.setattr(seams, "ORDERS", orders)
    first, second = order_plan()
    done_by_will(first)
    step("accept", first)
    clock("2027-01-01T06:00")
    project_cmd("reconcile", SCHEDULER, pid="pj:sales_order:SO-7")
    mg = step("makegood", second)[1]["item_id"]
    done_by_will(mg)
    assert step("accept", mg)[1]["delivery_status"] == "not_reported"
    assert [t[2] for t in orders.told] == [first]


# ── a closed or cancelled project's work does not move (AUDIT #3) ──────────────────────────

def test_a_cancelled_projects_work_does_not_move_or_bill(planned):
    done_by_will(I1)
    project_cmd("cancel", body={"reason": "client withdrew"})
    status, out = step("accept", I1)
    assert status == 409 and "cancelled" in out["detail"]
    assert item(I1)[1]["billing_status"] == "not_billed"
    assert project_cmd("client-link", body={"email": "kim@client.example"})[0] == 409


def test_a_short_item_is_not_finished_after_its_makegood(planned):
    done_by_will(I2)
    clock("2027-01-01T06:00")
    project_cmd("reconcile", SCHEDULER)
    assert step("makegood", I2)[0] == 200
    status, out = step("accept", I2)
    assert status == 409 and "closed and reconciled" in out["detail"]
    assert item(I2)[1]["billing_status"] == "not_billed"


# ── PJ-07, PJ-10, PJ-12, PJ-23 ──────────────────────────────────────────────────────────────

def test_reconcile_then_one_makegood(planned):
    done_by_will(I1)
    step("accept", I1)
    assert step("makegood", I2)[0] == 409                                    # not reconciled yet
    clock("2027-01-01T06:00")
    status, out = api("POST", "/projects/reconcile-due", SCHEDULER, {})
    assert status == 200 and out == {"reconciled": [{"project_id": P, "short": 1}]}
    assert api("POST", "/projects/reconcile-due", SCHEDULER, {})[1] == {"reconciled": []}
    project = api("GET", path("/projects/{p}", p=P), LEAD)[1]
    assert project["state"] == "closed"
    assert project["reconciliation"]["lines"] == [{"line_id": "L1", "sold": 2, "delivered": 1, "short": 1,
                                                   "over": 0, "descoped": 0}]
    assert [i["short"] for i in project["items"]] == [False, True]
    status, mg = step("makegood", I2)
    assert status == 200 and (mg["source"], mg["line_id"], mg["makegood_for"]) == ("makegood", "", I2)
    status, out = step("makegood", I2)
    assert status == 409 and "already being made good" in out["detail"]


def test_pj_23_acceptance_per_line_leaves_descoped_work_out(planned):
    """AUDIT #10: a descoped item is neither owed nor accepted; the line is accepted when what is left is."""
    done_by_will(I1)
    step("accept", I1)
    step("descope", I2, LEAD, {"reason": "replaced by change order CO-3"})
    lines = api("GET", "/acceptance", LEAD, {"source_kind": "contract", "source_id": "C-100"})[1]["lines"]
    assert [(x["line_id"], x["items"], x["accepted"], x["descoped"], x["all_accepted"]) for x in lines] == \
        [("L1", 1, 1, 1, True)]


def test_pj_10_descoped_is_counted_as_descoped(planned):
    step("descope", I2, LEAD, {"reason": "replaced by change order CO-3"})
    step("start", I1)
    clock("2027-01-01T06:00")
    project = project_cmd("reconcile", SCHEDULER)[1]
    assert project["reconciliation"]["lines"][0]["descoped"] == 1
    assert project["reconciliation"]["lines"][0]["short"] == 1


# ── PJ-08: the board ────────────────────────────────────────────────────────────────────────

def test_pj_08_the_board_puts_overdue_first_and_says_what_blocked_work_waits_on(planned):
    step("assign", I1, LEAD, {"due_on": "2026-10-30"})
    step("assign", I2, LEAD, {"due_on": "2026-11-04"})
    step("start", I2)
    step("block", I2, LEAD, {"reason": "waiting for assets"})
    board = api("GET", "/board", LEAD)[1]
    assert [r["item_id"] for r in board["overdue"]] == [I1]
    assert [(r["item_id"], r["waiting_on"]) for r in board["this_week"]] == [(I2, "waiting for assets")]
    assert api("GET", "/board", LEAD2)[1]["overdue"] == []


# ── AI steps: suggest only, and off means off ───────────────────────────────────────────────

def test_ai_steps_change_nothing_and_say_why(planned):
    out = api("GET", path("/projects/{p}/ai/draft-plan", p=P), LEAD)[1]
    assert out["applied"] is False and "no AI provider" in out["reason"]
    api("POST", "/settings/ai.ai_draft_plan.mode", MANAGER, {"value": "off"})
    assert api("GET", path("/projects/{p}/ai/draft-plan", p=P), LEAD)[1]["mode"] == "off"
    step("start", I1)
    step("block", I1, LEAD, {"reason": "waiting for brief"})
    assert api("GET", path("/items/{i}/ai/chase-brief", i=I1), LEAD)[0] == 409       # not yet past chase days
    clock("2026-11-06T09:00")
    assert api("GET", path("/items/{i}/ai/chase-brief", i=I1), LEAD)[1]["applied"] is False


# ── settings and state names ────────────────────────────────────────────────────────────────

def test_settings_are_checked_and_change_the_rules(planned):
    assert api("POST", "/settings/review.window_days", LEAD, {"value": 3})[0] == 403
    status, out = api("POST", "/settings/review.window_days", MANAGER, {"value": 0})
    assert status == 409 and "whole number" in out["detail"]
    assert api("POST", "/settings/review.window_days", MANAGER, {"value": 3})[1]["value"] == 3
    step("start", I1)
    step("submit", I1)
    assert item(I1)[1]["review_clock"]["window_ends_on"] == "2026-11-05"
    assert api("POST", "/states/in_review", MANAGER, {"name": "With client"})[1]["name"] == "With client"
    assert item(I1)[1]["state_name"] == "With client"


def test_a_setting_that_changes_the_signed_process_says_so():
    """AUDIT #9: switching deemed acceptance off removes a step the customer signed."""
    out = api("POST", "/settings/review.deemed_acceptance", MANAGER, {"value": False})[1]
    assert "changes the signed process" in out["warning"] and "erp-compose change propose" in out["warning"]
    assert api("POST", "/settings/review.window_days", MANAGER, {"value": 5})[1]["warning"]
    assert api("POST", "/settings/review.window_days", MANAGER, {"value": None})[1]["warning"] is None


# ── PJ-05: the client's link, and taking it back ────────────────────────────────────────────

def _door_get(p):
    import anyio
    import httpx
    door = public.app(use_cases.PJ.public_handlers, name="pj-public")

    async def get():
        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=door), base_url="http://door") as c:
            return await c.get(p)
    return anyio.run(get)


def test_pj_05_the_client_link_shows_one_project_read_only(planned):
    plan(source_id="C-300", bp_id="8000")
    status, link = project_cmd("client-link", body={"email": "Kim@Client.example"})
    assert status == 200 and link["url"].startswith(PUBLIC_BASE + "/p/") and link["recipient"] == "kim@client.example"
    page = _door_get(link["url"][len(PUBLIC_BASE):])
    assert page.status_code == 200
    assert "Implementation phase (1 of 2)" in page.text and "C-300" not in page.text
    assert _door_get("/p/not-a-real-token").status_code == 404


def test_a_client_link_can_be_taken_back(planned):
    """AUDIT #5: a contact left, or forwarded the link: revoke it and the page is the dead-link page at once."""
    link = project_cmd("client-link", body={"email": "kim@client.example"})[1]
    listed = api("GET", path("/projects/{p}/client-links", p=P), LEAD)[1]["links"]
    assert [(x["link_id"], x["recipient"]) for x in listed] == [(link["link_id"], "kim@client.example")]
    assert "url" not in listed[0]
    out = api("POST", path("/projects/{p}/client-links/{l}/revoke", p=P, l=link["link_id"]), LEAD, {})[1]
    assert out == {"link_id": link["link_id"], "revoked": True}
    assert _door_get(link["url"][len(PUBLIC_BASE):]).status_code == 404
    other = api("POST", path("/projects/{p}/client-links/{l}/revoke", p=P, l="0" * 32), LEAD, {})
    assert other[0] == 404