dodil/erp-brick-timesheets
DODIL ERP brick: timesheets
tests/test_migration.py
"""TS-22: the Odoo slice (workflow ts_migration_run): validated employee-weeks as approved revision-1 weeks, hours Odoo invoiced marked billed, the hours not yet
invoiced reconciled per order line and signed by the owner - read through a port that answers fake rows here (no live Odoo)."""
from decimal import Decimal
import pytest
from conftest import EMPLOYEE, FINANCE, OWNER, P1, P2, SYSTEM, api, must, on_day, refused
from sqlalchemy import func, select
from brick_timesheets.application import use_cases
from brick_timesheets.infrastructure import odoo as odoo_port
from brick_timesheets.infrastructure import tables as t
DB = "acme"
CUTOVER = "2026-11-01"
RUN = f"odoo:{DB}:{CUTOVER}"
ORDER = "odoo:acme:77"
def aal(i, emp, day, hours, project=21, so=None, invoice=None, valid=True):
return {"id": i, "name": f"work {i}", "date": day, "unit_amount": hours, "amount": -round(hours * 52, 2), "employee_id": [emp, f"Employee {emp}"],
"project_id": [project, f"Project {project}"], "task_id": False, "validated": valid, "validated_status": "validated" if valid else "draft",
"so_line": [so, "SO line"] if so else False, "timesheet_invoice_id": [invoice, "INV/2026/0001"] if invoice else False, "company_id": [1, "Acme"]}
class FakeOdoo:
def __init__(self):
self.currency = "GBP"
self.down = None
self.rows = [
aal(1, 11, "2026-11-16", 7.5, 21, so=301), # employee 11, W47: 7.50 on order line 301, not invoiced
aal(2, 11, "2026-11-17", 7.5, 21, so=301, invoice=900), # ... 7.50 more, invoiced in Odoo
aal(3, 11, "2026-11-18", 4.0, 22), # ... 4.00 on a project with no order line
aal(4, 12, "2026-11-16", 8.0, 21, so=301), # employee 12, W47: 8.00, not invoiced
aal(5, 12, "2026-11-09", 6.0, 21, so=301), # employee 12, W46 ...
aal(6, 12, "2026-11-10", 2.0, 21, so=301, valid=False), # ... with an unvalidated line: the week is not carried
aal(7, 11, "2025-12-01", 3.0, 21), # 2025: outside the years asked
]
def _check(self):
if self.down:
raise odoo_port.OdooUnavailable(502, self.down)
def company_currency(self, db):
self._check()
return self.currency
def timesheet_lines(self, db, years):
self._check()
return [dict(r) for r in self.rows]
@pytest.fixture
def odoo():
fake = FakeOdoo()
odoo_port.ports.reader = fake
yield fake
odoo_port.ports.reader = None
def body(**kw):
return {"odoo_db": DB, "cutover_date": CUTOVER, "company_code": "UK01", "years": [2026], "employee_map": {"11": 4410, "12": 4425},
"project_map": {"21": P1, "22": P2}, "so_line_map": {"301": {"sales_order_id": ORDER, "so_line_no": 2}}, "upstream_signed": True, **kw}
def load(who=None, **kw):
return api("POST", "/migration/odoo/load", who or SYSTEM, body(**kw))
def reconcile(run=RUN, who=None):
return api("POST", f"/migration/odoo/{run}/reconcile", who or SYSTEM)
def count(table):
with use_cases.repository() as repo:
return repo.conn.execute(select(func.count()).select_from(table)).scalar()
def test_ts_22_each_validated_employee_week_becomes_an_approved_revision_one_and_an_unvalidated_week_is_not_carried(odoo, nb):
got = must(load())
assert got["status"] == "loaded" and got["run_id"] == RUN
assert got["stats"]["weeks_carried"] == 2 and got["stats"]["weeks_not_validated"] == 1 and got["stats"]["odoo_lines"] == 6 # 2025 is outside the years
s = must(api("GET", "/sheets/odoo:acme:11:2026-W47", FINANCE))
assert (s["status"], s["revision"], s["person_id"], s["week"], s["worked_hours"]) == ("approved", 1, "4410", "2026-W47", "19.00")
assert [(ln["project_id"], ln["hours"], ln["billable"], ln["bill_via"], ln["sales_order_id"], ln["so_line_no"]) for ln in s["lines"]] == [
(P2, "4.00", False, "none", None, None), (P1, "15.00", True, "sales_order", ORDER, 2)] # lines in project order
assert [e["hours"] for e in s["lines"][1]["entries"]] == ["7.50", "7.50"]
assert api("GET", "/sheets/odoo:acme:12:2026-W46", FINANCE)[0] == 404 # W46 is not carried
assert must(api("GET", "/sheets/odoo:acme:12:2026-W47", FINANCE))["worked_hours"] == "8.00"
def test_ts_22_nothing_is_posted_or_sent_and_the_hours_are_marked_delivered_and_billed(odoo, nb):
must(load())
assert nb.journals == {} and nb.deliveries == {} and nb.invoices == {} and nb.journal_posts == 0 and nb.delivery_posts == 0 # gl's and the order's migrations carry these
assert count(t.ts_outbox) == 0
with use_cases.repository() as repo:
claims = {(c["sheet_id"], c["purpose"], c["line_no"]): c for c in repo.migration.claims_of(["odoo:acme:11:2026-W47", "odoo:acme:12:2026-W47"])}
d = claims[("odoo:acme:11:2026-W47", "delivery", 2)]
assert (d["quantity"], d["sales_order_id"], d["so_line_no"], d["delivery_ref"]) == (Decimal("15"), ORDER, 2, "odoo:acme:odoo:acme:11:2026-W47:2")
b = claims[("odoo:acme:11:2026-W47", "ar_run", 2)] # the 7.50 Odoo invoiced
assert (b["quantity"], b["run_id"]) == (Decimal("7.5"), f"tsbill:{P1}:{RUN}")
assert ("odoo:acme:12:2026-W47", "ar_run", 1) not in claims and ("odoo:acme:12:2026-W47", "ar_run", 2) not in claims
run = must(api("GET", f"/billing-runs/tsbill:{P1}:{RUN}", FINANCE))
assert run["status"] == "released" and run["lines"] == [] and run["claims"] == 1
hrs = must(api("GET", "/project-hours", FINANCE, {"project_id": P1, "from": "2026-11-01", "to": "2026-11-30"}))
assert (hrs["approved"], hrs["billed"], hrs["cost"]) == ("23.00", "7.50", "1196.00") and hrs["currency"] == "GBP" # Odoo's own cost: 23 x 52
def test_ts_22_a_second_load_writes_nothing_new_and_a_late_line_is_reported_not_edited_in(odoo, nb):
must(load())
before = (count(t.ts_sheet), count(t.ts_entry), count(t.ts_migration_map), count(t.ts_billing_claim))
again = must(load())
assert again["stats"]["weeks_carried"] == 0 and again["stats"]["weeks_unchanged"] == 2 and (count(t.ts_sheet), count(t.ts_entry), count(t.ts_migration_map),
count(t.ts_billing_claim)) == before
odoo.rows.append(aal(8, 11, "2026-11-19", 2.0, 21, so=301)) # a line added to a carried week in Odoo afterwards
odoo.rows[5]["validated"], odoo.rows[5]["validated_status"] = True, "validated" # and the W46 week is now validated
late = must(load())
assert late["stats"]["late_lines"] == 1 and late["stats"]["weeks_carried"] == 1 and count(t.ts_sheet) == 3
assert must(api("GET", "/sheets/odoo:acme:11:2026-W47", FINANCE))["worked_hours"] == "19.00" # an approved week is never edited
def test_ts_22_the_doors_must_answer_for_everything_odoo_holds(odoo, nb):
refused(load(upstream_signed=False), 409, "upstream_signed")
got = refused(load(employee_map={"11": 4410}, project_map={"21": P1}, so_line_map={}), 409, "doors_decided")
for word in ("employee 12", "project 22", "order line 301"):
assert word in str(got["detail"])
refused(load(activity_code="NOPE"), 409, "activity NOPE")
assert count(t.ts_sheet) == 0 and count(t.ts_migration_run) == 0
odoo.rows.append(aal(9, 11, "2026-11-20", 25.0, 21))
refused(load(), 409, "at most 24")
odoo.rows.pop()
odoo.rows.append(aal(10, 11, "2026-11-21", -1.0, 21))
refused(load(), 409, "zero or more")
def test_ts_22_odoo_down_or_unconfigured_is_a_reply_and_nothing_is_half_loaded(odoo, nb):
odoo.down = "Odoo did not answer"
refused(load(), 502, "did not answer")
assert count(t.ts_sheet) == 0
odoo_port.ports.reader = None
refused(load(), 409, "no Odoo connection")
def test_ts_22_reconcile_signs_what_is_equal_and_the_hours_not_yet_invoiced_equal_odoos_per_order_line(odoo, nb):
must(load())
got = must(reconcile())
assert got["status"] == "reconciled" and [c["check"] for c in got["checks"]] == ["unbilled_hours_per_order_line", "approved_hours_per_project", "weeks_carried_per_employee"]
c1, c2, c3 = got["checks"]
assert (c1["odoo"], c1["ours"], c1["equal"]) == ("15.5000", "15.5000", True) # 7.50 + 8.00 not yet invoiced on order line 301
assert c1["detail"] == {"301": {"odoo": "15.5000", "ours": "15.5000"}}
assert c2["detail"] == {P1: {"odoo": "23.0000", "ours": "23.0000"}, P2: {"odoo": "4.0000", "ours": "4.0000"}} and c2["equal"]
assert c3["detail"] == {"11": {"odoo": "1", "ours": "1"}, "12": {"odoo": "1", "ours": "1"}} and c3["equal"] and got["readiness"][0]["met"]
refused(api("POST", f"/migration/odoo/{RUN}/sign", FINANCE), 403, "ts:migration:sign")
signed = must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))
assert signed["status"] == "signed" and signed["signed_by"] and must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))["status"] == "signed"
refused(load(), 409, "signed")
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "oops"}), 409, "only a loaded or reconciled")
refused(reconcile(), 409, "only a loaded run")
def test_ts_22_a_difference_blocks_the_signature_until_the_owner_explains_it(odoo, nb):
must(load())
odoo.rows.append(aal(8, 11, "2026-11-19", 2.0, 21, so=301)) # Odoo changed after the load
got = must(reconcile())
c1 = got["checks"][0]
assert (c1["odoo"], c1["ours"], c1["equal"]) == ("17.5000", "15.5000", False) and not got["readiness"][0]["met"]
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "checks_reconciled")
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "weeks_carried_per_employee", "reason": "x"}), 409, "nothing to explain")
refused(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "unbilled_hours_per_order_line", "reason": " "}), 422, "reason")
must(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "unbilled_hours_per_order_line", "reason": "2.00 added to Odoo on the 19th, billed there by hand"}))
refused(api("POST", f"/migration/odoo/{RUN}/sign", OWNER), 409, "approved_hours_per_project") # the same line moved the project's hours too
must(api("POST", f"/migration/odoo/{RUN}/explain", OWNER, {"check": "approved_hours_per_project", "reason": "the same 2.00"}))
assert must(api("POST", f"/migration/odoo/{RUN}/sign", OWNER))["status"] == "signed"
def test_ts_22_a_loaded_run_is_discarded_cleanly_and_the_next_run_is_its_own(odoo, nb):
must(load())
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": " "}), 422, "reason")
refused(api("POST", f"/migration/odoo/{RUN}/discard", FINANCE, {"reason": "x"}), 403, "ts:migration:sign")
must(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "start over"}))
assert [count(x) for x in (t.ts_sheet, t.ts_line, t.ts_entry, t.ts_billing_claim, t.ts_migration_map, t.ts_billing_run)] == [0, 0, 0, 0, 0, 0]
assert must(api("GET", f"/migration/odoo/{RUN}", FINANCE))["status"] == "discarded"
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "again"}), 409, "only a loaded or reconciled")
assert must(load())["run_id"] == RUN + ":2" and count(t.ts_sheet) == 2
def test_ts_22_a_week_that_was_corrected_since_the_load_pins_the_run(odoo, nb):
must(load())
must(api("POST", "/sheets/odoo:acme:11:2026-W47/reopen", FINANCE, {"reason": "a correction"})) # finance reopens: revision 2 opens, nothing is sent
assert nb.delivery_posts == 0 and nb.journal_posts == 0 and nb.invoice_posts == 0
refused(api("POST", f"/migration/odoo/{RUN}/discard", OWNER, {"reason": "start over"}), 409, "nothing_moved")
assert count(t.ts_sheet) == 3
def test_ts_22_the_map_answers_what_was_migrated_and_omits_what_was_not(odoo, nb):
must(load())
got = must(api("GET", f"/migration-map/{DB}/account.analytic.line", FINANCE, {"ids": "1,2,6,99"}))
assert got == [{"odoo_model": "account.analytic.line", "odoo_id": "1", "ours": "odoo:acme:11:2026-W47:1:2:2026-11-16"},
{"odoo_model": "account.analytic.line", "odoo_id": "2", "ours": "odoo:acme:11:2026-W47:1:2:2026-11-17"}] # 6 is in a week not carried
refused(api("GET", f"/migration-map/{DB}/account.analytic.line", FINANCE, {"ids": "1;2"}), 422, "digits")
refused(api("GET", f"/migration-map/{DB}/account.analytic.line", EMPLOYEE, {"ids": "1"}), 403, "ts:migration:read")
def test_ts_22_no_billing_run_bills_what_odoo_invoiced_or_hours_an_order_delivers(odoo, nb):
must(load())
with on_day("2026-11-30T10:00:00+00:00"):
refused(api("POST", "/billing-runs", SYSTEM, {"project_id": P1, "period_id": "2026-11"}), 409, "bills_via_ar") # an order project: the order bills it
assert must(api("POST", "/billing-runs/tick", SYSTEM, {"period_id": "2026-11"}))["projects"] == [] # no carried hour waits for a run
assert nb.invoice_posts == 0
def test_ts_22_the_roles_are_exact(odoo, nb):
refused(load(who=FINANCE), 403, "ts:migration:load")
refused(load(who=OWNER), 403, "ts:migration:load")
refused(reconcile(who=OWNER), 403, "ts:migration:reconcile")
refused(api("GET", "/migration/odoo", EMPLOYEE), 403, "ts:migration:read")
assert must(api("GET", "/migration/odoo", FINANCE))["runs"] == []
must(load())
assert [r["status"] for r in must(api("GET", "/migration/odoo", OWNER))["runs"]] == ["loaded"]