import copy import json from pathlib import Path import subprocess import sys import tempfile import unittest import odds_compare as tool HERE = Path(__file__).resolve().parent class ComparisonTests(unittest.TestCase): def setUp(self): self.doc = json.loads((HERE / "fixtures.json").read_text()) self.doc["cases"] = [self.doc["cases"][0]] self.right = self.doc["cases"][0]["right"] def result(self): return tool.evaluate(self.doc)["results"][0] def held(self, code): result = self.result() self.assertFalse(result["eligible_for_comparison"]) self.assertIn(code, result["reason_codes"]) def test_same_contract_with_different_prices_is_eligible(self): case = self.doc["cases"][0] self.assertNotEqual(case["left"]["odds"], case["right"]["odds"]) self.assertTrue(self.result()["eligible_for_comparison"]) def test_selection_order_does_not_change_market_identity(self): self.right["selections"].reverse() self.assertTrue(self.result()["eligible_for_comparison"]) def test_exact_age_and_capture_skew_boundaries_are_inclusive(self): self.right["captured_at"] = "2026-09-12T11:59:57Z" self.right["price_updated_at"] = "2026-09-12T11:59:30Z" self.assertTrue(self.result()["eligible_for_comparison"]) self.right["price_updated_at"] = "2026-09-12T11:59:29.999999Z" self.held("right:PRICE_UPDATED_AT_STALE_BY_POLICY") def test_fresh_capture_does_not_refresh_price_timestamp(self): self.right["captured_at"] = "2026-09-12T12:00:00Z" self.right["price_updated_at"] = "2026-09-12T11:59:20Z" self.held("right:PRICE_UPDATED_AT_STALE_BY_POLICY") def test_old_capture_is_held_even_with_other_valid_data(self): self.right["captured_at"] = "2026-09-12T11:59:20Z" self.right["price_updated_at"] = "2026-09-12T11:59:15Z" self.held("right:CAPTURED_AT_STALE_BY_POLICY") def test_capture_skew_exceeds_policy(self): self.right["captured_at"] = "2026-09-12T11:59:56Z" self.held("pair:CAPTURE_SKEW_EXCEEDS_POLICY") def test_identity_dimensions_are_not_interchangeable(self): changes = [("event_namespace", "other-mapper-v1", "EVENT_NAMESPACE"), ("event_id", "fixture-other-match", "EVENT"), ("period", "FIRST_HALF", "PERIOD"), ("market", "TEAM_TOTAL_GOALS", "MARKET"), ("line", 3.5, "LINE"), ("settlement_rules", "extra-time-included", "SETTLEMENT_RULES"), ("in_play", True, "PHASE")] for field, value, code in changes: with self.subTest(field=field): original = self.right[field] self.right[field] = value self.held("pair:" + code + "_MISMATCH") self.right[field] = original def test_matching_unknown_rules_are_not_eligible(self): for side in ("left", "right"): self.doc["cases"][0][side]["settlement_rules"] = "unknown" self.held("right:SETTLEMENT_RULES_UNKNOWN") def test_missing_unknown_suspended_and_closed_states_hold(self): for status, code in [(None, "STATUS_UNKNOWN"), ("UNKNOWN", "STATUS_UNKNOWN"), ("SUSPENDED", "STATE_SUSPENDED"), ("CLOSED", "STATE_CLOSED")]: with self.subTest(status=status): self.right["status"] = status self.held("right:" + code) del self.right["status"] self.held("right:STATUS_UNKNOWN") def test_raw_delta_and_incomplete_selection_are_held(self): self.right["complete_snapshot"] = False self.held("right:INCOMPLETE_SNAPSHOT") self.right["complete_snapshot"] = True del self.right["odds"]["UNDER"] self.held("right:ODDS_INCOMPLETE") def test_duplicate_or_mismatched_selections_are_held(self): self.right["selections"] = ["OVER", "OVER"] self.held("right:SELECTIONS_INVALID") self.held("pair:SELECTIONS_MISMATCH") def test_bad_decimal_odds_are_rejected_including_boolean_and_nan(self): for price in [True, False, 1, 0, -2, float("nan"), float("inf"), "1.91", None, 10**400]: with self.subTest(price=repr(price)): self.right["odds"]["OVER"] = price self.held("right:DECIMAL_ODDS_INVALID") def test_numeric_flags_are_not_normalized_booleans(self): self.right["in_play"] = 0 self.held("right:PHASE_UNKNOWN") self.right["complete_snapshot"] = 1 self.held("right:INCOMPLETE_SNAPSHOT") def test_missing_naive_non_utc_and_invalid_datetimes(self): for field in ("captured_at", "price_updated_at"): original = self.right[field] for stamp, suffix in [(None, "MISSING"), ("2026-09-12T11:59:59", "NAIVE"), ("2026-09-12T12:59:59+01:00", "NOT_UTC"), ("bad", "INVALID")]: with self.subTest(field=field, stamp=stamp): self.right[field] = stamp self.held("right:" + field.upper() + "_" + suffix) self.right[field] = original def test_future_datetimes_are_held(self): for field in ("captured_at", "price_updated_at"): original = self.right[field] self.right[field] = "2026-09-12T12:00:01Z" self.held("right:" + field.upper() + "_FUTURE") self.right[field] = original def test_price_update_after_capture_is_inconsistent(self): self.right["price_updated_at"] = "2026-09-12T11:59:59Z" self.held("right:PRICE_AFTER_CAPTURE") def test_malformed_observation_and_undeclared_fields_hold(self): self.right["heartbeat_at"] = "2026-09-12T12:00:00Z" self.held("right:UNEXPECTED_FIELDS") self.doc["cases"][0]["right"] = [] self.held("right:OBSERVATION_INVALID") def test_invalid_policy_is_not_silently_defaulted(self): for value in [True, -1, float("nan"), 86401, "30"]: with self.subTest(value=repr(value)): self.doc["policy"]["max_age_seconds"] = value with self.assertRaisesRegex(ValueError, "POLICY_INVALID"): tool.evaluate(self.doc) def test_cases_and_duplicate_identifiers_are_bounded(self): self.doc["cases"] *= 33 with self.assertRaisesRegex(ValueError, "CASES_INVALID"): tool.evaluate(self.doc) self.doc["cases"] = self.doc["cases"][:2] with self.assertRaisesRegex(ValueError, "CASE_ID_INVALID"): tool.evaluate(self.doc) def cli(self, raw): with tempfile.TemporaryDirectory() as folder: path = Path(folder) / "input.json" path.write_bytes(raw) return subprocess.run([sys.executable, str(HERE / "odds_compare.py"), str(path)], capture_output=True, text=True) def test_exact_fixture_cli_has_expected_holds(self): proc = self.cli((HERE / "fixtures.json").read_bytes()) self.assertEqual(proc.returncode, 0, proc.stderr) result = json.loads(proc.stdout) expected = {"different_prices": None, "different_line": "pair:LINE_MISMATCH", "different_rules": "pair:SETTLEMENT_RULES_MISMATCH", "different_period": "pair:PERIOD_MISMATCH", "different_phase": "pair:PHASE_MISMATCH", "suspended": "right:STATE_SUSPENDED", "heartbeat_only": "right:PRICE_UPDATED_AT_STALE_BY_POLICY", "raw_delta": "right:INCOMPLETE_SNAPSHOT", "missing_status": "right:STATUS_UNKNOWN", "naive_price_time": "right:PRICE_UPDATED_AT_NAIVE", "future_capture": "right:CAPTURED_AT_FUTURE", "capture_skew": "pair:CAPTURE_SKEW_EXCEEDS_POLICY", "missing_price_time": "right:PRICE_UPDATED_AT_MISSING", "boolean_odds": "right:DECIMAL_ODDS_INVALID"} self.assertEqual({r["id"] for r in result["results"]}, set(expected)) for row in result["results"]: code = expected[row["id"]] self.assertEqual(row["eligible_for_comparison"], code is None) if code: self.assertIn(code, row["reason_codes"]) self.assertEqual((result["eligible_count"], result["held_count"]), (1, 13)) def test_cli_rejects_oversize_duplicate_keys_and_nonfinite_json(self): for raw, code in [(b" " * (tool.MAX_BYTES + 1), "INPUT_TOO_LARGE"), (b'{"now":1,"now":2}', "DUPLICATE_JSON_KEY"), (b'{"now":NaN}', "NONFINITE_JSON_NUMBER"), (b'not json', "INPUT_INVALID")]: with self.subTest(code=code): proc = self.cli(raw) self.assertEqual(proc.returncode, 2) self.assertEqual(proc.stdout, "") self.assertEqual(json.loads(proc.stderr), {"error": code}) if __name__ == "__main__": unittest.main()