"""PlusROI sheet-books transform (R2, ADR #016; sheet_books.py).

Exercises the sheet → DB step against tiny fixture CSVs (NOT the real
export) on a DEDICATED scratch database — isolated from the shared
harness/parity DB so it never pollutes the legacy-parity comparison
(the sheet step is opt-in via run_all(books_source=...); every other
transform test runs it OFF).

Covers: tenant + owners creation & idempotency, users linkage by
username, client match-vs-create, the Bowden-Works-Team + person-vendor
mappings, project status mapping + name collision, invoice
status/currency/original-amount mapping + name uniqueness + live↔archive
dedupe (id- and tuple-based), expense paid-by mapping + dedupe, blank-row
skipping, the reconciliation-fails-on-unaccounted-row contract, and a
guarded real-export smoke run.
"""

import os
from datetime import date
from decimal import Decimal
from pathlib import Path

import pytest
from sqlalchemy import create_engine, text

from tests.migrate.conftest import FIXTURE_SQL, rows, scalar

FIX_SHEET = Path(__file__).parent / "fixtures" / "sheet"
REAL_EXPORT = Path(__file__).resolve().parents[2] / "books-import" / "source"
SHEET_DB = "with_sheet_books_test"


# --- dedicated scratch DB (isolated from the harness DB) -------------------


@pytest.fixture(scope="module")
def sheet_url() -> str:
    from app.config import settings

    base = settings.with_db_url_host or settings.with_db_url
    if not base:
        pytest.fail("sheet-books tests need the sidecar Postgres (WITH_DB_URL_HOST)")
    admin = create_engine(
        base, isolation_level="AUTOCOMMIT", connect_args={"connect_timeout": 3}
    )
    try:
        with admin.connect() as conn:
            conn.execute(text(f"DROP DATABASE IF EXISTS {SHEET_DB} WITH (FORCE)"))
            conn.execute(text(f"CREATE DATABASE {SHEET_DB}"))
    except Exception as exc:
        pytest.fail(f"cannot reach the sidecar Postgres for sheet-books tests: {exc}")
    finally:
        admin.dispose()

    url = base.rsplit("/", 1)[0] + f"/{SHEET_DB}"
    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 ""
    yield url

    admin = create_engine(base, isolation_level="AUTOCOMMIT")
    with admin.connect() as conn:
        conn.execute(text(f"DROP DATABASE IF EXISTS {SHEET_DB} WITH (FORCE)"))
    admin.dispose()


@pytest.fixture(scope="module")
def sheet_engine(sheet_url):
    engine = create_engine(sheet_url)
    yield engine
    engine.dispose()


@pytest.fixture()
def books_run(sheet_engine):
    """Reload the legacy fixture, optionally seed users, run the full
    transform WITH the sheet step, and return the RunReport.

    Each _run starts from a VIRGIN public schema (parties wiped too):
    since ADR #021 the transform upserts parties by slug instead of
    bulk-deleting them, so without this reset every test after the first
    would see the previous test's parties as "matched" instead of
    "created". The reset keeps each test meaning "a fresh import on an
    empty database"; cross-run STABILITY is proven separately in
    test_app_owned_survival.py."""

    def _run(source: Path = FIX_SHEET, seed_users: tuple[str, ...] = ()):
        from app.services.migrate import run_all
        from app.services.migrate.transforms import TRUNCATE_TABLES

        with sheet_engine.begin() as conn:
            conn.execute(text("UPDATE users SET person_party_id = NULL"))
            conn.execute(text("TRUNCATE TABLE " + ", ".join(TRUNCATE_TABLES)))
            conn.execute(text("DELETE FROM parties"))
            conn.exec_driver_sql(FIXTURE_SQL.read_text(encoding="utf-8"))
        for username in seed_users:
            with sheet_engine.begin() as conn:
                conn.execute(
                    text(
                        "INSERT INTO users (id, idauth_username, email)"
                        " VALUES (gen_random_uuid(), :u, '')"
                        " ON CONFLICT (idauth_username) DO NOTHING"
                    ),
                    {"u": username},
                )
        return run_all(engine=sheet_engine, books_source=str(source))

    return _run


