diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..53d5684 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,852 @@ +# python_raft AI Brief + +> Single-file reference for AI tools. Covers architecture, API surface, configuration schemas, and common patterns for the Rapid Application Framework for Test (RAFT). + +## 1. Purpose + +python_raft is a Python 3.10+ test-automation framework for engineering-level device testing, primarily targeting RDK (Reference Design Kit) set-top boxes and embedded Linux devices. It provides: + +- YAML-driven rack/slot/device configuration (no test-code changes per environment) +- Unified console abstraction (SSH, Serial, Telnet) +- Power control across 8 switch types +- IR/RF remote control with key-mapping +- HDMI-CEC send/receive +- A/V sync measurement +- Video capture and OCR +- Two test harness styles: `testController` (custom lifecycle) and `RAFTUnitTestCase` (stdlib `unittest`) + +Repository: `rdkcentral/python_raft`, default branch `develop`. + +**Companion briefs (same format, designed to be used together):** +- [ut-core AI Brief](https://github.com/rdkcentral/ut-core/blob/develop/AGENTS.md) -- the on-target C/C++ unit-test framework whose binaries RAFT drives. +- [ut-control AI Brief](https://github.com/rdkcentral/ut-control/blob/develop/AGENTS.md) -- the support library (KVP profiles, logging, control plane) used by ut-core test binaries. + +Point an AI at all three briefs plus a component specification to generate a +working test suite -- see [15-from-a-spec-to-a-test-suite](#15-from-a-spec-to-a-test-suite). + +--- + +## 2. Directory Layout + +``` +python_raft/ + framework/ + core/ + testControl.py # testController class (custom test lifecycle) + deviceManager.py # deviceManager, deviceClass, consoleClass + rackController.py # rackController, rack, rackSlot + configParser.py # configParser (device/CPE config) + configParserBase.py # Base class for config parsing + decodeParams.py # CLI argument parser (argparse) + logModule.py # Logging wrapper with custom levels + singleton.py # Singleton shared state for unittest path + raftUnittest.py # RAFTUnitTestCase, RAFTUnitTestSuite + rcCodes.py # rcCode enum (remote control key constants) + commonRemote.py # commonRemoteClass, remoteControllerMapping + powerControl.py # powerControlClass (delegates to modules) + hdmiCECController.py # HDMICECController + avSyncController.py # AVSyncController (SyncONE2) + utPlaneController.py # utPlaneController (ut-core integration) + outboundClient.py # File download/upload client + capture.py # Video capture engine + webpageController.py # Selenium web driver wrapper + utilities.py # Shell command helpers, wait, etc. + streamToFile.py # Stream-to-file utility + commandModules/ + consoleInterface.py # Abstract base class for consoles + sshConsole.py # SSH via paramiko + serialClass.py # Serial via pyserial + telnetClass.py # Telnet via telnetlib + powerModules/ + none.py # No-op power + hs100.py # TP-Link HS100 + apc.py # APC (telnet-based) + apcAos.py # APC AOS firmware + olimex.py # Olimex relay board + kasaControl.py # TP-Link Kasa smart plugs/strips + tapoControl.py # TP-Link Tapo smart plugs + SLP.py # Server Technology SLP PDU + remoteControllerModules/ + remoteInterface.py # Abstract base for remotes + none.py # No-op remote + olimex.py # Olimex IR blaster + skyProc.py # Sky proprietary IR + arduino.py # Arduino-based IR + keySimulator.py # RDK middleware key simulator + redrat.py # RedRat IR hub + hdmicecModules/ # CEC-client, remote-cec-client, virtual-cec-client + avSyncModules/ # SyncONE2 serial driver + audioAmplifier/ # Audio amplifier controller + examples/ + code/ + example_ssh.py # testController example + example_ssh_unittest.py # RAFTUnitTestCase example + configs/ + getting_started_rack_config.yml + example_rack_config.yml + example_device_config.yml + example_keymap.yml + example_olimex_keys.yml + example_redrat_keymap.yml + installation/ + install_requirements.sh + requirements.txt +``` + +--- + +## 3. Core Classes + +### 3.1 testController (`framework/core/testControl.py`) + +The original test harness. Tests subclass `testController` and override lifecycle hooks. + +```python +class testController: + def __init__(self, testName="", qcId="", maxRunTime=TEST_MAX_RUN_TIME, + level=logModule.STEP, loop=1, log=None) +``` + +**Constructor flow:** +1. Creates `logModule` for this test +2. Parses CLI args via `decodeParams` +3. Decodes rack config via `rackController` +4. Decodes device config via `configParser` +5. Selects rack and slot (from CLI `--rack` / `--slot` / `--slotName`) +6. Constructs log paths under `logs////` +7. Creates `deviceManager` from slot device list +8. Sets up shortcut attributes: `self.dut`, `self.session`, `self.powerControl`, `self.commonRemote`, `self.hdmiCECController` +9. Optionally sets up `capture` (video), `webpageController` (Selenium) + +**Key attributes available in tests:** +- `self.session` -- console session (SSH/Serial/Telnet) for the DUT. The constructor auto-selects the console marked `enabled: true` in config, falling back to `default` (or the first available console) when none is flagged +- `self.dut` -- `deviceClass` instance for "dut" +- `self.powerControl` -- `powerControlClass` +- `self.commonRemote` -- `commonRemoteClass` +- `self.hdmiCECController` -- `HDMICECController` or None +- `self.capture` -- `capture` instance or None +- `self.log` -- `logModule` instance +- `self.cpe` -- device config CPE entry (dict) +- `self.slotInfo` -- `rackSlot` instance +- `self.config` -- `decodeParams` instance + +**Lifecycle hooks (override in subclass):** + +| Method | When | Default | +|---|---|---| +| `testPrepareFunction()` | Before test loop | returns True | +| `testFunction()` | Each loop iteration | returns True | +| `testEndFunction(powerOff=True)` | After all loops complete | closes session, powers off | +| `testExceptionCleanUp()` | On exception in testFunction | no-op | +| `waitForBoot()` | Before testPrepareFunction | returns True | + +**`run(powerOff=True)` execution order:** +1. `session.open()` +2. `waitForBoot()` +3. `testPrepareFunction()` +4. Loop `testFunction()` up to `self.loopCount` times (or until False/exception/maxRunTime) +5. `testEndFunction(powerOff)` + +**Other useful methods:** +- `pingTest(deviceName="dut")` -- ICMP ping check +- `waitForPrompt(prompt=None)` -- wait for shell prompt on session +- `syscmd(cmd)` -- run command on the local host +- `runHostCommand(command)` -- run command on host via subprocess + +### 3.2 deviceManager (`framework/core/deviceManager.py`) + +```python +class deviceManager: + def __init__(self, deviceConfig: dict, log: logModule, logPath: str = "") + def getDevice(self, deviceName: str = "dut") -> deviceClass +``` + +Iterates over the device list from slot config, creating a `deviceClass` for each. + +### 3.3 deviceClass + +Represents a single device. Created from the per-device dict in the rack config. + +```python +class deviceClass: + def __init__(self, log, logPath, devices: dict) +``` + +**Attributes:** +- `consoles` -- dict of `consoleClass` instances keyed by name (e.g. "default", "ssh") +- `powerControl` -- `powerControlClass` or None +- `outBoundClient` -- `outboundClientClass` or None +- `remoteController` -- `commonRemoteClass` or None +- `hdmiCECController` -- `HDMICECController` or None +- `avSyncController` -- `AVSyncController` or None +- `session` -- default console session + +**Key methods:** +- `getConsoleSession(consoleName="default")` -- returns the underlying console session object +- `getField(fieldName)` -- recursive dict search for a field in raw config +- `pingTest(logPingTime=False)` -- ICMP reachability check + +### 3.4 consoleClass + +Factory that creates the right console type based on `type` field in config. + +Supported types: `ssh`, `serial`, `telnet`. + +Consoles can be enabled/disabled per slot. A console entry marked +`enabled: false` is skipped at construction (no session is created). +`deviceClass.getConsoleSession(name)` falls back to the first available console +if the requested one is disabled or missing, and returns `None` if every console +is disabled. `testController` reads the same `enabled` flag to choose the active +`self.session` automatically. + +### 3.5 rackController / rack / rackSlot (`framework/core/rackController.py`) + +```python +class rackController: + def __init__(self, config) + def getRackByName(self, rackName) -> rack + def getRackByIndex(self, rackIndex) -> rack + +class rack: + def getSlot(self, slotIndex) -> rackSlot # 1-based index + def getSlotByName(self, slotName) -> rackSlot + +class rackSlot: + def getDevice(self, deviceName) -> dict + def getPlatform(deviceName="dut") -> str + def getDeviceAddress(deviceName="dut") -> str + def getRemoteKeyType(deviceName="dut") -> str +``` + +### 3.6 decodeParams (`framework/core/decodeParams.py`) + +Parses CLI arguments via argparse. The `--config` argument is **required**. + +| Argument | Description | +|---|---| +| `--config` / `-config` | **Required.** Path to rack config YAML | +| `--deviceConfig` / `-deviceConfig` / `--testConfig` | Device config YAML (overrides `includes.deviceConfig` in rack config) | +| `--rack` | Rack name to use | +| `--slot` | Slot number (1-based int) | +| `--slotName` | Slot name (string alternative to `--slot`) | +| `--loop` | Override loop count | +| `--debug` / `-debug` | Enable DEBUG log level | +| `--test` / `-test` | Enable test mode | +| `--buildInfo` | URL to build info YAML | +| `--overrideDeviceConfig` | URL to CPE config override YAML | +| `--job_id` | Optional job identifier | +| `--rack_job_execution_id` | Optional rack job execution identifier | + +### 3.7 logModule (`framework/core/logModule.py`) + +Wraps Python `logging` with custom levels and structured test output. + +**Custom log levels (in ascending order):** + +| Level | Name | Numeric | +|---|---|---| +| DEBUG | DEBUG | 10 | +| INFO | INFO | 20 | +| STEP | STEP | 21 | +| STEP_START | STEP_START | 22 | +| TEST_START | TEST_START | 23 | +| STEP_RESULT | STEP_RESULT | 24 | +| TEST_RESULT | TEST_RESULT | 25 | +| TEST_SUMMARY | TEST_SUMMARY | 26 | +| WARNING | WARNING | 30 | +| ERROR | ERROR | 40 | +| CRITICAL | CRITICAL | 50 | +| FATAL | FATAL | 100 | + +**Key methods:** +- `stepStart(message, expected=None)` -- begin a numbered test step +- `step(message)` -- log within a step +- `stepResult(result: bool, message)` -- record pass/fail for a step +- `testStart(testName, qcId, loops, maxRunTime)` -- begin test timing +- `testResult(message)` -- end test, compute pass/fail summary +- `setFilename(logPath, logFileName)` -- attach file handler + CSV logger +- `indent()` / `outdent()` -- visual indentation in log output +- `fatal(message)` -- logs and calls `os._exit(1)` + +**Log output format:** `YYYY-MM-DD HH:MM:SS, , : ` + +Parallel CSV output: `QcId, TestName, Result, Failed Step, Failure, Duration` + +**Log directory structure:** +``` +logs//// + test_summary.log # only when a parent/summary logger is passed in; + test_summary.log.csv # a standalone testController writes these INSIDE -/ instead + -/ + test-0.log + test-0.log.csv + screenImages/ # always created by constructTestPath() + captureImages/ # only present when video capture is enabled +``` + +### 3.8 Singleton (`framework/core/singleton.py`) + +Shared state for the `RAFTUnitTestCase` path. Created once at module import time as `SINGLETON`. + +```python +class Singleton: + # Class-level attributes shared by all tests: + config # decodeParams + _rackController # rackController + deviceConfig # configParser + rack # selected rack + slotInfo # selected slot + summaryLog # logModule for summary + testLog # logModule for test detail + devices # deviceManager (lazy init) +``` + +--- + +## 4. Console Types + +All consoles implement `consoleInterface` (abstract base class): + +```python +class consoleInterface(ABC): + def open(self) -> bool + def close(self) -> bool + def read_until(self, value: str, timeout: int = 10) -> str + def read_all(self) -> str + def write(self, message: list|str, lineFeed="\n", wait_for_prompt=False) -> bool + def waitForPrompt(self, prompt=None, timeout=10) -> bool + # property: timeout (int, seconds) +``` + +### 4.1 SSH (`sshConsole`) + +Uses `paramiko.SSHClient`. Supports key-based and password auth. + +Config fields: `type: "ssh"`, `address`/`ip`, `username`, `password`, `port` (default 22), `known_hosts`, `prompt`. + +Extra method: `open_interactive_shell()`, `read(timeout=10)`. + +### 4.2 Serial (`serialSession`) + +Uses `pyserial`. + +Config fields honored today: `type: "serial"`, `port` (e.g. "/dev/ttyUSB0"), `baudRate` (default 115200), `prompt`. Note: `serialSession` only passes `port` and `baudRate` (with a fixed 300s timeout) to `pyserial` — `dataBits`/`stopBits`/`parity`/`flowControl` are not applied even if present in the config (pyserial defaults are used: 8/N/1, no flow control). + +### 4.3 Telnet (`telnet`) + +Uses `telnetlib`. On Python 3.13+ (where `telnetlib` was removed from the +standard library) a built-in minimal `Telnet` compatibility shim is used +automatically, so telnet consoles keep working without the stdlib module. + +Config fields: `type: "telnet"`, `address`/`ip`, `username`, `password`, `port` (default 23), `username_prompt`, `password_prompt`, `prompt`. + +--- + +## 5. Power Modules + +`powerControlClass` delegates to a specific module based on `type` in config. + +```python +class powerControlClass: + def powerOn(self) -> bool + def powerOff(self) -> bool + def reboot(self) -> bool + def getPowerLevel(self) -> float # Watts + def getVoltageLevel(self) -> float # Volts + def getCurrentLevel(self) -> float # Amps +``` + +All operations support retry via `retryCount` (default 1) and `retryDelay` (default 30s) config fields. + +| Type | Module | Config fields | +|---|---|---| +| `"none"` | `powerNone` | (none) | +| `"hs100"` | `powerHS100` | `ip`, `port` | +| `"apc"` | `powerAPC` | `ip`, `username`, `password`, `outlet` | +| `"apcAos"` | `powerApcAos` | `ip`, `username`, `password`, `port` (23), `outlet` | +| `"olimex"` | `powerOlimex` | `ip`, `port`, `relay` | +| `"kasa"` | `powerKasa` | `ip`, `options` (args passed to the `kasa` CLI; defaults to `"--type plug"` when unset — use a strip form such as `"--type strip"` / `"--strip"` for power strips), `args` ("--index N") | +| `"tapo"` | `powerTapo` | `ip`, `username`, `password`, `outlet` | +| `"SLP"` | `powerSLP` | `ip`, `username`, `password`, `outlet_id`, `port` (23) | + +--- + +## 6. Remote Control + +`commonRemoteClass` provides a unified remote-control interface with key mapping. + +```python +class commonRemoteClass: + def sendKey(self, keycode: rcCode, delay=1, repeat=1, randomRepeat=0) + def setKeyMap(self, name) + def getKeyMap(self) -> dict +``` + +**Key codes** are defined in `rcCodes.py` as the `rcCode` enum: +```python +from framework.core.rcCodes import rcCode as rc +rc.ARROW_UP, rc.ARROW_DOWN, rc.OK, rc.BACK, rc.HOME, rc.POWER, +rc.NUM_0 .. rc.NUM_9, rc.RED, rc.GREEN, rc.BLUE, rc.YELLOW, etc. +``` + +**Remote types:** + +| Type | Module | Config fields | +|---|---|---| +| `"olimex"` | `remoteOlimex` | `ip`, `port`, `map`, `config` | +| `"sky_proc"` | `remoteSkyProc` | `map`, `config` | +| `"arduino"` | `remoteArduino` | `map`, `config` | +| `"keySimulator"` | `remoteKeySimulator` | `ip`, `port`, `username`, `password`, `map`, `config` | +| `"redrat"` | `remoteRedRat` | `hub_ip`, `hub_port`, `netbox_ip`, `netbox_name`, `netbox_mac`, `output`, `map`, `config` | +| (default) | `remoteNone` | -- | + +**Key mapping** is loaded from a YAML file specified by `config` field. Maps contain named translation tables with a `codes` dict and optional `prefix`. The `map` field selects which mapping to activate. + +--- + +## 7. HDMI-CEC Controller + +```python +class HDMICECController: + def __init__(self, log, config: dict) + def start(self) + def stop(self) + def sendMessage(self, sourceAddress, destAddress, opCode, payload=None) + def checkMessageReceived(self, sourceAddress, destAddress, opCode, + timeout=10, payload=None) -> bool + def listDevices(self) -> list[dict] +``` + +**CEC controller types:** + +| Type | Config fields | +|---|---| +| `"cec-client"` | `adaptor` (e.g. "/dev/ttyACM0") | +| `"remote-cec-client"` | `adaptor`, `address`, `username`, `password`, `port` (22), `prompt` | +| `"virtual-cec-client"` | `adaptor`, `address`, `username`, `password`, `port`, `prompt`, `control_port` (8080), `device_network_configuration` | + +--- + +## 8. A/V Sync Controller + +```python +class AVSyncController: + def __init__(self, log, config: dict) + def calibrate(self) + def start_measurements(self) + def stop_measurements(self) + def clear_results(self) + def get_results(self) -> list[dict] + # Properties: audio_trigger_level, video_trigger_level, frame_rate, mask_length, offset +``` + +Config: `type: "SyncOne2"`, `port`, `extended_mode` (bool), `audio_input` ("AUTO"|"EXTERNAL"|"INTERNAL"), `speaker_distance`. + +Results dict keys: `milliseconds`, `frames`, `avg_milliseconds`, `avg_frames`, `span_milliseconds`, `span_frames`. + +--- + +## 9. utPlaneController (ut-core Integration) + +Sends YAML commands to a ut-controller HTTP endpoint running on the DUT. + +```python +class utPlaneController: + def __init__(self, session, port=8080, log=None) + def sendMessage(self, yamlInput: str, isFile: bool = False) -> bool +``` + +Sends via `curl -X POST` to `http://localhost:/api/postKVP`. When `isFile=True`, uses `--data-binary @` for a file on the DUT; otherwise sends the YAML string inline. + +--- + +## 10. Configuration Schemas + +### 10.1 Rack Config YAML (the `--config` file) + +```yaml +globalConfig: + includes: + deviceConfig: "path/to/device_config.yml" # auto-loaded device config + local: + log: + directory: "./logs" + delimiter: "/" + # Optional sections: + # capture: + # ocrEnginePath: "/usr/bin/tesseract" + # resolution: "1080p" + # input: 0 + # webpageDriver: + # + +rackConfig: + rack1: + name: "rack1" + description: "my test rack" + slot1: + name: "slot1" + devices: + - dut: + ip: "192.168.1.100" + description: "Device under test" + platform: "llama" + consoles: + - default: + type: "ssh" # or "serial" or "telnet" + enabled: true # active console; testController auto-selects the enabled one + ip: "192.168.1.100" + port: 22 + username: "root" + password: "" + prompt: "root@device:~#" + - serial: + type: "serial" + enabled: false # disabled consoles are skipped at construction + port: "/dev/ttyUSB0" + baudRate: 115200 + powerSwitch: + type: "kasa" + ip: "192.168.1.50" + options: "--plug" + retryCount: 2 + retryDelay: 15 + remoteController: + type: "olimex" + ip: "192.168.1.60" + port: 7 + map: "llama_rc6" + config: "remote_commander.yml" + hdmiCECController: + type: "cec-client" + adaptor: "/dev/ttyACM0" + avSyncController: + type: "SyncOne2" + port: "/dev/ttyACM1" + outbound: + download_url: "http://server/images/" + upload_url: "http://server/uploads/" + workspaceDirectory: "~/workspace" + - pi2: + ip: "192.168.1.101" + platform: "pi4" + consoles: + - default: + type: "ssh" + port: 22 + username: "pi" +``` + +### 10.2 Device Config YAML + +```yaml +deviceConfig: + cpe1: + platform: "llama" + model: "ModelX" + prompt: "root@device:~#" + screenRegions: "path/to/screen_regions.yml" + validImage: + baseLocation: "http://images.example.com/" + image1: "firmware_v1.bin" + image2: "firmware_v2.bin" +``` + +The `platform` field links a CPE entry to a device in the rack config. + +### 10.3 Config include mechanism + +The `configParser.processIncludes()` method supports an `include` key at any level: + +```yaml +include: + - "path/to/extra_config.yml" + - "https://remote.server/config.yml" +``` + +Included files are merged into the parent dict. Supports local files and HTTP URLs. + +--- + +## 11. Test Lifecycle + +### 11.1 testController style + +``` +__init__() + | + v +run(powerOff=True) + |-> session.open() + |-> waitForBoot() + |-> testPrepareFunction() # override this + |-> loop 1..N: + | testFunction() # override this + | (break on False or exception) + |-> testEndFunction(powerOff) # override for cleanup +``` + +### 11.2 RAFTUnitTestCase style (unittest) + +```python +class MyTest(RAFTUnitTestCase): + def setUp(self): # standard unittest setUp + self.dut.session.open() + + def test_something(self): + self.dut.session.write("echo hello") + output = self.dut.session.read_all() + self.assertIn("hello", output) + + def tearDown(self): # standard unittest tearDown + self.dut.session.close() + +if __name__ == '__main__': + RAFTUnitTestMain() +``` + +`RAFTUnitTestCase` provides `self.log`, `self.devices`, `self.dut`, `self.cpe`, `self.utils` automatically via the `Singleton`. + +Run with: `python my_test.py --config rack_config.yml` + +--- + +## 12. CLI Usage + +```bash +# testController style +python my_test.py --config rack_config.yml [--rack rack1] [--slot 1] [--debug] [--loop 5] + +# unittest style +python my_test.py --config rack_config.yml [--rack rack1] [--slot 1] [--debug] +``` + +--- + +## 13. Dependencies (requirements.txt highlights) + +| Package | Purpose | +|---|---| +| `paramiko` | SSH console | +| `pyserial` | Serial console | +| `PyYAML` | Config parsing | +| `requests` | HTTP operations, remote YAML loading | +| `python-kasa` | Kasa smart plug control | +| `opencv-python` | Video capture | +| `pytesseract` / `pillow` | OCR | +| `selenium` | Web page control | +| `numpy` | Image processing | +| `fabric` / `invoke` | Remote execution utilities | +| `boto3` | AWS S3 integration | + +**Python:** 3.10+ (enforced by `installation/install_requirements.sh`). + +**Optional / Docker-safe:** `opencv-python`, `pytesseract`, `pillow` (capture/OCR) +and `selenium` (web control) are imported lazily. If they are absent (e.g. in a +slim Docker image) the framework still loads; the dependent feature raises a +clear `ImportError` only when actually used. Likewise telnet consoles work +without the stdlib `telnetlib` on Python 3.13+. + +--- + +## 14. Common Test Patterns + +### 14.1 Basic SSH command execution (testController) + +```python +class MyTest(testController): + def __init__(self): + super().__init__(testName="my_test", qcId="TC001") + + def testPrepareFunction(self): + self.session.prompt = self.cpe.get("prompt") + return True + + def testFunction(self): + self.log.stepStart("Run command on DUT") + self.session.write("cat /proc/version") + output = self.session.read_until(self.session.prompt) + result = "Linux" in output + self.log.stepResult(result, "Verify Linux kernel version string") + return result + + def testEndFunction(self, powerOff=False): + return super().testEndFunction(powerOff) + +if __name__ == "__main__": + test = MyTest() + test.run() +``` + +### 14.2 Power cycle and verify boot + +```python +def testFunction(self): + self.log.stepStart("Power cycle DUT") + self.powerControl.reboot() + self.log.step("Wait for device to come back") + self.utils.wait(30) # framework helper (no separate `import time` needed) + alive = self.pingTest() + self.log.stepResult(alive, "DUT responds to ping after reboot") + return alive +``` + +### 14.3 Send remote control keys + +```python +from framework.core.rcCodes import rcCode as rc + +def testFunction(self): + self.log.stepStart("Navigate to settings menu") + self.commonRemote.sendKey(rc.HOME, delay=2) + self.commonRemote.sendKey(rc.ARROW_DOWN, repeat=3, delay=1) + self.commonRemote.sendKey(rc.OK, delay=2) + return True +``` + +### 14.4 HDMI-CEC message exchange + +```python +def testFunction(self): + self.log.stepStart("Send CEC standby and verify acknowledgement") + self.hdmiCECController.sendMessage("0", "4", "0x36") # standby + received = self.hdmiCECController.checkMessageReceived( + "4", "0", "0x90", timeout=5, payload=["0x01"] + ) + self.log.stepResult(received, "Device acknowledged standby") + return received +``` + +### 14.5 Using utPlaneController + +```python +from framework.core.utPlaneController import utPlaneController + +def testFunction(self): + controller = utPlaneController(self.session, port=8080, log=self.log) + yaml_cmd = "key: value\ncommand: test_start" + result = controller.sendMessage(yaml_cmd) + self.log.stepResult(result, "Sent command to ut-controller") + return result +``` + +### 14.6 unittest style with assertions + +```python +from framework.core.raftUnittest import RAFTUnitTestCase, RAFTUnitTestMain + +class TestNetwork(RAFTUnitTestCase): + def setUp(self): + self.dut.session.open() + + def test_ping(self): + self.assertTrue(self.dut.pingTest(), "DUT should be reachable") + + def test_hostname(self): + self.dut.session.write("hostname") + output = self.dut.session.read_all() + self.assertIn("mydevice", output) + + def tearDown(self): + self.dut.session.close() + +if __name__ == "__main__": + RAFTUnitTestMain() +``` + +### 14.7 Multiple devices in one slot + +```python +# Access non-DUT device +pi_device = self.devices.getDevice("pi2") +pi_session = pi_device.getConsoleSession("default") +pi_session.open() +pi_session.write("ifconfig eth0") +output = pi_session.read_all() +``` + +### 14.8 Named console access + +```python +# Use a specific console on the DUT +serial_session = self.dut.getConsoleSession("serial") +serial_session.open() +serial_session.write("reboot") +``` + +--- + +## 15. From a Spec to a Test Suite + +This is the recipe an AI agent should follow when given a component +specification (e.g. an RDK HAL interface, an API contract, or acceptance +criteria) and asked to produce a RAFT test suite. + +**1. Decide the test layer.** +- *On-target unit / HAL conformance* (C/C++ assertions running on the device): + author the test logic with **ut-core** (see the ut-core companion brief). RAFT's + job is to deploy/run that binary and adjudicate its output. Device-specific + expected values come from a **ut-control** KVP profile (YAML) the binary loads. +- *Black-box / integration / system behaviour* driven over a console or remote: + author it directly in RAFT using the patterns below -- no on-target binary needed. + +**2. Pick the harness style.** +- `RAFTUnitTestCase` -- one `test_*` method per requirement, standard `unittest` + assertions, independent `setUp`/`tearDown`. Preferred for spec-driven suites + where each requirement maps to a discrete, independently-reported check. +- `testController` -- a single scripted scenario with `testPrepareFunction` / + `testFunction` / `testEndFunction`. Preferred for one end-to-end flow. + +**3. Map the spec to tests.** +- Each requirement / API behaviour → one `test_*` method (or one `stepStart`/ + `stepResult` block in a `testController`). +- Preconditions (device powered, image flashed, service running) → `setUp` / + `testPrepareFunction` / `waitForBoot`. +- Cleanup (remove artifacts, power off) → `tearDown` / `testEndFunction`. +- Negative/error cases in the spec → their own assertions. + +**4. Express the environment as config, not code.** +- Define the DUT console (mark the one you use `enabled: true`), `powerSwitch`, + and `remoteController` in the rack config; never hard-code addresses in tests. +- Put device-specific expected values (port counts, supported formats, etc.) in + a KVP profile consumed by the on-target ut-core binary -- the same value set the + spec parameterizes over. + +**5. Drive and assert.** +- Open `self.dut.session`, send commands or launch the ut-core binary with + `session.write(...)`, capture output with `read_until()` / `read_all()`. +- For on-target ut-core runs, invoke the binary in automated mode and parse its + pass/fail summary; assert in RAFT on that result. +- Report with `unittest` assertions (`RAFTUnitTestCase`) or `self.log.stepResult(passed, msg)` + (`testController`). Both feed the summary log. + +**Skeleton -- run an on-target ut-core suite and adjudicate it:** +```python +from framework.core.raftUnittest import RAFTUnitTestCase, RAFTUnitTestMain + +class HalConformance(RAFTUnitTestCase): + def test_hal_suite_passes(self): + self.dut.session.open() + # Launch the ut-core binary in automated mode with a device profile + self.dut.session.write( + "/opt/ut/hal_test -p /opt/ut/profiles/device.yaml -a") + output = self.dut.session.read_until(self.dut.session.prompt, timeout=120) + # ut-core prints a CUnit/GTest run summary; adjudicate it in RAFT. + # Use the exact summary markers documented in the ut-core brief; here we + # simply require that no failures were reported. + self.assertNotIn("FAILED", output.upper()) + +if __name__ == "__main__": + RAFTUnitTestMain() +``` + +The exact ut-core CLI flags, run modes, and output format are in the +[ut-core AI Brief](https://github.com/rdkcentral/ut-core/blob/develop/AGENTS.md); +the KVP profile schema the binary consumes is in the +[ut-control AI Brief](https://github.com/rdkcentral/ut-control/blob/develop/AGENTS.md). + +--- + +## 16. License + +Apache License 2.0. Copyright 2023 RDK Management. diff --git a/CHANGELOG.md b/CHANGELOG.md index da4e358..dae71d3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,12 +4,23 @@ All notable changes to this project will be documented in this file. Dates are d Generated by [`auto-changelog`](https://github.com/CookPete/auto-changelog). +#### [1.9.0](https://github.com/rdkcentral/python_raft/compare/1.8.2...1.9.0) + +- Fix virtual cec [`#220`](https://github.com/rdkcentral/python_raft/pull/220) +- Add AI-ingestible framework brief [`#207`](https://github.com/rdkcentral/python_raft/pull/207) +- - Add Python 3.13+ compatibility shim for removed telnetlib module [`#210`](https://github.com/rdkcentral/python_raft/pull/210) +- Add AI-ingestible framework brief for python_raft [`#206`](https://github.com/rdkcentral/python_raft/issues/206) +- Making big modules 'optional' for Docker [`c466bd6`](https://github.com/rdkcentral/python_raft/commit/c466bd6ab22e0e5f9aec43b8765612f24af2ae2d) +- Add : gh #206 : Refresh ai-brief for current develop + spec-to-suite recipe [`e5c5c50`](https://github.com/rdkcentral/python_raft/commit/e5c5c50b9a6d25e5c0b7016efcc603840ef28d6d) +- Update framework/core/commandModules/telnetClass.py [`0a7aa6f`](https://github.com/rdkcentral/python_raft/commit/0a7aa6f0d59df746034034ea67c64b48ea4313cb) + #### [1.8.2](https://github.com/rdkcentral/python_raft/compare/1.8.1...1.8.2) > 10 March 2026 - Enhance keySimulator sendKey to prevent missed keys [`#205`](https://github.com/rdkcentral/python_raft/pull/205) - Comms fixes [`#203`](https://github.com/rdkcentral/python_raft/pull/203) +- Bumped CHANGELOG [`0103b1b`](https://github.com/rdkcentral/python_raft/commit/0103b1bfceec6d724531bec12e6ea645060727d7) - Removed new line feed & updated the return result [`871cdde`](https://github.com/rdkcentral/python_raft/commit/871cdde0a490d669260e4beaa626bed4f4845ae6) - Merge tag '1.8.1' into develop [`13bf879`](https://github.com/rdkcentral/python_raft/commit/13bf87916437160f500b573c2166d68811ad7d85) diff --git a/framework/core/__init__.py b/framework/core/__init__.py index 2e44f9d..1a2fd90 100644 --- a/framework/core/__init__.py +++ b/framework/core/__init__.py @@ -32,13 +32,36 @@ #__all__ = ["testControl", "logModule", "rcCodes", "rackController", "slotInfo" ] #expose only the components required for the upper classes to operate -from . capture import capture +try: + from . capture import capture +except ModuleNotFoundError as e: + # Only treat known optional image-processing dependencies as optional. + _optional_capture_deps = {"cv2", "pytesseract", "PIL"} + if e.name in _optional_capture_deps: + class _CaptureMissingDependency: + def __call__(self, *args, **kwargs): + raise ImportError( + "Image capture functionality is unavailable because optional " + "dependencies (cv2, pytesseract, PIL) are not installed." + ) from e + capture = _CaptureMissingDependency() + else: + # Unexpected missing module (e.g., bug inside capture.py) - do not hide it. + raise from . commonRemote import commonRemoteClass from . deviceManager import deviceManager from . logModule import logModule from . logModule import DEBUG, INFO, WARNING, ERROR, CRITICAL from . testControl import testController -from . webpageController import webpageController +try: + from . webpageController import webpageController +except ModuleNotFoundError as e: + if e.name == "selenium": + # selenium not installed (e.g. Docker) + webpageController = None + else: + # Unexpected missing module (e.g., bug inside webpageController.py) - do not hide it. + raise from . rcCodes import rcCode as rc from . utilities import utilities diff --git a/framework/core/commandModules/serialClass.py b/framework/core/commandModules/serialClass.py index 7c617fd..f094fbd 100644 --- a/framework/core/commandModules/serialClass.py +++ b/framework/core/commandModules/serialClass.py @@ -200,5 +200,8 @@ def flush(self) -> bool: True if can successfully clear the serial console. """ self.log.info("Clearing Serial console log") - self.serialCon.reset_input_buffer() + if hasattr(self.serialCon, "reset_input_buffer"): + self.serialCon.reset_input_buffer() + else: + self.serialCon.flushInput() return True diff --git a/framework/core/commandModules/telnetClass.py b/framework/core/commandModules/telnetClass.py index f163386..0f52135 100644 --- a/framework/core/commandModules/telnetClass.py +++ b/framework/core/commandModules/telnetClass.py @@ -30,7 +30,126 @@ #* #* ****************************************************************************** -import telnetlib +try: + import telnetlib +except ModuleNotFoundError: + # telnetlib was removed in Python 3.13, provide a minimal compatibility layer + import socket + import time + + class Telnet: + def __init__(self, host=None, port=23, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): + self.host = host + self.port = port + self.timeout = timeout + self.sock = None + if host: + self.open(host, port, timeout) + + def open(self, host, port=23, timeout=socket._GLOBAL_DEFAULT_TIMEOUT): + self.host = host + self.port = port + self.sock = socket.create_connection((host, port), timeout) + + def close(self): + if self.sock: + self.sock.close() + self.sock = None + + def write(self, buffer): + if self.sock: + self.sock.sendall(buffer) + + def read_until(self, match, timeout=None): + if not self.sock: + return b'' + + buffer = b'' + start_time = time.time() + while True: + if timeout and (time.time() - start_time) > timeout: + break + try: + data = self.sock.recv(1024) + if not data: + break + buffer += data + if match in buffer: + break + except socket.timeout: + break + except Exception: + break + return buffer + + def read_all(self): + if not self.sock: + return b'' + buffer = b'' + try: + while True: + data = self.sock.recv(1024) + if not data: + break + buffer += data + except Exception: + pass + return buffer + + def read_very_eager(self): + """ + Read all data available on the socket without blocking. + This is a minimal approximation of telnetlib.Telnet.read_very_eager(). + """ + if not self.sock: + return b'' + + buffer = b'' + # Use non-blocking recv to drain any immediately available data. + self.sock.setblocking(False) + try: + while True: + try: + data = self.sock.recv(1024) + except BlockingIOError: + # No more data available without blocking. + break + if not data: + break + buffer += data + finally: + # Restore blocking mode for subsequent operations. + self.sock.setblocking(True) + return buffer + + def read_eager(self): + """ + Read any data available on the socket without significant blocking. + For this compatibility shim, delegate to read_very_eager(). + """ + return self.read_very_eager() + + def read_some(self): + """ + Read at least some data if available, without blocking if none is ready. + Returns an empty bytes object if no data is immediately available. + """ + if not self.sock: + return b'' + + self.sock.setblocking(False) + try: + try: + data = self.sock.recv(1024) + except BlockingIOError: + data = b'' + finally: + self.sock.setblocking(True) + return data + # Create a mock telnetlib module + class telnetlib: + Telnet = Telnet + import socket from .consoleInterface import consoleInterface @@ -153,7 +272,7 @@ def read_until(self,value: str, timeout: int = 10) -> str: message = value.encode() result = self.tn.read_until(message,self.timeout) return result.decode() - + def read_all(self) -> str: """Read all readily available information displayed in the console. @@ -161,7 +280,7 @@ def read_all(self) -> str: str: Information currently displayed in the console. """ return self.read_eager() - + def write(self,message:list|str, lineFeed:str="\r\n", wait_for_prompt:bool=False) -> bool: """Write a message into the session console. Optional: waits for prompt. diff --git a/framework/core/deviceManager.py b/framework/core/deviceManager.py index 5412183..cadf2a1 100644 --- a/framework/core/deviceManager.py +++ b/framework/core/deviceManager.py @@ -67,6 +67,11 @@ def __init__(self, log:logModule, logPath:str, configElements:dict): """ for element in configElements: config = configElements.get(element) + # Respect 'enabled: false' in console config — skip disabled consoles + if config.get("enabled") is False: + self.type = None + self.session = None + return self.type = config.get("type") self.prompt = config.get("prompt") # Create a new console since it hasn't been created @@ -172,7 +177,9 @@ def __init__(self, log:logModule, logPath:str, devices:dict): consoles = device.get("consoles") for element in consoles: for name in element: - self.consoles[name] = consoleClass(log, logPath, element ) + console = consoleClass(log, logPath, element) + if console.type is not None: # skip disabled consoles + self.consoles[name] = console config = device.get("outbound") if config != None: self.outBoundClient = outboundClientClass(log, **config) @@ -218,9 +225,15 @@ def getConsoleSession(self, consoleName:str="default" ): Returns: consoleClass: Console class, or None on failure """ - console = self.consoles[consoleName] - if console == None: - self.log.error("Invalid consoleName [{}]".format(consoleName)) + console = self.consoles.get(consoleName) + # If requested console was disabled/missing, fall back to first enabled console + if console is None and self.consoles: + fallback = next(iter(self.consoles)) + self.log.info("Console '{}' not available, falling back to '{}'".format(consoleName, fallback)) + console = self.consoles[fallback] + if console is None: + self.log.error("No consoles available (all disabled?)") + return None return console.session def pingTest(self, logPingTime=False): diff --git a/framework/core/hdmicecModules/virtualCECController.py b/framework/core/hdmicecModules/virtualCECController.py index 1f57ac7..7792887 100644 --- a/framework/core/hdmicecModules/virtualCECController.py +++ b/framework/core/hdmicecModules/virtualCECController.py @@ -94,6 +94,7 @@ def __init__(self, adaptor: str, logger: logModule, streamLogger: StreamToFile, except Exception as e: self._log.critical(f"Failed to load device configuration: {e}") raise + self.start() def loadCecDeviceNetworkConfiguration(self, configString: str): """ diff --git a/framework/core/logModule.py b/framework/core/logModule.py index d1e76ca..48d5953 100755 --- a/framework/core/logModule.py +++ b/framework/core/logModule.py @@ -118,9 +118,42 @@ def __init__(self, moduleName, level=INFO): def __del__(self): """Deletes the logger instance. """ - while self.log.hasHandlers(): - self.log.removeHandler(self.log.handlers[0]) - + # Ensure all handlers are flushed and closed to avoid leaking resources + for handler in list(getattr(self, "log", {}).handlers if getattr(self, "log", None) else []): + try: + # Not all handlers implement flush; guard just in case + if hasattr(handler, "flush"): + handler.flush() + except Exception: + # Best-effort cleanup; ignore flush errors during destruction + pass + try: + handler.close() + except Exception: + # Ignore close errors during destruction + pass + try: + self.log.removeHandler(handler) + except Exception: + # If removal fails, there's nothing more we can safely do here + pass + + # Also clean up CSV logger handlers, if this instance created one + if hasattr(self, "csvLogger") and getattr(self, "csvLogger") is not None: + for handler in list(self.csvLogger.handlers): + try: + if hasattr(handler, "flush"): + handler.flush() + except Exception: + pass + try: + handler.close() + except Exception: + pass + try: + self.csvLogger.removeHandler(handler) + except Exception: + pass def setFilename( self, logPath, logFileName ): """ Sets the filename for logging. @@ -134,11 +167,11 @@ def setFilename( self, logPath, logFileName ): return self.logPath = logPath logFileName = os.path.join(logPath + logFileName) - self.logFile = logging.FileHandler(logFileName) + self.logFile = logging.FileHandler(logFileName, encoding='utf-8') self.logFile.setFormatter( self.format ) self.log.addHandler( self.logFile ) #Create the CSV Logger module - self.csvLogFile = logging.FileHandler( logFileName+".csv" ) + self.csvLogFile = logging.FileHandler( logFileName+".csv", encoding='utf-8' ) self.csvLogger.addHandler( self.csvLogFile ) self.csvLogger.info("QcId, TestName, Result, Failed Step, Failure, Duration [hh:mm:ss]") self.log.info( "Log File: [{}]".format(logFileName) ) diff --git a/framework/core/testControl.py b/framework/core/testControl.py index cbe230b..30cba8b 100644 --- a/framework/core/testControl.py +++ b/framework/core/testControl.py @@ -33,7 +33,11 @@ import time import signal import random -import telnetlib +try: + import telnetlib +except ModuleNotFoundError: + # telnetlib was removed in Python 3.13, this is handled in telnetClass.py + telnetlib = None import os import shlex, subprocess import traceback @@ -52,8 +56,34 @@ from framework.core.configParser import configParser from framework.core.utilities import utilities from framework.core.decodeParams import decodeParams -from framework.core.capture import capture -from framework.core.webpageController import webpageController +try: + from framework.core.capture import capture +except ModuleNotFoundError as e: + # cv2/pytesseract/PIL are optional dependencies (e.g. in Docker images) + if e.name in ("cv2", "pytesseract", "PIL", "PIL.Image"): + def _capture_missing_dependency(*args, **kwargs): + raise ImportError( + "The 'capture' functionality requires optional dependencies " + "(cv2, pytesseract, and Pillow). One or more of these are not " + f"installed (missing module: {e.name!r}). Please install the " + "required packages to use capture-related features." + ) + + capture = _capture_missing_dependency + else: + raise +try: + from framework.core.webpageController import webpageController +except ModuleNotFoundError as e: + # selenium is an optional dependency (e.g. in Docker images) + if e.name == "selenium": + def _missing_webpage_controller(*args, **kwargs): + raise ImportError( + "webpageController requires the 'selenium' package, which is not installed" + ) + webpageController = _missing_webpage_controller + else: + raise from framework.core.deviceManager import deviceManager class testController(): @@ -149,9 +179,30 @@ def __init__(self, testName="", qcId="", maxRunTime=TEST_MAX_RUN_TIME, level=log #Start the rest of the testControl requirements self.devices = deviceManager(self.slotInfo.config.get("devices"), self.log, self.testLogPath) - # Set up the session from the default console + # Set up the session from the enabled console (not just "default") self.dut = self.devices.getDevice( "dut" ) - self.session = self.dut.getConsoleSession() + + # CRITICAL: Select console based on which is enabled, not hardcoded "default" + # Check console configuration to find the enabled one + console_name = "default" # Fallback to default + try: + # Get the raw config to check enabled status + dut_config = self.slotInfo.config.get("devices", [{}])[0].get("dut", {}) + consoles_config = dut_config.get("consoles", []) + + # Find the enabled console + for console_item in consoles_config: + for name, config in console_item.items(): + if config.get("enabled", False): + console_name = name + self.log.info(f"Using enabled console: {console_name} (type: {config.get('type')})") + break + if console_name != "default": + break + except Exception as e: + self.log.warn(f"Failed to detect enabled console, using default: {e}") + + self.session = self.dut.getConsoleSession(console_name) self.outboundClient = self.dut.outBoundClient self.powerControl = self.dut.powerControl self.commonRemote = self.dut.remoteController