#!/usr/bin/env python3
"""Verify static 200GB fixture files without executing SQL or using a network."""
from __future__ import annotations
import csv
from datetime import date, timedelta
from pathlib import Path

BASE = Path(__file__).resolve().parent

def rows(name):
    with (BASE / name).open(newline="", encoding="utf-8") as f:
        return list(csv.DictReader(f))

def require(condition, message):
    if not condition:
        raise AssertionError(message)

manifest = rows("partition-manifest-200GB-20260831.csv")
require(len(manifest) == 400, "manifest must contain 400 partitions")
dates = [date.fromisoformat(r["partition_date"]) for r in manifest]
require(len(set(dates)) == 400, "partition dates must be unique")
require(all(b - a == timedelta(days=1) for a, b in zip(dates, dates[1:])), "dates must be contiguous")
require(all(int(r["file_count"]) == 4 for r in manifest), "each partition must contain four files")
require(all(int(r["bytes_per_file"]) == 125_000_000 for r in manifest), "file bytes mismatch")
require(all(int(r["assumed_partition_bytes"]) == int(r["file_count"]) * int(r["bytes_per_file"]) for r in manifest), "partition multiplication mismatch")
require(sum(int(r["assumed_partition_bytes"]) for r in manifest) == 200_000_000_000, "manifest total mismatch")

window = [r for r in manifest if date(2026, 6, 1) <= date.fromisoformat(r["partition_date"]) < date(2026, 7, 1)]
require(len(window) == 30, "window must contain 30 partitions")
require(sum(int(r["file_count"]) for r in window) == 120, "window must contain 120 files")
require(sum(int(r["assumed_partition_bytes"]) for r in window) == 15_000_000_000, "window bytes mismatch")
require(30 * 200_000_000_000 // 400 == 15_000_000_000, "ratio arithmetic mismatch")

columns = rows("column-logical-bytes-200GB-20260831.csv")
require({r["column_name"] for r in columns} == {"channel", "revenue", "cost_of_goods", "variable_fulfillment_cost", "order_date", "is_internal"}, "selected columns mismatch")
require(sum(int(r["assumed_logical_bytes"]) for r in columns) == 15_000_000_000, "column allocation must conserve bounded logical bytes")
require(all("not physical read" in r["limitation"].lower() for r in columns), "column limitations must reject physical-read inference")

sql = (BASE / "query-plan-fixture-200GB-20260831.sql").read_text(encoding="utf-8")
sql_upper = sql.upper()
require("DO NOT EXECUTE" in sql_upper, "SQL must carry no-run warning")
require("SELECT *" not in sql_upper, "SQL must use explicit columns")
for token in ["CHANNEL", "REVENUE", "COST_OF_GOODS", "VARIABLE_FULFILLMENT_COST", "ORDER_DATE", "IS_INTERNAL"]:
    require(token in sql_upper, f"missing selected column: {token}")
for token in ["ORDER_DATE >= DATE '2026-06-01'", "ORDER_DATE < DATE '2026-07-01'", "IS_INTERNAL = FALSE", "SUM(REVENUE - COST_OF_GOODS - VARIABLE_FULFILLMENT_COST)", "GROUP BY CHANNEL", "ROW_NUMBER()", "CONTRIBUTION_RANK <= 5"]:
    require(token in sql_upper, f"missing SQL control: {token}")

control = rows("static-control-input-200GB-20260831.csv")
eligible = [r for r in control if r["is_internal"] == "false" and date(2026, 6, 1) <= date.fromisoformat(r["order_date"]) < date(2026, 7, 1)]
calculated = sorted(((r["channel"], int(r["revenue"]) - int(r["cost_of_goods"]) - int(r["variable_fulfillment_cost"])) for r in eligible), key=lambda x: (-x[1], x[0]))[:5]
expected = rows("expected-control-output-200GB-20260831.csv")
require([(r["channel"], int(r["contribution"])) for r in expected] == calculated, "top-five expected output mismatch")
require([int(r["contribution_rank"]) for r in expected] == [1, 2, 3, 4, 5], "rank sequence mismatch")

held = {r["field"]: r for r in rows("held-observation-fields-200GB-20260831.csv")}
for field in ["engine_plan", "pruning", "runtime", "processed_bytes", "scanned_bytes", "billed_bytes", "cost", "output"]:
    require(held[field]["value"] == "" and held[field]["state"] == "held_blank", f"{field} must remain blank")
for field in ["warehouse_connected", "execution_performed", "dry_run_performed", "second_run", "customer_data", "independent_validation", "sla_claimed", "product_warehouse_connection_capability", "product_200gb_processing_capability", "product_task_timeline_capability", "product_cancel_rerun_download_capability"]:
    require(held[field]["value"] == "false" and held[field]["state"] == "explicit_false", f"{field} must be explicitly false")

source_text = (BASE / "external-source-check-200GB-20260831.md").read_text(encoding="utf-8")
for url in [
    "https://cloud.google.com/bigquery/docs/samples/bigquery-query-dry-run",
    "https://docs.cloud.google.com/bigquery/docs/best-practices-costs",
    "https://docs.cloud.google.com/bigquery/docs/query-plan-explanation",
    "https://spark.apache.org/docs/latest/sql-ref-syntax-qry-explain.html",
    "https://hive.apache.org/docs/latest/language/languagemanual-explain/",
    "https://parquet.apache.org/docs/file-format/pageindex/",
    "https://docs.opensearch.org/latest/api-reference/search-apis/profile/",
    "https://grafana.com/docs/grafana/latest/explore/explore-inspector/",
]:
    require(url in source_text, f"missing bounded source: {url}")
require("None validates this fixture" in source_text, "source limitations missing")
print("PASS: 400 partitions; 1,600 files; 200,000,000,000 assumed bytes")
print("PASS: 30 window partitions; 120 files; 15,000,000,000 assumed bounded bytes")
print("PASS: selected-column allocation conserves 15,000,000,000 logical bytes")
print("PASS: SQL controls present; SQL was not executed")
print("PASS: synthetic top-five arithmetic matches expected output")
print("PASS: observation fields held blank and capability/evidence flags false")
print("PASS: bounded sources and limitations present")