Unit test for serial ports. Can be config with json

serial_unit_test.py

# /usr/bin/env python3

# MIT License
#
# Copyright (c) 2024 Magicwenli
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.

from dataclasses import dataclass, asdict
import json
import sys
import time
from typing import Dict, List, Literal, Tuple
import serial


@dataclass
class SerialConfig:
    port: str
    baudrate: int
    timeout: float = 1


@dataclass
class TestConfig:
    case_sensitive: bool = True
    read_size: int = 1024
    input_type: Literal["hex", "decimal", "string"] = "string"
    output_type: Literal["hex", "decimal", "string"] = "string"


@dataclass
class TestCase:
    input: str
    desired_output: str
    test_config: str


@dataclass
class TestCases:
    serial_config: SerialConfig
    test_config_map: Dict[str, TestConfig]
    test_cases: List[TestCase]

    def __post_init__(self):
        if not (type(self.serial_config) is SerialConfig):
            old_serial_config = self.serial_config
            self.serial_config = SerialConfig(**old_serial_config)
        for k, v in self.test_config_map.items():
            if not (type(v) is TestConfig):
                old_test_config = v
                self.test_config_map[k] = TestConfig(**old_test_config)
        for i, test_case in enumerate(self.test_cases):
            if not (type(test_case) is TestCase):
                old_test_case = test_case
                self.test_cases[i] = TestCase(**old_test_case)

    @classmethod
    def from_json(cls, file_path: str):
        with open(file_path, "r") as file:
            data = json.load(file)
        return cls(**data)

    def to_json(self, file_path: str):
        with open(file_path, "w") as file:
            json.dump(asdict(self), file, indent=2)

    @classmethod
    def default(cls):
        case_1 = TestCase(
            input="AT\r\n",
            desired_output="OK\r\n",
            test_config="default",
        )
        test_config_map = {
            "default": TestConfig(read_size=128),
        }
        serial_config = SerialConfig(port="/dev/ttyS3", baudrate=961200, timeout=1)

        return cls(
            test_cases=[case_1],
            serial_config=serial_config,
            test_config_map=test_config_map,
        )

    def run_tests(self):
        result_all = 0
        for i, test_case in enumerate(self.test_cases):
            test_config = self.test_config_map[test_case.test_config]
            result, response = test_serial(
                self.serial_config,
                test_config,
                test_case.input,
                test_case.desired_output,
            )
            print(f"Test case {i} - {result}")
            print(f"Response: {response}")
            result_all = result_all + 1 if result else result_all
        return result_all


def convert_to_type(value: str, input_type: str) -> bytes:
    if input_type == "hex":
        return bytes.fromhex(value)
    elif input_type == "decimal":
        return bytes([int(x) for x in value.split()])
    else:  # string
        return value.encode()


def test_serial(
    serial_config: SerialConfig,
    test_config: TestConfig,
    input: str,
    desired_output: str,
) -> Tuple[bool, bytes]:
    try:
        ser = serial.Serial(
            serial_config.port, serial_config.baudrate, timeout=serial_config.timeout
        )

        input_data = convert_to_type(input, test_config.input_type)
        ser.write(input_data)
        time.sleep(0.5)
        response = ser.read(test_config.read_size)
        response = response.replace(b"\x00", b"")

        is_match = False

        desired_output_data = convert_to_type(desired_output, test_config.output_type)
        is_match = desired_output_data in response

    except serial.SerialException as e:
        print(f"Serial connection error: {e}")
    finally:
        if ser.is_open:
            ser.close()

        return is_match, response


if __name__ == "__main__":
    try:
        test_cases = TestCases.from_json("test_case.json")
    except FileNotFoundError:
        test_cases = TestCases.default()
    sys.exit(test_cases.run_tests())  # 0 if all tests pass, non-zero otherwise