"""Transform behavior on the fixture legacy dataset (05 step 2 rules;
T-MIG-006/007/008/010/013/014 where unit-testable).

Every test reads the state produced by ONE clean run (session-scoped
clean_report fixture)."""

from tests.migrate.conftest import (
    IMP_1,
    IMP_3,
    INV_LEGACY,
    INV_MAY,
    INV_OPEN,
    PR_ADMIN,
    PR_BLOCK,
    PR_SITE,
    RO_ABS,
    RO_PCT,
    TE,
    rows,
    scalar,
)


def party_id(engine, name, kind="org"):
    return scalar(
        engine,
        "SELECT id FROM parties WHERE name = :name AND kind = :kind",
        name=name, kind=kind,
    )


# --- reconciliation + report (T-MIG-010) ----------------------------------


def test_every_legacy_table_reconciles(clean_report):
    assert clean_report.ok
    for table, entry in clean_report.reconciliation.items():
        assert entry["delta"] == 0, f"legacy.{table} unaccounted: {entry}"
    te = clean_report.reconciliation["time_entries"]
    assert te == {"source": 7, "dispositions": {"migrated": 7}, "delta": 0}
    assert clean_report.reconciliation["pending_imports"]["dispositions"] == {
        "dropped_staging": 1
    }
    assert clean_report.reconciliation["teams"]["dispositions"] == {
        "dissolved_dropped": 2
    }


def test_column_dispositions_verified(clean_report):
    check = clean_report.column_check
    assert check["unmapped"] == {}
    assert check["unknown_tables"] == []


def test_straggler_check_reports_zero_on_clean_data(clean_report):
    # T-MIG-006: computed and REPORTED; expected 0 after the 2026-05-25
    # backfill. (The nonzero path is test_straggler.py.)
    assert clean_report.straggler_count == 0
    assert clean_report.straggler_entry_ids == []


def test_currency_counts_are_all_cad(clean_report):
    # T-MIG-015 / 07 #21: per-currency row counts ride every report.
    for column, counts in clean_report.currency_counts.items():
        assert set(counts) <= {"CAD"}, f"{column}: {counts}"
    assert clean_report.currency_counts["time_entries.billout_currency"] == {"CAD": 5}
    assert clean_report.currency_counts["time_entries.cost_currency"] == {"CAD": 5}


def test_run_report_row_persisted(mig_engine, clean_report):
    assert scalar(mig_engine, "SELECT count(*) FROM migration_runs WHERE ok") >= 1


# --- 2.1 parties + users ----------------------------------------------------


def test_person_party_dedupe_by_lower_email(mig_engine, clean_report):
    # rian appears in two teams (one email UPPERCASED) -> ONE person party
    assert scalar(mig_engine, "SELECT count(*) FROM parties WHERE kind = 'person'") == 4
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM parties WHERE lower(email) = 'rian@rian.ca'",
        )
        == 1
    )


def test_operator_client_name_collision_merges(mig_engine, clean_report):
    # client "PlusROI" (under Bowden Works) merged into the PlusROI
    # operator party — one party wearing two hats
    assert (
        scalar(mig_engine, "SELECT count(*) FROM parties WHERE name = 'PlusROI'") == 1
    )
    # org parties: Bowden Works (tenant+operator), PlusROI (operator+client),
    # Brentwood, Internal
    assert scalar(mig_engine, "SELECT count(*) FROM parties WHERE kind = 'org'") == 4


def test_tenant_party_merges_with_operator(mig_engine, clean_report):
    tenant_party = scalar(
        mig_engine,
        "SELECT p.name FROM tenants t JOIN parties p ON p.id = t.party_id",
    )
    assert tenant_party == "Bowden Works"
    assert scalar(mig_engine, "SELECT count(*) FROM tenants") == 1


def test_users_upserted_with_identity_map(mig_engine, clean_report):
    users = {
        row[0]: row
        for row in rows(
            mig_engine,
            "SELECT idauth_username, email, is_super_admin,"
            " person_party_id IS NOT NULL FROM users",
        )
    }
    assert users["rian"][2] is True  # is_super_admin carried from profiles
    assert users["rian"][3] is True  # linked to his person party
    assert users["adi"][1] == "info@adipramono.com"
    assert "ghost" not in users  # unmapped profile creates nothing
    assert clean_report.reconciliation["profiles"]["dispositions"] == {
        "user_upserted": 2,
        "skipped_no_idauth_mapping": 1,
    }


# --- 2.2 direct_client edges -------------------------------------------------


