"""Seeded write-test graph (SQLite — the services under test use portable constructs; the bulk-safety suite runs on the migrated Postgres scratch in tests/bulk). Business-language fixtures per 04-architecture: rian (owner + super admin), adi (manager with cost 14 / billout 35 CAD terms + a worker-specific override), gary (the zero-cost billable AI), plus the PlusROI/Brentwood and BW/Internal entity chains, one invoice-locked row, and one unresolved raw-only row. """ from datetime import date, datetime from types import SimpleNamespace import pytest from sqlalchemy import create_engine from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.models import ( Base, CompensationTerm, Engagement, ImportBatch, Invoice, InvoiceLine, Party, Project, RateOverride, Tenant, TimeEntry, User, ) from app.services.authz.context import resolve_actor from tests.authz.conftest import fake_request TENANT_SLUG = "bowden-works" def _party(s, kind, name, slug, email=None): party = Party(kind=kind, name=name, slug=slug, email=email) s.add(party) s.flush() return party @pytest.fixture() def g(): engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) Base.metadata.create_all(engine) s = Session(engine) bw = _party(s, "org", "Bowden Works", TENANT_SLUG) plusroi = _party(s, "org", "PlusROI", "plusroi") brentwood = _party(s, "org", "Brentwood", "brentwood") internal = _party(s, "org", "Internal", "internal") tenant = Tenant(party_id=bw.id, settings={}) s.add(tenant) s.flush() p_rian = _party(s, "person", "Rian Bowden", "rian-bowden", "rian@rian.ca") p_adi = _party(s, "person", "Adi Pramono", "adi-pramono", "info@adipramono.com") p_gary = _party(s, "person", "Gary", "gary", "gary@rian.ca") users = { "rian": User( idauth_username="rian", email="rian@rian.ca", person_party_id=p_rian.id, is_super_admin=True, ), "adi": User( idauth_username="adi", email="info@adipramono.com", person_party_id=p_adi.id ), "gary": User(idauth_username="gary", email="gary@rian.ca", person_party_id=p_gary.id), "mallory": User(idauth_username="mallory", email="m@example.com"), } s.add_all(users.values()) s.flush() def worker_engagement(party, *, cmi=False, cost=None, billout=None): eng = Engagement( tenant_id=tenant.id, type="subcontract", party_a_id=bw.id, party_b_id=party.id, role_a="prime", role_b="subcontractor", can_manage_imports=cmi, ) s.add(eng) s.flush() for kind, rate in (("cost", cost), ("billout", billout)): if rate is not None: s.add( CompensationTerm( engagement_id=eng.id, kind=kind, model="hourly", rate_amount=rate, currency="CAD", effective_from=date(2026, 1, 1), ) ) s.flush() return eng s.add( Engagement( tenant_id=tenant.id, type="official_partnership", party_a_id=bw.id, party_b_id=p_rian.id, role_a="owner", role_b="owner", ) ) eng_rian = worker_engagement(p_rian, cost=50, billout=100) eng_adi = worker_engagement(p_adi, cmi=True, cost=14, billout=35) eng_gary = worker_engagement(p_gary, cost=0, billout=25) # entity chain: PlusROI -> Brentwood -> Site Rebuild; # BW -> Internal -> Admin for op, cl in ((plusroi, brentwood), (bw, internal)): s.add( Engagement( tenant_id=tenant.id, type="direct_client", party_a_id=op.id, party_b_id=cl.id, role_a="operator", role_b="client", ) ) pr_site = Project( tenant_id=tenant.id, operator_party_id=plusroi.id, client_party_id=brentwood.id, name="Site Rebuild", status="active", ) pr_admin = Project( tenant_id=tenant.id, operator_party_id=bw.id, client_party_id=internal.id, name="Admin", status="active", ) s.add_all([pr_site, pr_admin]) s.flush() # billout overrides on Site Rebuild: adi-specific absolute 20, # project-wide pct -20 (specificity precedence — T-ADJ-001) s.add_all( [ RateOverride( tenant_id=tenant.id, project_id=pr_site.id, worker_party_id=p_adi.id, override_rate=20, currency="CAD", ), RateOverride( tenant_id=tenant.id, project_id=pr_site.id, worker_party_id=None, override_pct=-20, ), ] ) batches = { "adi": ImportBatch(tenant_id=tenant.id, imported_by=users["adi"].id, source="clockify", name="adi batch"), "rian": ImportBatch(tenant_id=tenant.id, imported_by=users["rian"].id, source="clockify", name="rian batch"), } s.add_all(batches.values()) s.flush() invoice = Invoice(tenant_id=tenant.id, name="INV-1", status="open", currency="CAD") s.add(invoice) s.flush() line = InvoiceLine( invoice_id=invoice.id, kind="time", amount=35, snapshot_client_name="Brentwood", snapshot_project_name="Site Rebuild", ) s.add(line) s.flush() def entry(batch_key, **kw): defaults = dict( tenant_id=tenant.id, source="clockify", import_id=batches[batch_key].id if batch_key else None, start_at=datetime(2026, 6, 1, 9, 0, 0), end_at=datetime(2026, 6, 1, 10, 0, 0), duration_seconds=3600, billable=True, ) defaults.update(kw) e = TimeEntry(**defaults) s.add(e) return e entries = { "adi": entry( "adi", description="adi row", source_user_email="info@adipramono.com", source_user_name="Adi Pramono", worker_party_id=p_adi.id, via_engagement_id=eng_adi.id, project_id=pr_site.id, cost_amount=14, cost_currency="CAD", billout_amount=20, billout_currency="CAD", ), "locked": entry( "adi", description="locked row", source_user_email="info@adipramono.com", source_user_name="Adi Pramono", worker_party_id=p_adi.id, via_engagement_id=eng_adi.id, project_id=pr_site.id, invoice_line_id=line.id, cost_amount=14, cost_currency="CAD", billout_amount=20, billout_currency="CAD", ), "unresolved": entry( "adi", description="who did this", source_user_email="unknown@example.com", source_user_name="Unknown", raw_operator="PlusROI", raw_client=None, raw_project="Mystery", end_at=None, ), "gary": entry( "rian", description="automation run", source_user_email="gary@rian.ca", source_user_name="Gary", worker_party_id=p_gary.id, via_engagement_id=eng_gary.id, project_id=pr_admin.id, duration_seconds=7200, cost_amount=0, cost_currency="CAD", billout_amount=50, billout_currency="CAD", ), } s.commit() yield SimpleNamespace( session=s, engine=engine, tenant=tenant, users=users, parties=SimpleNamespace( bw=bw, plusroi=plusroi, brentwood=brentwood, internal=internal, rian=p_rian, adi=p_adi, gary=p_gary, ), engagements=SimpleNamespace(rian=eng_rian, adi=eng_adi, gary=eng_gary), projects=SimpleNamespace(site=pr_site, admin=pr_admin), batches=batches, invoice=invoice, line=line, entries=entries, ) s.close() engine.dispose() def actor_for(g, username: str): return resolve_actor(fake_request(username), g.session) @pytest.fixture() def rian(g): return actor_for(g, "rian") @pytest.fixture() def adi(g): return actor_for(g, "adi") @pytest.fixture() def gary(g): return actor_for(g, "gary")