dodil/erp-brick-expenses
DODIL ERP brick: expenses
tests/test_receipts.py
"""Slice B item 4: the receipt store (EXP-01's image half, EXP-16's duplicate flag): receipt images as DataK3 objects under a content-addressed key,
kept for the retention period, read back beside the line."""
import base64
from datetime import date
from conftest import EMPLOYEE, FINANCE, JPEG, MANAGER, add, api, image, must, new_report, person, refused, report, submit
from erp_platform import work
from brick_expenses.application import use_cases
from brick_expenses.domain import receipts as rc
from brick_expenses.infrastructure import objects
PNG = b"\x89PNG\r\n\x1a\n" + b"pixels" * 30
PDF = b"%PDF-1.7\n" + b"body" * 30
def upload(rid, who=None, **kw):
return api("POST", f"/reports/{rid}/receipts", who or EMPLOYEE, {**image(), **kw})
def draft(**kw):
rid = new_report()["report_id"]
add(rid, receipt_key="", **kw)
return rid
def test_EXP_01_a_receipt_is_stored_as_an_object_under_its_digest_and_attached_to_the_line(store):
rid = draft(amount="30.00")
got = must(upload(rid, line_no=1), 201)
sha = rc.sha256(JPEG)
assert got["receipt_key"] == f"raw/expenses/{rid}/1/{sha}.jpg" and got["attached"] and got["size_bytes"] == len(JPEG)
assert store.objects[got["receipt_key"]] == (JPEG, "image/jpeg") # the bytes are in the store under that key
assert report(rid)["lines"][0]["receipt_key"] == got["receipt_key"]
with use_cases.repository() as repo:
row = repo.receipts.get(got["receipt_key"])
assert (row["sha256"], row["report_id"], row["line_no"], row["uploaded_by"]) == (sha, rid, 1, "emp3301@uk.example")
assert row["retain_until"] == date(2032, 11, 4) # six years from the date of the cost (2026-11-04)
def test_EXP_01_a_receipt_on_a_line_above_the_threshold_satisfies_the_gate_and_the_line_is_submittable():
rid = draft(amount="30.00")
refused(api("POST", f"/reports/{rid}/submit", EMPLOYEE), 409, "receipts_present")
must(upload(rid, line_no=1), 201)
assert submit(rid)["status"] == "submitted"
def test_EXP_01_the_same_image_is_the_same_object_stored_once_and_one_row(store):
rid = draft(amount="30.00")
first = must(upload(rid, line_no=1), 201)
again = must(upload(rid, line_no=1), 201)
assert again["receipt_key"] == first["receipt_key"] and again["stored_again"] is True and len(store.objects) == 1
assert again["possible_duplicate"] is False # the same image on the same line is not a duplicate
other = must(upload(rid, line_no=1, **image(PNG, "image/png")), 201) # another image of the line is another key (evidence is never replaced)
assert other["receipt_key"].endswith(".png") and len(store.objects) == 2 and report(rid)["lines"][0]["receipt_key"] == other["receipt_key"]
def test_EXP_01_an_image_stored_first_is_the_key_of_the_next_line():
rid = draft(amount="30.00")
got = must(upload(rid), 201)
assert (got["line_no"], got["attached"]) == (2, False) and got["receipt_key"].startswith(f"raw/expenses/{rid}/2/")
line = add(rid, amount="40.00", receipt_key=got["receipt_key"])["lines"][1]
assert line["receipt_key"] == got["receipt_key"]
assert [x["line_no"] for x in report(rid)["lines"]] == [1, 2]
def test_EXP_01_pdfs_and_pngs_are_receipts_too_and_a_leap_day_keeps_its_retention_date():
rid = draft(amount="30.00")
assert must(upload(rid, line_no=1, **image(PDF, "application/pdf")), 201)["receipt_key"].endswith(".pdf")
assert must(upload(rid, line_no=1, **image(PNG, "image/png")), 201)["receipt_key"].endswith(".png")
assert rc.retain_until(date(2028, 2, 29), 6) == date(2034, 2, 28) and rc.retain_until(date(2026, 11, 4), 10) == date(2036, 11, 4)
def test_EXP_01_what_is_not_a_receipt_is_a_422_never_a_500(store):
rid = draft(amount="30.00")
ok = base64.b64encode(JPEG).decode()
for body, words in (({"content_type": "image/jpeg", "content_base64": "###not base64###"}, "base64"),
({"content_type": "image/jpeg", "content_base64": base64.b64encode(b"GIF89a....").decode()}, "do not start like one"),
({"content_type": "application/pdf", "content_base64": ok}, "do not start like one"),
({"content_type": "image/jpeg", "content_base64": base64.b64encode(b"\xff\xd8\xff" + b"x" * rc.MAX_BYTES).decode()}, "at most 3 MiB")):
refused(api("POST", f"/reports/{rid}/receipts", EMPLOYEE, body), 422, words)
refused(api("POST", f"/reports/{rid}/receipts", EMPLOYEE, {"content_type": "image/gif", "content_base64": ok}), 422)
refused(api("POST", f"/reports/{rid}/receipts", EMPLOYEE, {"content_type": "image/jpeg", "content_base64": ""}), 422)
assert store.objects == {} # nothing was stored for any of them
refused(upload(rid, line_no=9), 404, "no line 9")
def test_EXP_01_only_the_claimant_adds_to_a_draft_only(store):
rid = draft(amount="30.00")
refused(upload(rid, MANAGER, line_no=1), 404)
refused(upload(rid, person("nobody@uk.example", []), line_no=1), 403, "exp:report:write_own")
must(upload(rid, line_no=1), 201)
r2 = draft(amount="30.00")
must(upload(r2, line_no=1), 201)
add_receipt_then_submit = new_report()["report_id"]
add(add_receipt_then_submit, amount="5.00")
submit(add_receipt_then_submit)
refused(upload(add_receipt_then_submit, line_no=1), 409, "frozen")
def test_a_receipt_store_that_is_not_configured_or_down_is_a_reply_and_stores_nothing(store, monkeypatch):
rid = draft(amount="30.00")
store.down = objects.StoreDown(502, "the object store answered 500")
refused(upload(rid, line_no=1), 502, "answered 500")
assert report(rid)["lines"][0]["receipt_key"] is None
store.down = None
objects.ports.store = None
for name in ("OBJECT_ENDPOINT", "BUCKET", "OBJECT_ACCESS_KEY", "OBJECT_SECRET_KEY"):
monkeypatch.delenv(name, raising=False)
refused(upload(rid, line_no=1), 503, "OBJECT_ENDPOINT")
def test_the_s3_store_signs_a_put_and_does_not_overwrite_a_key_that_exists(monkeypatch):
import httpx
monkeypatch.setenv("BUCKET", "acme")
monkeypatch.setenv("OBJECT_ACCESS_KEY", "AK")
monkeypatch.setenv("OBJECT_SECRET_KEY", "SK")
seen = []
def fake(method):
def call(url, **kw):
seen.append((method, url, kw.get("headers", {})))
return httpx.Response(404 if method == "GET" else 200, request=httpx.Request(method, url), content=b"")
return call
monkeypatch.setattr(httpx, "get", fake("GET"))
monkeypatch.setattr(httpx, "put", fake("PUT"))
store = objects.S3Store()
assert store.configured and store.put("raw/expenses/R/1/abc.jpg", b"x", "image/jpeg") == "raw/expenses/R/1/abc.jpg"
assert [m for m, _, _ in seen] == ["GET", "PUT"] # looked first, then wrote
method, url, headers = seen[1]
assert url == "https://object.uk-lon-1.dodil.io/acme/raw/expenses/R/1/abc.jpg" and headers["authorization"].startswith("AWS4-HMAC-SHA256 Credential=AK/")
seen.clear()
monkeypatch.setattr(httpx, "get", lambda url, **kw: httpx.Response(200, request=httpx.Request("GET", url), content=b"already"))
assert store.put("raw/expenses/R/1/abc.jpg", b"new", "image/jpeg") == "raw/expenses/R/1/abc.jpg" and seen == [] # never overwritten
# ── EXP-16: duplicates ────────────────────────────────────────────────────────────────────────────
def test_EXP_16_the_same_receipt_added_by_another_employee_is_flagged_for_the_approver_and_nothing_is_refused(nb):
nb.employee(3308, "emp3308@uk.example", manager=nb.as_manager(2280, "mgr2280@uk.example"))
other = person("emp3308@uk.example", ["exp:report:read_own", "exp:report:write_own", "exp:report:submit"])
first = draft(amount="96.00", category_code="MEALS_CLIENT", merchant="The Ivy")
must(upload(first, line_no=1), 201)
second = must(api("POST", "/reports", other, {"currency": "GBP"}), 201)["report_id"]
add(second, other, amount="96.00", category_code="MEALS_CLIENT", merchant="The Ivy", receipt_key="")
got = must(api("POST", f"/reports/{second}/receipts", other, {**image(), "line_no": 1}), 201)
assert got["possible_duplicate"] is True and got["attached"] is True # flagged, not refused
mine = must(api("GET", f"/reports/{second}", other))["lines"][0]
assert mine["possible_duplicate"] is True and mine["possible_duplicate_of"] is None # the claimant is told, not shown another person's claim
assert must(api("POST", f"/reports/{second}/submit", other))["status"] == "submitted" # a person decides: a split bill is legitimate
seen = must(api("GET", f"/reports/{second}", MANAGER))["lines"][0]
assert seen["possible_duplicate"] is True and seen["possible_duplicate_of"] == f"{first}:1" # the approver sees which claim
assert must(api("GET", f"/reports/{second}", FINANCE))["lines"][0]["possible_duplicate_of"] == f"{first}:1"
def test_EXP_16_the_same_image_on_two_lines_of_one_person_is_flagged_too():
rid = new_report()["report_id"]
add(rid, receipt_key="", amount="30.00")
add(rid, receipt_key="", amount="30.00", merchant="Other")
must(upload(rid, line_no=1), 201)
must(upload(rid, line_no=2), 201)
lines = report(rid)["lines"]
assert [x["possible_duplicate"] for x in lines] == [False, True]
def test_EXP_16_a_different_image_is_no_duplicate():
rid = new_report()["report_id"]
add(rid, receipt_key="", amount="30.00")
add(rid, receipt_key="", amount="30.00", merchant="Other")
must(upload(rid, line_no=1), 201)
must(upload(rid, line_no=2, **image(PNG, "image/png")), 201)
assert [x["possible_duplicate"] for x in report(rid)["lines"]] == [False, False]
# ── reading it back ───────────────────────────────────────────────────────────────────────────────
def test_the_approver_reads_the_receipt_beside_the_line_and_a_stranger_gets_a_404():
rid = draft(amount="30.00")
must(upload(rid, line_no=1), 201)
submit(rid)
got = must(api("GET", f"/reports/{rid}/lines/1/receipt", MANAGER))
assert base64.b64decode(got["content_base64"]) == JPEG and got["content_type"] == "image/jpeg" and got["retain_until"] == "2032-11-04"
assert got["sha256"] == rc.sha256(JPEG) and must(api("GET", f"/reports/{rid}/lines/1/receipt", EMPLOYEE))["size_bytes"] == len(JPEG)
refused(api("GET", f"/reports/{rid}/lines/1/receipt", person("stranger@uk.example", ["exp:report:read_own"])), 404)
refused(api("GET", f"/reports/{rid}/lines/7/receipt", MANAGER), 404, "no receipt")
def test_a_receipt_key_that_was_not_stored_here_is_named_as_such_and_a_missing_object_is_a_404(store):
rid = new_report()["report_id"]
add(rid, amount="30.00") # the helper's key: shaped right, never uploaded
refused(api("GET", f"/reports/{rid}/lines/1/receipt", EMPLOYEE), 404, "was not stored through add_receipt")
r2 = draft(amount="30.00")
key = must(upload(r2, line_no=1), 201)["receipt_key"]
del store.objects[key]
body = refused(api("GET", f"/reports/{r2}/lines/1/receipt", EMPLOYEE), 404, "not in the store")
assert body["gate"] == "object_missing"
def test_the_retention_years_are_a_setting_not_code():
with use_cases.repository() as repo:
repo.settings.put("receipt_retention_years", "10", date(2026, 1, 1), "owner", work.now())
rid = draft(amount="30.00")
must(upload(rid, line_no=1), 201)
with use_cases.repository() as repo:
(row,) = repo.receipts.of_sha(rc.sha256(JPEG))
assert row["retain_until"] == date(2036, 11, 4) and JPEG