dodil/erp-brick-contracts

tests/test_doors.py

"""Both doors over the REAL app (CT.app(): the routes and /mcp) and the public door over the platform's public app,
storage on SQLite: the same outcome through the API and through MCP, 404 for a row the caller may not see, 403 without
the permission — and the security properties of the spec's tests.access list."""
import base64
import hashlib
import pathlib
import re

from sqlalchemy import event, select

from erp_platform import public, work

from brick_contracts.application import support as S
from brick_contracts.infrastructure import mail, tables as T
from conftest import ADMIN, headers
from test_flow import draft_nda, path_of

OWNER, OWNER2, MANAGER, DIRECTOR = headers("owner"), headers("owner", 2), headers("manager"), headers("director")
HR_ADMIN = headers("hr_admin")


def sent_nda(api, partners):
    c = draft_nda(api, partners)
    s, out = api("POST", f"/contracts/{c['contract_id']}/send", ADMIN, {})
    assert s == 200, out
    return c["contract_id"], out


def test_the_same_contract_through_both_doors(library, partners, tool):
    api = library
    c = draft_nda(api, partners)
    s, via_api = api("GET", f"/contracts/{c['contract_id']}", ADMIN)
    kind, via_mcp = tool(ADMIN, "show_contract", {"contract_id": c["contract_id"]})
    assert s == 200 and kind == "ok" and via_api == via_mcp
    assert isinstance(via_api["bp_id"], str)                     # a 64-bit id leaves as a string


def test_a_missing_permission_is_403_naming_it_through_both_doors(library, partners, tool):
    api = library
    c = draft_nda(api, partners)
    s, out = api("POST", f"/contracts/{c['contract_id']}/send", MANAGER, {})
    assert (s, out["detail"]) == (403, "missing permission: contract:contract:send")
    kind, msg = tool(MANAGER, "send_for_signature", {"contract_id": c["contract_id"]})
    assert kind == "refused" and msg == "403: missing permission: contract:contract:send"


def test_another_owners_commercial_contract_is_404_never_403(library, partners, tool):
    api = library
    bp = partners.add(7, "Mine Ltd", ["supplier"])
    s, c = api("POST", "/contracts", OWNER, {"type_code": "nda", "bp_id": bp, "form_key": "x1",
                                             "values": {"start_date": "2026-11-01"}})
    assert s == 200, c
    for path in ("", "/versions", "/status", "/executed"):
        assert api("GET", f"/contracts/{c['contract_id']}{path}", OWNER2)[0] == 404
    assert api("POST", f"/contracts/{c['contract_id']}/send", OWNER2, {})[0] == 404
    assert tool(OWNER2, "show_contract", {"contract_id": c["contract_id"]})[1].startswith("404")
    assert api("GET", f"/contracts/{c['contract_id']}", OWNER)[0] == 200
    listed = api("GET", "/contracts", OWNER2)[1]["contracts"]
    assert c["contract_id"] not in [x["contract_id"] for x in listed]


def test_an_employment_body_is_404_to_its_owner_and_a_manager_through_both_doors(library, partners, tool):
    api = library
    bp = partners.add(8, "Anna Berg", [], kind="person")
    s, c = api("POST", "/contracts", OWNER, {"type_code": "employment_contract", "bp_id": bp, "form_key": "e1",
                                             "values": {"job_title": "Designer", "start_date": "2026-11-02",
                                                        "rate_amount": "42000.00", "rate_basis": "year"}})
    assert s == 200 and c["body"] == "withheld", c
    cid = c["contract_id"]
    for who in (OWNER, MANAGER):
        s, head = api("GET", f"/contracts/{cid}", who)
        assert s == 200 and head["body"] == "withheld" and "sections" not in head and "variables" not in head
        assert head["key_dates"] and head["key_dates"][0]["kind"] == "PRINCIPAL_STATEMENT_DUE"
        for path in ("/versions", "/executed", "/versions/1/file"):
            assert api("GET", f"/contracts/{cid}{path}", who)[0] == 404, path
        assert tool(who, "show_versions", {"contract_id": cid})[1].startswith("404")
    assert api("GET", f"/contracts/{cid}/certificate", OWNER)[0] == 404
    s, full = api("GET", f"/contracts/{cid}", HR_ADMIN)
    assert s == 200 and full["body"] == "shown" and full["sections"]          # contract:employment:read sees it
    s, admin = api("GET", f"/contracts/{cid}", ADMIN)
    assert s == 200 and admin["body"] == "shown" and admin["sections"]


