hitl_tester.modules.bms.test_handler

  1"""Provides helper code for testing serial / SMBus."""
  2
  3from __future__ import annotations
  4
  5from typing import Any, ClassVar, Type, Literal
  6
  7import pytest
  8
  9from hitl_tester.modules.bms.bms_hw import BMSHardware
 10from hitl_tester.modules.bms.csv_tables import CSVBase
 11from hitl_tester.modules.logger import logger
 12
 13
 14_bms = BMSHardware(pytest.flags)  # type: ignore[arg-type]
 15
 16
 17class CSVRecordEvent:
 18    """Handler for serial tests."""
 19
 20    _failed: ClassVar[bool | None] = None  # Final failure status of test suite
 21    _registry: dict[str, list[Type[CSVRecordEvent]]] = {}  # Keep track of nested subclasses
 22    _registered_tests: list[Type[CSVRecordEvent]] = []
 23    _old_cycle_function: CSVBase = _bms.csv.cycle
 24    _failing_average: float = 0  # How successful a test is between 0 (pass) and 1 (fail)
 25    _samples: int = 0  # The amount of samples taken so far
 26    failure_status: bool = False  # Last failure status of a test
 27
 28    def __init_subclass__(cls):
 29        """Map nested subclasses to their parent class."""
 30        super().__init_subclass__()
 31        main_class, was_nested, _ = cls.__qualname__.partition(".")
 32        if was_nested:
 33            cls._registry[main_class] = cls._registry.get(main_class, []) + [cls]
 34
 35    @classmethod
 36    def current_test(cls, test_class: type | None):
 37        """Prepars to run a serial test."""
 38        CSVRecordEvent._old_cycle_function = _bms.csv.cycle  # Save logging function
 39        _bms.csv.cycle = _bms.csv.cycle_smbus  # Record serial and SMBus
 40        _bms.csv.cycle.postfix_fn = lambda: CSVRecordEvent.verify(  # Verify when a row is recorded in the CSV
 41            _bms.csv.cycle.last_row,  # type: ignore[attr-defined]
 42            _bms.csv.cycle.last_serial_data,  # type: ignore[attr-defined]
 43            _bms.csv.cycle.last_cell_data,  # type: ignore[attr-defined]
 44        )
 45        if test_class:
 46            for test_type in cls._registry.get(test_class.__name__, []):
 47                cls.register(test_type)
 48
 49    @classmethod
 50    def register(cls, test: Type[CSVRecordEvent]):
 51        """Register a test to run."""
 52        cls._registered_tests.append(test)
 53
 54    @classmethod
 55    def update_failing_average(cls, failed: bool):
 56        """Monitor how successful a test is from 0 (pass) to 1 (fail)."""
 57        cls._failing_average = (failed + cls._samples * cls._failing_average) / (cls._samples + 1)
 58        cls._samples += 1
 59
 60    @classmethod
 61    def fail_rate(cls):
 62        """Generate a failure rate string."""
 63        return f" [{cls._failing_average:.1%} failed]" if cls._failing_average > 0 else ""
 64
 65    @classmethod
 66    def clear_tests(cls):
 67        """Remove current serial tests."""
 68
 69        cls._failed = None  # Clear failure cache
 70        cls._registered_tests = []
 71        _bms.csv.cycle = cls._old_cycle_function  # Restore logging function (in case of modification)
 72
 73    @classmethod
 74    def failed(cls) -> bool:
 75        """Log results and whether the test failed."""
 76        if cls._failed is not None:
 77            return cls._failed
 78
 79        cls._failed = False
 80        for test in cls._registered_tests:
 81            cls._failed |= test.failed()
 82            result = test.result() + test.fail_rate()
 83            logger.write_result_to_html_report((f'<font color="#990000">{result}</font>' if test.failed() else result))
 84        return cls._failed
 85
 86    @classmethod
 87    def result(cls) -> str:
 88        """Overall failure or success message."""
 89        text = "anomalies detected."
 90        return text.capitalize() if cls.failed() else f"No {text}"
 91
 92    @classmethod
 93    def verify(cls, row: dict[str, Any], serial_data: dict[str, int | str], cell_data: dict[int, dict[str, int]]):
 94        """Verify all serial data is as expected."""
 95        for test in cls._registered_tests:
 96            try:
 97                test.verify(row, serial_data, cell_data)
 98                failed = test.failed()
 99                test.update_failing_average(failed)
100                if failed != test.failure_status:
101                    if failed:
102                        logger.write_warning_to_report(f"Now failing: {test.result()}")
103                    else:
104                        logger.write_info_to_report(f"Now passing: {test.result()}")
105                test.failure_status = failed
106            except ZeroDivisionError:  # FIXME(JA): fix in child class
107                pass
108
109    @classmethod
110    def cmp(
111        cls, a: float, sign: Literal["<", "<=", ">", ">=", "=="], b: float, unit: str = "", form: str = ".1%"
112    ) -> str:
113        """Generate a formatted string based on a comparison."""
114        sign_str = {
115            "<": ("≮", "<")[a < b],
116            "<=": ("≰", "≤")[a <= b],
117            ">": ("≯", ">")[a > b],
118            ">=": ("≱", "≥")[a >= b],
119            "==": ("≠", "=")[a == b],
120        }
121        return f"{a:{form}}{unit} {sign_str[sign]} {b:{form}}{unit}"