# --- helpers ---------------------------------------------------------------


def _sheet_step(report):
    return next(s for s in report.steps if s.name.startswith("2.10"))


def _pr_tenant(engine):
    return scalar(
        engine,
        "SELECT t.id FROM tenants t JOIN parties p ON p.id = t.party_id"
        " WHERE lower(p.name) = 'plusroi'",
    )


def _inv_status(engine, t, name):
    return scalar(
        engine,
        "SELECT status FROM invoices WHERE tenant_id = :t AND name = :n",
        t=t,
        n=name,
    )


def _counts(engine, tenant):
    out = {}
    for tbl in ("projects", "invoices", "expenses"):
        out[tbl] = scalar(
            engine,
            f"SELECT count(*) FROM {tbl} WHERE tenant_id = :t",  # noqa: S608
            t=tenant,
        )
    out["client_profiles"] = scalar(
        engine, "SELECT count(*) FROM client_profiles WHERE tenant_id = :t", t=tenant
    )
    out["project_billing"] = scalar(
        engine,
        "SELECT count(*) FROM project_billing pb JOIN projects p ON p.id = pb.project_id"
        " WHERE p.tenant_id = :t",
        t=tenant,
    )
    out["parties"] = scalar(engine, "SELECT count(*) FROM parties")
    out["tenants"] = scalar(engine, "SELECT count(*) FROM tenants")
    return out


# --- tenant + owners -------------------------------------------------------


def test_tenant_settings_and_owner_engagements(books_run, sheet_engine):
    report = books_run()
    assert report.ok, report.render()
    t = _pr_tenant(sheet_engine)
    settings = scalar(sheet_engine, "SELECT settings FROM tenants WHERE id = :t", t=t)
    assert settings["books_owner"] == "sheet"  # ADR #021 (supersedes parallel_run)
    assert "parallel_run" not in settings
    assert settings["partner_split"] == 0.5
    assert settings["bookkeeping_pct"] == 3.0
    assert settings["default_currency"] == "CAD"
    assert (
        scalar(
            sheet_engine,
            "SELECT count(*) FROM engagements"
            " WHERE tenant_id = :t AND type = 'official_partnership'",
            t=t,
        )
        == 3
    )
    # rian reused by email (one rian@rian.ca party); rob/danielle created
    assert scalar(
        sheet_engine, "SELECT count(*) FROM parties WHERE email = 'rian@rian.ca'"
    ) == 1
    assert scalar(
        sheet_engine,
        "SELECT count(*) FROM parties WHERE name = 'Rob Cooper' AND kind = 'person'",
    ) == 1
    assert scalar(
        sheet_engine,
        "SELECT count(*) FROM parties WHERE name = 'Danielle Cooper'"
        " AND kind = 'person' AND email = 'danielle@plusroi.com'",
    ) == 1


def test_users_linkage_by_username(books_run, sheet_engine):
    books_run(seed_users=("rob",))
    rob = scalar(
        sheet_engine,
        "SELECT id FROM parties WHERE name = 'Rob Cooper' AND kind = 'person'",
    )
    assert scalar(
        sheet_engine, "SELECT person_party_id FROM users WHERE idauth_username = 'rob'"
    ) == rob
    # rian keeps his existing BW person party — NOT relinked to a PlusROI party
    rian = scalar(sheet_engine, "SELECT id FROM parties WHERE email = 'rian@rian.ca'")
    assert scalar(
        sheet_engine, "SELECT person_party_id FROM users WHERE idauth_username = 'rian'"
    ) == rian


def test_idempotency_identical_counts(books_run, sheet_engine):
    books_run()
    counts1 = _counts(sheet_engine, _pr_tenant(sheet_engine))
    second = books_run()
    assert second.ok, second.render()
    counts2 = _counts(sheet_engine, _pr_tenant(sheet_engine))
    assert counts1 == counts2
    # every sheet tab reconciles on both runs
    for key, entry in second.reconciliation.items():
        if key.startswith("sheet."):
            assert entry["delta"] == 0, (key, entry)


