#!/usr/bin/env python3 """EZSTAT VERIFY — a customer-facing self-verification tool. PURPOSE. EzStat's differentiator is not "our arithmetic is correct" (every competitor claims that and none can prove it). It is "here is a tool — run it against YOUR key and YOUR data, and catch us if we are wrong." This script writes a small counter sequence whose total you can verify on paper, retries one write with the same idempotency key, then compares three surfaces: ground truth (what we just sent), the CSV export, and the API's reported value. If any surface disagrees with the arithmetic you can do on paper, it exits non-zero. CLAIM BOUNDARY (also printed at the end of every run): This proves: counter aggregation, idempotent retry, and agreement between ground truth, the CSV export and the API. This does NOT prove: gauge percentile behaviour, long-range downsampling, or cross-range consistency. DOWNLOAD (canonical, served by us over HTTPS — read it before you run it): curl -fsSL https://ezstat.dev/tools/ezstat-verify.py -o ezstat-verify.py Documentation: https://ezstat.dev/docs#verify USAGE: python3 ezstat-verify.py --key YOUR_API_KEY python3 ezstat-verify.py --key YOUR_API_KEY --dry-run # print plan, write nothing python3 ezstat-verify.py --key YOUR_API_KEY --json # machine-readable python3 ezstat-verify.py # key from env EZSTAT_API_KEY ENV: EZSTAT_API_KEY your API key (used if --key not given) EXIT CODES: 0 every surface reconciles with ground truth 1 at least one surface disagrees 2 could not complete (missing key, network failure) WRITES TO: one stat only (default: verify.reconcile_check — the dot lands it in your dashboard's "verify" folder). Delete it from the dashboard afterwards. """ import argparse import hashlib import json import os import sys import time import urllib.error import urllib.parse import urllib.request # --- Configuration --------------------------------------------------------- BASE_URL = 'https://api.ezstat.dev' INGEST_URL = f'{BASE_URL}/ez' DEFAULT_STAT = 'verify.reconcile_check' UA = 'ezstat-verify/1.0' # Exit codes (also documented in the module docstring). # These are the THREE outcomes a customer can rely on; do not collapse them. EXIT_OK = 0 # Every surface reconciles with ground truth EXIT_DISAGREEMENT = 1 # At least one surface disagrees with ground truth EXIT_INCOMPLETE = 2 # Could not complete (missing key, network, auth, no data) # The claim boundary text — single source of truth. Every exit path uses this: # - the main() `done()` function (normal exits, stdout) # - the KeyboardInterrupt handler (interrupted exits, stderr) # Do not edit the text: it is the contract with the customer about what was tested. _CLAIM_BOUNDARY_LINES = ( '=== CLAIM BOUNDARY ===', ' This proves: counter aggregation, idempotent retry, and agreement between', ' ground truth, the CSV export and the API.', ' This does NOT prove: gauge percentile behaviour, long-range downsampling, or', ' cross-range consistency.', ) def _print_boundary(stream): """Print the claim boundary to a text stream. Used by both the normal-exit path (done() → stdout) and the KeyboardInterrupt handler (stderr).""" for line in _CLAIM_BOUNDARY_LINES: stream.write(line + '\n') stream.flush() # The test sequence — values chosen so the customer can verify on paper: # 1 + 1 + 3 = 5, three points. The retry on the middle write (with the SAME idempotency # key) must be deduplicated; if it isn't, the total becomes 6 and count becomes 4 — the # exact failure mode this tool exists to catch. VALUES = [1, 1, 3] EXPECTED_SUM = sum(VALUES) # 5 EXPECTED_COUNT = len(VALUES) # 3 RETRY_INDEX = 1 # retry the middle write def make_ikey(stat, ts, label): """Per-write idempotency key (24 hex chars). The retry uses the SAME key as one of the originals so dedup is unambiguous.""" return hashlib.sha1(f'{stat}|{ts}|{label}'.encode()).hexdigest()[:24] def slots(now): """Three minute-boundary timestamps BEFORE now, so this run's writes don't collide with the next run's writes.""" base = (now // 60) * 60 return [base - 180, base - 120, base - 60] def http_post(url, form): body = urllib.parse.urlencode(form).encode() req = urllib.request.Request( url, data=body, headers={ 'Content-Type': 'application/x-www-form-urlencoded', 'User-Agent': UA, 'Accept': 'application/json', }, ) try: with urllib.request.urlopen(req, timeout=20) as r: return r.status, r.read() except urllib.error.HTTPError as e: return e.code, e.read() except urllib.error.URLError as e: return None, str(e).encode() def http_get(url, params=None, bearer=None): """GET helper. If `bearer` is given, send it as `Authorization: Bearer ` — EzStat's v1 READ endpoints reject query-string keys by design (a key in a URL lands in access logs, proxy logs, and browser history). The WRITE/INGEST endpoint uses a different scheme: `ezkey` form field, for StatHat wire compatibility.""" if params: url = f'{url}?{urllib.parse.urlencode(params)}' headers = { 'User-Agent': UA, 'Accept': '*/*', } if bearer: headers['Authorization'] = f'Bearer {bearer}' req = urllib.request.Request(url, headers=headers) try: with urllib.request.urlopen(req, timeout=20) as r: return r.status, r.read() except urllib.error.HTTPError as e: return e.code, e.read() except urllib.error.URLError as e: return None, str(e).encode() def looks_like_error_body(text): """This product returns HTTP 200 with an error BODY for wire compatibility. Detect it from the body so we don't falsely report success.""" if not text: return False head = text.strip()[:2048] # cheap substring checks if '"error"' in head or '"ok":false' in head.lower(): return True if head.startswith('error') or head.startswith('ERR:'): return True try: obj = json.loads(head) if isinstance(obj, dict): if obj.get('error'): return True if obj.get('ok') is False: return True except (json.JSONDecodeError, ValueError): pass return False def write_point(key, stat, count, ts, ikey): return http_post(INGEST_URL, { 'ezkey': key, 'stat': stat, 'count': str(count), 't': str(ts), 'ikey': ikey, }) def fetch_csv(key, stat): """READ. The REAL CSV export surface: /api/v1/stats//export?format=csv, with the API key as Authorization: Bearer (header only, by design — a key in a URL lands in access logs, proxy logs, browser history; the route deliberately has no ?ukey=). HISTORY (why this docstring matters): an earlier revision of this tool hit the plain data endpoint here instead, because export was session-cookie-only at the time — and fetch_reported hits that SAME data endpoint. The result was a verification tool that compared one API response with itself: it would have printed "verified" with the CSV export pipeline completely broken. Export has accepted Bearer since the dual-auth work; the workaround outlived the constraint. If export ever regresses to 401 for Bearer, this tool must FAIL loudly on this surface — never quietly fall back to the data endpoint, which un-verifies the comparison. Response shape: a `timestamp,value` header row; one RAW stored row per line (timestamps rendered in the account's timezone); then trailing `#` comment lines disclosing the applied window — including an explicit `# TRUNCATED` marker if the per-request cap was hit, which this tool treats as an incomplete comparison.""" return http_get( f'{BASE_URL}/api/v1/stats/{urllib.parse.quote(stat, safe=".")}/export?format=csv', bearer=key, ) def fetch_reported(key, stat): """READ. The dashboard/data surface — a genuinely DIFFERENT read path from the CSV export above, which is the point: two independent surfaces must agree with each other and with the ground-truth ledger. The summary's sum/count are the reported-value surface (the data array is downsampled; summary is authoritative).""" return http_get( f'{BASE_URL}/api/v1/stats/{urllib.parse.quote(stat, safe=".")}', bearer=key, ) def parse_csv(text): """Parse the REAL export CSV: skip the `timestamp,value` header row and the trailing `#` comment lines (applied-window disclosure / TRUNCATED marker), sum the value column, count the data rows. The export emits RAW stored rows — not a downsampled series — so the row count IS the stored point count. That makes this surface a stronger check than a summary object: the arithmetic is recomputed here, client-side, from the same rows a customer's warehouse would load.""" if not text: return [], 0 vals = [] for i, line in enumerate(text.splitlines()): line = line.strip() if not line or line.startswith('#'): continue if i == 0 and not line.split(',', 1)[-1].strip().replace('.', '', 1).lstrip('-').isdigit(): # header row (timestamp,value) continue parts = line.rsplit(',', 1) if len(parts) != 2: continue try: vals.append(float(parts[1])) except (TypeError, ValueError): pass return vals, len(vals) def parse_reported(text): """Parse the summary from the data endpoint's JSON. The summary object holds sum/count/min/max/avg; top-level sum/count are checked first for forwards compatibility with a flatter response shape.""" try: obj = json.loads(text) except (json.JSONDecodeError, ValueError): return None, None if not isinstance(obj, dict): return None, None s = obj.get('sum') or obj.get('total') c = obj.get('count') or obj.get('n') or obj.get('point_count') if s is None and c is None: # The data endpoint nests aggregations in `summary`. summary = obj.get('summary') if isinstance(summary, dict): s = summary.get('sum') or summary.get('total') c = summary.get('count') or summary.get('n') if s is None and c is None: # Generic nested result/data envelope. nested = obj.get('result') or obj.get('data') if isinstance(nested, dict): s = nested.get('sum') or nested.get('total') c = nested.get('count') or nested.get('n') return s, c def main(): global BASE_URL, INGEST_URL ap = argparse.ArgumentParser( description='EzStat customer-facing self-verification tool. Writes a small counter ' 'sequence, retries one write with the same idempotency key, then compares ' 'ground truth against the CSV export and the API reported value.', ) ap.add_argument('--key', default=os.environ.get('EZSTAT_API_KEY'), help='Your EzStat API key (or set EZSTAT_API_KEY).') ap.add_argument('--stat', default=DEFAULT_STAT, help=f'Stat name to write to (default: {DEFAULT_STAT}).') ap.add_argument('--base-url', default=BASE_URL, help=f'API base URL (default: {BASE_URL}).') ap.add_argument('--dry-run', action='store_true', help='Print the plan and exit. Writes nothing, reads nothing.') ap.add_argument('--cleanup', action='store_true', help='Delete the test stat after a successful run. OFF by default: a ' 'customer tool must never auto-delete data. Turn it on for CI and ' 'harness accounts, where a leftover stat breaks count-asserting ' 'gates (two correct designs — no-auto-delete here, an absolute ' 'stat-count assertion there — compose into a broken third thing).') ap.add_argument('--json', action='store_true', help='Emit exactly one JSON object on stdout (suitable for CI).') args = ap.parse_args() # Honor --base-url. Before this assignment existed the flag was accepted and silently # IGNORED — every fetch used the module default regardless of what the user passed. # A flag that does nothing is a false affordance, the same defect class as the # self-referential comparison this tool once shipped with. BASE_URL = args.base_url.rstrip('/') INGEST_URL = f'{BASE_URL}/ez' json_mode = args.json def say(*parts): if not json_mode: print(*parts, flush=True) def jout(obj): print(json.dumps(obj, indent=2, sort_keys=True), flush=True) def done(exit_code, **fields): """Single exit function. EVERY exit path in this tool MUST route through here. Prints the claim boundary on every exit (pass=0, disagreement=1, could-not-complete=2). The failure path is exactly where a reader is most likely to over-infer what was tested, so the boundary is unambiguous there. Future code paths cannot bypass the boundary because they must route through this function — that is the entire point of having one exit point.""" if json_mode: jout({**fields, 'exit': exit_code, 'claim_boundary': list(_CLAIM_BOUNDARY_LINES)}) else: _print_boundary(sys.stdout) return exit_code # --- Key --- if not args.key: say('ERROR: no API key. Pass --key YOUR_KEY or set EZSTAT_API_KEY.') return done(EXIT_INCOMPLETE, error='no-key') # --- Build the plan (always printed before any network call) --- now = int(time.time()) ts = slots(now) plan_writes = [] for i, v in enumerate(VALUES): plan_writes.append({ 'slot': i + 1, 'value': v, 't': ts[i], 'ikey': make_ikey(args.stat, ts[i], f'slot{i+1}'), }) retry = { 'slot': RETRY_INDEX + 1, 'value': VALUES[RETRY_INDEX], 't': ts[RETRY_INDEX], 'ikey': make_ikey(args.stat, ts[RETRY_INDEX], f'slot{RETRY_INDEX+1}'), 'note': 'same idempotency key as the original slot — must be deduplicated', } say('=== PLAN (printed before any network call) ===') say(f' Base URL : {args.base_url}') say(f' Stat to write : {args.stat} (lives in the "verify" folder via dot-namespace)') say(f' Values to write : {VALUES} (sum on paper = {EXPECTED_SUM}, count = {EXPECTED_COUNT})') say(f' HTTP writes : 4 (3 unique writes + 1 retry on slot{RETRY_INDEX+1} with same idempotency key)') say(f' Writes nothing else: TRUE (only this stat, only these fields, only these keys)') say(f' Then compare : ground truth vs CSV export vs API reported value') say('') say(' Per-write plan:') for w in plan_writes: say(f' slot{w["slot"]} value={w["value"]} t={w["t"]} ikey={w["ikey"]}') say(f' Retry plan : slot{retry["slot"]} value={retry["value"]} t={retry["t"]} ikey={retry["ikey"]}') say(f' (same ikey as slot{retry["slot"]} — dedup or fail)') plan_obj = { 'stat': args.stat, 'base_url': args.base_url, 'values': VALUES, 'expected_sum': EXPECTED_SUM, 'expected_count': EXPECTED_COUNT, 'writes': plan_writes, 'retry': retry, 'writes_nothing_else': True, 'surfaces_compared': ['ground_truth', 'csv_export', 'api_reported'], } if args.dry_run: say('') say('=== DRY RUN — nothing was written, nothing was read, nothing was deleted ===') return done(EXIT_OK, dry_run=True, plan=plan_obj) # --- WRITE THE SEQUENCE --- say('') say('=== WRITING THE SEQUENCE ===') write_log = [] any_write_failed = False # Failure categories tracked separately so the verdict text can name each one # precisely. The categories are: # write_failures — our writes did not land # auth_failures — a READ returned 401 (auth/access refused). Our arithmetic # was NOT tested in this case; we never read the surface. # Conflating this with reconciliation failure tells the # customer we computed wrongly when we merely could not read. # incomplete_reasons — other read-side problems: network, no data, unparseable # body, non-401 HTTP error. We have no value to compare. # reconciliation_failures— a READ returned 200 OK and the numbers disagree with # ground truth. This is the only category that maps to # EXIT_DISAGREEMENT; everything else is EXIT_INCOMPLETE. write_failures = [] auth_failures = [] incomplete_reasons = [] reconciliation_failures = [] for w in plan_writes: status, body = write_point(args.key, args.stat, w['value'], w['t'], w['ikey']) body_text = body.decode('utf-8', errors='replace') if isinstance(body, bytes) else str(body) ok = (status == 200 and not looks_like_error_body(body_text)) if not ok: any_write_failed = True write_failures.append(f'WRITE slot{w["slot"]}: HTTP {status or "network"}') write_log.append({ 'slot': f'slot{w["slot"]}', 'value': w['value'], 't': w['t'], 'ikey': w['ikey'], 'status': status, 'ok': ok, 'body': body_text[:160], }) say(f' slot{w["slot"]} value={w["value"]} t={w["t"]} ' f'ikey={w["ikey"][:12]}.. -> HTTP {status} {"OK" if ok else "FAIL"}') if not ok: say(f' body: {body_text[:200]}') say('') say(f'=== RETRYING slot{RETRY_INDEX+1} WITH THE SAME IDEMPOTENCY KEY (must be deduped) ===') status, body = write_point(args.key, args.stat, retry['value'], retry['t'], retry['ikey']) body_text = body.decode('utf-8', errors='replace') if isinstance(body, bytes) else str(body) ok = (status == 200 and not looks_like_error_body(body_text)) if not ok: any_write_failed = True write_failures.append(f'WRITE retry-slot{RETRY_INDEX+1}: HTTP {status or "network"}') write_log.append({ 'slot': f'RETRY-slot{RETRY_INDEX+1}', 'value': retry['value'], 't': retry['t'], 'ikey': retry['ikey'], 'status': status, 'ok': ok, 'body': body_text[:160], }) say(f' retry ikey={retry["ikey"][:12]}.. -> HTTP {status} {"OK" if ok else "FAIL"}') if not ok: say(f' body: {body_text[:200]}') say('') say('=== WAITING 6s FOR WRITES TO SETTLE ===') time.sleep(6) # --- THREE-SURFACE COMPARISON --- say('') say(f'=== THREE-SURFACE COMPARISON (ground truth: sum={EXPECTED_SUM} count={EXPECTED_COUNT}) ===') surfaces = { 'ground_truth': {'sum': EXPECTED_SUM, 'count': EXPECTED_COUNT, 'verdict': 'PASS', 'note': 'by construction — verifiable on paper'}, } # Surface 1/3: ground truth (printed last so it appears next to the others) say('') say(' surface sum count expected verdict') say(' -------------------- ------- ------ ------------ --------') # Surface 2/3: CSV export status, body = fetch_csv(args.key, args.stat) csv_text = body.decode('utf-8', errors='replace') if isinstance(body, bytes) else str(body) if status is None: say(f' CSV export ? ? 5, 3 FAIL (network)') incomplete_reasons.append('CSV export: network failure') surfaces['csv_export'] = {'verdict': 'FAIL', 'reason': 'network'} elif status == 401: # AUTH/ACCESS: we were refused entry. Our arithmetic was NOT tested. say(f' CSV export ? ? 5, 3 FAIL (HTTP 401 auth/access)') auth_failures.append(f'CSV export: HTTP 401 (auth/access refused)') surfaces['csv_export'] = {'verdict': 'FAIL', 'http': 401, 'reason': 'auth-refused', 'body': csv_text[:200]} elif status != 200 or looks_like_error_body(csv_text): # Non-401 HTTP error or 200 with error body -> could not complete (non-auth reason). say(f' CSV export ? ? 5, 3 FAIL (HTTP {status})') incomplete_reasons.append(f'CSV export: HTTP {status}') surfaces['csv_export'] = {'verdict': 'FAIL', 'http': status, 'body': csv_text[:200]} elif '# TRUNCATED' in csv_text: # The export hit its per-request cap: the rows we hold are a PREFIX, so any # sum/count comparison would be arithmetic over incomplete data. Say so — # never render a partial read as a verified number. say(f' CSV export ? ? 5, 3 FAIL (export truncated — partial data)') incomplete_reasons.append('CSV export: response carries the # TRUNCATED marker; comparison not run on partial data') surfaces['csv_export'] = {'verdict': 'FAIL', 'reason': 'truncated'} else: vals, n = parse_csv(csv_text) got_sum, got_count = sum(vals), n # Two reconciliation-failure modes we name explicitly: # 1. retry was not deduped -> sum=6 count=4 (sum=EXPECTED+VALUES[RETRY], count=EXPECTED+1) # 2. aggregation is wrong -> sum or count doesn't match EXPECTED if n == EXPECTED_COUNT and abs(got_sum - EXPECTED_SUM) < 1e-9: say(f' CSV export {got_sum:<7g} {got_count:<6} 5, 3 PASS') surfaces['csv_export'] = {'verdict': 'PASS', 'sum': got_sum, 'count': got_count} elif n == EXPECTED_COUNT + 1 and abs(got_sum - (EXPECTED_SUM + VALUES[RETRY_INDEX])) < 1e-9: say(f' CSV export {got_sum:<7g} {got_count:<6} 5, 3 FAIL ' f'(retry double-counted: got sum={got_sum:g} count={got_count})') reconciliation_failures.append( f'CSV export: retry double-counted (sum={got_sum:g}, count={got_count})') surfaces['csv_export'] = {'verdict': 'FAIL', 'sum': got_sum, 'count': got_count, 'reason': 'retry-double-counted'} elif n == 0 and got_sum == 0: # 200 OK but no data — stat doesn't exist or queries landed outside the window. # That is "no data", not a disagreement. say(f' CSV export ? ? 5, 3 FAIL (no data)') incomplete_reasons.append('CSV export: no data (stat empty or window mismatch)') surfaces['csv_export'] = {'verdict': 'FAIL', 'reason': 'no-data'} else: say(f' CSV export {got_sum:<7g} {got_count:<6} 5, 3 FAIL ' f'(got sum={got_sum:g} count={got_count})') reconciliation_failures.append( f'CSV export: sum={got_sum:g}, count={got_count} ' f'(expected {EXPECTED_SUM}, {EXPECTED_COUNT})') surfaces['csv_export'] = {'verdict': 'FAIL', 'sum': got_sum, 'count': got_count, 'expected_sum': EXPECTED_SUM, 'expected_count': EXPECTED_COUNT, 'note': 'if the stat had prior writes, delete it and re-run'} # Surface 3/3: API reported value status, body = fetch_reported(args.key, args.stat) api_text = body.decode('utf-8', errors='replace') if isinstance(body, bytes) else str(body) if status is None: say(f' API reported value ? ? 5, 3 FAIL (network)') incomplete_reasons.append('API reported: network failure') surfaces['api_reported'] = {'verdict': 'FAIL', 'reason': 'network'} elif status == 401: # AUTH/ACCESS: we were refused entry. Our arithmetic was NOT tested. say(f' API reported value ? ? 5, 3 FAIL (HTTP 401 auth/access)') auth_failures.append(f'API reported: HTTP 401 (auth/access refused)') surfaces['api_reported'] = {'verdict': 'FAIL', 'http': 401, 'reason': 'auth-refused', 'body': api_text[:200]} elif status != 200 or looks_like_error_body(api_text): say(f' API reported value ? ? 5, 3 FAIL (HTTP {status})') incomplete_reasons.append(f'API reported: HTTP {status}') surfaces['api_reported'] = {'verdict': 'FAIL', 'http': status, 'body': api_text[:200]} else: s, c = parse_reported(api_text) if s is None and c is None: # 200 OK but the body is not parseable as a sum/count report — no data to compare. say(f' API reported value ? ? 5, 3 FAIL (unparseable body)') incomplete_reasons.append(f'API reported: unparseable body (no sum/count): {api_text[:200]}') surfaces['api_reported'] = {'verdict': 'FAIL', 'reason': 'unparseable', 'raw': api_text[:500]} else: try: s_f = float(s) if s is not None else None c_i = int(c) if c is not None else None except (TypeError, ValueError): s_f, c_i = None, None if s_f == 0 and c_i == 0: # 200 OK, parseable, but the stat reports zero — "no data", not a disagreement. say(f' API reported value 0 0 5, 3 FAIL (no data)') incomplete_reasons.append('API reported: no data (sum=0 count=0)') surfaces['api_reported'] = {'verdict': 'FAIL', 'reason': 'no-data', 'sum': s_f, 'count': c_i} else: ok = (s_f is not None and abs(s_f - EXPECTED_SUM) < 1e-9) and (c_i == EXPECTED_COUNT) if ok: say(f' API reported value {s_f:<7g} {c_i:<6} 5, 3 PASS') surfaces['api_reported'] = {'verdict': 'PASS', 'sum': s_f, 'count': c_i} else: say(f' API reported value {str(s_f):<7s} {str(c_i):<6} 5, 3 FAIL') reconciliation_failures.append( f'API reported: sum={s_f}, count={c_i} ' f'(expected {EXPECTED_SUM}, {EXPECTED_COUNT})') surfaces['api_reported'] = {'verdict': 'FAIL', 'sum': s_f, 'count': c_i, 'expected_sum': EXPECTED_SUM, 'expected_count': EXPECTED_COUNT} # Surface 1/3 printed last for visual grouping with the other two say(f' ground truth {EXPECTED_SUM:<7g} {EXPECTED_COUNT:<6} 5, 3 PASS (by construction)') # --- VERDICT --- say('') say('=== VERDICT ===') # Exit-code logic. The contract is: # 0 = every surface reconciles with ground truth # 1 = at least one surface returned values that DISAGREE with ground truth # 2 = the run could not complete (missing key, network, auth, no data, write failure) # IMPORTANT: a 401 on a read is "could not complete" (exit 2), NOT "reconciliation # failed" (exit 1). 401 means we were refused access — our arithmetic was never # tested. Reporting that as exit 1 would tell the customer we computed wrongly when # we merely could not read. if (any_write_failed or write_failures or auth_failures or incomplete_reasons): exit_code = EXIT_INCOMPLETE elif reconciliation_failures: exit_code = EXIT_DISAGREEMENT else: exit_code = EXIT_OK if exit_code != EXIT_OK: # Name each category precisely. The ordering matters: AUTH/ACCESS first so the # reader knows "we never read the values" before seeing other failures. if auth_failures: say('') say(' AUTH/ACCESS FAILURE — EzStat refused access to one or more read surfaces.') say(' EzStat\'s arithmetic was NOT tested on this run: we never received') say(' the values. A 401 is not a disagreement; it is "could not read".') for f in auth_failures: say(f' {f}') if write_failures: say('') say(' WRITE FAILURE — EzStat did not accept our writes:') for f in write_failures: say(f' {f}') if incomplete_reasons: say('') say(' COULD NOT COMPLETE — at least one surface did not return values we could') say(' compare (network, no data, unparseable body, or a non-401 HTTP error):') for r in incomplete_reasons: say(f' {r}') if reconciliation_failures: say('') say(' RECONCILIATION FAILED — a surface returned 200 OK with values that') say(' disagree with ground truth:') for f in reconciliation_failures: say(f' {f}') say('') say(' If the stat was not fresh (prior writes exist), the test is inconclusive —') say(' delete it and re-run:') say(f' dashboard -> "verify" folder -> "{args.stat}" -> delete') return done(exit_code, dry_run=False, plan=plan_obj, writes=write_log, surfaces=surfaces, auth_failures=auth_failures, write_failures=write_failures, incomplete_reasons=incomplete_reasons, reconciliation_failures=reconciliation_failures, any_write_failed=any_write_failed) say(' every surface reconciles with ground truth.') say(' the retry was deduplicated (sum=5 count=3, NOT sum=6 count=4).') say('') say('=== CLEANUP ===') say(f' This tool wrote ONLY to the stat "{args.stat}", nothing else.') if args.cleanup: # DELETE via the same authenticated API; report the outcome honestly either way. del_url = f'{BASE_URL}/api/v1/stats/{urllib.parse.quote(args.stat, safe=".")}' req = urllib.request.Request(del_url, method='DELETE', headers={'Authorization': f'Bearer {args.key}', 'User-Agent': UA}) try: with urllib.request.urlopen(req, timeout=20) as r: dstatus = r.status except urllib.error.HTTPError as e: dstatus = e.code except urllib.error.URLError: dstatus = None if dstatus == 200: say(f' --cleanup: deleted "{args.stat}" (HTTP 200).') else: say(f' --cleanup: DELETE failed (HTTP {dstatus}) — the stat remains; delete it') say(' from the dashboard. (The verification verdict above is unaffected.)') else: say(' To delete it: open the EzStat dashboard, navigate to the "verify" folder,') say(f' find "{args.stat}", and delete it. This tool does NOT delete it — that is') say(' yours to do. (CI/harness accounts: pass --cleanup.)') return done(EXIT_OK, dry_run=False, plan=plan_obj, writes=write_log, surfaces=surfaces, auth_failures=[], write_failures=[], incomplete_reasons=[], reconciliation_failures=[], any_write_failed=False) if __name__ == '__main__': try: sys.exit(main()) except KeyboardInterrupt: sys.stderr.write('\ninterrupted\n') _print_boundary(sys.stderr) sys.exit(EXIT_INCOMPLETE)