test_inspect_mission_sdk.py 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. import hashlib
  2. from pathlib import Path
  3. import struct
  4. import sys
  5. import tempfile
  6. import unittest
  7. sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "tools"))
  8. from inspect_mission_sdk import Pack, SECTIONS, inspect_shard
  9. def empty_pack():
  10. data = bytearray(848)
  11. data[:8] = b"SRSDKP01"
  12. struct.pack_into("<IIQ", data, 8, 37, 848, len(data))
  13. struct.pack_into("<I", data, 152, 43)
  14. data[24:56] = hashlib.sha256(b"").digest()
  15. for index, stride in SECTIONS.values():
  16. struct.pack_into("<QII", data, 160 + index * 16, 848, 0, stride)
  17. return data
  18. class InspectionTests(unittest.TestCase):
  19. def setUp(self):
  20. self.directory = tempfile.TemporaryDirectory()
  21. self.addCleanup(self.directory.cleanup)
  22. self.path = Path(self.directory.name) / "fixture.pack"
  23. def open(self, data):
  24. self.path.write_bytes(data)
  25. return Pack(self.path)
  26. def test_absent_scenario_does_not_match_another(self):
  27. pack = self.open(empty_pack())
  28. try:
  29. with self.assertRaisesRegex(ValueError, "resolve exactly once"):
  30. pack.inspect(0x80800011)
  31. finally:
  32. pack.close()
  33. def test_version_and_truncation(self):
  34. data = empty_pack()
  35. struct.pack_into("<I", data, 8, 36)
  36. with self.assertRaisesRegex(ValueError, "unsupported"):
  37. self.open(data)
  38. with self.assertRaises(ValueError):
  39. self.open(empty_pack()[:-1])
  40. def test_payload_corruption(self):
  41. data = empty_pack() + b"payload"
  42. struct.pack_into("<Q", data, 16, len(data))
  43. with self.assertRaisesRegex(ValueError, "checksum"):
  44. self.open(data)
  45. def test_section_outside_file(self):
  46. data = empty_pack()
  47. struct.pack_into("<QII", data, 160 + 2 * 16, 848, 1, 48)
  48. with self.assertRaisesRegex(ValueError, "bounds"):
  49. self.open(data)
  50. def test_wrong_row_stride(self):
  51. data = empty_pack()
  52. struct.pack_into("<I", data, 160 + 2 * 16 + 12, 44)
  53. with self.assertRaisesRegex(ValueError, "stride"):
  54. self.open(data)
  55. def test_shard_identity_and_payload(self):
  56. data = bytearray(664)
  57. data[:8] = b"SRGWSHRD"
  58. struct.pack_into("<IIQII", data, 8, 13, 664, len(data), 0x80800011, 35)
  59. data[72:104] = hashlib.sha256(b"").digest()
  60. self.path.write_bytes(data)
  61. result = inspect_shard(self.path, 0x80800011)
  62. self.assertEqual(sum(result["authored_squad_context_counts"].values()), 0)
  63. with self.assertRaisesRegex(ValueError, "mismatched"):
  64. inspect_shard(self.path, 0x80800012)
  65. data[72] ^= 1
  66. self.path.write_bytes(data)
  67. with self.assertRaisesRegex(ValueError, "checksum"):
  68. inspect_shard(self.path, 0x80800011)
  69. if __name__ == "__main__":
  70. unittest.main()