# --- clients ---------------------------------------------------------------


def test_client_match_vs_create_and_archive(books_run, sheet_engine):
    report = books_run()
    disp = report.reconciliation["sheet.tab-Clients"]["dispositions"]
    assert disp["client_matched"] == 1  # Brentwood (existing org)
    assert disp["client_created"] == 2  # Acme Co, Zzz Archived
    assert disp["skipped_blank"] == 1
    t = _pr_tenant(sheet_engine)
    assert (
        scalar(
            sheet_engine,
            "SELECT cp.archived FROM client_profiles cp"
            " JOIN parties p ON p.id = cp.client_party_id"
            " WHERE cp.tenant_id = :t AND p.name = 'Zzz Archived'",
            t=t,
        )
        is True
    )
    assert (
        scalar(
            sheet_engine,
            "SELECT cp.currency FROM client_profiles cp"
            " JOIN parties p ON p.id = cp.client_party_id"
            " WHERE cp.tenant_id = :t AND p.name = 'Acme Co'",
            t=t,
        )
        == "USD"
    )


# --- vendors ---------------------------------------------------------------


def test_vendor_mappings(books_run, sheet_engine):
    report = books_run()
    disp = report.reconciliation["sheet.tab-Vendors"]["dispositions"]
    assert disp == {
        "mapped_bowden_works": 1,
        "matched_person": 1,  # Gary
        "created_person": 1,  # Brennan Doyle (pinned set)
        "created_org": 1,  # Google
        "matched_org": 1,  # Brentwood
        "skipped_blank": 1,
    }


# --- projects --------------------------------------------------------------


def test_project_status_mapping_and_collision(books_run, sheet_engine):
    report = books_run()
    disp = report.reconciliation["sheet.tab-Projects"]["dispositions"]
    assert disp["project_created"] == 5
    assert disp["project_client_created_on_fly"] == 1
    assert disp["skipped_blank"] == 1
    t = _pr_tenant(sheet_engine)

    def status(name):
        return scalar(
            sheet_engine,
            "SELECT status FROM projects WHERE tenant_id = :t AND name = :n",
            t=t,
            n=name,
        )

    assert status("Website") == "archived"  # Closed
    assert status("Support") == "archived"  # Paused
    paused_notes = scalar(
        sheet_engine,
        "SELECT notes FROM projects WHERE tenant_id = :t AND name = 'Support'",
        t=t,
    )
    assert paused_notes.startswith("[paused]")
    assert status("Oddball") == "active"  # unknown status → active + warn
    names = {
        r[0]
        for r in rows(
            sheet_engine,
            "SELECT name FROM projects WHERE tenant_id = :t AND name LIKE 'Retainer%'",
            t=t,
        )
    }
    assert {"Retainer", "Retainer (2)"} <= names
    warns = _sheet_step(report).details["warn_counts"]
    assert warns.get("project_status_unknown", 0) >= 1
    assert warns.get("project_name_suffixed", 0) >= 1
    assert warns.get("project_client_on_fly", 0) >= 1


# --- invoices --------------------------------------------------------------