def test_direct_client_engagement_per_legacy_client_row(mig_engine, clean_report):
    edges = rows(
        mig_engine,
        "SELECT pa.name, pb.name FROM engagements e"
        " JOIN parties pa ON pa.id = e.party_a_id"
        " JOIN parties pb ON pb.id = e.party_b_id"
        " WHERE e.type = 'direct_client' ORDER BY 1, 2",
    )
    assert edges == [
        ("Bowden Works", "Internal"),
        ("Bowden Works", "PlusROI"),  # the merged party, as CLIENT of BW
        ("PlusROI", "Brentwood"),
    ]


# --- 2.3 engagements + terms (kind model incl. Gary) ---------------------------


def test_one_worker_engagement_per_team_member_row(mig_engine, clean_report):
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM engagements WHERE type = 'subcontract'",
        )
        == 5
    )
    # rian's TWO team rows -> two engagements on ONE party
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM engagements e JOIN parties p"
            " ON p.id = e.party_b_id"
            " WHERE e.type = 'subcontract' AND p.email = 'rian@rian.ca'",
        )
        == 2
    )


def test_gary_kind_model_cost_zero_billout_25(mig_engine, clean_report):
    terms = rows(
        mig_engine,
        "SELECT ct.kind, ct.rate_amount, ct.currency, ct.model"
        " FROM compensation_terms ct"
        " JOIN engagements e ON e.id = ct.engagement_id"
        " JOIN parties p ON p.id = e.party_b_id"
        " WHERE p.email = 'gary@rian.ca' ORDER BY ct.kind",
    )
    assert [(k, str(r), c, m) for k, r, c, m in terms] == [
        ("billout", "25.00", "CAD", "hourly"),
        ("cost", "0.00", "CAD", "hourly"),      # zero is a REAL rate (07 #19)
    ]


def test_null_rate_means_no_term(mig_engine, clean_report):
    # rian row 1 has billout NULL -> cost term only; Matt has neither
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM compensation_terms ct"
            " JOIN engagements e ON e.id = ct.engagement_id"
            " JOIN parties p ON p.id = e.party_b_id"
            " WHERE p.email = 'matmoncoo@gmail.com'",
        )
        == 0
    )
    assert scalar(mig_engine, "SELECT count(*) FROM compensation_terms") == 7


def test_inactive_member_engagement_ended(mig_engine, clean_report):
    ended, consolidate = rows(
        mig_engine,
        "SELECT e.ended_on IS NOT NULL, e.consolidate_as FROM engagements e"
        " JOIN parties p ON p.id = e.party_b_id"
        " WHERE p.email = 'matmoncoo@gmail.com' AND e.type = 'subcontract'",
    )[0]
    assert ended is True  # is_active false -> ended at migration date
    assert consolidate == "Tingang"  # import-mapping config carried (02 §7)


def test_can_manage_imports_parity(mig_engine, clean_report):
    grants = dict(
        rows(
            mig_engine,
            "SELECT p.email, bool_or(e.can_manage_imports) FROM engagements e"
            " JOIN parties p ON p.id = e.party_b_id"
            " WHERE e.type = 'subcontract' GROUP BY p.email",
        )
    )
    assert grants["info@adipramono.com"] is True  # manager (canManageImports)
    assert grants["rian@rian.ca"] is True  # owner
    assert grants["gary@rian.ca"] is False
    assert grants["matmoncoo@gmail.com"] is False


def test_seeded_partnerships(mig_engine, clean_report):
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM engagements WHERE type = 'official_partnership'",
        )
        == 1
    )
    partner = rows(
        mig_engine,
        "SELECT pa.name, pb.name FROM engagements e"
        " JOIN parties pa ON pa.id = e.party_a_id"
        " JOIN parties pb ON pb.id = e.party_b_id"
        " WHERE e.type = 'partnership'",
    )
    assert partner == [("Bowden Works", "PlusROI")]  # no terms — R4's job


def test_effective_from_is_min_entry_date(mig_engine, clean_report):
    # Adi's entries start 2026-03-01 -> his terms backdate there (05 §2.3)
    dates = {
        str(d)
        for (d,) in rows(
            mig_engine,
            "SELECT ct.effective_from FROM compensation_terms ct"
            " JOIN engagements e ON e.id = ct.engagement_id"
            " JOIN parties p ON p.id = e.party_b_id"
            " WHERE p.email = 'info@adipramono.com'",
        )
    }
    assert dates == {"2026-03-01"}


# --- 2.4 projects + rate overrides ---------------------------------------------


