Test Framework Architecture Patterns

"""ECU interface abstraction: hide transport from test logic."""
from abc import ABC, abstractmethod
from typing import Optional

class EcuInterface(ABC):
    """Abstract ECU interface -- tests use this, not CAN/Ethernet directly."""

    @abstractmethod
    def send_signal(self, signal_name: str, value: float) -> None:
        pass

    @abstractmethod
    def read_signal(self, signal_name: str, timeout_s: float = 1.0) -> Optional[float]:
        pass

    @abstractmethod
    def send_uds_request(self, service_id: int, data: bytes) -> bytes:
        pass

class CanEcuInterface(EcuInterface):
    """CAN bus implementation of EcuInterface."""

    def __init__(self, channel: str, db_path: str):
        import cantools, can
        self.db = cantools.database.load_file(db_path)
        self.bus = can.interface.Bus(channel, bustype="socketcan")

    def send_signal(self, signal_name: str, value: float) -> None:
        msg = self.db.get_message_by_signal_name(signal_name)
        data = msg.encode({signal_name: value})
        self.bus.send(can.Message(arbitration_id=msg.frame_id, data=data))

    def read_signal(self, signal_name: str, timeout_s: float = 1.0) -> Optional[float]:
        deadline = __import__("time").time() + timeout_s
        while __import__("time").time() < deadline:
            msg = self.bus.recv(timeout=0.01)
            if msg:
                try:
                    decoded = self.db.decode_message(msg.arbitration_id, msg.data)
                    if signal_name in decoded:
                        return decoded[signal_name]
                except Exception:
                    pass
        return None

    def send_uds_request(self, service_id: int, data: bytes) -> bytes:
        raise NotImplementedError("UDS over CAN requires ISO 15765-2 transport")

"""pytest conftest.py: ECU test fixtures."""
import pytest
from ecu_interface import CanEcuInterface

@pytest.fixture(scope="session")
def ecu(request):
    """Session-scoped ECU connection -- connect once per test session."""
    channel = request.config.getoption("--can-channel", default="vcan0")
    db_path = request.config.getoption("--dbc", default="vehicle.dbc")
    interface = CanEcuInterface(channel, db_path)
    yield interface
    interface.bus.shutdown()

@pytest.fixture(scope="function")
def ecu_reset(ecu):
    """Function-scoped: reset ECU to known state before each test."""
    ecu.send_uds_request(0x11, bytes([0x01]))  # ECU reset
    __import__("time").sleep(2.0)  # wait for boot
    yield ecu
    # Teardown: clear DTCs after test
    ecu.send_uds_request(0x14, bytes([0xFF, 0xFF, 0xFF]))