Manages a LangBot subprocess for E2E testing.
| 17 | |
| 18 | |
| 19 | class LangBotProcess: |
| 20 | """Manages a LangBot subprocess for E2E testing.""" |
| 21 | |
| 22 | def __init__( |
| 23 | self, |
| 24 | project_root: Path, |
| 25 | work_dir: Path, |
| 26 | port: int = 15300, |
| 27 | timeout: int = 30, |
| 28 | collect_coverage: bool = True, |
| 29 | ): |
| 30 | self.project_root = project_root |
| 31 | self.work_dir = work_dir # Directory containing data/config.yaml |
| 32 | self.port = port |
| 33 | self.timeout = timeout |
| 34 | self.collect_coverage = collect_coverage |
| 35 | self.process: Optional[subprocess.Popen] = None |
| 36 | self._stdout_data: bytes = b'' |
| 37 | self._stderr_data: bytes = b'' |
| 38 | self._coverage_file: Optional[Path] = None |
| 39 | |
| 40 | def start(self) -> bool: |
| 41 | """Start LangBot process and wait for it to be ready.""" |
| 42 | import httpx |
| 43 | |
| 44 | # Prepare environment |
| 45 | env = os.environ.copy() |
| 46 | env['PYTHONPATH'] = str(self.project_root / 'src') |
| 47 | for proxy_key in ( |
| 48 | 'HTTP_PROXY', |
| 49 | 'HTTPS_PROXY', |
| 50 | 'ALL_PROXY', |
| 51 | 'http_proxy', |
| 52 | 'https_proxy', |
| 53 | 'all_proxy', |
| 54 | ): |
| 55 | env.pop(proxy_key, None) |
| 56 | env['NO_PROXY'] = '127.0.0.1,localhost' |
| 57 | env['no_proxy'] = '127.0.0.1,localhost' |
| 58 | |
| 59 | # Set API port via environment variable |
| 60 | env['API__PORT'] = str(self.port) |
| 61 | env['API__WEBHOOK_PREFIX'] = f'http://127.0.0.1:{self.port}' |
| 62 | |
| 63 | # Disable telemetry |
| 64 | env['SPACE__DISABLE_TELEMETRY'] = 'true' |
| 65 | env['SPACE__DISABLE_MODELS_SERVICE'] = 'true' |
| 66 | |
| 67 | # Build command |
| 68 | if self.collect_coverage: |
| 69 | # Use coverage.py to collect coverage data |
| 70 | # Set COVERAGE_PROCESS_START to enable coverage in subprocess |
| 71 | self._coverage_file = self.work_dir / '.coverage.e2e' |
| 72 | env['COVERAGE_PROCESS_START'] = str(self.project_root / '.coveragerc') |
| 73 | env['COVERAGE_FILE'] = str(self._coverage_file) |
| 74 | |
| 75 | # Create .coveragerc for subprocess |
| 76 | coveragerc_content = """ |
no outgoing calls