def test_invoice_status_currency_original_and_names(books_run, sheet_engine):
    report = books_run()
    t = _pr_tenant(sheet_engine)

    def one(name, cols):
        return rows(
            sheet_engine,
            f"SELECT {cols} FROM invoices WHERE tenant_id = :t AND name = :n",  # noqa: S608
            t=t,
            n=name,
        )[0]

    # CAD Paid, settlement month first-of-month, no original face
    assert one(
        "1001", "status, currency, manual_total, settlement_month, original_amount"
    ) == ("paid", "CAD", Decimal("1500.00"), date(2026, 5, 1), None)
    # USD Invoiced→sent: books CAD (rounded from 3dp), USD face preserved
    assert one(
        "US2001", "status, currency, manual_total, original_amount, original_currency"
    ) == ("sent", "CAD", Decimal("1320.56"), Decimal("1000.00"), "USD")
    # the full lifecycle map
    assert _inv_status(sheet_engine, t, "1003") == "paid"  # blank → paid
    assert _inv_status(sheet_engine, t, "1004") == "pending_approval"
    assert _inv_status(sheet_engine, t, "1005") == "ready"
    assert _inv_status(sheet_engine, t, "1006") == "cancelled"
    assert _inv_status(sheet_engine, t, "1007") == "delayed"
    # duplicate Invoice ID → name suffixed for uq_invoices_tenant_lower_name
    assert scalar(
        sheet_engine,
        "SELECT count(*) FROM invoices WHERE tenant_id = :t"
        " AND name IN ('1001', '1001 (2)')",
        t=t,
    ) == 2
    warns = _sheet_step(report).details["warn_counts"]
    assert warns.get("money_rounded", 0) >= 1
    assert warns.get("invoice_status_defaulted", 0) >= 1
    assert warns.get("invoice_name_suffixed", 0) >= 1


def test_invoice_dedupe_live_vs_archive(books_run, sheet_engine):
    report = books_run()
    disp = report.reconciliation["sheet.arch2022-Invoices"]["dispositions"]
    assert disp["deduped_live_wins_id"] == 1  # arch 1001 duped a live id
    assert disp["deduped_live_wins_tuple"] == 1  # arch Tuple Client duped a live tuple
    assert disp["archive_only"] == 1  # Old Client : Legacy A3001
    assert disp["skipped_blank"] == 1
    t = _pr_tenant(sheet_engine)
    assert scalar(
        sheet_engine,
        "SELECT count(*) FROM invoices WHERE tenant_id = :t AND external_id = 'A3001'",
        t=t,
    ) == 1
    # live wins: the two live 1001 rows import, the archive 1001 does NOT add a third
    assert scalar(
        sheet_engine,
        "SELECT count(*) FROM invoices WHERE tenant_id = :t AND external_id = '1001'",
        t=t,
    ) == 2
    dedupe = _sheet_step(report).details["dedupe"]["invoices"]
    assert dedupe["archive_dropped_by_id"] == 1
    assert dedupe["archive_dropped_by_tuple"] == 1


# --- expenses --------------------------------------------------------------


def test_expense_paidby_and_dedupe(books_run, sheet_engine):
    report = books_run()
    live = report.reconciliation["sheet.tab-Expenses"]["dispositions"]
    assert live == {"imported": 4, "skipped_blank": 1}
    arch = report.reconciliation["sheet.arch2022-Expenses"]["dispositions"]
    assert arch["deduped_live_wins_tuple"] == 1
    assert arch["archive_only"] == 1
    assert arch["skipped_blank"] == 1
    t = _pr_tenant(sheet_engine)
    pr_party = scalar(sheet_engine, "SELECT party_id FROM tenants WHERE id = :t", t=t)
    bw_party = scalar(
        sheet_engine, "SELECT id FROM parties WHERE lower(name) = 'bowden works'"
    )

    def paid_by(amount):
        return scalar(
            sheet_engine,
            "SELECT paid_by_party_id FROM expenses"
            " WHERE tenant_id = :t AND amount = :a",
            t=t,
            a=amount,
        )

    # PlusROI → PlusROI party; Bowden Works → BW party; other → NULL + warn
    assert paid_by(500.00) == pr_party
    assert paid_by(120.00) == bw_party
    assert paid_by(800.00) is None
    # blank vendor → NULL + warn; every row source='sheet'
    assert scalar(
        sheet_engine,
        "SELECT vendor_party_id FROM expenses WHERE tenant_id = :t AND amount = 55.00",
        t=t,
    ) is None
    assert scalar(
        sheet_engine,
        "SELECT count(*) FROM expenses WHERE tenant_id = :t AND source <> 'sheet'",
        t=t,
    ) == 0
    warns = _sheet_step(report).details["warn_counts"]
    assert warns.get("expense_paidby_unmapped", 0) >= 1
    assert warns.get("expense_vendor_blank", 0) >= 1