def test_projects_carry_ids_income_adjustments(mig_engine, clean_report):
    row = rows(
        mig_engine,
        "SELECT income, income_currency, billout_adjustment_pct,"
        " billout_adjustment_amount, notes FROM projects WHERE id = :id",
        id=PR_SITE,
    )[0]
    assert (str(row[0]), row[1], str(row[2]), str(row[3]), row[4]) == (
        "12000.00", "CAD", "10.0000", "250.00", "big one",
    )
    # operator chain: pr-admin belongs to client PlusROI (merged party)
    # under operator Bowden Works
    op, cl = rows(
        mig_engine,
        "SELECT po.name, pc.name FROM projects p"
        " JOIN parties po ON po.id = p.operator_party_id"
        " JOIN parties pc ON pc.id = p.client_party_id WHERE p.id = :id",
        id=PR_ADMIN,
    )[0]
    assert (op, cl) == ("Bowden Works", "PlusROI")


def test_rate_overrides_near_verbatim(mig_engine, clean_report):
    absolute = rows(
        mig_engine,
        "SELECT ro.override_rate, ro.currency, ro.override_pct, p.email"
        " FROM rate_overrides ro JOIN parties p ON p.id = ro.worker_party_id"
        " WHERE ro.id = :id",
        id=RO_ABS,
    )[0]
    assert (str(absolute[0]), absolute[1], absolute[2], absolute[3]) == (
        "20.00", "CAD", None, "info@adipramono.com",
    )
    pct = rows(
        mig_engine,
        "SELECT ro.override_rate, ro.currency, ro.override_pct,"
        " ro.worker_party_id FROM rate_overrides ro WHERE ro.id = :id",
        id=RO_PCT,
    )[0]
    # project-wide NULL-worker pct rule keeps its shape (judgment #25)
    assert (pct[0], pct[1], str(pct[2]), pct[3]) == (None, None, "-20.0000", None)


# --- 2.5 import batches -----------------------------------------------------------


def test_import_batches_carry_source_and_uploader(mig_engine, clean_report):
    batches = {
        str(r[0]): r
        for r in rows(
            mig_engine,
            "SELECT b.id, b.source, u.idauth_username, b.name"
            " FROM import_batches b LEFT JOIN users u ON u.id = b.imported_by",
        )
    }
    assert batches[IMP_1][1] == "clockify"
    assert batches[IMP_1][2] == "adi"  # imported_by re-keyed (load-bearing)
    assert batches[IMP_3][1] == "toggl"  # historical value VERBATIM (T-MIG-008)
    assert batches[IMP_3][2] is None


# --- 2.6 time entries ---------------------------------------------------------------


def test_stamps_copied_verbatim_with_currency(mig_engine, clean_report):
    # T-MIG-013: byte-identical stamps; currency is part of the stamp
    row = rows(
        mig_engine,
        "SELECT cost_amount, cost_currency, billout_amount, billout_currency,"
        " source_rate, source_amount FROM time_entries WHERE id = :id",
        id=TE[1],
    )[0]
    assert (str(row[0]), row[1], str(row[2]), row[3]) == ("14.00", "CAD", "35.00", "CAD")
    assert (str(row[4]), str(row[5])) == ("12.5000", "12.5000")  # 4dp provenance
    # NULL stamp -> NULL currency (never priced; judgment #19)
    baseline = rows(
        mig_engine,
        "SELECT cost_amount, cost_currency FROM time_entries WHERE id = :id",
        id=TE[5],
    )[0]
    assert baseline == (None, None)


def test_raw_triple_carries_verbatim_on_every_row(mig_engine, clean_report):
    # T-MIG-009 (the rename): raw_* equals the legacy text on ALL rows
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM time_entries te"
            " JOIN legacy.time_entries lte ON lte.id = te.id"
            " WHERE te.raw_operator IS DISTINCT FROM lte.operator"
            "    OR te.raw_client IS DISTINCT FROM lte.client"
            "    OR te.raw_project IS DISTINCT FROM lte.project",
        )
        == 0
    )
    blocked = rows(
        mig_engine,
        "SELECT raw_operator, raw_client, raw_project, worker_party_id,"
        " project_id FROM time_entries WHERE id = :id",
        id=TE[4],
    )[0]
    assert blocked == ("PlusROI", None, "Mystery", None, None)


def test_source_derivation(mig_engine, clean_report):
    sources = dict(
        rows(mig_engine, "SELECT id::text, source FROM time_entries")
    )
    assert sources[TE[1]] == "clockify"
    assert sources[TE[5]] == "manual"  # no import batch
    assert sources[TE[6]] == "backfill"  # historical value carried (T-MIG-008)
    assert sources[TE[7]] == "toggl"


