262 lines
10 KiB
Python
262 lines
10 KiB
Python
from tests.support import * # noqa: F401,F403
|
||
from nte_history_exporter import __version__
|
||
from nte_history_exporter.constants import EXPORTER_VERSION
|
||
from nte_history_exporter.decoder.achievement import AchievementRecord
|
||
from nte_history_exporter.export.json_export import build_achievement_export_json
|
||
from nte_history_exporter.live_capture.runner import _achievement_path
|
||
|
||
|
||
class ExportContractTests(unittest.TestCase):
|
||
def test_public_version_references_match(self):
|
||
self.assertEqual(__version__, "0.3.0")
|
||
self.assertEqual(EXPORTER_VERSION, __version__)
|
||
|
||
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("server_id", export)
|
||
self.assertNotIn("account_region", 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_export_includes_server_id_and_mapped_account_region(self):
|
||
annotated = annotate_groups(fixture_session().build_rows("permanent"))
|
||
export = build_export_json(annotated, [], server_id="23003")
|
||
|
||
self.assertEqual(export["server_id"], "23003")
|
||
self.assertEqual(export["account_region"], "EU")
|
||
|
||
def test_export_preserves_unknown_server_without_guessing_region(self):
|
||
annotated = annotate_groups(fixture_session().build_rows("permanent"))
|
||
export = build_export_json(annotated, [], server_id="23999")
|
||
|
||
self.assertEqual(export["server_id"], "23999")
|
||
self.assertNotIn("account_region", export)
|
||
|
||
def test_extracts_server_id_from_valid_initial_tcp_response(self):
|
||
payload = bytearray(204)
|
||
payload[0:4] = (200).to_bytes(4, "little")
|
||
payload[4:8] = (20).to_bytes(4, "little")
|
||
payload[96:100] = (23003).to_bytes(4, "little")
|
||
address = b"198.51.100.20"
|
||
payload[132:136] = len(address).to_bytes(4, "little")
|
||
payload[136 : 136 + len(address)] = address
|
||
|
||
self.assertEqual(extract_server_id(bytes(payload)), "23003")
|
||
|
||
def test_rejects_server_id_at_offset_without_valid_message_structure(self):
|
||
payload = bytearray(204)
|
||
payload[96:100] = (23003).to_bytes(4, "little")
|
||
|
||
self.assertIsNone(extract_server_id(bytes(payload)))
|
||
|
||
def test_live_session_detects_server_only_on_inbound_tcp(self):
|
||
payload = bytearray(204)
|
||
payload[0:4] = (200).to_bytes(4, "little")
|
||
payload[4:8] = (20).to_bytes(4, "little")
|
||
payload[96:100] = (23004).to_bytes(4, "little")
|
||
address = b"198.51.100.20"
|
||
payload[132:136] = len(address).to_bytes(4, "little")
|
||
payload[136 : 136 + len(address)] = address
|
||
session = LiveHistorySession("192.0.2.10")
|
||
|
||
session.process_packet(
|
||
UdpPacket(
|
||
1.0,
|
||
"198.51.100.20",
|
||
"192.0.2.10",
|
||
30000,
|
||
40000,
|
||
bytes(payload),
|
||
protocol="tcp",
|
||
)
|
||
)
|
||
|
||
self.assertEqual(session.server_id, "23004")
|
||
|
||
@patch("builtins.input", side_effect=["9", "3"])
|
||
def test_server_prompt_retries_then_returns_selected_server_id(self, _input):
|
||
self.assertEqual(console.prompt_server_id(), "23003")
|
||
|
||
@patch("builtins.input", return_value="")
|
||
def test_server_prompt_can_be_skipped(self, _input):
|
||
self.assertIsNone(console.prompt_server_id())
|
||
|
||
@patch("nte_history_exporter.console.wait_for_keypress")
|
||
def test_wait_for_close_waits_for_a_second_keypress(self, wait_for_keypress):
|
||
console.wait_for_close()
|
||
|
||
wait_for_keypress.assert_called_once_with()
|
||
|
||
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_achievement_path_includes_user_uid_and_timestamp(self):
|
||
path = _achievement_path("218216016349")
|
||
|
||
self.assertRegex(
|
||
path.name,
|
||
r"^218216016349_Achievements_\d{8}_\d{6}(?:_\d+)?\.json$",
|
||
)
|
||
|
||
def test_achievement_export_has_versioned_metadata_and_records(self):
|
||
export = build_achievement_export_json(
|
||
[
|
||
AchievementRecord("Battle_30", 10, 639_136_215_846_370_000),
|
||
AchievementRecord("Battle_25", 20, 0),
|
||
AchievementRecord("Playstation_017", 0, 0),
|
||
],
|
||
source="live_capture",
|
||
capture_source="npcap",
|
||
user_uid="218216016349",
|
||
server_id="23003",
|
||
)
|
||
|
||
self.assertEqual(export["format"], "nte-achievement-export")
|
||
self.assertEqual(export["format_version"], 1)
|
||
self.assertEqual(export["capture_source"], "npcap")
|
||
self.assertEqual(export["user_uid"], "218216016349")
|
||
self.assertEqual(export["account_region"], "EU")
|
||
self.assertEqual(
|
||
export["scan"],
|
||
{
|
||
"in_game": {
|
||
"total_achievements": 2,
|
||
"completed_achievements": 1,
|
||
"in_progress_achievements": 1,
|
||
},
|
||
"playstation": {
|
||
"total_achievements": 1,
|
||
"completed_achievements": 0,
|
||
"in_progress_achievements": 1,
|
||
},
|
||
},
|
||
)
|
||
battle_30, battle_25 = export["categories"]["battle"]
|
||
self.assertEqual(battle_30["name"], "Death Nova I")
|
||
self.assertEqual(battle_30["completed_at"], "2026-05-05 23:46:24")
|
||
self.assertEqual(battle_25["name"], "Devil Within II")
|
||
self.assertEqual(battle_25["description"], "Trigger Hexed ×50.")
|
||
self.assertEqual(battle_25["progress"], 20)
|
||
self.assertEqual(battle_25["target"], 50)
|
||
self.assertEqual(battle_25["quality"], "high")
|
||
self.assertEqual(
|
||
battle_25["rewards"], [{"item_id": "Annulith", "amount": 10}]
|
||
)
|
||
self.assertEqual(
|
||
export["categories"]["playstation"][0]["name"], "Speed Above All"
|
||
)
|
||
|
||
def test_unmapped_achievement_still_exports_capture_data(self):
|
||
export = build_achievement_export_json(
|
||
[AchievementRecord("FutureCategory_999", 7, 0)]
|
||
)
|
||
|
||
self.assertEqual(
|
||
export["categories"]["futurecategory"],
|
||
[
|
||
{
|
||
"id": "FutureCategory_999",
|
||
"platform": "in_game",
|
||
"status": "in_progress",
|
||
"progress": 7,
|
||
"completed": False,
|
||
"completed_at": None,
|
||
}
|
||
],
|
||
)
|
||
|
||
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)
|