#!/usr/bin/env python3 """Offline curl timing interval analyzer. It never opens a network connection.""" import argparse import json import math import re from pathlib import Path REQUIRED_SCOPE = { "connection": "new", "http_version": "1.1", "method": "GET", "follow_redirects": False, "retries": 0, "upload": False, "allow_1xx": False, } DEMO = { "scope": REQUIRED_SCOPE, "observations": [ {"id": "baseline", "curl_exit": 0, "http_status": "200", "content_valid": True, "connect": .08, "pretransfer": .32, "starttransfer": .62, "total": .82}, {"id": "setup", "curl_exit": 0, "http_status": "200", "content_valid": True, "connect": .08, "pretransfer": 1.52, "starttransfer": 1.82, "total": 2.02}, {"id": "wait", "curl_exit": 0, "http_status": "200", "content_valid": True, "connect": .08, "pretransfer": .32, "starttransfer": 2.12, "total": 2.32}, {"id": "body", "curl_exit": 0, "http_status": "200", "content_valid": True, "connect": .08, "pretransfer": .32, "starttransfer": .62, "total": 2.42}, {"id": "timeout", "curl_exit": 28, "http_status": "000", "content_valid": False, "total": 20.0}, ], } class InputError(ValueError): pass def number(value, name): if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0: raise InputError(f"{name} must be a finite non-negative number") return float(value) def milliseconds(value, name): converted = value * 1000 if not math.isfinite(converted): raise InputError(f"{name} is too large to convert to milliseconds") return round(converted, 3) def validate_scope(scope): if not isinstance(scope, dict) or set(scope) != set(REQUIRED_SCOPE): raise InputError("scope must exactly describe one new HTTP/1.1 GET, without redirects, retries, upload, or 1xx") for key, expected in REQUIRED_SCOPE.items(): value = scope[key] if type(value) is not type(expected) or value != expected: raise InputError("scope must exactly describe one new HTTP/1.1 GET, without redirects, retries, upload, or 1xx") def analyze(record): if not isinstance(record, dict) or not isinstance(record.get("id"), str) or not record["id"]: raise InputError("every observation needs a non-empty string id") exit_code = record.get("curl_exit") if isinstance(exit_code, bool) or not isinstance(exit_code, int) or exit_code < 0: raise InputError(f"{record['id']}: curl_exit must be a non-negative integer") raw_status = record.get("http_status") if not isinstance(raw_status, str) or not re.fullmatch(r"[0-9]{3}", raw_status): raise InputError(f"{record['id']}: http_status must be curl's three-digit response_code string") status = int(raw_status) result = {"id": record["id"], "curl_exit": exit_code, "http_status": status, "content_valid": record.get("content_valid")} if exit_code != 0 or not 200 <= status < 300 or record.get("content_valid") is not True: result["state"] = "failed_or_not_content_valid; raw timings retained without zero-filled intervals" raw = {key: record[key] for key in ("connect", "pretransfer", "starttransfer", "total") if key in record} if raw: result["raw_timing"] = raw return result values = [number(record.get(key), f"{record['id']}.{key}") for key in ("connect", "pretransfer", "starttransfer", "total")] if not values[0] <= values[1] <= values[2] <= values[3]: raise InputError(f"{record['id']}: cumulative times must satisfy connect <= pretransfer <= starttransfer <= total") connect, pretransfer, starttransfer, total = values result.update({ "state": "decomposed", "interval_ms": { "connect": milliseconds(connect, f"{record['id']}.connect"), "post_connect_setup": milliseconds(pretransfer - connect, f"{record['id']}.post_connect_setup"), "wait_first_byte": milliseconds(starttransfer - pretransfer, f"{record['id']}.wait_first_byte"), "after_first_byte": milliseconds(total - starttransfer, f"{record['id']}.after_first_byte"), }, }) return result def run(data): if not isinstance(data, dict): raise InputError("input must be a JSON object") validate_scope(data.get("scope")) observations = data.get("observations") if not isinstance(observations, list) or not observations: raise InputError("observations must be a non-empty list") seen = set() for row in observations: if not isinstance(row, dict) or not isinstance(row.get("id"), str) or not row["id"] or row["id"] in seen: raise InputError("every observation id must be unique") seen.add(row["id"]) return {"scope": "single new HTTP/1.1 GET only; user-declared scope is not traffic verification", "observations": [analyze(row) for row in observations]} def self_test(): output = run(DEMO) assert output["observations"][0]["interval_ms"] == {"connect": 80.0, "post_connect_setup": 240.0, "wait_first_byte": 300.0, "after_first_byte": 200.0} assert output["observations"][-1]["state"].startswith("failed") for mutate in ( lambda d: d["observations"][0].update(connect=-.1), lambda d: d["observations"][0].update(total=float("nan")), lambda d: d["observations"][0].pop("connect"), lambda d: d["observations"][0].update(pretransfer=.01), lambda d: d["scope"].update(connection="reused"), lambda d: d["scope"].update(retries=False), lambda d: d["scope"].update(follow_redirects=True), lambda d: d["observations"].append(d["observations"][0].copy()), ): specimen = json.loads(json.dumps(DEMO)) mutate(specimen) try: run(specimen) except InputError: pass else: raise AssertionError("invalid input was accepted") assert run(DEMO)["observations"][-1]["raw_timing"] == {"total": 20.0} redirect = json.loads(json.dumps(DEMO)) redirect["observations"][0]["http_status"] = "302" assert run(redirect)["observations"][0]["state"].startswith("failed") print("PASS: intervals, negative/NaN/missing values, non-monotonic times, strict scope types, reused connection, redirect, duplicate ID and retained failure handling") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("input", nargs="?", type=Path, help="strict JSON observation file; omit for synthetic demo") parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: self_test() return data = DEMO if args.input is None else json.loads( args.input.read_text(encoding="utf-8"), parse_constant=lambda value: (_ for _ in ()).throw(InputError(f"non-JSON numeric constant: {value}")), ) output = run(data) output["input"] = "synthetic demo only; no proxy, URL, credential, or network test was used" if args.input is None else str(args.input) print(json.dumps(output, indent=2, ensure_ascii=False)) if __name__ == "__main__": main()