clxud
43 lines
1.3 KiB
Python
43 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import msgpack
|
|
import zstandard as zstd
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from spectra_ws_payload import SpectraPayloadError, decode_spectra_ws_payload
|
|
|
|
|
|
def _zstd(payload: bytes) -> bytes:
|
|
return zstd.ZstdCompressor().compress(payload)
|
|
|
|
|
|
class SpectraWsPayloadTests(unittest.TestCase):
|
|
def test_decode_zstd_json_object(self) -> None:
|
|
replay = {"completed": [{"_id": 123, "players": {"1": {"nick": "A"}}}]}
|
|
|
|
self.assertEqual(decode_spectra_ws_payload(_zstd(json.dumps(replay).encode("utf-8"))), replay)
|
|
|
|
def test_decode_zstd_msgpack_object(self) -> None:
|
|
replay = {"completed": [{"_id": 123, "players": {"1": {"nick": "A"}}}]}
|
|
|
|
self.assertEqual(decode_spectra_ws_payload(_zstd(msgpack.packb(replay, use_bin_type=True))), replay)
|
|
|
|
def test_decode_plain_json_text_frame(self) -> None:
|
|
replay = {"_id": "abc", "teams": []}
|
|
|
|
self.assertEqual(decode_spectra_ws_payload(json.dumps(replay)), replay)
|
|
|
|
def test_unknown_payload_raises_clear_error(self) -> None:
|
|
with self.assertRaisesRegex(SpectraPayloadError, "neither JSON nor msgpack"):
|
|
decode_spectra_ws_payload(_zstd(b"not-json-or-msgpack"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|