dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
tests/fakes.py
"""In-memory stand-ins for the neighbours, registered as local seams (erp_platform.seam.register_local): hr (the person, the employment, the work
pattern, the derived manager), projects (what is bookable, the time terms), time-off (the approved absences), gl (the ledger, cost centres) and
sales-order (the delivery). They hold plain dicts the tests fill, answer the contract infrastructure/seams.py documents, and can be taken DOWN.
conftest.py checks every call against the CALLEE's signed spec."""
from datetime import date, timedelta
from decimal import Decimal
from typing import Optional
from fastapi import FastAPI, Response
from fastapi.responses import JSONResponse
class Neighbours:
def __init__(self):
self.persons: dict = {} # sub -> {person_id, bp_id, company_code, email}
self.starts_on: dict = {} # person_id -> first day hr says "employed"
self.left_on: dict = {} # person_id -> the day after which hr says "not employed"
self.managers: dict = {} # person_id -> [level 1, level 2]: each {person_id, bp_id, pool_user_sub, email} or None
self.cost_centres: dict = {} # person_id -> home cost centre
self.minutes: dict = {} # person_id -> contracted minutes a weekday (default 450)
self.holidays: dict = {} # ISO date -> the bank holiday's name (every employee: 0 expected)
self.hr_status: Optional[int] = None # 500 takes hr down
self.projects: dict = {} # project_id -> time terms (+ items, name)
self.project_calls = {"bookable": 0, "terms": 0}
self.projects_status: Optional[int] = None
self.absences: list = [] # time-off rows
self.absence_calls = 0
self.toff_status: Optional[int] = None
self.journals: dict = {} # journal_id -> body
self.journal_posts = 0 # every POST, replays included
self.closed_through: Optional[str] = None # a journal_date on or before this is refused 409 period_open
self.gl_status: Optional[int] = None # 500 takes gl down
self.gl_cost_centres: dict = {"CONS-LON": "open", "DELIV-1": "open", "DELIV-2": "open", "OPS-LON": "open"}
self.orders_book: dict = {} # sales_order_id -> {line_no: {"ordered", "delivered"}}
self.deliveries: dict = {} # delivery_ref -> body
self.delivery_posts = 0
self.orders_status: Optional[int] = None
self.prices: dict = {} # (product_id, bp_id) -> {"unit_price": "110.0000", ...}: product's answer for HUR
self.price_calls: list = [] # every priced_on asked
self.product_status: Optional[int] = None
self.invoices: dict = {} # source_id -> the raise body
self.invoice_posts = 0
self.ar_status: Optional[int] = None
self.hr = self._hr()
self.projects_app = self._projects()
self.time_off = self._time_off()
self.gl = self._gl()
self.orders = self._orders()
self.product = self._product()
self.ar = self._ar()
# ── builders ────────────────────────────────────────────────────────────────────────────────
def employee(self, person_id: int, email: str, *, company: str = "UK01", bp_id: Optional[int] = None, cost_centre: str = "CONS-LON",
manager: Optional[dict] = None, manager2: Optional[dict] = None) -> dict:
p = {"person_id": person_id, "bp_id": bp_id if bp_id is not None else 70000 + person_id, "company_code": company, "email": email,
"pool_user_sub": email}
self.persons[email] = p
self.cost_centres[person_id] = cost_centre
self.managers[person_id] = [manager, manager2]
return p
def as_manager(self, person_id: int, email: str) -> dict:
return {"person_id": person_id, "bp_id": 70000 + person_id, "pool_user_sub": email, "email": email}
def absence(self, person_id: int, ref: str, leave_type: str, dates: list, minutes: int = 450, status: str = "approved") -> dict:
row = {"absence_ref": ref, "revision": 1, "person_id": person_id, "leave_type_code": leave_type, "pay_treatment": "paid", "status": status,
"evidence_ok": True, "statutory": None, "changed_at": "2026-11-01T09:00:00Z",
"days": [{"date": d, "part": "full", "minutes": minutes} for d in dates]}
self.absences.append(row)
return row
def standard_projects(self) -> None:
"""P1 (a sales-order project, 120 HUR ordered on line 2, 40 delivered), P2 (a contract billed through ar), P3 fixed price, P4 non-billable."""
base = {"state": "active", "bp_id": 7311, "currency": "GBP", "lead_person_id": 3020}
self.projects["pj:sales_order:quote:Q-2026-0042@v2"] = {**base, "name": "Implementation", "billing": "time_and_materials", "bill_via": "sales_order",
"sales_order_id": "quote:Q-2026-0042@v2", "cost_centre": "DELIV-1",
"products": [{"rate_group": "SENIOR", "product_id": "CONSULT-SR", "so_line_no": 2}, {"rate_group": "JUNIOR", "product_id": "CONSULT-JR", "so_line_no": 3}],
"items": [{"item_id": "quote:Q-2026-0042@v2|2|1", "title": "Implementation", "state": "open"}]}
self.projects["pj:contract:CT-2026-0009"] = {**base, "name": "Support contract", "billing": "time_and_materials", "bill_via": "ar", "sales_order_id": None,
"cost_centre": "DELIV-2", "products": [{"rate_group": "SENIOR", "product_id": "CONSULT-SR", "so_line_no": None}], "items": []}
self.projects["pj:manual:FIXED-1"] = {**base, "name": "Fixed price", "billing": "fixed_price", "bill_via": "none", "sales_order_id": None,
"cost_centre": "DELIV-1", "products": [], "items": []}
self.projects["pj:manual:INTERNAL-1"] = {**base, "name": "Internal tooling", "billing": "non_billable", "bill_via": "none", "sales_order_id": None,
"cost_centre": "DELIV-2", "products": [], "items": []}
self.orders_book["quote:Q-2026-0042@v2"] = {2: {"ordered": Decimal("120"), "delivered": Decimal("40")}, 3: {"ordered": Decimal("50"), "delivered": Decimal("0")}}
# ── the apps ────────────────────────────────────────────────────────────────────────────────
def _hr(self) -> FastAPI:
app, me = FastAPI(), self
def person(pid: int):
return next((p for p in me.persons.values() if p["person_id"] == pid), None)
def employed(pid: int, on: str) -> bool:
start, gone = me.starts_on.get(pid), me.left_on.get(pid)
return not ((start and on < start) or (gone and on > gone))
@app.get("/persons")
def by_sub(response: Response, pool_user_sub: str):
"""hr HR-58: 200 {person_id, bp_id, company_code}; 404 when no person carries the sub."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
p = me.persons.get(pool_user_sub)
if p is None:
response.status_code = 404
return {"detail": "no such person"}
return {k: p[k] for k in ("person_id", "bp_id", "company_code")}
@app.get("/persons/{person_id}/employment")
def employment(person_id: int, on: str, response: Response):
"""hr HR-56: 200 {person_id, on, employed, engagement_id, worker_type, company_code, org_unit_id, location_code, cost_centre, bp_id}."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
p = person(person_id)
if p is None:
response.status_code = 404
return {"detail": "no such person"}
return {"person_id": person_id, "on": on, "employed": employed(person_id, on), "company_code": p["company_code"],
"cost_centre": me.cost_centres.get(person_id), "bp_id": p["bp_id"]}
@app.get("/persons/{person_id}/manager")
def manager(person_id: int, on: str, level: int, response: Response):
"""hr HR-57: 200 {person_id, on, level, manager: {...} | null, source}."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
chain = me.managers.get(person_id, [None, None])
return {"person_id": person_id, "on": on, "level": level, "manager": chain[level - 1] if level <= len(chain) else None, "source": "org_tree"}
@app.get("/persons/{person_id}/work-pattern")
def pattern(person_id: int, response: Response, from_: str = "", to: str = ""):
"""hr HR-62: 200 {person_id, from, to, days: [{date, employed, contracted_minutes, public_holiday, location_code, holiday_calendar, fte}]}."""
if me.hr_status:
response.status_code = me.hr_status
return {"detail": "down"}
if person(person_id) is None:
response.status_code = 404
return {"detail": "no such person"}
days, d = [], date.fromisoformat(from_)
while d <= date.fromisoformat(to):
iso = d.isoformat()
weekday = d.weekday() < 5
holiday = me.holidays.get(iso)
days.append({"date": iso, "employed": employed(person_id, iso), "contracted_minutes": 0 if (holiday or not weekday) else me.minutes.get(person_id, 450),
"public_holiday": holiday, "location_code": "LON", "holiday_calendar": "UK-ENG", "fte": 1.0})
d += timedelta(days=1)
return {"person_id": person_id, "from": from_, "to": to, "days": days}
return _alias_from(app)
def _projects(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/projects/bookable")
def bookable(person_id: int, on: str, response: Response):
"""projects PJ-25: 200 {person_id, on, projects: [{project_id, name, bp_id, billing, items: [{item_id, title, state}]}]}."""
me.project_calls["bookable"] += 1
if me.projects_status:
response.status_code = me.projects_status
return {"detail": "down"}
return {"person_id": person_id, "on": on, "projects": [
{"project_id": pid, "name": t["name"], "bp_id": t["bp_id"], "billing": t["billing"], "items": [i for i in t["items"] if i["state"] == "open"]}
for pid, t in me.projects.items() if t["state"] == "active"]}
@app.get("/projects/{project_id}/time-terms")
def terms(project_id: str, on: str, response: Response):
"""projects PJ-25: 200 {project_id, on, state, bp_id, currency, billing, bill_via, sales_order_id, products, cost_centre, lead_person_id}; 404 unknown."""
me.project_calls["terms"] += 1
if me.projects_status:
response.status_code = me.projects_status
return {"detail": "down"}
t = me.projects.get(project_id)
if t is None:
response.status_code = 404
return {"detail": "no such project"}
return {"project_id": project_id, "on": on, **{k: v for k, v in t.items() if k not in ("name", "items")}}
return app
def _time_off(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/absences")
def absences(company_code: str, response: Response, from_: str = "", to: str = "", person_id: Optional[int] = None, changed_since: Optional[str] = None):
"""time-off TOF-20: 200 {company_code, from, to, as_of, absences: [...]}; cancelled rows only with changed_since."""
me.absence_calls += 1
if me.toff_status:
response.status_code = me.toff_status
return {"detail": "down"}
rows = [a for a in me.absences if (person_id is None or a["person_id"] == person_id)
and (a["status"] != "cancelled" or changed_since) and any(from_ <= d["date"] <= to for d in a["days"])]
return {"company_code": company_code, "from": from_, "to": to, "as_of": "2026-11-20T15:00:00Z", "absences": rows}
return _alias_from(app)
def _gl(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/journal-entry/journals")
def journal(body: dict, response: Response):
"""gl GL-30 for timesheets: 201 {journal_id, replay: false}; 200 the same with replay: true; 409 {gate: period_open}."""
me.journal_posts += 1
if me.gl_status:
response.status_code = me.gl_status
return {"detail": "down"}
dr = sum(Decimal(x.get("debit", "0")) for x in body["lines"])
cr = sum(Decimal(x.get("credit", "0")) for x in body["lines"])
if dr != cr:
return JSONResponse({"gate": "balanced", "reason": f"{dr} against {cr}", "fix": "balance it"}, status_code=422)
if me.closed_through and body["journal_date"] <= me.closed_through:
return JSONResponse({"gate": "period_open", "reason": f"the period of {body['journal_date']} is closed", "fix": "post it in the next open period"},
status_code=409)
replay = body["journal_id"] in me.journals
me.journals[body["journal_id"]] = body
response.status_code = 200 if replay else 201
return {"journal_id": body["journal_id"], "status": "posted", "replay": replay, "lines": len(body["lines"])}
@app.get("/cost-centres/{cost_centre}")
def cost_centre(cost_centre: str, response: Response):
"""gl GL-61: 200 {cost_centre, name, status: open | closed, company_code}; 404 when unknown."""
if cost_centre not in me.gl_cost_centres:
response.status_code = 404
return {"detail": "no such cost centre"}
return {"cost_centre": cost_centre, "name": cost_centre, "status": me.gl_cost_centres[cost_centre], "company_code": "UK01"}
return app
def _orders(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/sales-orders/{sales_order_id}/deliveries")
def delivery(sales_order_id: str, body: dict, response: Response):
"""sales-order SO-51: the order view; 409 {gate: service_line | within_ordered | not_below_invoiced, reason, fix}; a repeated delivery_ref replays."""
me.delivery_posts += 1
if me.orders_status:
response.status_code = me.orders_status
return {"detail": "down"}
if body["delivery_ref"] in me.deliveries:
return {"sales_order_id": sales_order_id, "replay": True}
line = (me.orders_book.get(sales_order_id) or {}).get(body["line_no"])
if line is None:
return JSONResponse({"gate": "service_line", "reason": f"line {body['line_no']} is not a service line in HUR", "fix": "deliver a service line"},
status_code=409)
qty = Decimal(body["quantity"])
if qty > 0 and line["delivered"] + qty > line["ordered"]:
return JSONResponse({"gate": "within_ordered", "reason": f"{line['delivered'] + qty} of {line['ordered']} ordered", "fix": "amend the order"},
status_code=409)
if qty < 0 and not body.get("corrects_ref"):
return JSONResponse({"gate": "not_below_invoiced", "reason": "a negative delivery corrects an earlier one", "fix": "name it"}, status_code=409)
line["delivered"] += qty
me.deliveries[body["delivery_ref"]] = {"sales_order_id": sales_order_id, **body}
return {"sales_order_id": sales_order_id, "replay": False, "delivered": str(line["delivered"])}
return app
def _product(self) -> FastAPI:
app, me = FastAPI(), self
@app.get("/products/{product_id}/price")
def price(product_id: str, response: Response, quantity: str = "1", unit_code: str = "HUR", bp_id: Optional[int] = None, currency: str = "GBP",
priced_on: str = ""):
"""product PRD-31: 200 {product_id, currency, unit_price, pricing_quantity, pricing_unit_code, amount, priced_from, line: {tax_category}}; 404 no price."""
me.price_calls.append(priced_on)
if me.product_status:
response.status_code = me.product_status
return {"detail": "down"}
row = me.prices.get((product_id, bp_id))
if row is None:
response.status_code = 404
return {"detail": "no price"}
amount = (Decimal(row["unit_price"]) * Decimal(quantity)).quantize(Decimal("0.01"))
return {"product_id": product_id, "currency": currency, "unit_price": row["unit_price"], "pricing_quantity": row.get("pricing_quantity", 1),
"pricing_unit_code": row.get("pricing_unit_code", "HUR"), "amount": str(amount),
"priced_from": {"price_list_id": "LIST-GBP", "rule_seq": 1, "row": 1}, "line": {"tax_category": row.get("tax_category", "S"), "unit_code": "HUR"}}
return app
def _ar(self) -> FastAPI:
app, me = FastAPI(), self
@app.post("/invoices/raise")
def raise_invoice(body: dict, response: Response):
"""ar AR-20 for timesheets: 201 the draft invoice; 200 the same for the same (source_kind, source_id); 422 {gate: not_negative}."""
me.invoice_posts += 1
if me.ar_status:
response.status_code = me.ar_status
return {"detail": "down"}
total = sum(Decimal(x["quantity"]) * Decimal(x["unit_price"]) for x in body["lines"])
if total < 0:
return JSONResponse({"gate": "not_negative", "reason": f"the draft totals {total}", "fix": "credit by hand"}, status_code=422)
key = (body["source_kind"], body["source_id"])
if key in me.invoices:
response.status_code = 200
return {"invoice_id": me.invoices[key]["invoice_id"], "status": "draft", "replay": True}
me.invoices[key] = {**body, "invoice_id": f"INV-{len(me.invoices) + 1:04d}"}
response.status_code = 201
return {"invoice_id": me.invoices[key]["invoice_id"], "status": "draft", "replay": False}
return app
def _alias_from(app: FastAPI) -> FastAPI:
"""The seams send `from` (a Python keyword); the fakes read it as `from_`: rewrite the query key on the way in."""
@app.middleware("http")
async def rename(request, call_next):
q = request.scope.get("query_string", b"").decode()
if "from=" in q:
request.scope["query_string"] = q.replace("from=", "from_=").encode()
return await call_next(request)
return app