pw_hdlc_lite/py/decode_test.py - pigweed/pigweed - Git at Google

 #!/usr/bin/env python
 # Copyright 2020 The Pigweed Authors
 #
 # 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
 #
 #     https://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.
 """Contains the Python decoder tests and generates C++ decoder tests."""

 from collections import defaultdict
 from pathlib import Path
 from typing import Any, Callable, Dict, Iterator, List, NamedTuple, Union
 from typing import Optional, TextIO, Tuple
 import unittest
 import sys

 from pw_hdlc_lite.decode import Frame, FrameDecoder, FrameStatus, NO_ADDRESS
 from pw_hdlc_lite.protocol import frame_check_sequence as fcs


 def _encode(address: int, control: int, data: bytes) -> bytes:
     frame = bytearray([address, control]) + data
     frame += fcs(frame)
     frame = frame.replace(b'\x7d', b'\x7d\x5d')
     frame = frame.replace(b'\x7e', b'\x7d\x5e')
     return b''.join([b'\x7e', frame, b'\x7e'])


 class Expected(NamedTuple):
     address: int
     control: bytes
     data: bytes
     status: FrameStatus = FrameStatus.OK

     def __eq__(self, other) -> bool:
         """Define == so an Expected and a Frame can be compared."""
         return (self.address == other.address and self.control == other.control
                 and self.data == other.data and self.status is other.status)


 _PARTIAL = fcs(b'\x0ACmsg\x5e')
 _ESCAPED_FLAG_TEST_CASE = (
     b'\x7e\x0ACmsg\x7d\x7e' + _PARTIAL + b'\x7e',
     [
         Expected(0xA, b'C', b'', FrameStatus.INCOMPLETE),
         Expected(_PARTIAL[0], _PARTIAL[1:2], b'', FrameStatus.INCOMPLETE),
     ],
 )

 TestCase = Tuple[bytes, List[Expected]]
 TestCases = Tuple[Union[str, TestCase], ...]

 TEST_CASES: TestCases = (
     'Empty payload',
     (_encode(0, 0, b''), [Expected(0, b'\0', b'')]),
     (_encode(55, 0x99, b''), [Expected(55, b'\x99', b'')]),
     (_encode(55, 0x99, b'') * 3, [Expected(55, b'\x99', b'')] * 3),
     'Simple one-byte payload',
     (_encode(0, 0, b'\0'), [Expected(0, b'\0', b'\0')]),
     (_encode(123, 0, b'A'), [Expected(123, b'\0', b'A')]),
     'Simple multi-byte payload',
     (_encode(0, 0, b'Hello, world!'), [Expected(0, b'\0', b'Hello, world!')]),
     (_encode(123, 0, b'\0\0\1\0\0'), [Expected(123, b'\0', b'\0\0\1\0\0')]),
     'Escaped one-byte payload',
     (_encode(1, 2, b'\x7e'), [Expected(1, b'\2', b'\x7e')]),
     (_encode(1, 2, b'\x7d'), [Expected(1, b'\2', b'\x7d')]),
     (_encode(1, 2, b'\x7e') + _encode(1, 2, b'\x7d'),
      [Expected(1, b'\2', b'\x7e'),
       Expected(1, b'\2', b'\x7d')]),
     'Escaped address',
     (_encode(0x7e, 0, b'A'), [Expected(0x7e, b'\0', b'A')]),
     (_encode(0x7d, 0, b'B'), [Expected(0x7d, b'\0', b'B')]),
     'Escaped control',
     (_encode(0, 0x7e, b'C'), [Expected(0, b'\x7e', b'C')]),
     (_encode(0, 0x7d, b'D'), [Expected(0, b'\x7d', b'D')]),
     'Escaped address and control',
     (_encode(0x7e, 0x7d, b'E'), [Expected(0x7e, b'\x7d', b'E')]),
     (_encode(0x7d, 0x7e, b'F'), [Expected(0x7d, b'\x7e', b'F')]),
     (_encode(0x7e, 0x7e, b'\x7e'), [Expected(0x7e, b'\x7e', b'\x7e')]),
     'Multiple frames separated by single flag',
     (_encode(0, 0, b'A')[:-1] + _encode(1, 2, b'123'),
      [Expected(0, b'\0', b'A'),
       Expected(1, b'\2', b'123')]),
     (_encode(0xff, 0, b'Yo')[:-1] * 3 + b'\x7e',
      [Expected(0xff, b'\0', b'Yo')] * 3),
     'Ignore empty frames',
     (b'\x7e\x7e', []),
     (b'\x7e' * 10, []),
     (b'\x7e\x7e' + _encode(1, 2, b'3') + b'\x7e' * 5,
      [Expected(1, b'\2', b'3')]),
     (b'\x7e' * 10 + _encode(1, 2, b':O') + b'\x7e' * 3 + _encode(3, 4, b':P'),
      [Expected(1, b'\2', b':O'),
       Expected(3, b'\4', b':P')]),
     'Cannot escape flag',
     (b'\x7e\xAA\x7d\x7e\xab\x00Hello' + fcs(b'\xab\0Hello') + b'\x7e', [
         Expected(0xAA, b'', b'', FrameStatus.INCOMPLETE),
         Expected(0xab, b'\0', b'Hello'),
     ]),
     _ESCAPED_FLAG_TEST_CASE,
     'Frame too short',
     (b'\x7e1\x7e', [Expected(ord('1'), b'', b'', FrameStatus.INCOMPLETE)]),
     (b'\x7e12\x7e', [Expected(ord('1'), b'2', b'', FrameStatus.INCOMPLETE)]),
     (b'\x7e12345\x7e', [Expected(ord('1'), b'2', b'',
                                  FrameStatus.INCOMPLETE)]),
     'Incorrect frame check sequence',
     (b'\x7e123456\x7e',
      [Expected(ord('1'), b'2', b'', FrameStatus.FCS_MISMATCH)]),
     (b'\x7e\1\2msg\xff\xff\xff\xff\x7e',
      [Expected(0x1, b'\2', b'msg', FrameStatus.FCS_MISMATCH)]),
     (_encode(0xA, 0xB, b'???')[:-2] + _encode(1, 2, b'def'), [
         Expected(0xA, b'\x0B', b'??', FrameStatus.FCS_MISMATCH),
         Expected(1, b'\2', b'def'),
     ]),
     'Invalid escape in address',
     (b'\x7e\x7d\x7d\0' + fcs(b'\x5d\0') + b'\x7e',
      [Expected(0,
                fcs(b'\x5d\0')[0:1], b'', FrameStatus.INVALID_ESCAPE)]),
     'Invalid escape in control',
     (b'\x7e\0\x7d\x7d' + fcs(b'\0\x5d') + b'\x7e',
      [Expected(0,
                fcs(b'\0\x5d')[0:1], b'', FrameStatus.INVALID_ESCAPE)]),
     'Invalid escape in data',
     (b'\x7e\0\1\x7d\x7d' + fcs(b'\0\1\x5d') + b'\x7e',
      [Expected(0, b'\1', b'', FrameStatus.INVALID_ESCAPE)]),
     'Frame ends with escape',
     (b'\x7e\x7d\x7e', [Expected(NO_ADDRESS, b'', b'',
                                 FrameStatus.INCOMPLETE)]),
     (b'\x7e\1\x7d\x7e', [Expected(1, b'', b'', FrameStatus.INCOMPLETE)]),
     (b'\x7e\1\2abc\x7d\x7e', [Expected(1, b'\2', b'',
                                        FrameStatus.INCOMPLETE)]),
     (b'\x7e\1\2abcd\x7d\x7e',
      [Expected(1, b'\2', b'', FrameStatus.INCOMPLETE)]),
     (b'\x7e\1\2abcd1234\x7d\x7e',
      [Expected(1, b'\2', b'abcd', FrameStatus.INCOMPLETE)]),
     'Inter-frame data is only escapes',
     (b'\x7e\x7d\x7e\x7d\x7e', [
         Expected(NO_ADDRESS, b'', b'', FrameStatus.INCOMPLETE),
         Expected(NO_ADDRESS, b'', b'', FrameStatus.INCOMPLETE),
     ]),
     (b'\x7e\x7d\x7d\x7e\x7d\x7d\x7e', [
         Expected(NO_ADDRESS, b'', b'', FrameStatus.INVALID_ESCAPE),
         Expected(NO_ADDRESS, b'', b'', FrameStatus.INVALID_ESCAPE),
     ]),
     'Data before first flag',
     (b'\0\1' + fcs(b'\0\1'), []),
     (b'\0\1' + fcs(b'\0\1') + b'\x7e',
      [Expected(0, b'\1', b'', FrameStatus.INCOMPLETE)]),
     'No frames emitted until flag',
     (_encode(1, 2, b'3')[:-1], []),
     (b'\x7e' + _encode(1, 2, b'3')[1:-1] * 2, []),
 )  # yapf: disable
 # Formatting for the above tuple is very slow, so disable yapf.


 def _sort_test_cases(test_cases: TestCases) -> Dict[str, List[TestCase]]:
     cases: Dict[str, List[TestCase]] = defaultdict(list)
     message = ''

     for case in test_cases:
         if isinstance(case, str):
             message = case
         else:
             cases[message].append(case)

     return cases


 def _expected(frames: List[Frame]) -> Iterator[str]:
     for i, frame in enumerate(frames, 1):
         if frame.ok():
             yield f'      Frame(kDecodedFrame{i:02}),'
         else:
             yield f'      Status::DATA_LOSS,  // Frame {i}'


 _CPP_HEADER = f"""\
 // Copyright 2020 The Pigweed Authors
 //
 // 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
 //
 //     https://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.

 // AUTOGENERATED by {Path(__file__).name} - DO NOT EDIT

 #include "pw_hdlc_lite/decoder.h"

 #include <array>
 #include <cstddef>
 #include <variant>

 #include "gtest/gtest.h"
 #include "pw_bytes/array.h"

 namespace pw::hdlc_lite {{
 namespace {{

 // clang-format off
 """

 _CPP_FOOTER = """\
 // clang-format on

 }  // namespace
 }  // namespace pw::hdlc_lite
 """


 def _cpp_test(group: str, count: Optional[int], data: bytes) -> Iterator[str]:
     """Generates a C++ test for the provided test data."""
     frames = list(FrameDecoder().process(data))
     data_bytes = ''.join(rf'\x{byte:02x}' for byte in data)

     name = ''.join(w.capitalize() for w in group.replace('-', ' ').split(' '))
     name = ''.join(c if c.isalnum() else '_' for c in name)
     name = name if count is None else name + f'_{count}'

     yield f'TEST(Decoder, {name}) {{'
     yield f'  static constexpr auto kData = bytes::String("{data_bytes}");\n'

     for i, frame in enumerate(frames, 1):
         if frame.status is FrameStatus.OK:
             frame_bytes = ''.join(rf'\x{byte:02x}' for byte in frame.raw)
             yield (f'  static constexpr auto kDecodedFrame{i:02} = '
                    f'bytes::String("{frame_bytes}");')
         else:
             yield f'  // Frame {i}: {frame.status.value}'

     yield ''

     expected = '\n'.join(_expected(frames)) or '      // No frames'
     decoder_size = max(len(data), 8)  # Make sure large enough for a frame

     yield f"""\
   DecoderBuffer<{decoder_size}> decoder;

   static constexpr std::array<std::variant<Frame, Status>, {len(frames)}> kExpected = {{
 {expected}
   }};

   size_t decoded_frames = 0;

   decoder.Process(kData, [&](const Result<Frame>& result) {{
     ASSERT_LT(decoded_frames++, kExpected.size());
     auto& expected = kExpected[decoded_frames - 1];

     if (std::holds_alternative<Status>(expected)) {{
       EXPECT_EQ(Status::DATA_LOSS, result.status());
     }} else {{
       ASSERT_EQ(Status::OK, result.status());

       const Frame& decoded_frame = result.value();
       const Frame& expected_frame = std::get<Frame>(expected);
       EXPECT_EQ(expected_frame.address(), decoded_frame.address());
       EXPECT_EQ(expected_frame.control(), decoded_frame.control());
       ASSERT_EQ(expected_frame.data().size(), decoded_frame.data().size());
       EXPECT_EQ(std::memcmp(expected_frame.data().data(),
                             decoded_frame.data().data(),
                             expected_frame.data().size()),
                 0);
     }}
   }});

   EXPECT_EQ(decoded_frames, kExpected.size());
 }}"""


 def _define_cc_tests(test_cases: TestCases, output: TextIO) -> None:
     """Writes C++ tests for all test cases."""
     output.write(_CPP_HEADER)

     for group, test_list in _sort_test_cases(test_cases).items():
         for i, (data, _) in enumerate(test_list, 1):
             count = i if len(test_list) > 1 else None
             for line in _cpp_test(group, count, data):
                 output.write(line)
                 output.write('\n')

             output.write('\n')

     output.write(_CPP_FOOTER)


 def _define_py_test(group: str,
                     data: bytes,
                     expected_frames: List[Expected],
                     count: int = None) -> Callable[[Any], None]:
     def test(self) -> None:
         # Decode in one call
         self.assertEqual(expected_frames,
                          list(FrameDecoder().process(data)),
                          msg=f'{group}: {data!r}')

         # Decode byte-by-byte
         decoder = FrameDecoder()
         decoded_frames = []
         for i in range(len(data)):
             decoded_frames += decoder.process(data[i:i + 1])

         self.assertEqual(expected_frames,
                          decoded_frames,
                          msg=f'{group} (byte-by-byte): {data!r}')

     name = 'test_' + ''.join(c if c.isalnum() else '_' for c in group.lower())
     test.__name__ = name if count is None else name + f'_{count}'
     return test


 def _define_py_tests(test_cases: TestCases = TEST_CASES):
     """Generates a Python test function for each test case."""
     tests: Dict[str, Callable[[Any], None]] = {}

     for group, test_list in _sort_test_cases(test_cases).items():
         for i, (data, expected_frames) in enumerate(test_list, 1):
             count = i if len(test_list) > 1 else None
             test = _define_py_test(group, data, expected_frames, count)

             assert test.__name__ not in tests, f'Duplicate! {test.__name__}'
             tests[test.__name__] = test

     return tests


 # Class that tests all cases in TEST_CASES.
 DecoderTest = type('DecoderTest', (unittest.TestCase, ), _define_py_tests())

 if __name__ == '__main__':
     # If --generate-cc-decode-test is provided, generate the C++ test file.
     if len(sys.argv) >= 2 and sys.argv[1] == '--generate-cc-decode-test':
         with Path(sys.argv[2]).open('w') as file:
             _define_cc_tests(TEST_CASES, file)
     else:  # Otherwise, run the unit tests.
         unittest.main()
	#!/usr/bin/env python
	# Copyright 2020 The Pigweed Authors
	#
	# 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
	#
	# https://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.
	"""Contains the Python decoder tests and generates C++ decoder tests."""

	from collections import defaultdict
	from pathlib import Path
	from typing import Any, Callable, Dict, Iterator, List, NamedTuple, Union
	from typing import Optional, TextIO, Tuple
	import unittest
	import sys

	from pw_hdlc_lite.decode import Frame, FrameDecoder, FrameStatus, NO_ADDRESS
	from pw_hdlc_lite.protocol import frame_check_sequence as fcs


	def _encode(address: int, control: int, data: bytes) -> bytes:
	frame = bytearray([address, control]) + data
	frame += fcs(frame)
	frame = frame.replace(b'\x7d', b'\x7d\x5d')
	frame = frame.replace(b'\x7e', b'\x7d\x5e')
	return b''.join([b'\x7e', frame, b'\x7e'])


	class Expected(NamedTuple):
	address: int
	control: bytes
	data: bytes
	status: FrameStatus = FrameStatus.OK

	def __eq__(self, other) -> bool:
	"""Define == so an Expected and a Frame can be compared."""
	return (self.address == other.address and self.control == other.control
	and self.data == other.data and self.status is other.status)


	_PARTIAL = fcs(b'\x0ACmsg\x5e')
	_ESCAPED_FLAG_TEST_CASE = (
	b'\x7e\x0ACmsg\x7d\x7e' + _PARTIAL + b'\x7e',
	[
	Expected(0xA, b'C', b'', FrameStatus.INCOMPLETE),
	Expected(_PARTIAL[0], _PARTIAL[1:2], b'', FrameStatus.INCOMPLETE),
	],
	)

	TestCase = Tuple[bytes, List[Expected]]
	TestCases = Tuple[Union[str, TestCase], ...]

	TEST_CASES: TestCases = (
	'Empty payload',
	(_encode(0, 0, b''), [Expected(0, b'\0', b'')]),
	(_encode(55, 0x99, b''), [Expected(55, b'\x99', b'')]),
	(_encode(55, 0x99, b'') * 3, [Expected(55, b'\x99', b'')] * 3),
	'Simple one-byte payload',
	(_encode(0, 0, b'\0'), [Expected(0, b'\0', b'\0')]),
	(_encode(123, 0, b'A'), [Expected(123, b'\0', b'A')]),
	'Simple multi-byte payload',
	(_encode(0, 0, b'Hello, world!'), [Expected(0, b'\0', b'Hello, world!')]),
	(_encode(123, 0, b'\0\0\1\0\0'), [Expected(123, b'\0', b'\0\0\1\0\0')]),
	'Escaped one-byte payload',
	(_encode(1, 2, b'\x7e'), [Expected(1, b'\2', b'\x7e')]),
	(_encode(1, 2, b'\x7d'), [Expected(1, b'\2', b'\x7d')]),
	(_encode(1, 2, b'\x7e') + _encode(1, 2, b'\x7d'),
	[Expected(1, b'\2', b'\x7e'),
	Expected(1, b'\2', b'\x7d')]),
	'Escaped address',
	(_encode(0x7e, 0, b'A'), [Expected(0x7e, b'\0', b'A')]),
	(_encode(0x7d, 0, b'B'), [Expected(0x7d, b'\0', b'B')]),
	'Escaped control',
	(_encode(0, 0x7e, b'C'), [Expected(0, b'\x7e', b'C')]),
	(_encode(0, 0x7d, b'D'), [Expected(0, b'\x7d', b'D')]),
	'Escaped address and control',
	(_encode(0x7e, 0x7d, b'E'), [Expected(0x7e, b'\x7d', b'E')]),
	(_encode(0x7d, 0x7e, b'F'), [Expected(0x7d, b'\x7e', b'F')]),
	(_encode(0x7e, 0x7e, b'\x7e'), [Expected(0x7e, b'\x7e', b'\x7e')]),
	'Multiple frames separated by single flag',
	(_encode(0, 0, b'A')[:-1] + _encode(1, 2, b'123'),
	[Expected(0, b'\0', b'A'),
	Expected(1, b'\2', b'123')]),
	(_encode(0xff, 0, b'Yo')[:-1] * 3 + b'\x7e',
	[Expected(0xff, b'\0', b'Yo')] * 3),
	'Ignore empty frames',
	(b'\x7e\x7e', []),
	(b'\x7e' * 10, []),
	(b'\x7e\x7e' + _encode(1, 2, b'3') + b'\x7e' * 5,
	[Expected(1, b'\2', b'3')]),
	(b'\x7e' * 10 + _encode(1, 2, b':O') + b'\x7e' * 3 + _encode(3, 4, b':P'),
	[Expected(1, b'\2', b':O'),
	Expected(3, b'\4', b':P')]),
	'Cannot escape flag',
	(b'\x7e\xAA\x7d\x7e\xab\x00Hello' + fcs(b'\xab\0Hello') + b'\x7e', [
	Expected(0xAA, b'', b'', FrameStatus.INCOMPLETE),
	Expected(0xab, b'\0', b'Hello'),
	]),
	_ESCAPED_FLAG_TEST_CASE,
	'Frame too short',
	(b'\x7e1\x7e', [Expected(ord('1'), b'', b'', FrameStatus.INCOMPLETE)]),
	(b'\x7e12\x7e', [Expected(ord('1'), b'2', b'', FrameStatus.INCOMPLETE)]),
	(b'\x7e12345\x7e', [Expected(ord('1'), b'2', b'',
	FrameStatus.INCOMPLETE)]),
	'Incorrect frame check sequence',
	(b'\x7e123456\x7e',
	[Expected(ord('1'), b'2', b'', FrameStatus.FCS_MISMATCH)]),
	(b'\x7e\1\2msg\xff\xff\xff\xff\x7e',
	[Expected(0x1, b'\2', b'msg', FrameStatus.FCS_MISMATCH)]),
	(_encode(0xA, 0xB, b'???')[:-2] + _encode(1, 2, b'def'), [
	Expected(0xA, b'\x0B', b'??', FrameStatus.FCS_MISMATCH),
	Expected(1, b'\2', b'def'),
	]),
	'Invalid escape in address',
	(b'\x7e\x7d\x7d\0' + fcs(b'\x5d\0') + b'\x7e',
	[Expected(0,
	fcs(b'\x5d\0')[0:1], b'', FrameStatus.INVALID_ESCAPE)]),
	'Invalid escape in control',
	(b'\x7e\0\x7d\x7d' + fcs(b'\0\x5d') + b'\x7e',
	[Expected(0,
	fcs(b'\0\x5d')[0:1], b'', FrameStatus.INVALID_ESCAPE)]),
	'Invalid escape in data',
	(b'\x7e\0\1\x7d\x7d' + fcs(b'\0\1\x5d') + b'\x7e',
	[Expected(0, b'\1', b'', FrameStatus.INVALID_ESCAPE)]),
	'Frame ends with escape',
	(b'\x7e\x7d\x7e', [Expected(NO_ADDRESS, b'', b'',
	FrameStatus.INCOMPLETE)]),
	(b'\x7e\1\x7d\x7e', [Expected(1, b'', b'', FrameStatus.INCOMPLETE)]),
	(b'\x7e\1\2abc\x7d\x7e', [Expected(1, b'\2', b'',
	FrameStatus.INCOMPLETE)]),
	(b'\x7e\1\2abcd\x7d\x7e',
	[Expected(1, b'\2', b'', FrameStatus.INCOMPLETE)]),
	(b'\x7e\1\2abcd1234\x7d\x7e',
	[Expected(1, b'\2', b'abcd', FrameStatus.INCOMPLETE)]),
	'Inter-frame data is only escapes',
	(b'\x7e\x7d\x7e\x7d\x7e', [
	Expected(NO_ADDRESS, b'', b'', FrameStatus.INCOMPLETE),
	Expected(NO_ADDRESS, b'', b'', FrameStatus.INCOMPLETE),
	]),
	(b'\x7e\x7d\x7d\x7e\x7d\x7d\x7e', [
	Expected(NO_ADDRESS, b'', b'', FrameStatus.INVALID_ESCAPE),
	Expected(NO_ADDRESS, b'', b'', FrameStatus.INVALID_ESCAPE),
	]),
	'Data before first flag',
	(b'\0\1' + fcs(b'\0\1'), []),
	(b'\0\1' + fcs(b'\0\1') + b'\x7e',
	[Expected(0, b'\1', b'', FrameStatus.INCOMPLETE)]),
	'No frames emitted until flag',
	(_encode(1, 2, b'3')[:-1], []),
	(b'\x7e' + _encode(1, 2, b'3')[1:-1] * 2, []),
	) # yapf: disable
	# Formatting for the above tuple is very slow, so disable yapf.


	def _sort_test_cases(test_cases: TestCases) -> Dict[str, List[TestCase]]:
	cases: Dict[str, List[TestCase]] = defaultdict(list)
	message = ''

	for case in test_cases:
	if isinstance(case, str):
	message = case
	else:
	cases[message].append(case)

	return cases


	def _expected(frames: List[Frame]) -> Iterator[str]:
	for i, frame in enumerate(frames, 1):
	if frame.ok():
	yield f' Frame(kDecodedFrame{i:02}),'
	else:
	yield f' Status::DATA_LOSS, // Frame {i}'


	_CPP_HEADER = f"""\
	// Copyright 2020 The Pigweed Authors
	//
	// 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
	//
	// https://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.

	// AUTOGENERATED by {Path(__file__).name} - DO NOT EDIT

	#include "pw_hdlc_lite/decoder.h"

	#include <array>
	#include <cstddef>
	#include <variant>

	#include "gtest/gtest.h"
	#include "pw_bytes/array.h"

	namespace pw::hdlc_lite {{
	namespace {{

	// clang-format off
	"""

	_CPP_FOOTER = """\
	// clang-format on

	} // namespace
	} // namespace pw::hdlc_lite
	"""


	def _cpp_test(group: str, count: Optional[int], data: bytes) -> Iterator[str]:
	"""Generates a C++ test for the provided test data."""
	frames = list(FrameDecoder().process(data))
	data_bytes = ''.join(rf'\x{byte:02x}' for byte in data)

	name = ''.join(w.capitalize() for w in group.replace('-', ' ').split(' '))
	name = ''.join(c if c.isalnum() else '_' for c in name)
	name = name if count is None else name + f'_{count}'

	yield f'TEST(Decoder, {name}) {{'
	yield f' static constexpr auto kData = bytes::String("{data_bytes}");\n'

	for i, frame in enumerate(frames, 1):
	if frame.status is FrameStatus.OK:
	frame_bytes = ''.join(rf'\x{byte:02x}' for byte in frame.raw)
	yield (f' static constexpr auto kDecodedFrame{i:02} = '
	f'bytes::String("{frame_bytes}");')
	else:
	yield f' // Frame {i}: {frame.status.value}'

	yield ''

	expected = '\n'.join(_expected(frames)) or ' // No frames'
	decoder_size = max(len(data), 8) # Make sure large enough for a frame

	yield f"""\
	DecoderBuffer<{decoder_size}> decoder;

	static constexpr std::array<std::variant<Frame, Status>, {len(frames)}> kExpected = {{
	{expected}
	}};

	size_t decoded_frames = 0;

	decoder.Process(kData, [&](const Result<Frame>& result) {{
	ASSERT_LT(decoded_frames++, kExpected.size());
	auto& expected = kExpected[decoded_frames - 1];

	if (std::holds_alternative<Status>(expected)) {{
	EXPECT_EQ(Status::DATA_LOSS, result.status());
	}} else {{
	ASSERT_EQ(Status::OK, result.status());

	const Frame& decoded_frame = result.value();
	const Frame& expected_frame = std::get<Frame>(expected);
	EXPECT_EQ(expected_frame.address(), decoded_frame.address());
	EXPECT_EQ(expected_frame.control(), decoded_frame.control());
	ASSERT_EQ(expected_frame.data().size(), decoded_frame.data().size());
	EXPECT_EQ(std::memcmp(expected_frame.data().data(),
	decoded_frame.data().data(),
	expected_frame.data().size()),
	0);
	}}
	}});

	EXPECT_EQ(decoded_frames, kExpected.size());
	}}"""


	def _define_cc_tests(test_cases: TestCases, output: TextIO) -> None:
	"""Writes C++ tests for all test cases."""
	output.write(_CPP_HEADER)

	for group, test_list in _sort_test_cases(test_cases).items():
	for i, (data, _) in enumerate(test_list, 1):
	count = i if len(test_list) > 1 else None
	for line in _cpp_test(group, count, data):
	output.write(line)
	output.write('\n')

	output.write('\n')

	output.write(_CPP_FOOTER)


	def _define_py_test(group: str,
	data: bytes,
	expected_frames: List[Expected],
	count: int = None) -> Callable[[Any], None]:
	def test(self) -> None:
	# Decode in one call
	self.assertEqual(expected_frames,
	list(FrameDecoder().process(data)),
	msg=f'{group}: {data!r}')

	# Decode byte-by-byte
	decoder = FrameDecoder()
	decoded_frames = []
	for i in range(len(data)):
	decoded_frames += decoder.process(data[i:i + 1])

	self.assertEqual(expected_frames,
	decoded_frames,
	msg=f'{group} (byte-by-byte): {data!r}')

	name = 'test_' + ''.join(c if c.isalnum() else '_' for c in group.lower())
	test.__name__ = name if count is None else name + f'_{count}'
	return test


	def _define_py_tests(test_cases: TestCases = TEST_CASES):
	"""Generates a Python test function for each test case."""
	tests: Dict[str, Callable[[Any], None]] = {}

	for group, test_list in _sort_test_cases(test_cases).items():
	for i, (data, expected_frames) in enumerate(test_list, 1):
	count = i if len(test_list) > 1 else None
	test = _define_py_test(group, data, expected_frames, count)

	assert test.__name__ not in tests, f'Duplicate! {test.__name__}'
	tests[test.__name__] = test

	return tests


	# Class that tests all cases in TEST_CASES.
	DecoderTest = type('DecoderTest', (unittest.TestCase, ), _define_py_tests())

	if __name__ == '__main__':
	# If --generate-cc-decode-test is provided, generate the C++ test file.
	if len(sys.argv) >= 2 and sys.argv[1] == '--generate-cc-decode-test':
	with Path(sys.argv[2]).open('w') as file:
	_define_cc_tests(TEST_CASES, file)
	else: # Otherwise, run the unit tests.
	unittest.main()