84 lines
2.6 KiB
Python
84 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Verify the todo-census artifact: re-fetch the live API and assert the on-disk
|
|
CSV/markdown match it, then exit 0 only if everything is consistent.
|
|
|
|
Usage: python3 verify_todo_census.py
|
|
"""
|
|
import csv
|
|
import json
|
|
import os
|
|
import sys
|
|
from urllib.error import URLError
|
|
from urllib.request import urlopen
|
|
|
|
CSV_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
|
"todo-census-2026-11-01.csv")
|
|
API_URL = "https://jsonplaceholder.typicode.com/todos"
|
|
|
|
|
|
def fetch(path):
|
|
req = urlopen(path, timeout=30)
|
|
raw = req.read()
|
|
req.close()
|
|
return json.loads(raw.decode("utf-8"))
|
|
|
|
|
|
def read_csv_rows(path):
|
|
with open(path, encoding="utf-8") as f:
|
|
return list(csv.DictReader(f, delimiter="|"))
|
|
|
|
|
|
def main():
|
|
errors = []
|
|
|
|
try:
|
|
live = fetch(API_URL)
|
|
except (URLError, OSError, ValueError) as exc:
|
|
# A verifier that re-fetches a live API cannot grade offline. That is
|
|
# rc 2 ("could not grade"), not a traceback and not a failed artifact.
|
|
print(f"CANNOT GRADE: fetch {API_URL} failed: {exc}")
|
|
sys.exit(2)
|
|
if not isinstance(live, list):
|
|
print("FAIL: API did not return a list", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
disk = read_csv_rows(CSV_PATH)
|
|
|
|
# 1. Row count
|
|
if len(live) != len(disk):
|
|
errors.append(f"row count mismatch: API={len(live)} CSV={len(disk)}")
|
|
|
|
# 2. Completed/pending tallies
|
|
live_done = sum(1 for r in live if r["completed"])
|
|
if live_done != sum(1 for r in disk if r["completed"] == "1"):
|
|
errors.append("completed tally mismatch")
|
|
|
|
# 3. Every disk row exists in the live set (id+user_id+title+completed)
|
|
live_map = {(str(r["id"]), str(r["userId"]), r["title"], "1" if r["completed"] else "0"): 1
|
|
for r in live}
|
|
for i, r in enumerate(disk, start=1):
|
|
key = (r["id"], r["user_id"], r["title"], r["completed"])
|
|
if key not in live_map:
|
|
errors.append(f"row {i} not present in live API: {key}")
|
|
|
|
# 4. Sampling: first, mid, last ids match the live API
|
|
for idx in (0, len(disk) // 2, len(disk) - 1):
|
|
r = disk[idx]
|
|
live_row = next((x for x in live if x["id"] == int(r["id"])), None)
|
|
if live_row is None or live_row["title"] != r["title"]:
|
|
errors.append(f"sample id={r['id']} mismatch")
|
|
|
|
if errors:
|
|
print("VERIFY FAILED")
|
|
for e in errors:
|
|
print(" -", e)
|
|
sys.exit(1)
|
|
|
|
print(f"VERIFY OK: {len(disk)} rows match the live API "
|
|
f"(completed={sum(1 for r in disk if r['completed']=='1')})")
|
|
sys.exit(0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|