"""Harness parity + idempotency on the fixture dataset.

T-MIG-001/002/003/004 in miniature (the full-history versions run
against real data in the nightly rehearsal), T-MIG-011 (second run is a
no-op), and the id-stability / raw-key contracts from spec.md."""

import json

from harness.compare import DEFAULT_ALLOWLIST, load_allowlist, run_compare
from tests.migrate.conftest import INV_MAY, PR_SITE, TE, extract_both, rows, scalar

ALLOWLIST = load_allowlist(DEFAULT_ALLOWLIST)


def _stripped(doc: dict) -> str:
    return json.dumps({k: v for k, v in doc.items() if k != "meta"}, sort_keys=True)


def test_compare_green_after_clean_run(mig_url, clean_report):
    old_doc, new_doc = extract_both(mig_url)
    report = run_compare(old_doc, new_doc, ALLOWLIST)
    failures = [(d.path, d.old, d.new) for d in report.failures]
    assert report.ok, f"harness RED on the fixture dataset: {failures}"
    # meta is the ONLY allowed difference class on a clean run
    assert all(d.path.startswith("meta/") for d in report.diffs if d.allowed)


def test_unassigned_group_key_byte_matches(mig_url, clean_report):
    old_doc, new_doc = extract_both(mig_url)
    key = 'unassigned:["PlusROI",null,"Mystery"]'  # raw triple, verbatim
    assert key in old_doc["months"]["2026-03"]
    assert key in new_doc["months"]["2026-03"]
    assert old_doc["months"]["2026-03"][key] == new_doc["months"]["2026-03"][key]


def test_id_stability_contracts(mig_url, clean_report):
    old_doc, new_doc = extract_both(mig_url)
    # project ids ARE the month group keys on both sides
    assert PR_SITE in new_doc["months"]["2026-03"]
    # attached-entry id sets compare by legacy uuid (05 step 2.6 id-carry)
    assert new_doc["invoices"][INV_MAY]["attached_entry_ids"] == [TE[1]]
    assert (
        old_doc["invoices"][INV_MAY]["attached_entry_ids"]
        == new_doc["invoices"][INV_MAY]["attached_entry_ids"]
    )


def test_second_run_is_a_noop(mig_url, mig_engine, clean_report):
    # T-MIG-011: rebuild-from-legacy twice -> identical canonical output
    # ("cutover day is the hundredth run"). Party uuids regenerate, but
    # the canonical document keys by names/emails/carried ids.
    from app.services.migrate import run_all

    _, first = extract_both(mig_url)
    report = run_all(engine=mig_engine)
    assert report.ok, report.render()
    assert report.straggler_count == 0
    _, second = extract_both(mig_url)
    assert _stripped(first) == _stripped(second)


def test_user_preferences_survive_reruns(mig_url, mig_engine, clean_report):
    # users are upserted column-scoped, never truncated: a login-owned
    # field (preferences) must survive a rebuild.
    from sqlalchemy import text

    from app.services.migrate import run_all

    with mig_engine.begin() as conn:
        conn.execute(text(
            "UPDATE users SET preferences = '{\"theme\": \"dark\"}'::jsonb"
            " WHERE idauth_username = 'rian'"
        ))
    report = run_all(engine=mig_engine)
    assert report.ok
    prefs, linked = rows(
        mig_engine,
        "SELECT preferences, person_party_id IS NOT NULL FROM users"
        " WHERE idauth_username = 'rian'",
    )[0]
    assert prefs == {"theme": "dark"}
    assert linked is True  # re-linked to the freshly rebuilt party
    assert scalar(mig_engine, "SELECT count(*) FROM users") >= 2