def test_a_signing_link_reaches_only_its_own_page_and_a_read_link_cannot_sign(library, partners, door):
    api = library
    cid, out = sent_nda(api, partners)
    url = api("POST", f"/contracts/{cid}/signers/1/copy-link", ADMIN, {})[1]["url"]
    other = draft_nda(api, partners, key="F2")
    api("POST", f"/contracts/{other['contract_id']}/send", ADMIN, {})
    s, page = door("GET", path_of(url))
    assert s == 200 and other["contract_id"] not in page
    s, page = door("POST", path_of(url), form={"action": "sign", "typed_name": "Sam Nine", "consent": "yes",
                                                "version": "1", "sha256": out["sha256"]})
    assert s == 200
    signer = {x["signer_no"]: x for x in api("GET", f"/contracts/{cid}", ADMIN)[1]["signers"]}[1]
    assert signer["link"]["state"] == "used" and signer["link"]["used_at"]          # SIGN is single use
    api("POST", f"/contracts/{cid}/sign", DIRECTOR, {"version": 1, "sha256": out["sha256"], "typed_name": "Dana D",
                                                     "consent": True})
    read = api("POST", f"/contracts/{cid}/signers/1/copy-link", ADMIN, {"which": "copy"})[1]["url"]
    r = door("GET", path_of(read), raw=True)
    assert r.status_code == 200 and r.headers["content-type"] == "application/pdf"
    assert hashlib.sha256(r.content).hexdigest() == api("GET", f"/contracts/{cid}", ADMIN)[1]["executed"]["sha256"]
    s, _ = door("POST", path_of(read), form={"action": "sign"})
    assert s == 404                                              # READ does not act
    s, _ = door("GET", path_of(url))
    assert s == 404                                              # the SIGN link is spent


def test_the_signing_page_needs_no_script_and_shows_the_fingerprint_consent_and_retention(library, partners, door):
    api = library
    cid, out = sent_nda(api, partners)
    url = api("POST", f"/contracts/{cid}/signers/1/copy-link", ADMIN, {})[1]["url"]
    r = door("GET", path_of(url), raw=True)
    assert "<script" not in r.text and out["sha256"] in r.text and "kept for 6 years" in r.text
    assert "I agree to sign this document electronically" in r.text
    assert "script-src" not in r.headers["content-security-policy"] or "'none'" in r.headers["content-security-policy"]
    pdf = door("POST", path_of(url), form={"action": "download", "version": "1", "sha256": out["sha256"]}, raw=True)
    assert pdf.headers["content-type"] == "application/pdf" and hashlib.sha256(pdf.content).hexdigest() == out["sha256"]


def test_ip_and_user_agent_are_fields_of_every_public_event(library, partners, door):
    api = library
    cid, out = sent_nda(api, partners)
    url = api("POST", f"/contracts/{cid}/signers/1/copy-link", ADMIN, {})[1]["url"]
    door("GET", path_of(url), headers={"user-agent": "Firefox/140", "x-forwarded-for": "198.51.100.7"})
    door("POST", path_of(url), headers={"user-agent": "Firefox/140", "x-forwarded-for": "198.51.100.7"},
         form={"action": "sign", "typed_name": "Sam Nine", "consent": "yes", "version": "1", "sha256": out["sha256"]})
    ev = [e for e in api("GET", f"/contracts/{cid}/certificate", ADMIN)[1]["events"]
          if e["kind"] in ("OPENED", "CONSENTED", "SIGNED")]
    assert ev and all(e["ip"] == "198.51.100.7" and e["user_agent"] == "Firefox/140" for e in ev)
    assert all(e["detail"] is None or "198.51" not in e["detail"] for e in ev)          # never free text


