#!/usr/bin/env python3 """Run a small, synthetic WOK inventory contract exercise without an API key. Save this file beside inventory-contract-fixtures.json, then run: python3 inventory-contract-check.py An optional filename loads your edited copy of that exercise instead. Only that local JSON file is read. There are no network calls, source imports, credentials, retries or writes. This is a teaching subset, not a JSON Schema validator or proof of current ownership, market availability or sale proceeds. It covers full grouped USD responses (top=0), public-only, without pagination, search, trade-view enrichment, currency conversion or summary-only batches. Native status, compatibility HTTP status and each batch pair are independent. """ from __future__ import annotations import json import math import sys from decimal import Decimal, ROUND_HALF_UP from pathlib import Path MAX_BYTES = 131072 MAX_CASES = 32 UNAVAILABLE = {"private", "notfound", "limited", "error"} PAIR_STATUS = {"ok": 200, "empty": 200, "private": 403, "notfound": 404, "limited": 429, "error": 502} def require(condition, code): if not condition: raise ValueError(code) def integer(value): return type(value) is int and value >= 0 def money(value): require(type(value) in (int, float) and math.isfinite(value) and value >= 0, "invalid_money") return Decimal(str(value)) def dollars(value): return float(value.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)) def rows_summary(rows): require(isinstance(rows, list) and len(rows) <= 1000, "invalid_items") units = unpriced = eligible = 0 subtotal = Decimal(0) for row in rows: require(isinstance(row, dict), "invalid_item") require(isinstance(row.get("markethashname"), str) and row["markethashname"], "missing_item_name") count = row.get("count") require(integer(count) and count > 0, "invalid_quantity") for flag in ("tradable", "marketable"): require(type(row.get(flag)) in (int, bool) and row[flag] in (0, 1), "invalid_eligibility") units += count sellable = bool(row["tradable"] and row["marketable"]) eligible += count if sellable else 0 price = row.get("pricemedian") if price is None: unpriced += count else: price = money(price) require(price > 0, "zero_is_not_missing_price") require(isinstance(row.get("pricesource"), str) and row["pricesource"], "missing_price_source") observed = row.get("priceupdatedat") require(observed is None or money(observed) > 0, "invalid_price_time") if sellable: subtotal += price * count return {"units": units, "eligible_units": eligible, "unpriced_units": unpriced, "known_subtotal_usd": dollars(subtotal)} def classified(rows, total): summary = rows_summary(rows) require(dollars(money(total)) == summary["known_subtotal_usd"], "subtotal_mismatch") state = ("partial_value" if summary["unpriced_units"] else "valued" if summary["units"] else "empty") return {"state": state, **summary, "inventory_value_usd": None if summary["unpriced_units"] else summary["known_subtotal_usd"]} def native(body): require(isinstance(body, dict) and body.get("currency", "USD") == "USD", "unsupported_native_response") status = body.get("status") require(status in {"ok", "empty"} | UNAVAILABLE, "unsupported_native_status") require(isinstance(body.get("items"), list), "invalid_items") if status in UNAVAILABLE: # The server's numeric default total=0 does not establish a real value. return {"state": "unavailable", "inventory_value_usd": None} summary = classified(body["items"], body.get("total")) for field, expected in (("items_total", summary["units"]), ("sellable", summary["eligible_units"]), ("unpriced", summary["unpriced_units"])): require(integer(body.get(field)) and body[field] == expected, "inventory_count_mismatch") require(status != "empty" or summary["units"] == 0, "empty_with_items") meta = body.get("meta") require(isinstance(meta, dict) and integer(meta.get("cache_age")) and integer(meta.get("cache_ttl")), "missing_inventory_age") return summary def analyze(case): require(isinstance(case, dict), "invalid_case") kind, status, body = case.get("kind"), case.get("http_status"), case.get("body") require(type(status) is int, "invalid_http_status") if kind == "native": require(status == 200, "unsupported_native_http_status") return native(body) if kind == "compatibility": if status != 200: require(status in {403, 404, 410, 429, 502} and isinstance(body, dict) and isinstance(body.get("error"), str), "unsupported_compatibility_error") return {"state": "unavailable", "inventory_value_usd": None} headers = case.get("headers", {}) require(headers.get("X-Wok-Price-Currency") == "USD", "unsupported_currency") try: total = Decimal(headers["X-Wok-Total"]) except (KeyError, TypeError, ArithmeticError): raise ValueError("missing_compatibility_total") from None require(total.is_finite() and total >= 0, "invalid_compatibility_total") return classified(body, float(total)) require(kind == "batch" and status == 200 and isinstance(body, dict), "unsupported_response") pairs, meta = body.get("results"), body.get("meta") require(isinstance(pairs, list) and 1 <= len(pairs) <= 80 and isinstance(meta, dict), "invalid_batch") results, seen = [], set() for pair in pairs: require(isinstance(pair, dict), "invalid_pair") identity = (pair.get("steamid"), pair.get("game")) require(all(isinstance(value, str) and value for value in identity) and identity not in seen, "duplicate_or_missing_pair") seen.add(identity) expected = PAIR_STATUS.get(pair.get("status")) require(expected is not None and type(pair.get("http_status")) is int and pair["http_status"] == expected and type(pair.get("ok")) is bool and pair["ok"] == (expected == 200), "pair_status_mismatch") if not pair["ok"]: require(isinstance(pair.get("error"), dict) and isinstance(pair["error"].get("code"), str) and pair["error"]["code"], "missing_pair_error") results.append(native(pair)) succeeded = sum(pair["ok"] for pair in pairs) for field, expected in (("requested", len(pairs)), ("succeeded", succeeded), ("failed", len(pairs) - succeeded)): require(integer(meta.get(field)) and meta[field] == expected, "batch_count_mismatch") return {"state": "partial_batch" if succeeded < len(pairs) else "complete_batch", "succeeded": succeeded, "failed": len(pairs) - succeeded, "pairs": results} def exercise(document): require(isinstance(document, dict) and document.get("synthetic") is True and type(document.get("version")) is int and document["version"] == 1, "not_this_synthetic_exercise") cases = document.get("cases") require(isinstance(cases, list) and 1 <= len(cases) <= MAX_CASES, "invalid_case_count") results = [] for index, case in enumerate(cases, 1): result = analyze(case) require(result == case.get("expected"), "expected_interpretation_mismatch") # Do not print response content, identifiers, source bodies or local paths. results.append({"case": index, "state": result["state"]}) return {"ok": True, "synthetic": True, "cases": results} def main(argv=None): args = sys.argv[1:] if argv is None else argv if args == ["--help"]: print(__doc__) return 0 if len(args) > 1: print('{"ok":false,"error":"usage_one_local_fixture_file"}') return 1 path = Path(args[0]) if args else Path(__file__).with_name("inventory-contract-fixtures.json") try: with path.open("rb") as handle: raw = handle.read(MAX_BYTES + 1) require(len(raw) <= MAX_BYTES, "fixture_too_large") document = json.loads(raw, parse_constant=lambda _: require(False, "nonfinite_json")) result = exercise(document) except (OSError, ValueError, TypeError, ArithmeticError, RecursionError): print('{"ok":false,"error":"invalid_exercise"}') return 1 print(json.dumps(result, separators=(",", ":"), allow_nan=False)) return 0 if __name__ == "__main__": raise SystemExit(main())