#!/usr/bin/env python3 """Check an experiment record offline. Never connects to Shopify or a proxy.""" import argparse import csv import json import re import sys from datetime import datetime, timezone from pathlib import Path from urllib.parse import urlsplit INPUT_KEYS = {"entry_url_ref", "exit_ref", "browser_language", "session_state", "shipping_fixture_ref"} CHANGES = {"ip": "exit_ref", "url": "entry_url_ref", "language": "browser_language", "saved": "session_state", "province": "shipping_fixture_ref", "cross_border": "shipping_fixture_ref"} COMMON_KEYS = ("product_cart_fixture", "browser_version", "viewport", "timezone", "login_state", "browser_geolocation") CONFIG_TEXT = ("snapshot_id", "active_markets_and_currencies", "domain_strategy", "published_languages_by_market", "selector_and_saved_choice_setup", "shipping_setup_and_rules", "third_party_redirects_or_none", "configuration_evidence_ref", "test_window") + COMMON_KEYS def filled(value): return isinstance(value, str) and bool(value.strip()) def utc_time(value): if not filled(value): return False try: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed.tzinfo is not None and parsed.utcoffset() == timezone.utc.utcoffset(parsed) except ValueError: return False def safe_url(value): if not filled(value): return False try: parts = urlsplit(value) return parts.scheme in {"http", "https"} and bool(parts.hostname) and parts.username is None and parts.password is None except ValueError: return False def expected_actual_inputs(inputs, config, exits): return { "entry_url": config.get(inputs.get("entry_url_ref")), "exit_country": exits.get(inputs.get("exit_ref"), {}).get("observed_country"), "egress_id": exits.get(inputs.get("exit_ref"), {}).get("egress_id"), "browser_language": inputs.get("browser_language"), "session_state": inputs.get("session_state"), "shipping_fixture_ref": inputs.get("shipping_fixture_ref"), "configuration_snapshot_id": config.get("snapshot_id"), **{key: config.get(key) for key in COMMON_KEYS}, } def validate(record, observations=False): errors = [] def require(condition, message): if not condition: errors.append(message) if not isinstance(record, dict): return ["record must be a JSON object"], 0 require(record.get("schema_version") == 1, "schema_version must be 1") config = record.get("configuration") exits = record.get("exits") fixtures = record.get("address_fixtures") pairs = record.get("pairs") if not all(isinstance(x, dict) for x in (config, exits, fixtures)) or not isinstance(pairs, list): return errors + ["configuration, exits and address_fixtures must be objects; pairs must be an array"], 0 require(config.get("storefront_kind") == "Shopify Online Store", "scope is Shopify Online Store; headless/custom storefronts need their own protocol") require(config.get("login_state") == "signed-out", "configuration.login_state must be signed-out for this protocol") require(config.get("browser_geolocation") == "denied", "configuration.browser_geolocation must be denied; granted coordinates add another location input") for key in CONFIG_TEXT: require(filled(config.get(key)), "configuration." + key + " is missing") require(utc_time(config.get("captured_at_utc")), "configuration.captured_at_utc must be an ISO timestamp with UTC offset") for key in ("country_redirection_enabled", "language_redirection_enabled"): require(type(config.get(key)) is bool, "configuration." + key + " must be true or false") require(config.get("domain_strategy") in {"shared", "dedicated", "mixed"}, "configuration.domain_strategy must be shared, dedicated or mixed") ids = [p.get("id") if isinstance(p, dict) else None for p in pairs] require(len(ids) == len(CHANGES) and set(str(x) for x in ids) == set(CHANGES), "retain each of the six comparison IDs exactly once; mark unsuitable comparisons inapplicable") active = 0 required_exits = set() for pair in pairs: if not isinstance(pair, dict): errors.append("each comparison must be an object") continue pid = str(pair.get("id")) applicable = pair.get("applicable") require(type(applicable) is bool, pid + ": choose applicable true or false") if applicable is False: require(filled(pair.get("inapplicable_reason")), pid + ": explain why it is inapplicable") continue if applicable is not True: continue active += 1 for key in ("expected_rule", "rule_evidence_ref"): require(filled(pair.get(key)), pid + ": " + key + " is missing") require(pair.get("changed_input") == CHANGES.get(pid), pid + ": changed_input does not match this comparison") visits = pair.get("visits") if not isinstance(visits, list) or len(visits) != 2 or not all(isinstance(v, dict) for v in visits): errors.append(pid + ": exactly two visit objects are required") continue inputs = [v.get("inputs") for v in visits] if not all(isinstance(i, dict) and set(i) == INPUT_KEYS for i in inputs): errors.append(pid + ": each visit must declare exactly the five controlled inputs") continue diff = {key for key in INPUT_KEYS if inputs[0][key] != inputs[1][key]} require(diff == {CHANGES.get(pid)}, pid + ": change only " + str(CHANGES.get(pid)) + "; differences are " + ", ".join(sorted(diff))) if pid == "ip": require(all(i["entry_url_ref"] == "neutral_entry_url" for i in inputs), "ip: both visits must use the same neutral entry URL") require([i["exit_ref"] for i in inputs] == ["CA", "control"], "ip: compare CA with the non-CA control") require(all(i["shipping_fixture_ref"] == "none" for i in inputs), "ip: observe first-visit localization before any shipping address is entered") if pid == "url": require([i["entry_url_ref"] for i in inputs] == ["neutral_entry_url", "canada_entry_url"], "url: compare neutral and configured Canada entries") require(config.get("neutral_entry_url") != config.get("canada_entry_url"), "url: URLs are identical; this is not a URL contrast, mark inapplicable if appropriate") if pid == "language": require([i["browser_language"] for i in inputs] == ["en-CA", "fr-CA"], "language: compare en-CA and fr-CA preferences") if pid == "saved": require([i["session_state"] for i in inputs] == ["fresh", "manual-choice-retained"], "saved: compare a fresh context with the deliberately prepared retained choice") if pid == "province": require([i["shipping_fixture_ref"] for i in inputs] == ["ON", "QC"], "province: compare approved Ontario and Quebec delivery fixtures") if pid == "cross_border": require([i["shipping_fixture_ref"] for i in inputs] == ["control", "ON"], "cross_border: compare the control-country and Ontario delivery fixtures") require(all(i["entry_url_ref"] == "control_market_entry_url" and i["exit_ref"] == "control" for i in inputs), "cross_border: keep the control-market entry and control exit fixed") for number, (visit, inp) in enumerate(zip(visits, inputs), start=1): label = pid + " visit " + str(number) require(visit.get("id") == pid + ("-A" if number == 1 else "-B"), label + ": retain the unique visit ID") require(filled(visit.get("expected_outcome")), label + ": write expected_outcome before execution") url_key = inp["entry_url_ref"] require(url_key in {"neutral_entry_url", "canada_entry_url", "control_market_entry_url"} and safe_url(config.get(url_key)), label + ": entry URL must be a configured HTTP(S) URL without embedded credentials") exit_key = inp["exit_ref"] require(exit_key in {"CA", "control"}, label + ": exit_ref must be CA or control") required_exits.add(str(exit_key)) require(filled(inp["browser_language"]), label + ": browser language is missing") if pid != "saved": require(inp["session_state"] == "fresh", label + ": fresh context required") fixture_key = inp["shipping_fixture_ref"] if fixture_key != "none": fixture = fixtures.get(fixture_key) if not isinstance(fixture, dict): errors.append(label + ": unknown shipping fixture") else: require(filled(fixture.get("private_fixture_ref")), label + ": approved full address fixture reference is missing") require(filled(fixture.get("country")) and filled(fixture.get("region")), label + ": shipping fixture country/region is missing") if fixture_key in {"ON", "QC"}: require(fixture.get("country") == "CA" and fixture.get("region") == fixture_key, label + ": fixture must match the named Canadian province") if fixture_key == "control": require(fixture.get("country") != "CA", label + ": control fixture must be outside Canada") if not observations: continue obs = visit.get("observation") if not isinstance(obs, dict): errors.append(label + ": observation must be an object") continue require(obs.get("status") == "observed", label + ": observation status is not observed") require(utc_time(obs.get("executed_at_utc")), label + ": observation UTC timestamp is missing") for key in ("final_url", "selected_country", "market", "market_evidence", "language", "currency", "evidence_ref"): require(filled(obs.get(key)), label + ": observed " + key + " is missing; use inconclusive notes if the measurement cannot be obtained") require(safe_url(obs.get("final_url")), label + ": observed final_url must be HTTP(S) without embedded credentials") require(obs.get("outcome") in {"match", "mismatch", "inconclusive"}, label + ": outcome must be match, mismatch or inconclusive") if any(obs.get(key) == "unavailable" for key in ("selected_country", "market", "market_evidence", "language", "currency")): require(obs.get("outcome") == "inconclusive", label + ": an unavailable required measurement must be inconclusive") if obs.get("outcome") == "inconclusive": require(filled(obs.get("notes")), label + ": explain the inconclusive measurement") if fixture_key != "none": require(filled(obs.get("delivery_options")), label + ": record delivery options, including an explicit none if absent") actual = obs.get("actual_inputs") expected = expected_actual_inputs(inp, config, exits) if not isinstance(actual, dict): errors.append(label + ": actual_inputs are missing") else: for key, value in expected.items(): require(actual.get(key) == value and value is not None, label + ": actual input " + key + " differs from the controlled plan or is missing") require(active > 0, "no applicable comparison has been selected") for key in sorted(required_exits): item = exits.get(key) if not isinstance(item, dict): errors.append("exits." + key + " must be an object") continue for field in ("intended_country", "observed_country"): require(bool(re.fullmatch(r"[A-Z]{2}", str(item.get(field, "")))), "exits." + key + "." + field + " must be a two-letter country code") require(item.get("intended_country") == item.get("observed_country"), "exits." + key + ": requested country and observed country differ") require(item.get("observed_country") == "CA" if key == "CA" else item.get("observed_country") != "CA", "exits." + key + ": CA/control country assignment is invalid") for field in ("egress_id", "target_or_check_service", "ip_evidence_ref"): require(filled(item.get(field)), "exits." + key + "." + field + " is missing") require(utc_time(item.get("checked_at_utc")), "exits." + key + ".checked_at_utc must be an ISO UTC timestamp") if {"CA", "control"} <= required_exits and all(isinstance(exits.get(k), dict) for k in ("CA", "control")): require(exits["CA"].get("egress_id") != exits["control"].get("egress_id"), "CA and control must have different observed egress IDs") return errors, active def export_csv(record, path): """Export a visit view. Keep the configuration JSON/CSV alongside it.""" fields = ["comparison", "visit", "applicable", "inapplicable_reason", "snapshot_id", "changed_input", "entry_url_ref", "entry_url", "exit_ref", "browser_language", "session_state", "shipping_fixture_ref", "expected_rule", "expected_outcome", "observation_status", "observed_at_utc", "actual_inputs_json", "final_url", "selected_country", "market", "market_evidence", "language", "currency", "delivery_options", "outcome", "evidence_ref", "notes"] with Path(path).open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fields) writer.writeheader() for pair in record["pairs"]: for visit in pair["visits"]: inp, obs = visit["inputs"], visit["observation"] writer.writerow({"comparison": pair["id"], "visit": visit["id"], "applicable": pair["applicable"], "inapplicable_reason": pair["inapplicable_reason"], "snapshot_id": record["configuration"]["snapshot_id"], "changed_input": pair["changed_input"], "entry_url_ref": inp["entry_url_ref"], "entry_url": record["configuration"].get(inp["entry_url_ref"]), "exit_ref": inp["exit_ref"], "browser_language": inp["browser_language"], "session_state": inp["session_state"], "shipping_fixture_ref": inp["shipping_fixture_ref"], "expected_rule": pair["expected_rule"], "expected_outcome": visit["expected_outcome"], "observation_status": obs["status"], "observed_at_utc": obs["executed_at_utc"], "actual_inputs_json": json.dumps(obs["actual_inputs"]) if obs.get("actual_inputs") is not None else "", **{key: obs.get(key) for key in ("final_url", "selected_country", "market", "market_evidence", "language", "currency", "delivery_options", "outcome", "evidence_ref", "notes")}}) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("worksheet", type=Path) parser.add_argument("--observations", action="store_true", help="also require observation records and check their declared actual inputs") parser.add_argument("--export-csv", type=Path, help="write a review view; keep the JSON configuration alongside it") args = parser.parse_args() try: record = json.loads(args.worksheet.read_text(encoding="utf-8")) errors, count = validate(record, args.observations) if args.export_csv: export_csv(record, args.export_csv) except (OSError, ValueError, TypeError, KeyError, AttributeError) as exc: print(json.dumps({"result": "INVALID_INPUT", "error": str(exc)})) return 2 result = "INCOMPLETE_RECORD" if errors else ("RECORD_COMPLETE" if args.observations else "PLAN_COMPLETE") print(json.dumps({"result": result, "applicable_comparisons": count, "errors": errors, "meaning": "Offline completeness and declared-control checks only. Does not verify Shopify behavior, address eligibility, IP geolocation or truth of observations."}, indent=2)) return 2 if errors else 0 if __name__ == "__main__": sys.exit(main())