"""summary_by_project semantics over the migrated fixture graph (T-SUM-001/002/004/007/008/009) + the month math (T-SUM-003/004).""" from datetime import date, datetime from decimal import Decimal from sqlalchemy import text from app.services.reporting.summary import MonthWindow, parse_month, summary_by_project SITE_REBUILD = "07000000-0000-0000-0000-000000000001" def row_for(result, project_name): return next(r for r in result["rows"] if r["project"] == project_name) # ------------------------------------------------------------ month math def test_t_sum_003_month_parsing(): today = date(2026, 7, 7) assert parse_month("2026-7", today=today) == MonthWindow(2026, 7) assert parse_month("2026-07", today=today) == MonthWindow(2026, 7) assert parse_month("2026-3", today=today) == MonthWindow(2026, 3) # absent / malformed / out-of-range fall back silently to now for bad in (None, "", "garbage", "2026-13", "2026-0", "26-07"): assert parse_month(bad, today=today) == MonthWindow(2026, 7) def test_t_sum_004_month_bounds_and_rolls(): july = MonthWindow(2026, 7) assert july.start == datetime(2026, 7, 1) assert july.end == datetime(2026, 8, 1) assert july.shift(-1) == MonthWindow(2026, 6) # December rolls to January year+1; January back to December year-1 assert MonthWindow(2026, 12).end == datetime(2027, 1, 1) assert MonthWindow(2026, 1).shift(-1) == MonthWindow(2025, 12) assert MonthWindow(2026, 12).shift(1) == MonthWindow(2027, 1) # ----------------------------------------------------------- the windows def test_t_sum_001_windows_by_start_at(rpt_session, rian): """this=2026-03, prev=2026-02: Site Rebuild has te-1 (1h) + te-2 (0.5h) in March, te-7 (1.5h) in Feb, and nothing else all-time.""" result = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) site = row_for(result, "Site Rebuild") assert site["this_seconds"] == 5400 assert site["prev_seconds"] == 5400 assert site["all_seconds"] == 10800 def test_t_sum_002_dollars_are_stamp_sums(rpt_session, rian): """No rate lookup at read time: March cost = 14.00 + 7.00 stamps.""" result = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) site = row_for(result, "Site Rebuild") assert site["this_cost"]["amount"] == "21.00" assert site["this_billout"]["amount"] == "52.50" # 35 + 17.50 def test_t_sum_009_adjusted_billout_per_window(rpt_session, rian): """Site Rebuild: pct=10, amount=250 -> adjusted = raw*1.1 + 250 in EVERY window (the fixed amount lands in full per window, ADR #036 parity) — including a zero-raw window.""" result = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) site = row_for(result, "Site Rebuild") assert site["this_billout_adjusted"]["amount"] == "307.75" # 52.5*1.1+250 assert site["prev_billout_adjusted"]["amount"] == "349.00" # 90*1.1+250 assert site["all_billout_adjusted"]["amount"] == "406.75" # 142.5*1.1+250 # zero-raw window: May has no Site Rebuild entries -> adjusted = 250 may = summary_by_project(rpt_session, rian, MonthWindow(2026, 5)) assert row_for(may, "Site Rebuild")["this_billout"]["amount"] == "0.00" assert row_for(may, "Site Rebuild")["this_billout_adjusted"]["amount"] == "250.00" def test_t_sum_008_unassigned_rows_group_by_raw_triple_null_labels(rpt_session, rian): """te-4 (unresolved) groups on its raw triple with NULL client — placeholders are the frontend's job now.""" result = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) unassigned = next(r for r in result["rows"] if r["project_id"] is None) assert unassigned["operator"] == "PlusROI" assert unassigned["client"] is None assert unassigned["project"] == "Mystery" assert unassigned["this_seconds"] == 900 # never matches a project entity -> adjustments are zero-effect assert ( unassigned["this_billout_adjusted"]["amount"] == unassigned["this_billout"]["amount"] ) def test_t_sum_007_null_duration_rows_are_invisible_here(rpt_session, rian): """B14: the summary scan requires duration_seconds IS NOT NULL — a NULL-duration row changes NOTHING (no window, no group, no total).""" clean = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) nested = rpt_session.begin_nested() try: rpt_session.execute( text( "INSERT INTO time_entries (id, tenant_id, source, start_at," " description, duration_seconds)" " SELECT gen_random_uuid(), id, 'manual'," " '2026-03-15 09:00:00', 'no-duration row', NULL" " FROM tenants LIMIT 1" ) ) with_row = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) assert with_row == clean finally: nested.rollback() def test_totals_equal_sum_of_rows(rpt_session, rian): result = summary_by_project(rpt_session, rian, MonthWindow(2026, 3)) assert result["totals"]["this_seconds"] == sum( r["this_seconds"] for r in result["rows"] ) assert Decimal(result["totals"]["this_billout_adjusted"]["amount"]) == sum( Decimal(r["this_billout_adjusted"]["amount"]) for r in result["rows"] ) assert Decimal(result["totals"]["all_cost"]["amount"]) == sum( Decimal(r["all_cost"]["amount"]) for r in result["rows"] ) def test_t_sum_013_pivots_narrow_rows_and_totals(rpt_session, rian): result = summary_by_project(rpt_session, rian, MonthWindow(2026, 3), operator="roi") names = {r["project"] for r in result["rows"]} assert "Admin" not in names # Bowden Works-operated assert result["totals"]["all_seconds"] == sum( r["all_seconds"] for r in result["rows"] ) def test_non_owner_summary_has_no_billout_keys_and_is_scoped(rpt_session, adi): result = summary_by_project(rpt_session, adi, MonthWindow(2026, 3)) assert result["rows"], "adi has March rows of his own" for row in result["rows"]: assert "this_billout" not in row assert "this_billout_adjusted" not in row assert "this_cost" in row # cost was never owner-only assert "this_billout" not in result["totals"] # importer scoping: only adi's batch (te-1..4) contributes assert result["totals"]["all_seconds"] == 3600 + 1800 + 7200 + 900