# --- blank rows ------------------------------------------------------------


def test_blank_rows_skipped_every_tab(books_run):
    report = books_run()
    for key in (
        "sheet.tab-Clients",
        "sheet.tab-Vendors",
        "sheet.tab-Projects",
        "sheet.tab-Invoices",
        "sheet.arch2022-Invoices",
        "sheet.tab-Expenses",
        "sheet.arch2022-Expenses",
    ):
        entry = report.reconciliation[key]
        assert entry["dispositions"].get("skipped_blank", 0) >= 1
        # the reconciliation contract: dispositions partition every CSV row
        assert sum(entry["dispositions"].values()) == entry["source"]
        assert entry["delta"] == 0


# --- reconciliation failure (constructed) ----------------------------------


def test_reconciliation_fails_on_unaccounted_row():
    # the sheet step reconciles each tab with RunReport.reconcile; an
    # unaccounted row FAILS the run (which rolls the transaction back).
    from app.services.migrate.report import RunReport

    report = RunReport()
    report.reconcile("sheet.tab-Invoices", 5, {"imported": 4})  # 1 unaccounted
    assert not report.ok
    assert any("unaccounted" in e for e in report.errors)


# --- real-export smoke -----------------------------------------------------


@pytest.mark.skipif(
    not (REAL_EXPORT / "tab-Invoices.csv").exists(),
    reason="no real PlusROI export on this box",
)
def test_real_export_smoke(books_run):
    report = books_run(source=REAL_EXPORT)
    assert report.ok, report.render()
    sheet_tabs = {k: e for k, e in report.reconciliation.items() if k.startswith("sheet.")}
    assert sheet_tabs, "no sheet reconciliation entries — step did not run"
    for key, entry in sheet_tabs.items():
        # no hardcoded totals: assert dispositions sum to the CSV row count
        assert sum(entry["dispositions"].values()) == entry["source"], key
        assert entry["delta"] == 0, key
    step = _sheet_step(report)
    assert step.details["written"]["invoices"] > 0
    assert step.details["written"]["expenses"] > 0
    assert step.details["settlement_summary"]["invoices_by_month"]


# --- fail-fast semantics (adversarial-review fixes, 2026-07-11) -------------


def test_missing_source_dir_is_fatal(books_run):
    """A requested-but-absent export FAILS the run (rolls back) instead of
    committing a skip that would delete the PlusROI tenant (the CRITICAL
    review finding)."""
    report = books_run(source=Path("/nonexistent/books-source"))
    assert not report.ok
    assert any("books source not found" in e for e in report.errors)


def test_missing_required_tab_is_fatal(books_run, tmp_path):
    import shutil

    partial = tmp_path / "source"
    shutil.copytree(FIX_SHEET, partial)
    (partial / "tab-Vendors.csv").unlink()
    report = books_run(source=partial)
    assert not report.ok
    assert any("incomplete" in e and "tab-Vendors.csv" in e for e in report.errors)


def test_shrunken_export_is_fatal(books_run, tmp_path, sheet_engine):
    """A >20% row-count shrink vs the last OK run means a truncated/corrupt
    export — refuse it (the partial-write review finding)."""
    import shutil

    # The module's real-export smoke may have left an ok run with 2,711-row
    # counts; the guard would (correctly) refuse the tiny fixtures against
    # that baseline — clear history so THIS test controls the baseline.
    with sheet_engine.begin() as conn:
        conn.exec_driver_sql("DELETE FROM migration_runs")
    ok_report = books_run()  # establishes the last-good row counts
    assert ok_report.ok, ok_report.errors

    truncated = tmp_path / "source"
    shutil.copytree(FIX_SHEET, truncated)
    expenses = truncated / "tab-Expenses.csv"
    lines = expenses.read_text(encoding="utf-8").splitlines(keepends=True)
    expenses.write_text("".join(lines[:2]), encoding="utf-8")  # header + 1 row
    report = books_run(source=truncated)
    assert not report.ok
    assert any("shrink" in e for e in report.errors)