Integration class for Docker-based QEMU testing.
| 19 | |
| 20 | |
| 21 | class QEMUTestIntegration: |
| 22 | """Integration class for Docker-based QEMU testing.""" |
| 23 | |
| 24 | def __init__(self): |
| 25 | """Initialize QEMU test integration.""" |
| 26 | self.docker_available = self._check_docker() |
| 27 | |
| 28 | def _check_docker(self) -> bool: |
| 29 | """Check if Docker is available.""" |
| 30 | try: |
| 31 | result = subprocess.run( |
| 32 | [get_docker_command(), "version"], capture_output=True, timeout=5 |
| 33 | ) |
| 34 | return result.returncode == 0 |
| 35 | except (subprocess.SubprocessError, FileNotFoundError): |
| 36 | return False |
| 37 | |
| 38 | def select_runner(self) -> str: |
| 39 | """Select the best available runner. |
| 40 | |
| 41 | Returns: |
| 42 | 'docker' if available, otherwise 'none' |
| 43 | """ |
| 44 | if self.docker_available: |
| 45 | return "docker" |
| 46 | else: |
| 47 | return "none" |
| 48 | |
| 49 | def run_qemu_test( |
| 50 | self, |
| 51 | firmware_path: str | Path, |
| 52 | timeout: int = 30, |
| 53 | interrupt_regex: Optional[str] = None, |
| 54 | machine: str = "esp32", |
| 55 | ) -> int: |
| 56 | """Run QEMU test using Docker. |
| 57 | |
| 58 | Args: |
| 59 | firmware_path: Path to firmware or build directory |
| 60 | timeout: Test timeout in seconds |
| 61 | interrupt_regex: Pattern to interrupt on success |
| 62 | machine: QEMU machine type (esp32, esp32c3, esp32s3) |
| 63 | |
| 64 | Returns: |
| 65 | Exit code (0 for success) |
| 66 | """ |
| 67 | firmware_path = Path(firmware_path) |
| 68 | |
| 69 | runner_type = self.select_runner() |
| 70 | print(f"Selected QEMU runner: {runner_type}") |
| 71 | |
| 72 | if runner_type == "docker": |
| 73 | return self._run_docker_qemu( |
| 74 | firmware_path, timeout, interrupt_regex, machine |
| 75 | ) |
| 76 | else: |
| 77 | print("ERROR: Docker is not available!", file=sys.stderr) |
| 78 | print("Please install Docker to run QEMU tests", file=sys.stderr) |