Files

21 lines
675 B
Python

import unittest
from pi_dictate_service.protocol import MAX_FRAME_BYTES, FrameDecoder
class FrameDecoderRegressionTests(unittest.TestCase):
def test_accepts_pipelined_bounded_frames_larger_than_aggregate_limit(self) -> None:
decoder = FrameDecoder()
frame = b'{"version":2,"type":"ping","requestId":"request"}\n'
payload = frame * ((MAX_FRAME_BYTES // len(frame)) + 1)
self.assertGreater(len(payload), MAX_FRAME_BYTES + 1)
decoded = decoder.feed(payload)
self.assertEqual(len(decoded), payload.count(b"\n"))
self.assertEqual(decoded[0]["requestId"], "request")
if __name__ == "__main__":
unittest.main()