"""Project income + display-only billout-adjustment EDITING (owner-only) — the M8 last-parity-gap write surface (services/projects_admin). Closes tier-A T-ADJ-020, T-PRJ-028, T-PRJ-029 and the owner-only 403 matrix + audit + adjustment-math-parity rows. Runs over the migrated reporting scratch on a SAVEPOINT-isolated session so the service's own commit()s roll back after each test — the shared scratch DB the other reporting tests read is never polluted. """ import uuid from decimal import Decimal import pytest from sqlalchemy import select from sqlalchemy.orm import Session from app.models import AuditLog, Project from app.services import projects_admin from app.services.authz.context import Actor from app.services.errors import ServiceError from app.services.reporting.filters import EntryFilters from app.services.reporting.projects import project_summary from tests.reporting.conftest import actor_named # ids carry unchanged through the migration (transforms §2.4) SITE_REBUILD = uuid.UUID("07000000-0000-0000-0000-000000000001") ADMIN = uuid.UUID("07000000-0000-0000-0000-000000000002") @pytest.fixture() def iso_session(rpt_engine): """SAVEPOINT-isolated session: service commit()s stay inside an outer transaction that rolls back after the test (mirrors the team-admin gate fixture), so write tests never pollute the shared scratch.""" conn = rpt_engine.connect() trans = conn.begin() session = Session(bind=conn, join_transaction_mode="create_savepoint") yield session session.close() trans.rollback() conn.close() @pytest.fixture() def owner(iso_session): actor = actor_named(iso_session, "rian") assert actor.allowed("can_transfer") return actor def _project(session, project_id): return session.scalar(select(Project).where(Project.id == project_id)) def _row(session, actor, project_name): result = project_summary(session, actor, EntryFilters()) return next(r for r in result["rows"] if r["project"] == project_name) # ---------------------------------------------------------------- parse / clear def test_t_prj_028_income_and_adjustments_parse(iso_session, owner): """income '1,234.50' + pct '-10' + amount '5' -> stored 1234.50 / -10 / 5, income currency paired; clearing all three writes NULLs (T-PRJ-028).""" res = projects_admin.update_project_financials( iso_session, owner, ADMIN, {"income": "$1,234.50", "billout_adjustment_pct": "-10", "billout_adjustment_amount": "5"}, ) assert res["ok"] and res["changed"] p = _project(iso_session, ADMIN) assert p.income == Decimal("1234.50") assert p.income_currency == "CAD" assert p.billout_adjustment_pct == Decimal("-10.00") assert p.billout_adjustment_amount == Decimal("5.00") # clear everything (T-ADJ-020 twin path) projects_admin.update_project_financials(iso_session, owner, ADMIN, {}) p2 = _project(iso_session, ADMIN) assert p2.income is None and p2.income_currency is None assert p2.billout_adjustment_pct is None assert p2.billout_adjustment_amount is None def test_t_adj_020_blank_clears_all(iso_session, owner): """A save with every field blank clears income + both adjustments to NULL, even on a project that HAD all three set (Site Rebuild carries income 12000 / pct 10 / amount 250 from the fixture).""" before = _project(iso_session, SITE_REBUILD) assert before.income is not None and before.billout_adjustment_pct is not None projects_admin.update_project_financials( iso_session, owner, SITE_REBUILD, {"income": "", "billout_adjustment_pct": "", "billout_adjustment_amount": " "}, ) p = _project(iso_session, SITE_REBUILD) assert p.income is None and p.income_currency is None assert p.billout_adjustment_pct is None and p.billout_adjustment_amount is None # read side reflects the clear: billout_adjusted falls back to raw row = _row(iso_session, owner, "Site Rebuild") assert row["billout_adjusted"]["amount"] == "142.50" # == raw, no adjustment assert row["income"] is None and row["margin"] is None # ------------------------------------------------- adjustment math parity def test_adjustment_math_parity_and_never_stamped(iso_session, owner): """The written adjustment flows into the read-time formula raw*(1+pct/100)+amount (adjustments-and-rates B32) and re-stamps NOTHING — the per-entry billout sum is byte-unchanged (ADR #040).""" projects_admin.update_project_financials( iso_session, owner, SITE_REBUILD, {"income": "20000", "billout_adjustment_pct": "20", "billout_adjustment_amount": "100"}, ) row = _row(iso_session, owner, "Site Rebuild") # raw billout 142.50 (unchanged), adjusted = 142.50 * 1.20 + 100 = 271.00 assert row["billout"]["amount"] == "142.50" assert row["billout_adjusted"]["amount"] == "271.00" assert row["income"]["amount"] == "20000.00" assert row["margin"]["amount"] == "19926.50" # 20000 - cost 73.50 def test_pct_and_amount_coexist_not_xor(iso_session, owner): """DELIBERATE parity (ADR #010): pct and amount are NOT mutually exclusive — both persist and COMBINE (the T-ADJ-018 shape). No XOR is enforced on the display adjustment (only rate_overrides is XOR).""" res = projects_admin.update_project_financials( iso_session, owner, ADMIN, {"billout_adjustment_pct": "10", "billout_adjustment_amount": "50"}, ) assert res["billout_adjustment_pct"] == "10.00" assert res["billout_adjustment_amount"]["amount"] == "50.00" p = _project(iso_session, ADMIN) assert p.billout_adjustment_pct == Decimal("10.00") assert p.billout_adjustment_amount == Decimal("50.00") def test_income_currency_pairing_holds_on_clear(iso_session, owner): """The real R1 constraint (ck_projects_income_currency_pair): setting income pairs a currency; clearing income clears the currency even when an adjustment is kept — so the CHECK can never be violated on commit.""" projects_admin.update_project_financials( iso_session, owner, ADMIN, {"income": "500", "billout_adjustment_pct": "5"} ) p = _project(iso_session, ADMIN) assert p.income == Decimal("500.00") and p.income_currency == "CAD" # clear income but keep an adjustment -> income+currency both NULL, pct kept projects_admin.update_project_financials( iso_session, owner, ADMIN, {"income": "", "billout_adjustment_pct": "5"} ) p2 = _project(iso_session, ADMIN) assert p2.income is None and p2.income_currency is None assert p2.billout_adjustment_pct == Decimal("5.00") # ---------------------------------------------------------------- validation @pytest.mark.parametrize("bad", ["-1", "-0.01", "abc"]) def test_income_rejects_negative_and_nonnumeric(iso_session, owner, bad): with pytest.raises(ServiceError) as exc: projects_admin.update_project_financials( iso_session, owner, ADMIN, {"income": bad} ) assert exc.value.code == "BAD_INCOME" def test_adjustment_rejects_nonnumeric(iso_session, owner): with pytest.raises(ServiceError) as exc: projects_admin.update_project_financials( iso_session, owner, ADMIN, {"billout_adjustment_pct": "ten"} ) assert exc.value.code == "BAD_ADJUSTMENT" def test_t_prj_029_unassigned_group_has_no_write_surface(iso_session, owner): """The legacy 'Fix the entries first' NULL-tuple guard is structural in R1: financials attach to a project id, so a missing/unassigned rollup group has no id to target — an unknown id is a clean PROJECT_NOT_FOUND (never a silent no-op).""" with pytest.raises(ServiceError) as exc: projects_admin.update_project_financials( iso_session, owner, uuid.uuid4(), {"income": "100"} ) assert exc.value.code == "PROJECT_NOT_FOUND" assert "fix the entries" in (exc.value.detail or "").lower() # ---------------------------------------------------------------- permissions def _actor_with_role(iso_session, role: str) -> Actor: """A synthetic tenant actor of a given role, capabilities per the R1 truth table — pins the owner-only 403 matrix without needing a fixture user for every role.""" from app.services.authz.capabilities import CAPABILITIES, capability_allowed tenant_id = actor_named(iso_session, "rian").tenant_id caps = { c: capability_allowed(c, role=role, is_super_admin=False, is_viewing_as=False) for c in CAPABILITIES } return Actor( tenant_id=tenant_id, tenant_name="Bowden Works", tenant_slug="bowden-works", real_user_id=None, real_username="x", real_email="x@y", real_is_super_admin=False, user_id=None, username="x", email="x@y", is_super_admin=False, person_party_id=None, role=role, role_reason="test", capabilities=caps, ) @pytest.mark.parametrize("role", ["manager", "worker", "none"]) def test_owner_only_403_matrix(iso_session, role): """T-PRJ-030 / T-ADJ-025 shape: every non-owner is refused with the exact owner-only message — before any project read.""" actor = _actor_with_role(iso_session, role) with pytest.raises(ServiceError) as exc: projects_admin.update_project_financials( iso_session, actor, SITE_REBUILD, {"income": "1"} ) assert exc.value.status == 403 assert exc.value.code == "NOT_ALLOWED" assert exc.value.summary == "Only an owner can edit project income." def test_migrated_manager_adi_is_refused(iso_session): """The real migrated manager (Adi) is refused — the truth table holds on production-shaped identity, not just synthetic actors.""" adi = actor_named(iso_session, "adi") assert not adi.allowed("can_transfer") with pytest.raises(ServiceError) as exc: projects_admin.update_project_financials( iso_session, adi, SITE_REBUILD, {"income": "1"} ) assert exc.value.code == "NOT_ALLOWED" # ---------------------------------------------------------------- audit def test_save_writes_audit_before_image(iso_session, owner): projects_admin.update_project_financials( iso_session, owner, SITE_REBUILD, {"income": "15000", "billout_adjustment_pct": "5"} ) audit = iso_session.scalars( select(AuditLog) .where(AuditLog.action == "projects.update_financials") .order_by(AuditLog.created_at.desc()) ).first() assert audit is not None assert audit.entity_type == "project" and audit.entity_id == SITE_REBUILD # before-image carries the fixture's prior financials (income 12000) assert audit.before["income"] == "12000.00" assert audit.after["income"] == "15000.00" assert audit.after["billout_adjustment_pct"] == "5.00"