def test_baseline_entry_survives(mig_engine, clean_report):
    # the 2025-12-31 sentinel is load-bearing data, not Toggl residue
    row = rows(
        mig_engine,
        "SELECT project_id::text, duration_seconds, billout_amount, start_at"
        " FROM time_entries WHERE id = :id",
        id=TE[5],
    )[0]
    assert row[0] == PR_BLOCK
    assert row[1] == 36000
    assert str(row[2]) == "100.00"
    assert str(row[3]) == "2025-12-31 00:00:00"  # naive wall-clock preserved


def test_worker_and_via_engagement_resolution(mig_engine, clean_report):
    row = rows(
        mig_engine,
        "SELECT p.email, e.type FROM time_entries te"
        " JOIN parties p ON p.id = te.worker_party_id"
        " JOIN engagements e ON e.id = te.via_engagement_id WHERE te.id = :id",
        id=TE[1],
    )[0]
    assert row == ("info@adipramono.com", "subcontract")


# --- 2.7 invoices (lock + parties + audit values) --------------------------------------


def test_lock_carries_exactly(mig_engine, clean_report):
    # T-MIG-007: every previously-locked row is locked via a line; no
    # previously-unlocked row became locked
    locked = {
        r[0]
        for r in rows(
            mig_engine,
            "SELECT id::text FROM time_entries WHERE invoice_line_id IS NOT NULL",
        )
    }
    assert locked == {TE[1], TE[6]}


def test_invoice_from_to_parties_pinned_algorithm(mig_engine, clean_report):
    bw = party_id(mig_engine, "Bowden Works")
    plusroi = party_id(mig_engine, "PlusROI")
    brentwood = party_id(mig_engine, "Brentwood")
    invoices = {
        str(r[0]): r
        for r in rows(
            mig_engine,
            "SELECT id, from_party_id, to_party_id, manual_total, currency,"
            " status FROM invoices",
        )
    }
    # from_party = the tenant for EVERY legacy invoice
    assert all(r[1] == bw for r in invoices.values())
    # operator chip != BW -> that operator party
    assert invoices[INV_MAY][2] == plusroi
    # derived operator == BW -> client party (derived from attached entries)
    assert invoices[INV_LEGACY][2] == plusroi
    # no operator scope; client chip -> client party
    assert invoices[INV_OPEN][2] == brentwood
    assert str(invoices[INV_MAY][3]) == "5000.00"  # manual_total verbatim
    assert invoices[INV_MAY][4] == "CAD"


def test_attach_audit_values_carry(mig_engine, clean_report):
    # T-MIG-014: invoice_applied_at/_by -> attached_at/attached_by VERBATIM
    row = rows(
        mig_engine,
        "SELECT il.attached_at AT TIME ZONE 'UTC', u.idauth_username,"
        " il.snapshot_client_name, il.snapshot_project_name, il.kind"
        " FROM time_entries te"
        " JOIN invoice_lines il ON il.id = te.invoice_line_id"
        " LEFT JOIN users u ON u.id = il.attached_by WHERE te.id = :id",
        id=TE[1],
    )[0]
    assert str(row[0]) == "2026-05-25 10:00:00"
    assert row[1] == "rian"
    assert (row[2], row[3], row[4]) == ("Brentwood", "Site Rebuild", "time")


# --- 2.8 CC + 2.9 comments ----------------------------------------------------------


def test_cc_carries_with_cad_and_null_locks(mig_engine, clean_report):
    assert rows(
        mig_engine, "SELECT name, currency FROM cc_expense_batches"
    ) == [("May statement", "CAD")]
    assert (
        scalar(
            mig_engine,
            "SELECT count(*) FROM cc_expense_lines WHERE invoice_line_id IS NOT NULL",
        )
        == 0
    )
    assert scalar(mig_engine, "SELECT count(*) FROM cc_expense_lines") == 3
    assert scalar(mig_engine, "SELECT count(*) FROM cc_highlight_keywords") == 2
    assert scalar(mig_engine, "SELECT count(*) FROM cc_auto_rules") == 1


def test_comments_rekeyed_authors(mig_engine, clean_report):
    result = rows(
        mig_engine,
        "SELECT ua.idauth_username, ur.idauth_username FROM comments c"
        " LEFT JOIN users ua ON ua.id = c.author_id"
        " LEFT JOIN users ur ON ur.id = c.resolved_by ORDER BY c.created_at",
    )
    assert result == [(None, "adi"), ("rian", None)]  # ghost -> NULL