"""Import-pipeline test graph (04-architecture: fixtures speak the business's language). SQLite is enough — the pipeline uses portable constructs; the entity chain + terms + overrides mirror the real graph: tenant Bowden Works (party = operator, two hats) operators: Bowden Works, PlusROI clients: BW -> Brentwood, BW -> Victoria; PlusROI -> DaxTech projects: Brentwood:Website (BW), DaxTech:Website (PlusROI), Victoria:Blocks (BW) workers (active subcontract engagements, terms per the kind model): adi info@adipramono.com cost 18 / billout 25 (CAD) gary gary@rian.ca cost 0 / billout 25 (Gary the AI — judgment #12: zero-cost billable worker) zed zed@example.com NO terms (never priced — judgment #19) rate overrides on Brentwood:Website: adi-specific absolute 30 CAD + project-wide absolute 28 CAD rate override on DaxTech:Website: project-wide pct -20 """ from datetime import date 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, Party, Project, RateOverride, Tenant, User, ) from tests.authz.conftest import actor_for, fake_request # noqa: F401 TENANT_SLUG = "bowden-works" EFFECTIVE = date(2026, 1, 1) 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 def _worker(s, tenant, tenant_party, person, *, cost, billout, can_manage_imports=False): engagement = Engagement( tenant_id=tenant.id, type="subcontract", party_a_id=tenant_party.id, party_b_id=person.id, role_a="prime", role_b="subcontractor", can_manage_imports=can_manage_imports, ) s.add(engagement) s.flush() for kind, rate in (("cost", cost), ("billout", billout)): if rate is None: continue # NULL = never priced -> no term (judgment #19) s.add( CompensationTerm( engagement_id=engagement.id, kind=kind, model="hourly", rate_amount=rate, currency="CAD", effective_from=EFFECTIVE, ) ) return engagement @pytest.fixture() def graph(): 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") victoria = _party(s, "org", "Victoria", "victoria") daxtech = _party(s, "org", "DaxTech", "daxtech") 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") p_zed = _party(s, "person", "Zed", "zed", "zed@example.com") 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() # roles: rian owner (official_partnership); adi manager-grade 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", ) ) engagements = { "adi": _worker( s, tenant, bw, p_adi, cost=18, billout=25, can_manage_imports=True ), "gary": _worker(s, tenant, bw, p_gary, cost=0, billout=25), "zed": _worker(s, tenant, bw, p_zed, cost=None, billout=None), } # client -> operator edges for operator, client in ((bw, brentwood), (bw, victoria), (plusroi, daxtech)): s.add( Engagement( tenant_id=tenant.id, type="direct_client", party_a_id=operator.id, party_b_id=client.id, role_a="operator", role_b="client", ) ) s.flush() projects = {} for key, operator, client, name in ( ("brentwood_website", bw, brentwood, "Website"), ("daxtech_website", plusroi, daxtech, "Website"), ("victoria_blocks", bw, victoria, "Blocks"), ): project = Project( tenant_id=tenant.id, operator_party_id=operator.id, client_party_id=client.id, name=name, status="active", ) s.add(project) s.flush() projects[key] = project # billout overrides (judgment #25 precedence fixtures) s.add_all( [ RateOverride( tenant_id=tenant.id, project_id=projects["brentwood_website"].id, worker_party_id=p_adi.id, override_rate=30, currency="CAD", ), RateOverride( tenant_id=tenant.id, project_id=projects["brentwood_website"].id, worker_party_id=None, override_rate=28, currency="CAD", ), RateOverride( tenant_id=tenant.id, project_id=projects["daxtech_website"].id, worker_party_id=None, override_pct=-20, ), ] ) s.commit() yield SimpleNamespace( session=s, engine=engine, tenant=tenant, users=users, engagements=engagements, projects=projects, parties=SimpleNamespace( bw=bw, plusroi=plusroi, brentwood=brentwood, victoria=victoria, daxtech=daxtech, rian=p_rian, adi=p_adi, gary=p_gary, zed=p_zed, ), ) s.close() engine.dispose() # ------------------------------------------------------- CSV builders HEADER = ( "Project,Client,Description,Task,User,Group,Email,Tags,Billable," "Start Date,Start Time,End Date,End Time,Duration (h),Duration (decimal)," "Billable Rate (USD),Billable Amount (USD)" ) FIELD_ORDER = ( "Project", "Client", "Description", "Task", "User", "Group", "Email", "Tags", "Billable", "Start Date", "Start Time", "End Date", "End Time", "Duration (h)", "Duration (decimal)", "Billable Rate (USD)", "Billable Amount (USD)", ) ROW_DEFAULTS = { "Project": "Brentwood : Website", "Client": "Bowden Works", "Description": "Something", "Task": "", "User": "Adi Pramono", "Group": "", "Email": "info@adipramono.com", "Tags": "", "Billable": "Yes", "Start Date": "6/2/2026", "Start Time": "9:00:00 AM", "End Date": "6/2/2026", "End Time": "10:30:00 AM", "Duration (h)": "1:30:00", "Duration (decimal)": "1.5", "Billable Rate (USD)": "", "Billable Amount (USD)": "", } def csv_row(**overrides: str) -> str: fields = {**ROW_DEFAULTS, **overrides} return ",".join(fields[key] for key in FIELD_ORDER) def csv_of(*rows: str) -> str: return "\n".join([HEADER, *rows])