dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
tests/test_billing.py
"""TS-13 / TS-14: the month's approved billable hours of a contract project, priced from product and billed through ar - once, however often the run
is made, retried or raced; a correction bills the difference; the invoice request leaves only after the commit, from a pending row."""
import json
from decimal import Decimal
from conftest import (
EMP,
EMPLOYEE,
FINANCE,
FINANCE2,
LEADER,
MANAGER,
P1,
P2,
ROLES,
SHEET,
SYSTEM,
api,
approve,
approved_week,
hours,
line,
must,
on_day,
open_week,
person,
refused,
submit,
)
from sqlalchemy import select
from brick_timesheets.application import use_cases
from brick_timesheets.domain import billing as bd
from brick_timesheets.infrastructure import tables as t
MONTH_END = "2026-11-30T10:00:00+00:00"
RUN, SOURCE = f"tsbill:{P2}:2026-11", f"ts:{P2}:2026-11"
def create(project=P2, period="2026-11", at=MONTH_END, who=None):
with on_day(at):
return api("POST", "/billing-runs", who or SYSTEM, {"project_id": project, "period_id": period})
def made(**kw):
got = must(create(**kw))
assert got["created"], got
return got["run"]
def release(run_id=RUN, who=None, at=MONTH_END):
with on_day(at):
return api("POST", f"/billing-runs/{run_id}/release", who or FINANCE)
def claim_rows():
with use_cases.repository() as repo:
return [dict(r) for r in repo.conn.execute(select(t.ts_billing_claim).where(t.ts_billing_claim.c.purpose == "ar_run")).mappings()]
def correct(p2_hours, admin_hours=None):
"""Finance reopens the week; the employee puts P2's Thursday at `p2_hours` (0 clears it) and the rest of the day on ADMIN (the day stays covered);
the manager approves revision 2."""
must(api("POST", f"/sheets/{SHEET}/reopen", FINANCE, {"reason": "P2 hours were wrong"}))
hours(SHEET, 4, {"2026-11-19": p2_hours})
if admin_hours:
hours(SHEET, 3, {"2026-11-19": admin_hours})
submit()
must(approve())
# ── TS-13: one run per project and month, billed once ────────────────────────────────────────────
def test_ts_13_the_month_end_run_prices_each_line_from_product_and_ar_gets_one_invoice_on_release(nb):
approved_week()
run = made()
assert (run["run_id"], run["status"], run["source_id"], run["bp_id"], run["currency"]) == (RUN, "draft", SOURCE, "7311", "GBP")
assert run["lines"] == [{"description": "CONSULT-SR - time worked 2026-11", "quantity": "7.5000", "unit_code": "HUR", "unit_price": "110.0000",
"product_id": "CONSULT-SR", "tax_category": "S"}]
assert run["total_amount"] == "825.00" and run["claims"] == 1
assert nb.price_calls == ["2026-11-19"] and nb.invoice_posts == 0 # priced on the work day; nothing leaves before release
(c,) = claim_rows()
assert (c["sheet_id"], c["revision"], c["line_no"], c["run_id"], c["quantity"], c["unit_price"]) == (SHEET, 1, 4, RUN, Decimal("7.5"), Decimal("110"))
assert json.loads(c["priced_from"])["price_list_id"] == "LIST-GBP"
got = must(release())
assert got["status"] == "released" and got["invoice_id"] == "INV-0001" and got["invoice_request"]["status"] == "sent"
(sent,) = nb.invoices.values()
assert (sent["source_kind"], sent["source_id"], sent["bp_id"], sent["currency"], sent["kind"]) == ("timesheet", SOURCE, 7311, "GBP", "regular")
assert sent["lines"][0]["quantity"] == "7.5000" and sent["lines"][0]["unit_price"] == "110.0000"
def test_ts_13_a_retry_of_the_release_and_a_second_run_send_and_bill_nothing_again(nb):
approved_week()
made()
must(release())
assert must(release())["status"] == "released" # the retry: the same answer
assert nb.invoice_posts == 1 and len(nb.invoices) == 1
again = must(create())
assert again["created"] is False and len(claim_rows()) == 1
assert len(nb.price_calls) == 1 # the second run priced nothing: the hours are claimed
def test_ts_13_the_invoice_request_is_a_pending_row_that_survives_ar_being_down(nb):
approved_week()
made()
nb.ar_status = 503
got = must(release())
assert got["status"] == "released" and got["invoice_id"] is None and got["invoice_request"]["status"] == "pending"
assert "503" in got["invoice_request"]["last_error"] or "failed" in got["invoice_request"]["last_error"]
nb.ar_status = None
refused(api("POST", "/billing-runs/invoices/send", FINANCE), 403, "ts:billing_run:create")
sent = must(api("POST", "/billing-runs/invoices/send", SYSTEM))["sent"]
assert [s["status"] for s in sent] == ["sent"]
assert must(api("GET", f"/billing-runs/{RUN}", FINANCE))["invoice_id"] == "INV-0001" and len(nb.invoices) == 1
assert must(api("POST", "/billing-runs/invoices/send", SYSTEM))["sent"] == []
def test_ts_13_a_rolled_back_release_sends_nothing(nb, monkeypatch):
approved_week()
made()
from brick_timesheets.infrastructure.tables import OutboxRepository
kept = OutboxRepository.add_many
def boom(self, rows, at):
raise RuntimeError("the commit fails")
monkeypatch.setattr(OutboxRepository, "add_many", boom)
try:
release()
except Exception:
pass
assert nb.invoice_posts == 0
monkeypatch.setattr(OutboxRepository, "add_many", kept)
assert must(api("GET", f"/billing-runs/{RUN}", FINANCE))["status"] == "draft"
def test_ts_13_non_billable_time_is_never_claimed(nb):
open_week()
n = line(SHEET, project_id=P2, activity_code="DEV", billable=False)["lines"][-1]["line_no"]
hours(SHEET, n, {"2026-11-19": "7.50"})
a = line(SHEET, project_id=P1, activity_code="DEV")["lines"][-1]["line_no"] # a sales-order line: the order bills it, never ar
hours(SHEET, a, {"2026-11-16": "7.50", "2026-11-17": "7.50", "2026-11-18": "7.50", "2026-11-20": "0"})
submit()
must(approve())
assert must(create())["created"] is False and claim_rows() == []
def test_ts_13_a_correction_is_billed_as_its_difference_in_the_next_run(nb):
approved_week()
made()
must(release())
correct("9.50") # +2.00 on P2 (Thursday is over-covered: allowed)
run2 = made()
assert run2["run_id"] == RUN + ":2" and run2["source_id"] == SOURCE + ":2"
assert [(x["quantity"], x["unit_price"]) for x in run2["lines"]] == [("2.0000", "110.0000")] and run2["total_amount"] == "220.00"
must(release(run2["run_id"]))
assert {k[1] for k in nb.invoices} == {SOURCE, SOURCE + ":2"}
assert sum(Decimal(c["quantity"]) for c in claim_rows()) == Decimal("9.5") # billed = the latest approved revision's hours
assert must(create())["created"] is False
def test_ts_13_hours_taken_away_are_a_negative_line_and_a_negative_run_waits_for_finance(nb):
approved_week()
made()
must(release())
nb.prices[("CONSULT-SR", 7311)] = {"unit_price": "150.0000"} # the price list moved since
correct("5.00", "2.50") # -2.50: a write-off of hours already billed
run2 = made()
assert [(x["quantity"], x["unit_price"]) for x in run2["lines"]] == [("-2.5000", "110.0000")] # taken back at the price it was billed at
assert run2["total_amount"] == "-275.00"
refused(release(run2["run_id"]), 409, "not_negative")
assert nb.invoice_posts == 1
must(api("POST", f"/billing-runs/{run2['run_id']}/cancel", FINANCE, {"reason": "credit note by hand in ar"}))
assert sum(Decimal(c["quantity"]) for c in claim_rows()) == Decimal("7.5") # the claim was let go
run3 = made() # ... so the next run takes the difference again
assert run3["run_id"] == RUN + ":3" and run3["total_amount"] == "-275.00"
def test_ts_13_hours_written_off_in_a_correction_are_taken_back_whole(nb):
approved_week()
made()
must(release())
correct("0", "7.50") # no use case unticks a billed line: the write-off is the hours taken away
run2 = made()
assert run2["lines"][0]["quantity"] == "-7.5000"
# ── the gates ─────────────────────────────────────────────────────────────────────────────────────
def test_ts_13_the_gates_of_a_run(nb):
approved_week()
refused(create(project=P1), 409, "bills_via_ar")
refused(create(project="pj:contract:NOPE"), 404, "does not know")
refused(create(period="2026-13"), 422, "billing month")
refused(create(period="2026-12", at=MONTH_END), 409, "has not started")
refused(create(who=EMPLOYEE), 403, "ts:billing_run:create")
refused(create(who=FINANCE), 403, "ts:billing_run:create")
assert claim_rows() == [] and must(api("GET", "/billing-runs", FINANCE))["runs"] == []
def test_ts_14_no_price_stops_the_run_and_nothing_is_claimed(nb):
approved_week()
del nb.prices[("CONSULT-SR", 7311)]
refused(create(), 409, "price")
assert claim_rows() == [] and must(api("GET", "/billing-runs", FINANCE))["runs"] == []
nb.prices[("CONSULT-SR", 7311)] = {"unit_price": "3000.0000", "pricing_quantity": 60, "pricing_unit_code": "HUR"}
refused(create(), 409, "price_unit")
nb.prices[("CONSULT-SR", 7311)] = {"unit_price": "110.0000"}
nb.product_status = 503
refused(create(), 502, "product failed")
nb.product_status = None
assert made()["total_amount"] == "825.00"
def test_ts_13_the_releaser_is_never_a_person_whose_hours_are_in_the_run(nb):
approved_week()
made()
mine = person(EMP, [*ROLES["employee"], "ts:billing_run:release"]) # 4410 is also in finance
refused(release(who=mine), 409, "second_person")
assert must(api("GET", f"/billing-runs/{RUN}", FINANCE))["status"] == "draft"
assert must(release(who=FINANCE2))["status"] == "released"
def test_ts_13_hold_cancel_and_release_only_move_forward(nb):
approved_week()
made()
refused(api("POST", f"/billing-runs/{RUN}/hold", FINANCE, {"reason": " "}), 422, "reason")
assert must(api("POST", f"/billing-runs/{RUN}/hold", FINANCE, {"reason": "customer disputes"}))["status"] == "held"
refused(api("POST", f"/billing-runs/{RUN}/hold", FINANCE, {"reason": "again"}), 409, "only a draft")
assert must(release())["status"] == "released" # a held run is released from held
refused(api("POST", f"/billing-runs/{RUN}/cancel", FINANCE, {"reason": "too late"}), 409, "only a draft or held")
refused(api("POST", f"/billing-runs/{RUN}/hold", FINANCE, {"reason": "too late"}), 409, "only a draft")
assert must(api("GET", f"/billing-runs/{RUN}", FINANCE))["status"] == "released"
def test_ts_13_a_cancelled_run_lets_go_of_its_hours_and_is_terminal(nb):
approved_week()
made()
must(api("POST", f"/billing-runs/{RUN}/cancel", FINANCE, {"reason": "wrong project"}))
assert claim_rows() == []
refused(release(), 409, "is cancelled")
refused(api("POST", f"/billing-runs/{RUN}/cancel", FINANCE, {"reason": "again"}), 409, "is cancelled")
run = made() # the lines fall into the next run
assert run["run_id"] == RUN + ":2" and run["total_amount"] == "825.00"
def test_ts_13_two_runs_racing_for_the_same_hours_leave_one_claim(nb):
approved_week()
made()
with use_cases.repository() as repo: # a second worker that read before the first committed
import pytest
from sqlalchemy.exc import IntegrityError
with pytest.raises(IntegrityError):
repo.billing.add_many([{"sheet_id": SHEET, "revision": 1, "line_no": 4, "purpose": "ar_run", "run_id": RUN + ":2", "quantity": Decimal("7.5"),
"unit_price": Decimal("110")}], use_cases._now())
assert len(claim_rows()) == 1
def test_ts_13_the_tick_makes_a_run_for_every_project_with_hours_waiting_and_is_safe_twice(nb):
approved_week()
with on_day(MONTH_END):
first = must(api("POST", "/billing-runs/tick", SYSTEM, {"period_id": "2026-11"}))
second = must(api("POST", "/billing-runs/tick", SYSTEM, {"period_id": "2026-11"}))
assert first["projects"] == [{"project_id": P2, "created": True, "run_id": RUN}]
assert second["projects"] == [{"project_id": P2, "created": False, "run_id": None}]
assert len(claim_rows()) == 1
def test_a_lead_sees_no_billing_runs_and_finance_lists_them(nb):
approved_week()
made()
refused(api("GET", "/billing-runs", LEADER), 403, "ts:billing_run:release")
refused(api("GET", "/billing-runs", MANAGER), 403, "ts:billing_run:release")
assert [r["run_id"] for r in must(api("GET", "/billing-runs", FINANCE, {"status": "draft"}))["runs"]] == [RUN]
assert must(api("GET", "/billing-runs", FINANCE, {"project_id": P1}))["runs"] == []
refused(api("GET", "/billing-runs", FINANCE, {"status": "nope"}), 422, "status")
def test_the_arithmetic_rounds_each_invoice_line_once_and_refuses_what_an_invoice_cannot_hold():
lines, total = bd.invoice_lines([{"product_id": "X", "quantity": Decimal("0.3333"), "unit_price": Decimal("110.0000"), "tax_category": ""},
{"product_id": "X", "quantity": Decimal("0.3333"), "unit_price": Decimal("110.0000"), "tax_category": ""}], "2026-11")
assert lines[0]["quantity"] == "0.6666" and total == Decimal("73.33") # (0.6666 x 110) rounded once, not two rounded halves
import pytest
from brick_timesheets.domain.errors import RuleBroken
with pytest.raises(RuleBroken):
bd.invoice_lines([{"product_id": "X", "quantity": Decimal("99999999999.0000"), "unit_price": Decimal("999999999.0000"), "tax_category": ""}], "2026-11")