# Copyright 2024 Google LLC # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. from __future__ import absolute_import import unittest from pebble.pulse2 import framing class TestEncodeFrame(unittest.TestCase): def test_empty_frame(self): # CRC-32 of nothing is 0 # COBS encoding of b'\0\0\0\0' is b'\x01\x01\x01\x01\x01' (5 bytes) self.assertEqual(framing.encode_frame(b''), b'\x55\x01\x01\x01\x01\x01\x55') def test_simple_data(self): self.assertEqual(framing.encode_frame(b'abcdefg'), b'\x55\x0cabcdefg\xa6\x6a\x2a\x31\x55') def test_flag_in_datagram(self): # ASCII 'U' is 0x55 hex self.assertEqual(framing.encode_frame(b'QUACK'), b'\x55\x0aQ\0ACK\xdf\x8d\x80\x74\x55') def test_flag_in_fcs(self): # crc32(b'R') -> 0x5767df55 # Since there is an \x55 byte in the FCS, it must be substituted, # just like when that byte value is present in the datagram itself. self.assertEqual(framing.encode_frame(b'R'), b'\x55\x06R\0\xdf\x67\x57\x55') class TestFrameSplitter(unittest.TestCase): def setUp(self): self.splitter = framing.FrameSplitter() def test_basic_functionality(self): self.splitter.write(b'\x55abcdefg\x55foobar\x55asdf\x55') self.assertEqual(list(self.splitter), [b'abcdefg', b'foobar', b'asdf']) def test_wait_for_sync(self): self.splitter.write(b'garbage data\x55frame 1\x55') self.assertEqual(list(self.splitter), [b'frame 1']) def test_doubled_flags(self): self.splitter.write(b'\x55abcd\x55\x55efgh\x55') self.assertEqual(list(self.splitter), [b'abcd', b'efgh']) def test_multiple_writes(self): self.splitter.write(b'\x55ab') self.assertEqual(list(self.splitter), []) self.splitter.write(b'cd\x55') self.assertEqual(list(self.splitter), [b'abcd']) def test_lots_of_writes(self): for char in b'\x55abcd\x55ef': self.splitter.write(bytearray([char])) self.assertEqual(list(self.splitter), [b'abcd']) def test_iteration_pops_frames(self): self.splitter.write(b'\x55frame 1\x55frame 2\x55frame 3\x55') self.assertEqual(next(iter(self.splitter)), b'frame 1') self.assertEqual(list(self.splitter), [b'frame 2', b'frame 3']) def test_stopiteration_latches(self): # The iterator protocol requires that once an iterator raises # StopIteration, it must continue to do so for all subsequent calls # to its next() method. self.splitter.write(b'\x55frame 1\x55') iterator = iter(self.splitter) self.assertEqual(next(iterator), b'frame 1') with self.assertRaises(StopIteration): next(iterator) next(iterator) self.splitter.write(b'\x55frame 2\x55') with self.assertRaises(StopIteration): next(iterator) self.assertEqual(list(self.splitter), [b'frame 2']) def test_max_frame_length(self): splitter = framing.FrameSplitter(max_frame_length=6) splitter.write( b'\x5512345\x55123456\x551234567\x551234\x5512345678\x55') self.assertEqual(list(splitter), [b'12345', b'123456', b'1234']) def test_dynamic_max_length_1(self): self.splitter.write(b'\x5512345') self.splitter.max_frame_length = 6 self.splitter.write(b'6\x551234567\x551234\x55') self.assertEqual(list(self.splitter), [b'123456', b'1234']) def test_dynamic_max_length_2(self): self.splitter.write(b'\x551234567') self.splitter.max_frame_length = 6 self.splitter.write(b'89\x55123456\x55') self.assertEqual(list(self.splitter), [b'123456']) class TestDecodeTransparency(unittest.TestCase): def test_easy_decode(self): self.assertEqual(framing.decode_transparency(b'\x06abcde'), b'abcde') def test_escaped_flag(self): self.assertEqual(framing.decode_transparency(b'\x06Q\0ACK'), b'QUACK') def test_flag_byte_in_frame(self): with self.assertRaises(framing.DecodeError): framing.decode_transparency(b'\x06ab\x55de') def test_truncated_cobs_block(self): with self.assertRaises(framing.DecodeError): framing.decode_transparency(b'\x0aabc') class TestStripFCS(unittest.TestCase): def test_frame_too_short(self): with self.assertRaises(framing.CorruptFrame): framing.strip_fcs(b'abcd') def test_good_fcs(self): self.assertEqual(framing.strip_fcs(b'abcd\x11\xcd\x82\xed'), b'abcd') def test_frame_corrupted(self): with self.assertRaises(framing.CorruptFrame): framing.strip_fcs(b'abce\x11\xcd\x82\xed') def test_fcs_corrupted(self): with self.assertRaises(framing.CorruptFrame): framing.strip_fcs(b'abcd\x13\xcd\x82\xed') class TestDecodeFrame(unittest.TestCase): def test_it_works(self): # Not much to test; decode_frame is just chained decode_transparency # with strip_fcs, and both of those have already been tested separately. self.assertEqual(framing.decode_frame(b'\x0aQ\0ACK\xdf\x8d\x80t'), b'QUACK')