diff --git a/tests/__init__.py b/tests/__init__.py new file mode 100644 index 0000000..3f4d23c --- /dev/null +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Test package for the NTE history exporter.""" diff --git a/tests/fixtures/README.md b/tests/fixtures/README.md new file mode 100644 index 0000000..7c54fd3 --- /dev/null +++ b/tests/fixtures/README.md @@ -0,0 +1,23 @@ +# Synthetic network fixture + +`synthetic_history_network.json` is the committed end-to-end packet fixture for +the exporter test suite. It contains UDP payloads for Permanent Board and Arc +history requests/responses, plus isolated protocol edge cases. + +The fixture is safe to publish: + +- endpoints use the RFC 5737 documentation ranges `192.0.2.0/24` and + `198.51.100.0/24`; +- packet and history timestamps are deterministic synthetic values; +- it contains no game user UID, account/session data, cookies, tokens, or raw + capture metadata; +- the test suite checks every payload with the UID extractor before accepting + it. + +Tests load the fixture through helpers in `tests/support.py`. The `replay` +scenario is fed through `LiveHistorySession`; `protocol_sample` packets cover +points gifts, chase rewards, batched pages, and bit-packed responses. + +Do not replace this file with a real `.pcap`, `.flows`, or exported account +history. Add new cases by constructing the smallest relevant payload, replacing +all timestamps and endpoints, and extending the privacy assertions. diff --git a/tests/fixtures/synthetic_history_network.json b/tests/fixtures/synthetic_history_network.json new file mode 100644 index 0000000..4f3e531 --- /dev/null +++ b/tests/fixtures/synthetic_history_network.json @@ -0,0 +1,322 @@ +{ + "type": "nte-synthetic-network-fixture", + "schema_version": 1, + "description": "Synthetic UDP history transcript for tests. Contains no real account or network identity data.", + "privacy": { + "synthetic": true, + "contains_user_uid": false, + "contains_raw_account_session": false, + "addresses": "RFC 5737 documentation ranges", + "timestamps": "deterministic synthetic values" + }, + "local_ip": "192.0.2.10", + "expected": { + "permanent_records": 25, + "arc_records": 25, + "permanent_first_uid": "bdf0395ad3fb21f89b62ca92aaef7e36", + "arc_first_uid": "221b6accc3de16a145293b8804761c7c" + }, + "packets": [ + { + "type": "packet", + "label": "permanent-page-1-request", + "scenario": "replay", + "timestamp": 1.0, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "00000000000000000000000000000000000000000000000000000000000000040000007c100000000400000000" + }, + { + "type": "packet", + "label": "permanent-page-1-response", + "scenario": "replay", + "timestamp": 1.01, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b23" + }, + { + "type": "packet", + "label": "permanent-page-2-request", + "scenario": "replay", + "timestamp": 1.02, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "00000000000000000000000000000000000000000000000000000000000000080000007c100000000400000000" + }, + { + "type": "packet", + "label": "permanent-page-2-response", + "scenario": "replay", + "timestamp": 1.03, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b23" + }, + { + "type": "packet", + "label": "permanent-page-3-request", + "scenario": "replay", + "timestamp": 1.04, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000c0000007c100000000400000000" + }, + { + "type": "packet", + "label": "permanent-page-3-response", + "scenario": "replay", + "timestamp": 1.05, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b23" + }, + { + "type": "packet", + "label": "permanent-page-4-request", + "scenario": "replay", + "timestamp": 1.06, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "00000000000000000000000000000000000000000000000000000000000000100000007c100000000400000000" + }, + { + "type": "packet", + "label": "permanent-page-4-response", + "scenario": "replay", + "timestamp": 1.07, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b23" + }, + { + "type": "packet", + "label": "permanent-page-5-request", + "scenario": "replay", + "timestamp": 1.08, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "00000000000000000000000000000000000000000000000000000000000000140000007c100000000400000000" + }, + { + "type": "packet", + "label": "permanent-page-5-response", + "scenario": "replay", + "timestamp": 1.09, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b2300c805000014000000100000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c800440000000c85c99141bdbdb17d3995dd49bdb195010000f3095f138b23" + }, + { + "type": "packet", + "label": "arc-page-1-request", + "scenario": "replay", + "timestamp": 1.1, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "0000000000000000000000000000000000000000000000000c080000000200000000" + }, + { + "type": "packet", + "label": "arc-page-1-response", + "scenario": "replay", + "timestamp": 1.11, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000016000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc511" + }, + { + "type": "packet", + "label": "arc-page-2-request", + "scenario": "replay", + "timestamp": 1.12, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "0000000000000000000000000000000000000000000000000c080000000400000000" + }, + { + "type": "packet", + "label": "arc-page-2-response", + "scenario": "replay", + "timestamp": 1.1300000000000001, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000016000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc511" + }, + { + "type": "packet", + "label": "arc-page-3-request", + "scenario": "replay", + "timestamp": 1.1400000000000001, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "0000000000000000000000000000000000000000000000000c080000000600000000" + }, + { + "type": "packet", + "label": "arc-page-3-response", + "scenario": "replay", + "timestamp": 1.1500000000000001, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000016000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc511" + }, + { + "type": "packet", + "label": "arc-page-4-request", + "scenario": "replay", + "timestamp": 1.1600000000000001, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "0000000000000000000000000000000000000000000000000c080000000800000000" + }, + { + "type": "packet", + "label": "arc-page-4-response", + "scenario": "replay", + "timestamp": 1.1700000000000002, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000016000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc511" + }, + { + "type": "packet", + "label": "arc-page-5-request", + "scenario": "replay", + "timestamp": 1.1800000000000002, + "protocol": "udp", + "src_ip": "192.0.2.10", + "dst_ip": "198.51.100.20", + "src_port": 49152, + "dst_port": 30031, + "payload_hex": "0000000000000000000000000000000000000000000000000c080000000a00000000" + }, + { + "type": "packet", + "label": "arc-page-5-response", + "scenario": "replay", + "timestamp": 1.1900000000000002, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000016000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc51116000000ccdee4d6bedcdedcdee600280000008cdee4d698dee8e8cae4f2be82dc90eadca2ea000000cdd9418bc511" + }, + { + "type": "packet", + "label": "limited-points-gift-1", + "scenario": "protocol_sample", + "timestamp": 11.0, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e4050000140000000000000014000000c4c0c8c000000000000400000014000000c4c0c8c0004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23" + }, + { + "type": "packet", + "label": "limited-chase-reward", + "scenario": "protocol_sample", + "timestamp": 12.0, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000009806000014000000fcffffff4b00000010a58d957dd1a58dad95d17dc1c4b0ccc00000000000040000003c00000010a58d957dd1a58dad95d17dc1c4004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23" + }, + { + "type": "packet", + "label": "limited-batched-pages", + "scenario": "protocol_sample", + "timestamp": 13.0, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e805000014000000140000002c00000098bdc9ad7db9bdb9bdcd0100000000040000002c00000098bdc9ad7db9bdb9bdcd014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b230400000014000000c4c0c8c000000000000400000014000000c4c0c8c0004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23100000002c00000098bdc9ad7db9bdb9bdcd0100000000040000002c00000098bdc9ad7db9bdb9bdcd014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23040000002800000098bdc9ad7dd9a5b9950100000000080000003c00000010a58d957dd1a58dad95d17dc1c4002800000098bdc9ad7dd9a5b995014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23040000003400000098bdc9ad7d91d5cdd189a5b90100000000040000003400000098bdc9ad7d91d5cdd189a5b9014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23100126675671c6e0ecfcb769070000000c08000000100000003e0800000004000000486c0000001835bdb9bdc1bdb1e531bdd1d195c9e549958dbdc9911185d18501000406000014000000100000003000000098bdc9ad7de585bd9185bd0100000000040000003000000098bdc9ad7de585bd9185bd014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b230c00000014000000c4c0d4c80000000000080000003c00000010a58d957dd1a58dad95d17dc1c40014000000c4c0d4c8004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23140000004800000010a58d957dd1a58dad95d17dc1c8b0d4c00000000000040000003c00000010a58d957dd1a58dad95d17dc1c8004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b231800000014000000c4c0c0e000000000000400000014000000c4c0c0e0004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23100000003c00000098bdc9ad7d41c9bdad85c9e5bdb90100000000040000003c00000098bdc9ad7d41c9bdad85c9e5bdb9014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23" + }, + { + "type": "packet", + "label": "limited-bitpacked-source", + "scenario": "protocol_sample", + "timestamp": 14.0, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000004c06000014000000180000003c00000098bdc9ad7d41c9bdad85c9e5bdb90100000000080000003c00000010a58d957dd1a58dad95d17dc1c4003c00000098bdc9ad7d41c9bdad85c9e5bdb9014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b230000000014000000c4c0c4e400000000000400000014000000c4c0c4e4004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23040000003400000098bdc9ad7d91d5cdd189a5b90100000000080000003c00000010a58d957dd1a58dad95d17dc1c4003400000098bdc9ad7d91d5cdd189a5b9014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23100000003c00000098bdc9ad7d85c1c1b1a585b98d950100000000040000003c00000098bdc9ad7d85c1c1b1a585b98d95014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b231000000014000000c4c0cccc00000000000400000014000000c4c0cccc004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23005406000014000000080000004000000098bdc9ad7dddd5a1d585add585b99d0100000000040000004000000098bdc9ad7dddd5a1d585add585b99d014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b230c0000002c00000010a58d95b1a5b5a5d1950100000000040000002c00000010a58d95b1a5b5a5d195014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23040000003400000098bdc9ad7d91d5cdd189a5b90100000000040000003400000098bdc9ad7d91d5cdd189a5b9014c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23180000004800000010a58d957dd1a58dad95d17dc1c8b0ccc00000000000040000003c00000010a58d957dd1a58dad95d17dc1c8004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23180000004800000010a58d957dd1a58dad95d17dc1c8b0ccc00000000000040000003c00000010a58d957dd1a58dad95d17dc1c8004c0000000c85c99141bdbdb17d0da185c9858dd195c9010000f3095f138b23" + }, + { + "type": "packet", + "label": "limited-bitpacked-response", + "scenario": "protocol_sample", + "timestamp": 15.0, + "protocol": "udp", + "src_ip": "198.51.100.20", + "dst_ip": "192.0.2.10", + "src_port": 30031, + "dst_port": 49152, + "payload_hex": "15000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000080c9000080020000000300008007000000b337b9b52f28b9b7b530b9bc373700000000000100008007000000a2b4b1b22fbab4b1b532ba2f98188007000000b337b9b52f28b9b7b530b9bc37378009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b62710400000080020000801898981c000000008000000080020000801898981c8009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b6271840000008006000000b337b9b52fb2ba393ab1343700000000000100008007000000a2b4b1b22fbab4b1b532ba2f98188006000000b337b9b52fb2ba393ab134378009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b6271040200008007000000b337b9b5af303838b6b430b7b13200000000800000008007000000b337b9b5af303838b6b430b7b1328009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b62710402000080020000801898991900000000800000008002000080189899198009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b62710480ca000080020000000100000008000000b337b9b5afbb3ab4bab0b5ba30b73300000000800000000008000000b337b9b5afbb3ab4bab0b5ba30b7338009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b6271840100008005000000a2b4b132b6b4b634ba3200000000800000008005000000a2b4b132b6b4b634ba328009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b6271840000008006000000b337b9b52fb2ba393ab1343700000000800000008006000000b337b9b52fb2ba393ab134378009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b6271040300000009000000a2b4b1b22fbab4b1b532ba2f181996191800000000800000008007000000a2b4b1b22fbab4b1b532ba2f18198009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b6271040300000009000000a2b4b1b22fbab4b1b532ba2f181996191800000000800000008007000000a2b4b1b22fbab4b1b532ba2f18198009000080a1303932a8b737b6af21b430b9b031ba32390000603ee16b627104" + } + ] +} + diff --git a/tests/support.py b/tests/support.py new file mode 100644 index 0000000..d64467d --- /dev/null +++ b/tests/support.py @@ -0,0 +1,93 @@ +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" +FIXTURES = ROOT / "tests" / "fixtures" +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 decode_single_record(record_hex): + return decode_response_records(bytes(0x50) + bytes.fromhex(record_hex))[0] + + +def load_network_fixture(): + path = FIXTURES / "synthetic_history_network.json" + with path.open(encoding="utf-8") as f: + return json.load(f) + + +def fixture_packets(scenario=None): + fixture = load_network_fixture() + packets = fixture["packets"] + if scenario is not None: + packets = [packet for packet in packets if packet["scenario"] == scenario] + return [ + UdpPacket( + timestamp=packet["timestamp"], + src_ip=packet["src_ip"], + dst_ip=packet["dst_ip"], + src_port=packet["src_port"], + dst_port=packet["dst_port"], + payload=bytes.fromhex(packet["payload_hex"]), + protocol=packet["protocol"], + ) + for packet in packets + ] + + +def fixture_payload(label): + fixture = load_network_fixture() + packet = next(packet for packet in fixture["packets"] if packet["label"] == label) + return bytes.fromhex(packet["payload_hex"]) + + +def fixture_session(): + fixture = load_network_fixture() + session = LiveHistorySession(fixture["local_ip"]) + for packet in fixture_packets("replay"): + session.process_packet(packet) + return session diff --git a/tests/test_arc_decoding.py b/tests/test_arc_decoding.py new file mode 100644 index 0000000..4a422d9 --- /dev/null +++ b/tests/test_arc_decoding.py @@ -0,0 +1,120 @@ +from tests.support import * # noqa: F401,F403 + + +class ArcDecodingTests(unittest.TestCase): + def test_arc_key_timestamp_and_uid_match_fixture(self): + fixture = load_network_fixture() + row = fixture_session().build_rows("arc_miracle_box")[0] + self.assertEqual(decode_arc_key(bytes.fromhex(row["reward_key_hex"])), "fork_nonos") + _ticks, _unix, decoded = decode_arc_timestamp(bytes.fromhex(row["timestamp_raw_hex"])) + self.assertEqual(decoded, "2030-01-02 00:00:00") + self.assertEqual(row["uid"], fixture["expected"]["arc_first_uid"]) + + def test_arc_response_parser_matches_fixture_first_page(self): + decoded = parse_arc_response(fixture_payload("arc-page-1-response")) + self.assertEqual(len(decoded), 5) + self.assertEqual([row["reward_id"] for row in decoded], ["fork_nonos"] * 5) + self.assertEqual(decoded[0]["reward_type"], "arc") + self.assertEqual(decode_arc_key(bytes.fromhex(decoded[0]["reward_key_hex"])), "fork_nonos") + + 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): + decoded = fixture_session().build_rows("arc_miracle_box") + 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): + pairs = fixture_session().pairs_for_kind("arc_miracle_box") + full_pairs = pairs[:2] + trunc_pairs = pairs[:1] + ts = parse_arc_response(fixture_payload("arc-page-1-response"))[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): + response = fixture_payload("arc-page-1-response") + 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): + rows = fixture_session().build_rows("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)) diff --git a/tests/test_boundary_export.py b/tests/test_boundary_export.py deleted file mode 100644 index 7d82cc9..0000000 --- a/tests/test_boundary_export.py +++ /dev/null @@ -1,1046 +0,0 @@ -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() diff --git a/tests/test_capture_backends.py b/tests/test_capture_backends.py new file mode 100644 index 0000000..d768c70 --- /dev/null +++ b/tests/test_capture_backends.py @@ -0,0 +1,92 @@ +from tests.support import * # noqa: F401,F403 +from nte_history_exporter.live_capture.libpcap import _extract_ipv4_frame, _load_library + + +class CaptureBackendTests(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) diff --git a/tests/test_export_contract.py b/tests/test_export_contract.py new file mode 100644 index 0000000..24cc919 --- /dev/null +++ b/tests/test_export_contract.py @@ -0,0 +1,108 @@ +from tests.support import * # noqa: F401,F403 +from nte_history_exporter import __version__ + + +class ExportContractTests(unittest.TestCase): + def test_sanitized_export_omits_raw_packet_fields(self): + annotated = annotate_groups(fixture_session().build_rows("permanent")) + 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): + annotated = annotate_groups(fixture_session().build_rows("permanent")) + 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) diff --git a/tests/test_live_session.py b/tests/test_live_session.py new file mode 100644 index 0000000..b92cc92 --- /dev/null +++ b/tests/test_live_session.py @@ -0,0 +1,214 @@ +from tests.support import * # noqa: F401,F403 + + +class LiveSessionTests(unittest.TestCase): + 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) + diff --git a/tests/test_mappings_and_updates.py b/tests/test_mappings_and_updates.py new file mode 100644 index 0000000..4b9cb35 --- /dev/null +++ b/tests/test_mappings_and_updates.py @@ -0,0 +1,122 @@ +from tests.support import * # noqa: F401,F403 + + +class MappingAndUpdateTests(unittest.TestCase): + 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(""), "") + diff --git a/tests/test_network_fixture.py b/tests/test_network_fixture.py new file mode 100644 index 0000000..1a4be10 --- /dev/null +++ b/tests/test_network_fixture.py @@ -0,0 +1,74 @@ +from tests.support import * # noqa: F401,F403 + + +class SyntheticNetworkFixtureTests(unittest.TestCase): + def test_fixture_contains_only_synthetic_network_identity(self): + fixture = load_network_fixture() + + self.assertTrue(fixture["privacy"]["synthetic"]) + self.assertFalse(fixture["privacy"]["contains_user_uid"]) + self.assertFalse(fixture["privacy"]["contains_raw_account_session"]) + self.assertEqual(fixture["local_ip"], "192.0.2.10") + + allowed_ips = {"192.0.2.10", "198.51.100.20"} + for packet, decoded in zip(fixture["packets"], fixture_packets()): + with self.subTest(label=packet["label"]): + self.assertIn(packet["src_ip"], allowed_ips) + self.assertIn(packet["dst_ip"], allowed_ips) + self.assertEqual(packet["protocol"], "udp") + self.assertIsNone(extract_user_uid(decoded.payload)) + + def test_fixture_replays_permanent_and_arc_history(self): + fixture = load_network_fixture() + session = fixture_session() + + permanent = annotate_groups(session.build_rows("permanent")) + arc = session.build_rows("arc_miracle_box") + + self.assertIsNone(session.user_uid) + self.assertEqual(len(permanent), fixture["expected"]["permanent_records"]) + self.assertEqual(len(arc), fixture["expected"]["arc_records"]) + self.assertEqual(permanent[0]["uid"], fixture["expected"]["permanent_first_uid"]) + self.assertEqual(arc[0]["uid"], fixture["expected"]["arc_first_uid"]) + self.assertEqual(sorted({row["page"] for row in permanent}), [1, 2, 3, 4, 5]) + self.assertEqual(sorted({row["page"] for row in arc}), [1, 2, 3, 4, 5]) + self.assertEqual({row["timestamp_decoded"] for row in permanent}, {"2030-01-01 00:00:00"}) + self.assertEqual({row["timestamp_decoded"] for row in arc}, {"2030-01-02 00:00:00"}) + + # Repeating one manufactured record prevents the replay transcript from + # preserving a real account's pull sequence while still exercising page + # ordering, timestamp ordinals, UID stability, and export behavior. + self.assertEqual(len({row["record_hex"] for row in permanent}), 1) + self.assertEqual(len({row["record_hex"] for row in arc}), 1) + + def test_fixture_protocol_samples_cover_edge_cases(self): + points = decode_response_records(fixture_payload("limited-points-gift-1")) + chase = decode_response_records(fixture_payload("limited-chase-reward")) + batched = decode_response_records(fixture_payload("limited-batched-pages")) + aligned = decode_response_records(fixture_payload("limited-bitpacked-source")) + bitpacked = decode_response_records(fixture_payload("limited-bitpacked-response")) + + self.assertEqual(len(points), 1) + self.assertEqual(points[0]["result_type"], "points_gift") + self.assertEqual(points[0]["reward_id"], "1020") + self.assertEqual(len(chase), 1) + self.assertEqual(chase[0]["result_type"], "chase_reward") + self.assertEqual(chase[0]["reward_id"], "Dice_ticket_01") + self.assertEqual(chase[0]["quantity"], 30) + self.assertEqual(len(batched), 10) + self.assertEqual(len(aligned), 10) + self.assertEqual( + [row["record_hex"] for row in bitpacked], + [row["record_hex"] for row in aligned], + ) + + def test_fixture_exports_sanitized_json_with_stable_uids(self): + session = fixture_session() + rows = annotate_groups(session.build_rows("permanent")) + export = build_export_json(rows, []) + + self.assertNotIn("user_uid", export) + self.assertEqual(export["format_version"], 1) + self.assertEqual(export["records"][0]["uid"], load_network_fixture()["expected"]["permanent_first_uid"]) + self.assertNotIn("record_hex", export["records"][0]) + self.assertNotIn("timestamp_raw_hex", export["records"][0]) diff --git a/tests/test_protocol_decoding.py b/tests/test_protocol_decoding.py new file mode 100644 index 0000000..22b1840 --- /dev/null +++ b/tests/test_protocol_decoding.py @@ -0,0 +1,125 @@ +from tests.support import * # noqa: F401,F403 + + +class ProtocolDecodingTests(unittest.TestCase): + 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_fixture_prefixed_points_gift_overrides_visible_dice(self): + decoded = decode_response_records(fixture_payload("limited-points-gift-1"))[0] + 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"], "1020") + + def test_fixture_prefixed_chase_reward_overrides_visible_dice_and_quantity(self): + decoded = decode_response_records(fixture_payload("limited-chase-reward"))[0] + + 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) + + 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): + decoded = decode_response_records(fixture_payload("limited-batched-pages")) + + self.assertEqual(len(decoded), 10) + self.assertEqual(decoded[5]["dice"], 4) + self.assertEqual(decoded[5]["result_type"], "dice") + + def test_monopoly_response_parser_realigns_bit_packed_payload(self): + reference_rows = decode_response_records(fixture_payload("limited-bitpacked-source")) + decoded = decode_response_records(fixture_payload("limited-bitpacked-response")) + + 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]) diff --git a/tests/test_uid_compatibility.py b/tests/test_uid_compatibility.py new file mode 100644 index 0000000..ed7d3af --- /dev/null +++ b/tests/test_uid_compatibility.py @@ -0,0 +1,153 @@ +from tests.support import * # noqa: F401,F403 + + +class UidCompatibilityTests(unittest.TestCase): + def test_uid_source_matches_committed_network_fixture(self): + fixture = load_network_fixture() + rows = annotate_groups(fixture_session().build_rows("permanent")) + self.assertEqual(rows[0]["uid"], fixture["expected"]["permanent_first_uid"]) + + 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): + annotated = annotate_groups(fixture_session().build_rows("permanent")) + + 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_committed_network_scan_exports_all_rows(self): + fixture = load_network_fixture() + annotated = annotate_groups(fixture_session().build_rows("permanent")) + exported = [row for row in annotated if row["export_record"] is True] + + self.assertEqual(len(annotated), fixture["expected"]["permanent_records"]) + self.assertEqual(len(exported), fixture["expected"]["permanent_records"]) + + def test_monopoly_uid_compatibility_vectors_are_frozen(self): + vectors = [ + ("Lottery_Permanent", "0000000000000000", 0, "5da57468adb23fcfde530fd849a25767"), + ("Lottery_Permanent", "00a243eb689b7a23", 0, "8f9fb9bc92a867f6f41e5839e94c7991"), + ("Lottery_Permanent", "00a243eb689b7a23", 1, "6a0bea71d091b9b7063780fcc86238d5"), + ("Lottery_LimitedCharacter", "00a243eb689b7a23", 0, "b8f5d3c27315c70ceafbe8eb2bab0fc2"), + ("Lottery_LimitedCharacter", "ffffffffffffffff", 9, "66916e1f077fdf7664f0f0edc4c64eb8"), + ] + + for pool_group_id, timestamp_raw_hex, ordinal, expected_uid in vectors: + with self.subTest( + pool_group_id=pool_group_id, + timestamp=timestamp_raw_hex, + ordinal=ordinal, + ): + row = { + "pool_group_id": pool_group_id, + "timestamp_raw_hex": timestamp_raw_hex, + } + self.assertEqual(make_uid(row, ordinal), expected_uid) + + def test_arc_uid_compatibility_vectors_are_frozen(self): + vectors = [ + ("0000000000000000", 0, "faa0b07cefd71758d4a8d64bc2a2f78e"), + ("00a243eb689b7a23", 0, "7a7ef75281fcced043f691d269477cfa"), + ("00a243eb689b7a23", 1, "f798c076ea1d137a08fe62df859e8a7f"), + ("ffffffffffffffff", 9, "d114cc6c313a15f55ca12910801d78ea"), + ] + + for timestamp_raw_hex, ordinal, expected_uid in vectors: + with self.subTest(timestamp=timestamp_raw_hex, ordinal=ordinal): + self.assertEqual(make_arc_uid(timestamp_raw_hex, ordinal), expected_uid)