|
2 | 2 |
|
3 | 3 | import subprocess |
4 | 4 | import pytest |
| 5 | +import os |
5 | 6 |
|
6 | 7 |
|
7 | | -@pytest.fixture |
8 | | -def demo_terminal(): |
9 | | - """Run demo/terminal.py with pre-written input and capture output.""" |
10 | | - import os |
11 | | - import tempfile |
| 8 | +class DemoTimeoutError(Exception): |
| 9 | + """Custom exception for demo timeouts with screen debugging.""" |
| 10 | + |
| 11 | + def __init__(self, message, stdout="", stderr="", height=24): |
| 12 | + super().__init__(message) |
| 13 | + self.stdout = stdout |
| 14 | + self.stderr = stderr |
| 15 | + |
| 16 | + # Extract last `height` rows from stdout for screen debugging |
| 17 | + if stdout: |
| 18 | + lines = stdout.split("\n") |
| 19 | + last_lines = lines[-height:] if len(lines) > height else lines |
| 20 | + screen_content = "\n".join(last_lines) |
| 21 | + enhanced_message = f"{message}\n\nLast {len(last_lines)} rows of screen:\n{screen_content}" |
| 22 | + else: |
| 23 | + enhanced_message = f"{message}\n\nNo stdout captured" |
| 24 | + |
| 25 | + self.args = (enhanced_message,) |
| 26 | + |
12 | 27 |
|
13 | | - # Path to demo script relative to project root |
| 28 | +def _run_demo(input_commands, timeout=2.0): |
| 29 | + """Internal function to run demo and return output.""" |
14 | 30 | demo_path = os.path.join(os.path.dirname(__file__), "..", "..", "demo", "terminal.py") |
15 | 31 |
|
16 | | - # Create a temporary file for output |
17 | | - with tempfile.NamedTemporaryFile(mode="w+", delete=False) as output_file: |
18 | | - output_path = output_file.name |
| 32 | + try: |
| 33 | + result = subprocess.run( |
| 34 | + ["python3", demo_path], |
| 35 | + input=input_commands, |
| 36 | + capture_output=True, |
| 37 | + text=True, |
| 38 | + timeout=timeout, |
| 39 | + cwd=os.path.dirname(demo_path), |
| 40 | + ) |
| 41 | + return result.stdout |
| 42 | + except subprocess.TimeoutExpired as e: |
| 43 | + # Convert to our custom exception with screen debugging |
| 44 | + stdout = e.stdout.decode("utf-8") if e.stdout else "" |
| 45 | + stderr = e.stderr.decode("utf-8") if e.stderr else "" |
| 46 | + raise DemoTimeoutError(f"Demo timed out after {timeout}s", stdout=stdout, stderr=stderr) from e |
| 47 | + |
| 48 | + |
| 49 | +@pytest.fixture |
| 50 | +def assert_demo_output(): |
| 51 | + """Assert that demo output contains expected text, with nice screen dump on failure.""" |
| 52 | + |
| 53 | + def _assert(commands, expected, timeout=2.0): |
| 54 | + """Run demo with commands and assert output contains expected text. |
| 55 | +
|
| 56 | + Args: |
| 57 | + commands: Commands to send to demo (include \r\n for newlines) |
| 58 | + expected: String or list of strings that should appear in output |
| 59 | + timeout: Timeout in seconds (default 2.0) |
| 60 | + """ |
| 61 | + # Ensure commands end with exit |
| 62 | + if not commands.strip().endswith("exit"): |
| 63 | + commands = commands.rstrip() + "\r\nexit\r\n" |
19 | 64 |
|
20 | | - class DemoRunner: |
21 | | - def __init__(self, demo_path, output_path): |
22 | | - self.demo_path = demo_path |
23 | | - self.output_path = output_path |
| 65 | + output = _run_demo(commands, timeout) |
24 | 66 |
|
25 | | - def run_with_input(self, input_commands): |
26 | | - """Run demo with given input commands and return output.""" |
27 | | - process = subprocess.Popen( |
28 | | - f"printf '{input_commands}' | python3 {self.demo_path} > {self.output_path}", |
29 | | - shell=True, |
30 | | - cwd=os.path.dirname(self.demo_path), |
31 | | - ) |
32 | | - process.wait() |
| 67 | + # Handle both string and list expectations |
| 68 | + if isinstance(expected, str): |
| 69 | + expected = [expected] |
33 | 70 |
|
34 | | - with open(self.output_path, "r") as f: |
35 | | - return f.read() |
| 71 | + # Check each expected string |
| 72 | + for exp in expected: |
| 73 | + if exp not in output: |
| 74 | + # Pretty screen dump on failure |
| 75 | + lines = output.split("\n") |
| 76 | + screen_display = "\n".join(lines[-24:] if len(lines) > 24 else lines) |
36 | 77 |
|
37 | | - runner = DemoRunner(demo_path, output_path) |
| 78 | + pytest.fail( |
| 79 | + f"Expected '{exp}' not found in output\n\n" |
| 80 | + f"=== Last {min(24, len(lines))} rows of screen ===\n" |
| 81 | + f"{screen_display}\n" |
| 82 | + f"=== End of screen ===\n\n" |
| 83 | + f"Full output ({len(output)} chars):\n{repr(output)}" |
| 84 | + ) |
38 | 85 |
|
39 | | - yield runner |
| 86 | + return output # Return for additional checks if needed |
40 | 87 |
|
41 | | - # Cleanup |
42 | | - os.unlink(output_path) |
| 88 | + return _assert |
0 commit comments