def test_with_a_mail_provider_the_link_goes_by_mail_and_delivered_is_recorded(library, partners, monkeypatch):
    api = library
    sent_mail = []
    monkeypatch.setattr(mail, "provider", lambda: "gws")
    monkeypatch.setattr(mail, "send", lambda key, **kw: sent_mail.append((key, kw)) or mail.Sent("sent", "msg-1"))
    cid, out = sent_nda(api, partners)
    assert out["deliveries"] == [{"signer_no": 1, "channel": "provider"}]
    (key, kw), = sent_mail
    assert key == f"{cid}|1|INVITE|1" and kw["to"] == "sam@studionine.example"
    assert re.search(r"https://public\.contracts\.test/p/[A-Za-z0-9_-]{43}", kw["body"])
    ev = api("GET", f"/contracts/{cid}/certificate", ADMIN)[1]["events"]
    assert [e["kind"] for e in ev if e["signer_no"] == 1] == ["INVITED", "DELIVERED"]
    c = api("GET", f"/contracts/{cid}", ADMIN)[1]
    assert c["tasks"] == [] and c["deliveries"][0]["provider_ref"] == "msg-1"
    stored = str([d for d in c["deliveries"]])
    assert "/p/" not in stored                                   # the link is never stored


def test_a_request_with_no_gateway_identity_is_rejected(library):
    s, _ = library("GET", "/contracts")
    assert s == 401


def test_the_config_snapshot_is_read_once_per_change_not_per_request(library, engine):
    api = library
    api("GET", "/contracts", ADMIN)
    seen = []

    @event.listens_for(engine, "before_cursor_execute")
    def count(conn, cursor, statement, *a):
        seen.append(statement)
    api("GET", "/contracts", ADMIN)
    event.remove(engine, "before_cursor_execute", count)
    assert not [s for s in seen if "FROM ct_type" in s or "FROM ct_approval_policy" in s or "FROM ct_settings" in s]
    assert any("ct_config_version" in s for s in seen)
    assert len(seen) <= 6, seen


def test_an_unchanged_edit_writes_nothing(library, partners, engine):
    api = library
    c = draft_nda(api, partners)
    api("POST", f"/contracts/{c['contract_id']}/variables", ADMIN, {"values": {"notice_days": 30}})
    seen = []

    @event.listens_for(engine, "before_cursor_execute")
    def count(conn, cursor, statement, *a):
        seen.append(statement)
    api("POST", f"/contracts/{c['contract_id']}/variables", ADMIN, {"values": {"notice_days": 30}})
    event.remove(engine, "before_cursor_execute", count)
    assert not [s for s in seen if s.lstrip().upper().startswith(("INSERT", "UPDATE", "DELETE"))], seen


def test_nothing_shown_claims_more_than_signed_electronically(library, partners, door):
    """The page, the certificate, the executed PDF and every message say "signed electronically" — never advanced,
    qualified, legally binding or equivalent to a wet signature."""
    from brick_contracts.domain.services import FORBIDDEN_CLAIMS
    api = library
    cid, out = sent_nda(api, partners)
    url = api("POST", f"/contracts/{cid}/signers/1/copy-link", ADMIN, {})[1]["url"]
    shown = [door("GET", path_of(url))[1]]
    shown.append(door("POST", path_of(url), form={"action": "sign", "typed_name": "Sam Nine", "consent": "yes",
                                                   "version": "1", "sha256": out["sha256"]})[1])
    api("POST", f"/contracts/{cid}/sign", DIRECTOR, {"version": 1, "sha256": out["sha256"], "typed_name": "Dana D",
                                                     "consent": True})
    cert = api("GET", f"/contracts/{cid}/certificate", ADMIN)[1]
    shown.append(cert["text"])
    shown += [d["message"] + d["subject"] for d in api("GET", f"/contracts/{cid}", ADMIN)[1]["deliveries"]]
    import io
    from pypdf import PdfReader
    pdf = base64.b64decode(api("GET", f"/contracts/{cid}/executed", ADMIN)[1]["content_base64"])
    shown += [p.extract_text() for p in PdfReader(io.BytesIO(pdf)).pages]
    text = "\n".join(shown).lower()
    assert "signed electronically" in text
    assert not [w for w in FORBIDDEN_CLAIMS if w in text], [w for w in FORBIDDEN_CLAIMS if w in text]