`verify_recovery.py`

109 lines
#!/usr/bin/env python3
"""Independent read-only E2E verifier for completed TK-10066 recovery."""
import argparse
import json
import os
import subprocess

import recover_tk10066 as recovery


def sql_rows(sql):
    result = subprocess.run(
        ["psql", "host=/tmp dbname=dw_unified", "-X", "-qAt", "-F", "\t", "-c", sql],
        capture_output=True, text=True, check=True)
    return [line.split("\t") for line in result.stdout.splitlines() if line]


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("manifest")
    parser.add_argument("--output", required=True)
    args = parser.parse_args()
    manifest = json.load(open(args.manifest))
    failures, checks = [], []

    def check(name, condition, detail=None):
        checks.append({"name": name, "verdict": "PASS" if condition else "FAIL", "detail": detail})
        if not condition:
            failures.append({"name": name, "detail": detail})

    for item in manifest["same_identity"]:
        retained = recovery.product(item["retain_product_id"])
        archived = recovery.product(item["archive_product_ids"][0])
        check("same_identity_retained_draft", retained and retained["status"] == "DRAFT",
              {"sku": item["sku"], "product_id": item["retain_product_id"]})
        check("same_identity_twin_archived", archived and archived["status"] == "ARCHIVED",
              {"sku": item["sku"], "product_id": item["archive_product_ids"][0]})

    for item in manifest["cross_identity"]:
        moved = recovery.product(item["move_product"]["product_id"])
        retained_pid = item["retain_identity"]["product_id"]
        retained_pid = retained_pid if retained_pid.startswith("gid://") else f"gid://shopify/Product/{retained_pid}"
        retained = recovery.product(retained_pid)
        moved_skus = {v["sku"] for v in moved["variants"]["nodes"]} if moved else set()
        retained_skus = {v["sku"] for v in retained["variants"]["nodes"]} if retained else set()
        check("cross_identity_mover_reskud",
              moved and moved["status"] == "DRAFT" and item["new_sku"] + "-Sample" in moved_skus,
              {"old": item["old_sku"], "new": item["new_sku"], "product_id": item["move_product"]["product_id"]})
        check("cross_identity_owner_retained",
              retained and retained["status"] == "DRAFT" and item["old_sku"] + "-Sample" in retained_skus,
              {"sku": item["old_sku"], "product_id": retained_pid})

    for item in manifest["handle_repairs"]:
        target, expected = item["target"], item["shopify_product"]
        rows = sql_rows(
            "SELECT COALESCE(shopify_product_id,''),COALESCE(shopify_handle,''),on_shopify::text "
            f"FROM rwltd_catalog WHERE mfr_sku={recovery.q(target['mfr_sku'])} "
            f"AND dw_sku={recovery.q(target['sku'])};")
        check("handle_link_catalog", rows == [[expected["id"], item["handle"], "true"]],
              {"mfr_sku": target["mfr_sku"], "rows": rows})

    catalog_cross = sql_rows("""
      SELECT dw_sku,COUNT(DISTINCT UPPER(TRIM(mfr_sku)))::text
      FROM rwltd_catalog WHERE COALESCE(dw_sku,'')<>''
      GROUP BY dw_sku HAVING COUNT(DISTINCT UPPER(TRIM(mfr_sku)))>1 ORDER BY dw_sku;
    """)
    check("catalog_has_no_cross_identity_duplicate_skus", not catalog_cross, catalog_cross[:10])
    registry_dupes = sql_rows("""
      SELECT dw_sku,COUNT(*)::text FROM dw_sku_registry
      WHERE vendor_prefix='DWKR' GROUP BY dw_sku HAVING COUNT(*)>1 ORDER BY dw_sku;
    """)
    check("registry_dwkr_skus_unique", not registry_dupes, registry_dupes[:10])

    affected_skus = ([x["sku"] for x in manifest["same_identity"]] +
                     [x["old_sku"] + "-Sample" for x in manifest["cross_identity"]] +
                     [x["new_sku"] + "-Sample" for x in manifest["cross_identity"]])
    live_collisions = []
    for sku in affected_skus:
        owners = [v for v in recovery.sku_matches(sku)
                  if v.get("product", {}).get("status") in ("ACTIVE", "DRAFT")]
        if len(owners) != 1:
            live_collisions.append({"sku": sku, "owners": owners})
    check("affected_active_or_draft_skus_have_one_owner", not live_collisions, live_collisions[:10])

    output = {
        "schema": "tk-10066-recovery-e2e/v1",
        "intent": "Verify approved Reid Witlin recovery across Shopify, rwltd_catalog, and dw_sku_registry.",
        "risk_tier": "R4",
        "environment": "live Shopify read API + canonical dw_unified",
        "manifest": os.path.realpath(args.manifest),
        "summary": {"checks": len(checks), "failures": len(failures),
                    "same_identity": len(manifest["same_identity"]),
                    "cross_identity": len(manifest["cross_identity"]),
                    "link_repairs": len(manifest["handle_repairs"])},
        "checks": checks, "failures": failures,
        "cleanup": "Duplicate products retained as ARCHIVED for reversible recovery; no deletions.",
        "verdict": "PASS" if not failures else "FAIL",
    }
    with open(args.output, "w") as fh:
        json.dump(output, fh, indent=2, sort_keys=True)
        fh.write("\n")
    print(json.dumps({"output": args.output, **output["summary"], "verdict": output["verdict"]}))
    if failures:
        raise SystemExit(1)


if __name__ == "__main__":
    main()