import csv import json import sys import unittest from pathlib import Path from unittest.mock import Mock, patch from tempfile import TemporaryDirectory ROOT = Path(__file__).resolve().parents[1] SRC = ROOT / "src" EXPORTS = ROOT / "exports" V7_EXPORTS = EXPORTS / "1" ARC_EXPORTS = EXPORTS / "arc" if str(SRC) not in sys.path: sys.path.insert(0, str(SRC)) from nte_history_exporter import __version__ from nte_history_exporter.decoder.boundary import annotate_groups, make_uid from nte_history_exporter.constants import LIMITED_CHARACTER_MARKER, MARKER from nte_history_exporter.decoder.boundary import select_continuous_run_from_page_1 from nte_history_exporter.decoder.protocol import decode_response_records, history_request_kind from nte_history_exporter.constants import POOL_META from nte_history_exporter.mappings import ARC_META, CHARACTERS, ITEMS, REWARDS_BY_ID from nte_history_exporter.decoder.protocol import decode_reward_key, infer_reward_type from nte_history_exporter.decoder.user_uid import extract_user_uid from nte_history_exporter.export.csv_export import write_csv from nte_history_exporter.decoder.arc import ( arc_request_page, build_arc_rows_from_pairs, decode_arc_key, decode_arc_timestamp, is_arc_history_request, make_arc_uid, parse_arc_response, ) from nte_history_exporter.live_capture.session import LiveHistorySession, UdpPacket from nte_history_exporter.live_capture.libpcap import ( DLT_EN10MB, DLT_LINUX_SLL, DLT_LINUX_SLL2, DLT_LOOP, DLT_RAW, _extract_ipv4_frame, _load_library, LibpcapUnavailable, ) from nte_history_exporter.live_capture.windows_raw import parse_ipv4_packet from nte_history_exporter.live_capture.backends import open_capture_backend from nte_history_exporter.export.json_export import build_export_json from nte_history_exporter.live_capture.runner import export_paths from nte_history_exporter.pool_mappings import load_pool_mappings, pool_meta_from_mapping from nte_history_exporter.update_check import UpdateInfo, check_for_update, is_newer_version def load_reference_csv(name): path = EXPORTS / name if not path.exists(): path = EXPORTS / "old" / name if not path.exists(): raise unittest.SkipTest(f"private reference fixture not present: {path}") rows = [] with path.open(newline="", encoding="utf-8") as f: for row in csv.DictReader(f): normalized = dict(row) for key in ["page", "offset", "row", "dice", "dice_raw_u32", "quantity"]: if normalized.get(key) not in ("", None): normalized[key] = int(normalized[key]) rows.append(normalized) return rows def decode_single_record(record_hex): return decode_response_records(bytes(0x50) + bytes.fromhex(record_hex))[0] def load_v7_row(name, one_based_index): path = V7_EXPORTS / name if not path.exists(): raise unittest.SkipTest(f"private reference fixture not present: {path}") with path.open(newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) return rows[one_based_index - 1] def load_arc_csv(name): path = ARC_EXPORTS / name if not path.exists(): raise unittest.SkipTest(f"private reference fixture not present: {path}") with path.open(newline="", encoding="utf-8") as f: return list(csv.DictReader(f)) class BoundaryExportTests(unittest.TestCase): def test_windows_auto_prefers_npcap(self): capture = Mock(device=r"\Device\NPF_test") with ( patch("nte_history_exporter.live_capture.backends.sys.platform", "win32"), patch( "nte_history_exporter.live_capture.backends.open_libpcap_capture", return_value=capture, ), patch("nte_history_exporter.live_capture.backends.RawSocketCapture") as raw_capture, ): selected = open_capture_backend("192.0.2.1") self.assertIs(selected, capture) self.assertEqual(selected.name, "npcap") self.assertEqual(selected.fallback_reason, "") raw_capture.assert_not_called() def test_windows_auto_falls_back_to_raw_socket_without_npcap(self): fallback = Mock() with ( patch("nte_history_exporter.live_capture.backends.sys.platform", "win32"), patch( "nte_history_exporter.live_capture.backends.open_libpcap_capture", side_effect=LibpcapUnavailable("Npcap could not be loaded"), ), patch( "nte_history_exporter.live_capture.backends.RawSocketCapture", return_value=fallback, ) as raw_capture, ): selected = open_capture_backend("192.0.2.1") self.assertIs(selected, fallback) raw_capture.assert_called_once_with( "192.0.2.1", fallback_reason="Npcap could not be loaded", ) def test_explicit_libpcap_does_not_fall_back(self): with ( patch("nte_history_exporter.live_capture.backends.sys.platform", "win32"), patch( "nte_history_exporter.live_capture.backends.open_libpcap_capture", side_effect=LibpcapUnavailable("Npcap could not be loaded"), ), patch("nte_history_exporter.live_capture.backends.RawSocketCapture") as raw_capture, self.assertRaises(LibpcapUnavailable), ): open_capture_backend("192.0.2.1", "libpcap") raw_capture.assert_not_called() def test_windows_libpcap_load_prefers_system_then_path(self): system_dir = Path("C:/Windows/System32/Npcap") loaded = Mock() with ( patch("nte_history_exporter.live_capture.libpcap.sys.platform", "win32"), patch("nte_history_exporter.live_capture.libpcap._windows_npcap_directory", return_value=system_dir), patch("nte_history_exporter.live_capture.libpcap.Path.is_dir", return_value=True), patch("nte_history_exporter.live_capture.libpcap.os.add_dll_directory", return_value=Mock(), create=True), patch( "nte_history_exporter.live_capture.libpcap.ctypes.CDLL", side_effect=[OSError("system missing"), loaded], ) as cdll, ): self.assertIs(_load_library(), loaded) self.assertEqual( [call.args[0] for call in cdll.call_args_list], [ str(system_dir / "wpcap.dll"), "wpcap.dll", ], ) def test_libpcap_link_layers_extract_ipv4_packets(self): ip_packet = bytes.fromhex("4500001c0000000040110000c0000201c6336402") + bytes(8) ethernet = bytes(12) + bytes.fromhex("0800") + ip_packet loop = (2).to_bytes(4, sys.byteorder) + ip_packet linux_sll = bytes(14) + bytes.fromhex("0800") + ip_packet linux_sll2 = bytes.fromhex("0800") + bytes(18) + ip_packet self.assertEqual(_extract_ipv4_frame(ethernet, DLT_EN10MB), ip_packet) self.assertEqual(_extract_ipv4_frame(ip_packet, DLT_RAW), ip_packet) self.assertEqual(_extract_ipv4_frame(loop, DLT_LOOP), ip_packet) self.assertEqual(_extract_ipv4_frame(linux_sll, DLT_LINUX_SLL), ip_packet) self.assertEqual(_extract_ipv4_frame(linux_sll2, DLT_LINUX_SLL2), ip_packet) def test_pool_mapping_json_files_have_uniform_shape(self): required_top_level = {"pool_key", "game", "system", "banner", "request", "response"} for pool_key, mapping in load_pool_mappings().items(): with self.subTest(pool_key=pool_key): self.assertEqual(set(required_top_level) - set(mapping), set()) self.assertEqual(mapping["pool_key"], pool_key) self.assertIn("id", mapping["system"]) self.assertIn("name", mapping["system"]) self.assertIn("id", mapping["banner"]) self.assertIn("name", mapping["banner"]) self.assertIn("shared_pity", mapping["banner"]) self.assertIn("family", mapping["request"]) self.assertIn("length", mapping["request"]) self.assertIn("constant", mapping["request"]) self.assertIn("cursor_step", mapping["request"]) def test_pool_mapping_json_matches_runtime_pool_meta(self): for pool_key, mapping in load_pool_mappings().items(): with self.subTest(pool_key=pool_key): self.assertEqual(pool_meta_from_mapping(mapping), POOL_META[pool_key]) def test_update_version_comparison_handles_release_tags(self): self.assertTrue(is_newer_version("v0.1.7", "0.1.6")) self.assertTrue(is_newer_version("0.2.0", "0.1.6")) self.assertTrue(is_newer_version("v0.1.10", "0.1.9")) self.assertFalse(is_newer_version("v0.1.6", "0.1.6")) self.assertFalse(is_newer_version("v0.1.5", "0.1.6")) self.assertFalse(is_newer_version("latest", "0.1.6")) def test_update_check_reports_newer_github_release(self): latest = { "tag_name": "v0.1.7", "html_url": "https://github.com/Golumpa/nte-exporter/releases/tag/v0.1.7", } with patch("nte_history_exporter.update_check.fetch_latest_release", return_value=latest): update = check_for_update("0.1.6", timeout=0.1) self.assertEqual( update, UpdateInfo( current_version="0.1.6", latest_version="v0.1.7", release_url="https://github.com/Golumpa/nte-exporter/releases/tag/v0.1.7", ), ) def test_update_check_ignores_prerelease(self): latest = { "tag_name": "v0.1.8-dev-branch.123", "html_url": "https://github.com/Golumpa/nte-exporter/releases/tag/v0.1.8-dev-branch.123", "prerelease": True, } with patch("nte_history_exporter.update_check.fetch_latest_release", return_value=latest): self.assertIsNone(check_for_update("0.1.7", timeout=0.1)) def test_update_check_is_quiet_when_unavailable_or_current(self): with patch("nte_history_exporter.update_check.fetch_latest_release", side_effect=OSError("offline")): self.assertIsNone(check_for_update("0.1.6", timeout=0.1)) with patch("nte_history_exporter.update_check.fetch_latest_release", return_value={"tag_name": "v0.1.6"}): self.assertIsNone(check_for_update("0.1.6", timeout=0.1)) def test_reward_mapping_files_have_expected_shape(self): self.assertTrue(ARC_META) for arc_id, meta in ARC_META.items(): with self.subTest(arc_id=arc_id): self.assertTrue(arc_id.startswith("fork_")) self.assertIn("name", meta) self.assertIn(meta.get("rank"), ("S", "A", "B")) self.assertTrue(CHARACTERS) for character_id, info in CHARACTERS.items(): with self.subTest(character_id=character_id): self.assertTrue(character_id.isdigit()) self.assertIn("name", info) self.assertIn(info.get("rank"), ("S", "A")) self.assertTrue(ITEMS) for item_id, info in ITEMS.items(): with self.subTest(item_id=item_id): self.assertIn(info.get("type"), ("item", "cosmetic")) self.assertIn("name", info) def test_rewards_by_id_merges_all_mapping_files(self): for reward_id in (*ARC_META, *CHARACTERS, *ITEMS): with self.subTest(reward_id=reward_id): reward = REWARDS_BY_ID[reward_id] self.assertEqual(reward["id"], reward_id) self.assertIn(reward["type"], ("arc", "character", "item", "cosmetic")) def test_decode_reward_key_round_trips_observed_keys(self): observed = { "98bdc9ad7dd9a5b99501": "fork_vine", "98bdc9ad7d41c9bdad85c9e5bdb901": "fork_Prokaryon", "98bdc9ad7ddda1d585add585b99d01": "fork_whuakuang", "98bdc9ad7dddd5a1d585add585b99d01": "fork_wuhuakuang", "10a58d9539bdc9b585b101": "DiceNormal", "10a58d957dd1a58dad95d17dc1c400": "Dice_ticket_01", "10a58d957dd1a58dad95d17dc1c800": "Dice_ticket_02", "10a58d95b1a5b5a5d19501": "Dicelimite", "1885cda1a5bdb97d1db1a591957dc5c0c4c000": "Fashion_Glide_1010", "1885cda1a5bdb97dd995a1a58db1957dc5c0c4c07c59c1c0e000": "Fashion_vehicle_1010_V008", "c4c0cccc00": "1033", "c4c0dcc000": "1070", "c4c0dcc0": "1070", "c4c0c8c4": "1021", } for key_hex, expected_id in observed.items(): with self.subTest(key_hex=key_hex): self.assertEqual(decode_reward_key(bytes.fromhex(key_hex)), expected_id) def test_infer_reward_type_for_unmapped_ids(self): self.assertEqual(infer_reward_type("fork_newarc"), "arc") self.assertEqual(infer_reward_type("1099"), "character") self.assertEqual(infer_reward_type("Fashion_hat_2000"), "cosmetic") self.assertEqual(infer_reward_type("Dice_ticket_03"), "item") self.assertEqual(infer_reward_type(""), "") def test_uid_source_matches_v4_reference(self): rows = load_reference_csv("monopoly_history_poc_10_all_44_pages_v4.csv") first = rows[0] self.assertEqual(make_uid(first, 0), "f2c72f0a80b79216bf15661521620693") def test_uid_uses_pool_timestamp_and_ordinal_only(self): row = { "pool_group_id": "Lottery_LimitedCharacter", "timestamp_raw_hex": "40e93247c3097b23", "dice": 5, "reward_key_hex": "10a58d957dd1a58dad95d17dc1c800", "quantity": 50, } changed_content = { **row, "dice": 1, "reward_key_hex": "98bdc9ad7dd9a5b99501", "quantity": 1, } changed_pool = {**row, "pool_group_id": "Lottery_Permanent"} self.assertEqual(make_uid(row, 0), "74a9ef4aacde549dfe8e8e7cc6ddd65b") self.assertEqual(make_uid(changed_content, 0), make_uid(row, 0)) self.assertNotEqual(make_uid(changed_pool, 0), make_uid(row, 0)) self.assertNotEqual(make_uid(row, 1), make_uid(row, 0)) def test_pages_1_to_5_exports_every_row(self): rows = load_reference_csv("monopoly_history_poc_13_pages_1_to_5_v4.csv") annotated = annotate_groups(rows) exported = [row for row in annotated if row["export_record"] is True] # Every decoded row is exported; boundary groups are never dropped. self.assertEqual(len(annotated), 25) self.assertEqual(len(exported), 25) @staticmethod def _synthetic_row(page, timestamp_hex, result_type): return { "page": page, "timestamp_raw_hex": timestamp_hex, "timestamp_decoded": f"ts-{timestamp_hex}", "result_type": result_type, "dice": 4 if result_type == "dice" else 0, "reward_key_hex": "10a58d9539bdc9b585b101", "quantity": 1, } def test_oldest_group_with_partial_dice_count_exports_without_warning(self): rows = [self._synthetic_row(1, "aa", "dice") for _ in range(5)] rows += [self._synthetic_row(2, "bb", "dice") for _ in range(5)] rows += [self._synthetic_row(3, "bb", "dice") for _ in range(4)] rows += [self._synthetic_row(3, "bb", "points_gift")] annotated = annotate_groups(rows) exported = [row for row in annotated if row["export_record"] is True] # Oldest group is a partially captured 10-pull on a full final page. Its # captured prefix is ordinal-stable, so it is exported with stable UIDs. self.assertEqual(len(exported), 15) oldest = [row for row in annotated if row["timestamp_raw_hex"] == "bb"] self.assertTrue(all(row["uid"] for row in oldest)) self.assertTrue(all(row["uid_status"] == "stable" for row in oldest)) self.assertEqual([row["timestamp_group_ordinal"] for row in oldest], list(range(10))) def test_incomplete_oldest_prefix_keeps_stable_uids(self): full = [self._synthetic_row(1, "aa", "dice") for _ in range(5)] full += [self._synthetic_row(2, "bb", "dice") for _ in range(3)] full += [self._synthetic_row(3, "bb", "dice") for _ in range(2)] truncated = [r for r in full if r["page"] in (1, 2)] full_rows = annotate_groups([dict(r) for r in full]) trunc_rows = annotate_groups([dict(r) for r in truncated]) full_uids = [r["uid"] for r in full_rows if r["timestamp_raw_hex"] == "bb"][:3] trunc_uids = [r["uid"] for r in trunc_rows if r["timestamp_raw_hex"] == "bb"] # Capturing only the first 3 of a 5-record oldest group yields the same # UIDs those rows have in the full capture. self.assertEqual(len(trunc_uids), 3) self.assertEqual(trunc_uids, full_uids) def test_oldest_group_with_ten_dice_exports_on_full_final_page(self): rows = [self._synthetic_row(1, "aa", "dice") for _ in range(5)] rows += [self._synthetic_row(2, "bb", "dice") for _ in range(5)] rows += [self._synthetic_row(3, "bb", "dice") for _ in range(5)] annotated = annotate_groups(rows) exported = [row for row in annotated if row["export_record"] is True] self.assertEqual(len(exported), 15) def test_run_selection_anchors_to_page_1_and_keeps_newest(self): # Page 2's response was lost: captured pages 1, 3, 4, 5. pairs = [(p, p * 2, 0, 0, 0, 0, b"", "permanent") for p in (1, 3, 4, 5)] run, warnings = select_continuous_run_from_page_1(pairs) # The page-1 run (just page 1, the newest history) is kept; later pages are # ignored with a gap warning, never silently discarding page 1. self.assertEqual([p[0] for p in run], [1]) self.assertEqual(len(warnings), 1) self.assertEqual(warnings[0]["code"], "PAGE_GAP_DETECTED") self.assertEqual(warnings[0]["ignored_pages"], [3, 4, 5]) def test_run_selection_warns_when_page_1_missing(self): pairs = [(p, p * 2, 0, 0, 0, 0, b"", "permanent") for p in (3, 4, 5)] run, warnings = select_continuous_run_from_page_1(pairs) self.assertEqual([p[0] for p in run], [3, 4, 5]) self.assertEqual(warnings[0]["code"], "DID_NOT_START_AT_PAGE_1") def test_full_reference_scan_exports_all_rows(self): rows = load_reference_csv("monopoly_history_poc_10_all_44_pages_v4.csv") annotated = annotate_groups(rows) exported = [row for row in annotated if row["export_record"] is True] json_path = EXPORTS / "monopoly_history_export_10_all_44_pages_v4.json" if not json_path.exists(): json_path = EXPORTS / "old" / "monopoly_history_export_10_all_44_pages_v4.json" if not json_path.exists(): raise unittest.SkipTest(f"private reference fixture not present: {json_path}") with json_path.open(encoding="utf-8") as f: reference = json.load(f) self.assertEqual(len(annotated), reference["scan"]["decoded_records"]) self.assertEqual(len(exported), reference["scan"]["exported_records"]) def test_sanitized_export_omits_raw_packet_fields(self): rows = load_reference_csv("monopoly_history_poc_13_pages_1_to_5_v4.csv") annotated = annotate_groups(rows) export = build_export_json(annotated, []) self.assertEqual(export["format"], "nte-history-export") self.assertIn("exporter", export) self.assertNotIn("user_uid", export) self.assertNotIn("record_hex", export["records"][0]) self.assertNotIn("request_msg", export["records"][0]) self.assertNotIn("response_msg", export["records"][0]) def test_export_includes_user_uid_when_provided(self): rows = load_reference_csv("monopoly_history_poc_13_pages_1_to_5_v4.csv") annotated = annotate_groups(rows) export = build_export_json( annotated, [], capture_source="npcap", user_uid="123456789", ) self.assertEqual(list(export).index("user_uid"), list(export).index("records") - 1) self.assertEqual(export["capture_source"], "npcap") self.assertEqual(export["user_uid"], "123456789") def test_debug_csv_includes_exporter_version(self): with TemporaryDirectory() as tmp: path = Path(tmp) / "debug.csv" write_csv(path, [{"uid": "abc123"}]) with path.open(newline="", encoding="utf-8") as f: rows = list(csv.DictReader(f)) self.assertEqual(rows[0]["exporter_version"], __version__) self.assertEqual(rows[0]["uid"], "abc123") def test_export_paths_include_user_uid_banner_and_timestamp(self): _csv_path, json_path = export_paths("limited_character", "218216016349") self.assertRegex( json_path.name, r"^218216016349_Limited_\d{8}_\d{6}(?:_\d+)?\.json$", ) def test_extracts_user_uid_from_record_context(self): payload = ( b"\x00" * 24 + (218216016349).to_bytes(8, "little") + b"\x00\x00\x00\x00\x09\x00\x00\x00TagOthers\x00" ) self.assertEqual(extract_user_uid(payload), "218216016349") def test_extracts_user_uid_from_private_spawn_record_context(self): payload = ( b"\x88\x00\x00\x00\x10\x00\x00\x00" + (218216016349).to_bytes(8, "little") + b"\x08\x00\x0c\x00\x07\x00\x08\x00\x08\x00\x00\x00" + b"\x00\x00\x00\x01\x08\x00\x00\x00\x04\x00\x04\x00" + b"\x04\x00\x00\x00\x16\x00\x00\x00PrivateSpawnInfoRecord\x00" ) self.assertEqual(extract_user_uid(payload), "218216016349") def test_does_not_extract_user_uid_from_wrong_record_offset(self): payload = ( b"\x00" * 28 + (218216016349).to_bytes(8, "little") + b"\x00\x00\x00\x00TagOthers\x00" ) self.assertIsNone(extract_user_uid(payload)) def test_does_not_extract_old_eight_digit_false_positive_as_user_uid(self): payload = ( b"WholeVehicleData\x00\x00\x00\x00\x00o<\x00\x00\x05\x00\x00\x00" b"\x0b\x00\x00\x00Vehicle015\x00\x0b\x00\x00\x00buyvehicle\x00" b"\x09\x00\x00\x0015363624\x00\x06\x00\x00\x00" ) self.assertIsNone(extract_user_uid(payload)) def test_ipv4_parser_extracts_tcp_payload_for_user_uid_detection(self): payload = ( (218216016349).to_bytes(8, "little") + b"\x00\x00\x00\x00\x09\x00\x00\x00TagOthers\x00" ) tcp_header = bytearray(20) tcp_header[0:2] = (40000).to_bytes(2, "big") tcp_header[2:4] = (30000).to_bytes(2, "big") tcp_header[12] = 5 << 4 total_len = 20 + len(tcp_header) + len(payload) ip_header = bytearray(20) ip_header[0] = 0x45 ip_header[2:4] = total_len.to_bytes(2, "big") ip_header[9] = 6 ip_header[12:16] = bytes([192, 0, 2, 1]) ip_header[16:20] = bytes([198, 51, 100, 2]) packet = parse_ipv4_packet(bytes(ip_header) + bytes(tcp_header) + payload) self.assertIsNotNone(packet) self.assertEqual(packet.protocol, "tcp") self.assertEqual(packet.payload, payload) def test_limited_selector_and_marker_decode(self): request = bytearray(45) request[31:35] = (4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (8).to_bytes(4, "little") self.assertEqual(history_request_kind(bytes(request)), "limited_character") def test_monopoly_request_allows_coalesced_trailing_payload(self): request = bytearray(45) request[31:35] = (25 * 4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (4).to_bytes(4, "little") coalesced = bytes(request) + bytes.fromhex( "007c669610062038461bc40100000872a34b93821a0219aa933b0a6b2ba34a6b" ) self.assertEqual(history_request_kind(coalesced), "permanent") def test_arc_request_allows_coalesced_trailing_payload(self): request = bytearray(34) request[24:28] = (2060).to_bytes(4, "little") request[29:33] = (7 * 2).to_bytes(4, "little") self.assertTrue(is_arc_history_request(bytes(request) + bytes(32))) self.assertEqual(arc_request_page(bytes(request) + bytes(32)), 7) response = bytearray(220) response[0x50:0x54] = (4).to_bytes(4, "little") response[0x54:0x58] = (20).to_bytes(4, "little") response[0x58:0x5d] = bytes.fromhex("c4c0c4c000") marker_offset = 0x5d response[marker_offset:marker_offset + len(LIMITED_CHARACTER_MARKER)] = LIMITED_CHARACTER_MARKER timestamp_raw = (2556647947780680000).to_bytes(8, "little") response[marker_offset + len(LIMITED_CHARACTER_MARKER):marker_offset + len(LIMITED_CHARACTER_MARKER) + 8] = timestamp_raw rows = decode_response_records(bytes(response)) self.assertEqual(len(rows), 1) self.assertEqual(rows[0]["reward_id"], "1010") self.assertEqual(rows[0]["reward_name"], "Nanally") def test_v7_prefixed_points_gift_rows_override_visible_dice(self): cases = [ ("limited_all_04_v7.csv", 41, "1020"), ("monopoly_history_poc_10_all_44_pages_v7.csv", 31, "1033"), ("monopoly_history_poc_10_all_44_pages_v7.csv", 86, "fork_PaperPlane"), ("monopoly_history_poc_10_all_44_pages_v7.csv", 141, "fork_Kite"), ("monopoly_history_poc_10_all_44_pages_v7.csv", 196, "1021"), ] for filename, row_index, reward_id in cases: with self.subTest(filename=filename, row_index=row_index): reference = load_v7_row(filename, row_index) decoded = decode_single_record(reference["record_hex"]) self.assertEqual(decoded["result_type"], "points_gift") self.assertEqual(decoded["result_source_raw"], 0) self.assertEqual(decoded["dice"], 0) self.assertEqual(decoded["dice_raw_u32"], 0) self.assertEqual(decoded["reward_id"], reward_id) def test_v7_prefixed_chase_reward_overrides_visible_dice_and_quantity(self): reference = load_v7_row("limited_all_04_v7.csv", 61) decoded = decode_single_record(reference["record_hex"]) decoded.update( { "pool_group_id": "Lottery_LimitedCharacter", "timestamp_group_ordinal": int(reference["timestamp_group_ordinal"]), } ) self.assertEqual(decoded["result_type"], "chase_reward") self.assertEqual(decoded["result_source_raw"], -4) self.assertEqual(decoded["dice"], -4) self.assertEqual(decoded["dice_raw_u32"], -4) self.assertEqual(decoded["reward_id"], "Dice_ticket_01") self.assertEqual(decoded["quantity"], 30) self.assertEqual(make_uid(decoded, int(reference["timestamp_group_ordinal"])), "23dc293f39f18e94a81ccaaf7e1a67eb") def test_warp_piece_chase_subrecord_without_prefix_marker_is_chase_reward(self): decoded = decode_single_record( "c1c4b0ccc00000000000040000003c00000010a58d957dd1a58dad95d17dc1c400" "4c0000000c85c99141bdbdb17d0da185c9858dd195c90140eb2c2dd7227b23" ) self.assertEqual(decoded["result_type"], "chase_reward") self.assertEqual(decoded["result_source_raw"], -4) self.assertEqual(decoded["dice"], -4) self.assertEqual(decoded["dice_raw_u32"], -4) self.assertEqual(decoded["reward_id"], "Dice_ticket_01") self.assertEqual(decoded["reward_name"], "Warp Piece") self.assertEqual(decoded["quantity"], 30) def test_page_first_prefix_uses_real_dice_field(self): cases = [ ( "003006000014000000040000002800000098bdc9ad7dd9a5b995010000000008000000" "3c00000010a58d957dd1a58dad95d17dc1c4002800000098bdc9ad7dd9a5b995014c" "0000000c85c99141bdbdb17d0da185c9858dd195c901c0dd53bd2b137b23", 1, "fork_vine", ), ( "00c8060000140000001000000014000000c4c0d4d400000000000400000014000000" "c4c0d4d400440000000c85c99141bdbdb17d3995dd49bdb1950100d929e115087b23", 4, "1055", ), ] for record_hex, expected_dice, reward_id in cases: with self.subTest(reward_id=reward_id): decoded = decode_single_record(record_hex) self.assertEqual(decoded["dice"], expected_dice) self.assertEqual(decoded["dice_raw_u32"], expected_dice * 4) self.assertEqual(decoded["dice_offset_in_record"], 9) self.assertEqual(decoded["result_type"], "dice") self.assertEqual(decoded["reward_id"], reward_id) def test_batched_monopoly_response_normalizes_embedded_page_header(self): page_7 = [load_v7_row("limited_all_04_v7.csv", row) for row in range(31, 36)] page_8 = [load_v7_row("limited_all_04_v7.csv", row) for row in range(36, 41)] embedded_header = bytes.fromhex( "100126675671c6e0ecfcb769070000000c08000000100000003e0800000004000000" "486c0000001835bdb9bdc1bdb1e531bdd1d195c9e549958dbdc9911185d18501" ) response = ( bytes(0x50) + b"".join(bytes.fromhex(row["record_hex"]) for row in page_7) + embedded_header + b"".join(bytes.fromhex(row["record_hex"]) for row in page_8) ) decoded = decode_response_records(response) self.assertEqual(len(decoded), 10) self.assertEqual(decoded[5]["reward_id"], page_8[0]["reward_id"]) self.assertEqual(decoded[5]["dice"], 4) self.assertEqual(decoded[5]["result_type"], "dice") self.assertEqual(decoded[5]["record_hex"], page_8[0]["record_hex"]) def test_monopoly_response_parser_realigns_bit_packed_payload(self): reference_rows = [ load_v7_row("limited_all_04_v7.csv", row) for row in range(51, 61) ] response = bytes(0x50) + b"".join( bytes.fromhex(row["record_hex"]) for row in reference_rows ) packed = bytearray() carry = 0b10101 for byte in response: packed.append(carry | ((byte << 5) & 0xFF)) carry = byte >> 3 packed.append(carry) decoded = decode_response_records(bytes(packed)) self.assertEqual(len(decoded), 10) self.assertEqual( [row["record_hex"] for row in decoded], [row["record_hex"] for row in reference_rows], ) def test_page_gap_warning_reports_ignored_pages(self): pairs = [(p, p * 2, 0, 0, 0, 0, b"", "permanent") for p in (1, 2, 3, 5)] run, warnings = select_continuous_run_from_page_1(pairs) self.assertEqual([p[0] for p in run], [1, 2, 3]) self.assertEqual(warnings[0]["code"], "PAGE_GAP_DETECTED") self.assertEqual(warnings[0]["ignored_pages"], [5]) def test_arc_key_timestamp_and_uid_match_reference(self): row = load_arc_csv("arc_pull_10_all_pages_v2.csv")[0] self.assertEqual(decode_arc_key(bytes.fromhex(row["arc_key_hex"])), "fork_nonos") _ticks, _unix, decoded = decode_arc_timestamp(bytes.fromhex(row["timestamp_raw_hex"])) self.assertEqual(decoded, "2026-06-10 23:46:29") self.assertEqual(make_arc_uid(row["timestamp_raw_hex"], int(row["timestamp_group_ordinal"])), "4435d9729fa8fd0eaf1b1ad7aa4d2172") def test_arc_response_parser_matches_reference_first_page(self): reference_rows = load_arc_csv("arc_pull_10_all_pages_v2.csv")[:5] response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in reference_rows) decoded = parse_arc_response(response) self.assertEqual(len(decoded), 5) self.assertEqual([row["reward_id"] for row in decoded[:3]], ["fork_nonos", "fork_nonos", "fork_Prokaryon"]) self.assertEqual(decoded[0]["reward_type"], "arc") self.assertEqual(decoded[0]["reward_key_hex"], reference_rows[0]["arc_key_hex"]) def test_arc_response_parser_rejects_invalid_timestamp_noise(self): response = bytearray(0x4C) response += (10).to_bytes(4, "little") response += bytes.fromhex("ccdee4d6be") response += (8).to_bytes(4, "little") response += b"garb" response += (0xFFFFFFFFFFFFFFFF).to_bytes(8, "little") self.assertEqual(parse_arc_response(bytes(response)), []) def test_arc_partial_timestamp_group_is_exported_without_warning(self): rows = load_arc_csv("arc_pages_1_to_5_v2.csv") pairs = [] for page in range(1, 6): page_rows = [row for row in rows if int(row["page"]) == page] response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in page_rows) pairs.append((page, page * 2, page, 1.0, page + 100, 1.1, response)) decoded = build_arc_rows_from_pairs(pairs) exported = [row for row in decoded if row["export_record"] is True] # The oldest group is a 10-pull split by stopping at page 5 (5 of 10 rows). # Its captured prefix is ordinal-stable, so every row is exported with a # stable UID and no warning. self.assertEqual(len(decoded), 25) self.assertEqual(len(exported), 25) self.assertTrue(all(row["uid"] for row in decoded)) self.assertTrue(all(row["uid_status"] == "stable" for row in decoded)) def test_arc_incomplete_prefix_keeps_stable_uids(self): rows = load_arc_csv("arc_pull_10_all_pages_v2.csv") # Build a full 2-page (10-record) group, then a truncated 1-page version. full_pairs, trunc_pairs = [], [] for page in (1, 2): page_rows = [r for r in rows if int(r["page"]) == page] response = bytes(0x4C) + b"".join(bytes.fromhex(r["record_hex"]) for r in page_rows) full_pairs.append((page, page * 2, page, 1.0, page + 100, 1.1, response)) if page == 1: trunc_pairs.append((page, page * 2, page, 1.0, page + 100, 1.1, response)) ts = rows[0]["timestamp_raw_hex"] full = [r for r in build_arc_rows_from_pairs(full_pairs) if r["timestamp_raw_hex"] == ts] trunc = [r for r in build_arc_rows_from_pairs(trunc_pairs) if r["timestamp_raw_hex"] == ts] self.assertTrue(trunc) self.assertEqual([r["uid"] for r in trunc], [r["uid"] for r in full[: len(trunc)]]) def test_arc_row_builder_accepts_live_pairs_with_kind(self): reference_rows = load_arc_csv("arc_pull_10_all_pages_v2.csv")[:5] response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in reference_rows) decoded = build_arc_rows_from_pairs([(1, 2, 1, 1.0, 2, 1.1, response, "arc_miracle_box")]) self.assertEqual(len(decoded), 5) self.assertEqual(decoded[0]["reward_id"], "fork_nonos") def test_arc_export_is_shared_pity(self): reference_rows = load_arc_csv("arc_pull_10_all_pages_v2.csv")[:10] response = bytes(0x4C) + b"".join(bytes.fromhex(row["record_hex"]) for row in reference_rows) rows = build_arc_rows_from_pairs([(1, 2, 1, 1.0, 2, 1.1, response, "arc_miracle_box")]) export = build_export_json(rows, []) self.assertEqual(export["banner"]["id"], "Arc_MiracleBox") self.assertIs(export["banner"]["shared_pity"], True) def test_group_detection_counts_only_dice_records_but_uid_ordinals_keep_all_rows(self): rows = [ { "page": 1, "timestamp_raw_hex": "aa", "timestamp_decoded": "2026-01-01 00:00:00", "result_type": "dice", "dice": 1, "reward_key_hex": "k1", "quantity": 1, }, { "page": 1, "timestamp_raw_hex": "aa", "timestamp_decoded": "2026-01-01 00:00:00", "result_type": "points_gift", "dice": 0, "reward_key_hex": "k2", "quantity": 1, }, { "page": 1, "timestamp_raw_hex": "aa", "timestamp_decoded": "2026-01-01 00:00:00", "result_type": "chase_reward", "dice": -4, "reward_key_hex": "k3", "quantity": 30, }, { "page": 1, "timestamp_raw_hex": "aa", "timestamp_decoded": "2026-01-01 00:00:00", "result_type": "dice", "dice": 2, "reward_key_hex": "k4", "quantity": 1, }, { "page": 1, "timestamp_raw_hex": "bb", "timestamp_decoded": "2026-01-01 00:01:00", "result_type": "dice", "dice": 3, "reward_key_hex": "k5", "quantity": 1, }, ] annotated = annotate_groups(rows) # Ordinals cover every row in the group, but the dice-only count drives # timestamp_group_size_seen (2 dice in the 4-record group). self.assertEqual([row["timestamp_group_ordinal"] for row in annotated[:4]], [0, 1, 2, 3]) self.assertEqual({row["timestamp_group_size_seen"] for row in annotated[:4]}, {2}) self.assertEqual({row["timestamp_group_record_size_seen"] for row in annotated[:4]}, {4}) self.assertTrue(all(row["export_record"] for row in annotated)) def test_live_session_pairs_request_and_response(self): session = LiveHistorySession("192.168.0.10") request = bytearray(45) request[31:35] = (4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (4).to_bytes(4, "little") response = bytearray(220) response[0x50:0x50 + len(MARKER)] = MARKER response[0x50 + len(MARKER):0x50 + len(MARKER) + 8] = ( 2556647947780680000 ).to_bytes(8, "little") self.assertFalse( session.process_packet( UdpPacket( timestamp=1.0, src_ip="192.168.0.10", dst_ip="203.0.113.5", src_port=50000, dst_port=40000, payload=bytes(request), ) ) ) self.assertTrue( session.process_packet( UdpPacket( timestamp=1.2, src_ip="203.0.113.5", dst_ip="192.168.0.10", src_port=40000, dst_port=50000, payload=bytes(response), ) ) ) self.assertEqual(len(session.pairs), 1) self.assertEqual(session.last_page_seen, 1) def test_live_session_pairs_pipelined_and_batched_pages(self): session = LiveHistorySession("192.168.0.10") def request_packet(page, timestamp): request = bytearray(45) request[31:35] = (page * 4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (4).to_bytes(4, "little") return UdpPacket( timestamp=timestamp, src_ip="192.168.0.10", dst_ip="203.0.113.5", src_port=50000, dst_port=40000, payload=bytes(request), ) timestamp = (2556647947780680000).to_bytes(8, "little") def response_packet(record_count, packet_timestamp): response = bytearray(0x50) for _ in range(record_count): response += bytes(4) + MARKER + timestamp return UdpPacket( timestamp=packet_timestamp, src_ip="203.0.113.5", dst_ip="192.168.0.10", src_port=40000, dst_port=50000, payload=bytes(response), ) for page in range(1, 9): session.process_packet(request_packet(page, 1.0 + page / 10)) self.assertTrue(session.process_packet(response_packet(10, 2.0))) self.assertEqual([pair[0] for pair in session.pairs], [1, 2]) self.assertEqual([pair[8:10] for pair in session.pairs], [(0, 5), (5, 5)]) self.assertTrue(session.process_packet(response_packet(10, 2.1))) self.assertEqual([pair[0] for pair in session.pairs], [1, 2, 3, 4]) self.assertEqual(session.missing_pages("permanent"), []) rows = session.build_rows("permanent") self.assertEqual(len(rows), 20) self.assertEqual( {page: sum(row["page"] == page for row in rows) for page in range(1, 5)}, {1: 5, 2: 5, 3: 5, 4: 5}, ) session.process_packet(request_packet(9, 2.2)) self.assertTrue(session.process_packet(response_packet(4, 2.3))) self.assertEqual(session.pairs[-1][0], 9) self.assertEqual(session.pairs[-1][8:10], (0, 4)) def test_live_session_new_page_one_starts_clean_recovery_cycle(self): session = LiveHistorySession("192.168.0.10") def request_packet(page, timestamp): request = bytearray(45) request[31:35] = (page * 4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (4).to_bytes(4, "little") return UdpPacket( timestamp=timestamp, src_ip="192.168.0.10", dst_ip="203.0.113.5", src_port=50000, dst_port=40000, payload=bytes(request), ) session.process_packet(request_packet(7, 1.0)) session.process_packet(request_packet(8, 1.1)) session.process_packet(request_packet(1, 2.0)) self.assertEqual([request.page for request in session.pending], [1]) self.assertEqual( session.missing_page_reason("permanent", 7), "request captured; no matching response page was captured", ) timestamp = (2556647947780680000).to_bytes(8, "little") response = bytearray(0x50) for _ in range(5): response += bytes(4) + MARKER + timestamp self.assertTrue( session.process_packet( UdpPacket( timestamp=2.1, src_ip="203.0.113.5", dst_ip="192.168.0.10", src_port=40000, dst_port=50000, payload=bytes(response), ) ) ) self.assertEqual(session.pairs[-1][0], 1) self.assertEqual(session.missing_pages("permanent"), []) def test_live_session_reports_unrecognized_response_candidate(self): session = LiveHistorySession("192.168.0.10") def request_packet(page, timestamp): request = bytearray(45) request[31:35] = (page * 4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (4).to_bytes(4, "little") return UdpPacket( timestamp=timestamp, src_ip="192.168.0.10", dst_ip="203.0.113.5", src_port=50000, dst_port=40000, payload=bytes(request), ) session.process_packet(request_packet(1, 1.0)) session.process_packet( UdpPacket( timestamp=1.1, src_ip="203.0.113.5", dst_ip="192.168.0.10", src_port=40000, dst_port=50000, payload=bytes(220), ) ) session.process_packet(request_packet(2, 1.2)) self.assertEqual( session.missing_page_reason("permanent", 1), "1 matching inbound UDP packet(s) captured but not recognized as history response (lengths: 220)", ) def test_live_session_ignores_non_history_udp_packets(self): session = LiveHistorySession("192.168.0.10") request = bytearray(45) request[31:35] = (4).to_bytes(4, "little") request[35:39] = (4220).to_bytes(4, "little") request[40:44] = (4).to_bytes(4, "little") noise = b"not-a-history-response" * 20 session.process_packet( UdpPacket( timestamp=1.0, src_ip="192.168.0.10", dst_ip="203.0.113.5", src_port=50000, dst_port=40000, payload=bytes(request), ) ) self.assertFalse( session.process_packet( UdpPacket( timestamp=1.1, src_ip="203.0.113.5", dst_ip="192.168.0.10", src_port=40000, dst_port=50000, payload=noise, ) ) ) self.assertEqual(len(session.pairs), 0) if __name__ == "__main__": unittest.main()