dodil/erp-brick-expenses
DODIL ERP brick: expenses
tests/test_ai.py
"""Slice B item 5: the receipt reader's consent (EXP-02) and the step itself (EXP-01) against its port: suggest only, after consent, every field marked until
the person confirms it, audited, and the fallback of typing the fields. Each test is written from the story's Given/When/Then (module.yaml)."""
import base64
import json
import pytest
from conftest import EMPLOYEE, FINANCE, JPEG, MANAGER, add, api, must, new_report, on_day, person, refused, report, submit
from sqlalchemy import text
from brick_expenses.application import use_cases
from brick_expenses.infrastructure import readers
HOTEL = {"date": "2026-11-05", "merchant": "Manchester Grand Hotel", "total": "264.00", "currency": "GBP", "vat": "44.00", "vat_number": "GB 987 6543 21",
"category_code": "ACCOMMODATION", "category_reason": "a hotel invoice"}
class FakeReader:
def __init__(self):
self.calls = []
self.fields = dict(HOTEL)
self.down = None
self.model = "kimi-vision-eu"
def read(self, image, content_type, categories):
self.calls.append((image, content_type, categories))
if self.down:
raise readers.ReaderDown(self.down)
return {"model": self.model, "fields": self.fields, "confidence": {"total": 99, "vat": 97}}
@pytest.fixture
def reader():
r = FakeReader()
readers.ports.reader = r
yield r
readers.ports.reader = None
def consent(decision="given", who=None):
return api("POST", "/ai/consent", who or EMPLOYEE, {"decision": decision})
def receipt(rid, who=None, body=JPEG, **kw):
return api("POST", f"/reports/{rid}/receipts", who or EMPLOYEE, {"content_type": "image/jpeg", "content_base64": base64.b64encode(body).decode(), **kw})
# ── EXP-02 ───────────────────────────────────────────────────────────────────────────────────────
def test_EXP_02_asked_once_with_the_text_agrees_and_withdraws_and_the_latest_answer_decides():
first = must(api("GET", "/ai/consent", EMPLOYEE))
assert first["decision"] is None and first["mode"] == "suggest" and first["read_by_a_model_now"] is False
assert "sovereign" in first["text"] and "Nobody can agree for you" in first["text"] and first["text_version"]
got = must(consent("given"), 201)
assert got["decision"] == "given" and got["read_by_a_model_now"] is True
assert must(consent("given"), 201)["decision"] == "given" # the same answer again adds no row
with use_cases.repository() as repo:
assert repo.conn.execute(text("select count(*) from exp_consent")).scalar() == 1
with on_day("2026-12-01T09:00:00+00:00"):
assert must(consent("withdrawn"), 201)["read_by_a_model_now"] is False
assert must(api("GET", "/ai/consent", EMPLOYEE))["decision"] == "withdrawn"
with use_cases.repository() as repo:
rows = repo.conn.execute(text("select decision, text_version from exp_consent order by given_at")).all()
assert [r[0] for r in rows] == ["given", "withdrawn"] and rows[0][1] == rows[1][1] # append-only: both answers and the text seen
def test_EXP_02_nobody_answers_for_someone_else_finance_included_and_hr_must_know_the_person():
refused(consent("given", FINANCE), 403, "exp:ai:consent")
refused(api("GET", "/ai/consent", FINANCE), 403, "exp:ai:consent")
assert must(api("POST", "/ai/consent", EMPLOYEE, {"decision": "given", "person_id": "2280"}), 201)["decision"] == "given" # a named person is ignored: it is mine
assert must(api("GET", "/ai/consent", MANAGER))["decision"] is None # 2280's answer was never recorded
with use_cases.repository() as repo:
repo.conn.execute(text("delete from exp_consent"))
refused(consent("given", person("stranger@uk.example", ["exp:ai:consent"])), 409, "hr holds no person")
refused(consent("maybe"), 422)
with use_cases.repository() as repo:
assert repo.conn.execute(text("select count(*) from exp_consent")).scalar() == 0
must(consent("given", MANAGER), 201)
assert must(api("GET", "/ai/consent", EMPLOYEE))["decision"] is None # the manager's yes is the manager's
# ── EXP-01 ───────────────────────────────────────────────────────────────────────────────────────
def test_EXP_01_the_reader_suggests_a_line_from_the_photo_each_field_marked_until_confirmed(reader, store):
must(consent(), 201)
rid = new_report()["report_id"]
got = must(receipt(rid), 201)
assert got["suggested"] is True and got["attached"] is True and got["line_no"] == 1
line = got["line"]
assert (line["category_code"], line["expense_date"], line["merchant"], line["amount"], line["vat_amount"]) == (
"ACCOMMODATION", "2026-11-05", "Manchester Grand Hotel", "264.00", "44.00")
assert line["unconfirmed_fields"] == ["category_code", "expense_date", "merchant", "amount", "vat_amount", "vat_number"] # every field marked suggested
assert line["receipt_key"].startswith(f"raw/expenses/{rid}/1/") and line["receipt_key"] in store.objects
assert len(reader.calls) == 1 and reader.calls[0][0] == JPEG and reader.calls[0][1] == "image/jpeg"
assert {"category_code": "ACCOMMODATION", "name": "Hotels"} in reader.calls[0][2] # the company's categories are what the model chooses from
body = refused(api("POST", f"/reports/{rid}/submit", EMPLOYEE), 409, "lines_ready") # the line is ready only when each field is confirmed
assert body["gate"] == "lines_ready"
assert report(rid)["lines"][0]["ai"]["model"] == "kimi-vision-eu" and report(rid)["lines"][0]["ai"]["filled"][0] == "category_code"
def test_EXP_01_confirming_or_correcting_each_field_makes_the_line_ready_and_what_changed_is_kept_for_the_approver(reader):
must(consent(), 201)
rid = new_report()["report_id"]
must(receipt(rid), 201)
must(api("POST", f"/reports/{rid}/lines/1", EMPLOYEE, {"merchant": "Manchester Grand Hotel Ltd", "confirm_fields": ["category_code", "expense_date"]}))
mid = report(rid)["lines"][0]
assert mid["unconfirmed_fields"] == ["amount", "vat_amount", "vat_number"] and mid["ai"]["changed"] == ["merchant"]
refused(api("POST", f"/reports/{rid}/submit", EMPLOYEE), 409, "lines_ready")
must(api("POST", f"/reports/{rid}/lines/1", EMPLOYEE, {"vat_amount": "40.00", "confirm_fields": ["amount"],
"receipt": {"supplier_name": "Manchester Grand Hotel Ltd", "supplier_vat_number": "GB987654321",
"supplier_address": {"lines": ["1 Deansgate"], "city": "Manchester", "country": "GB"}}}))
done = report(rid)["lines"][0]
assert done["unconfirmed_fields"] == [] and done["ai"]["changed"] == ["merchant", "vat_amount", "vat_number"] # the VAT number as typed differs from "GB 987 6543 21"
assert submit(rid)["status"] == "submitted"
seen = must(api("GET", f"/reports/{rid}", MANAGER))["lines"][0]["ai"] # the approver sees which fields the model filled and which changed
assert seen["filled"] == ["category_code", "expense_date", "merchant", "amount", "vat_amount", "vat_number"] and seen["changed"] == ["merchant", "vat_amount", "vat_number"]
with use_cases.repository() as repo:
(f,) = repo.ai.of_report(rid)
read, confirmed = json.loads(f["fields_json"]), json.loads(f["confirmed_json"])
assert (f["model"], f["step_key"], f["decided_by"], len(f["inputs_digest"])) == ("kimi-vision-eu", "read_receipt", "emp3301@uk.example", 64)
assert read["fields"]["merchant"] == "Manchester Grand Hotel" and read["fields"]["total"] == "264.00" and read["confidence"] == {"total": 99, "vat": 97} # what the model read
assert confirmed["merchant"] == {"value": "Manchester Grand Hotel Ltd", "changed": True} and confirmed["amount"] == {"value": "264.00", "changed": False} # what the person did
def test_EXP_01_with_the_step_off_or_no_consent_or_no_reader_the_fields_are_empty_and_the_image_goes_to_no_model(reader, store):
rid = new_report()["report_id"]
got = must(receipt(rid), 201)
assert got["suggested"] is False and got["attached"] is False and "have not agreed" in got["suggestion_note"] and reader.calls == [] # never asked
line = add(rid, amount="264.00", category_code="ACCOMMODATION", receipt_key=got["receipt_key"])["lines"][0] # the employee types it
assert line["unconfirmed_fields"] == [] and line["receipt_key"] == got["receipt_key"] and got["receipt_key"] in store.objects
must(consent(), 201)
must(api("POST", "/settings", person("owner@uk.example", ["exp:settings:controls"]), {"key": "ai_read_receipt_mode", "value": "off", "valid_from": "2026-11-07"}), 201)
off = must(receipt(rid, body=JPEG + b"2"), 201)
assert off["suggested"] is False and "off" in off["suggestion_note"] and reader.calls == []
readers.ports.reader = None
must(api("POST", "/settings", person("owner@uk.example", ["exp:settings:controls"]), {"key": "ai_read_receipt_mode", "value": "suggest", "valid_from": "2026-11-08"}), 201)
with on_day("2026-11-09T09:00:00+00:00"):
none = must(receipt(rid, body=JPEG + b"3"), 201)
assert none["suggested"] is False and "no receipt reader is configured" in none["suggestion_note"]
def test_EXP_01_a_receipt_attached_to_an_existing_line_is_never_read(reader):
must(consent(), 201)
rid = new_report()["report_id"]
add(rid, amount="30.00", receipt_key="")
got = must(receipt(rid, line_no=1), 201)
assert got["suggested"] is False and got["attached"] is True and reader.calls == []
def test_EXP_01_an_answer_no_person_could_have_typed_is_no_suggestion_and_the_image_is_kept(reader, store):
must(consent(), 201)
rid = new_report()["report_id"]
for k, (patch, words) in enumerate((({"date": "2026-12-25"}, "future"), ({"date": "yesterday"}, "not a date"), ({"total": "-5"}, "not an amount"),
({"total": "abc"}, "not an amount"), ({"category_code": "MADE_UP"}, "category (MADE_UP)"), ({"merchant": ""}, "did not read"),
({"currency": "pounds"}, "did not read"), ({"vat": "999.00"}, "VAT"), ({"currency": "EUR"}, "EUR to GBP rate"))):
reader.fields = {**HOTEL, **patch}
got = must(receipt(rid, body=JPEG + bytes([k])), 201)
assert got["suggested"] is False and words in got["suggestion_note"], (patch, got)
assert got["receipt_key"] in store.objects # the evidence is kept whatever the model said
assert report(rid)["lines"] == []
def test_EXP_01_a_reader_that_is_down_falls_back_to_typing(reader):
must(consent(), 201)
rid = new_report()["report_id"]
reader.down = "the model timed out"
got = must(receipt(rid), 201)
assert got["suggested"] is False and "timed out" in got["suggestion_note"] and report(rid)["lines"] == []
def test_EXP_01_a_foreign_currency_suggestion_is_converted_at_gls_rate(reader, nb):
must(consent(), 201)
nb.fx_rates["EUR"] = [("2026-11-01", "0.870000")]
reader.fields = {**HOTEL, "currency": "EUR", "total": "238.00", "vat": "39.67"}
rid = new_report()["report_id"]
line = must(receipt(rid), 201)["line"]
assert (line["receipt_amount"], line["receipt_currency"], line["amount"], line["fx_rate"], line["vat_amount"]) == ("238.00", "EUR", "207.06", "0.870000", "34.51")
def test_EXP_01_only_a_new_receipt_by_the_claimant_who_agreed_is_read_and_a_colleagues_yes_is_not_mine(reader, nb):
must(consent(who=MANAGER), 201)
rid = new_report()["report_id"]
assert must(receipt(rid), 201)["suggested"] is False and reader.calls == [] # the manager agreed; this claimant did not
refused(receipt(rid, who=MANAGER), 404)
def test_SC_EXP_05_consent_withdrawn_no_call_to_the_model_the_fields_stay_empty_and_earlier_findings_remain(reader):
with on_day("2026-10-01T09:00:00+00:00"):
must(consent("given"), 201)
rid = new_report()["report_id"]
with on_day("2026-11-05T09:00:00+00:00"):
first = must(receipt(rid), 201)
assert first["suggested"] is True
must(api("POST", f"/reports/{rid}/lines/1", EMPLOYEE, {"confirm_fields": ["category_code", "expense_date", "merchant", "amount", "vat_amount", "vat_number"]}))
with on_day("2026-12-01T09:00:00+00:00"):
must(consent("withdrawn"), 201)
calls = len(reader.calls)
with on_day("2026-12-02T09:00:00+00:00"):
later = must(receipt(rid, body=JPEG + b"later"), 201)
assert later["suggested"] is False and "withdrew" in later["suggestion_note"] and len(reader.calls) == calls # no call to the model
got = report(rid)
assert [x["line_no"] for x in got["lines"]] == [1] and got["lines"][0]["merchant"] == "Manchester Grand Hotel" and got["lines"][0]["unconfirmed_fields"] == []
with use_cases.repository() as repo:
(f,) = repo.ai.of_report(rid)
assert json.loads(f["confirmed_json"])["amount"]["value"] == "264.00" # what was read before stays as the person confirmed it
def test_a_suggested_claim_runs_on_to_the_ledger_with_the_confirmed_values(reader, nb):
must(consent(), 201)
rid = new_report()["report_id"]
must(receipt(rid), 201)
must(api("POST", f"/reports/{rid}/lines/1", EMPLOYEE, {"confirm_fields": ["category_code", "expense_date", "merchant", "amount", "vat_amount", "vat_number"]}))
submit(rid)
must(api("POST", f"/reports/{rid}/approve", MANAGER))
(j,) = nb.journals.values()
assert j["lines"][0] == {"purpose": "accommodation", "cost_centre": "SALES-N", "debit": "264.00"}