"""M7 CC-expenses test harness. Two fixture families: * ``sg`` — a small SQLite graph (BW tenant, rian=owner+super-admin, gary=worker, two BW projects with clients, plus a second-tenant project for the cross-tenant guard). Fast, no Postgres — carries the service-semantics tests (CCE-A / CCE-V / CCE-B / permissions). * ``cc_engine`` / ``s`` / ``cc_owner`` — a commit-safe clone of the REAL migrated data in its OWN scratch DB (``with_cc_test``), built by REUSING the M6 clone machinery (tests/invoices/conftest helpers): alembic head + the real ``legacy`` snapshot copied table-by-table + the transforms run once. Carries the byte-diff GATE and the end-to-end paste->rules->assign->export test on the 102 real statement lines. """ import os import uuid from types import SimpleNamespace import pytest from sqlalchemy import create_engine, select, text from sqlalchemy.orm import Session from sqlalchemy.pool import StaticPool from app.models import Base, Engagement, Party, Project, Tenant, User from app.services.authz.context import resolve_actor from tests.authz.conftest import fake_request # Reuse the M6 clone machinery verbatim (09-build-plan M7 gate directive). from tests.invoices.conftest import ( # noqa: F401 LEGACY_TABLES, _copy_real_legacy, _dsn, _sidecar_url, actor_named, ) TENANT_SLUG = "bowden-works" TEST_DB = "with_cc_test" # ---------------------------------------------------------- SQLite graph def _party(s, kind, name, slug): party = Party(kind=kind, name=name, slug=slug) s.add(party) s.flush() return party @pytest.fixture() def sg(): """Bowden Works + two projects (Tuscany/Venetian : Monthly Services) + a second tenant's project. rian owner+super-admin; gary worker.""" engine = create_engine( "sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool ) # SQLite ignores ON DELETE CASCADE unless FK enforcement is on — turn # it on so batch-delete cascade (CCE-A-003) behaves like Postgres. from sqlalchemy import event @event.listens_for(engine, "connect") def _fk_pragma(dbapi_conn, _rec): # pragma: no cover - trivial dbapi_conn.execute("PRAGMA foreign_keys=ON") Base.metadata.create_all(engine) s = Session(engine) bw = _party(s, "org", "Bowden Works", TENANT_SLUG) tingang = _party(s, "org", "Tingang", "tingang") tenant = Tenant(party_id=bw.id, settings={"default_currency": "CAD"}) tenant2 = Tenant(party_id=tingang.id, settings={}) s.add_all([tenant, tenant2]) s.flush() p_rian = _party(s, "person", "Rian Bowden", "rian-bowden") p_gary = _party(s, "person", "Gary", "gary") tuscany = _party(s, "org", "Tuscany", "tuscany") venetian = _party(s, "org", "Venetian", "venetian") other_client = _party(s, "org", "OtherClient", "other-client") s.add_all( [ User( idauth_username="rian", email="rian@rian.ca", person_party_id=p_rian.id, is_super_admin=True, ), User(idauth_username="gary", email="gary@rian.ca", person_party_id=p_gary.id), ] ) s.add_all( [ 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", ), Engagement( tenant_id=tenant.id, type="subcontract", party_a_id=bw.id, party_b_id=p_gary.id, role_a="prime", role_b="subcontractor", ), ] ) s.flush() def project(tenant_id, operator, client, name): p = Project( tenant_id=tenant_id, operator_party_id=operator.id, client_party_id=client.id, name=name, status="active", ) s.add(p) s.flush() return p proj_tuscany = project(tenant.id, bw, tuscany, "Monthly Services") proj_venetian = project(tenant.id, bw, venetian, "Monthly Services") proj_other = project(tenant2.id, tingang, other_client, "Other Work") s.commit() yield SimpleNamespace( session=s, tenant=tenant, rian=resolve_actor(fake_request("rian"), s), gary=resolve_actor(fake_request("gary"), s), proj_tuscany=proj_tuscany.id, proj_venetian=proj_venetian.id, proj_other=proj_other.id, # different tenant -> cross-org guard ) s.close() engine.dispose() # ------------------------------------------------ real-data clone (gate) @pytest.fixture(scope="session") def cc_url() -> str: base = _sidecar_url() admin = create_engine( base, isolation_level="AUTOCOMMIT", connect_args={"connect_timeout": 3} ) from sqlalchemy import text try: with admin.connect() as conn: conn.execute(text(f"DROP DATABASE IF EXISTS {TEST_DB} WITH (FORCE)")) conn.execute(text(f"CREATE DATABASE {TEST_DB}")) except Exception as exc: # noqa: BLE001 pytest.fail(f"cannot reach the sidecar Postgres for CC tests: {exc}") finally: admin.dispose() url = base.rsplit("/", 1)[0] + f"/{TEST_DB}" from pathlib import Path from alembic.config import Config from alembic import command root = Path(__file__).resolve().parents[2] cfg = Config(str(root / "alembic.ini")) cfg.set_main_option("script_location", str(root / "alembic")) previous = os.environ.get("WITH_DB_URL") os.environ["WITH_DB_URL"] = url try: command.upgrade(cfg, "head") finally: os.environ["WITH_DB_URL"] = previous if previous is not None else "" return url @pytest.fixture(scope="session") def cc_engine(cc_url): _copy_real_legacy(_dsn(_sidecar_url()), _dsn(cc_url)) engine = create_engine(cc_url) from app.services.migrate import run_all report = run_all(engine=engine) assert report.ok, report.render() yield engine engine.dispose() @pytest.fixture() def s(cc_engine): """SAVEPOINT-isolated session: service commit()s stay inside the outer transaction, which rolls back after the test (e2e writes leave no trace).""" conn = cc_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 cc_owner(s): return actor_named(s, "rian") def raw_lines(session, limit: int | None = None) -> list[str]: """The REAL migrated statement lines, batch/line order (102 exist).""" sql = "SELECT raw_line FROM cc_expense_lines ORDER BY batch_id, line_index" if limit is not None: sql += f" LIMIT {int(limit)}" return [r.raw_line for r in session.execute(text(sql))] def first_batch_id(session) -> uuid.UUID: from app.models import CcExpenseBatch return session.scalars(select(CcExpenseBatch.id).limit(1)).first()