| from __future__ import annotations | |
| import subprocess | |
| from backend import process_scanner | |
| def test_parse_ps_output_normalizes_fields_and_detects_harness_hints(): | |
| output = """ | |
| PID PPID TT ELAPSED COMMAND COMMAND | |
| 101 100 pts/4 12 codex /usr/bin/codex resume thread-1 | |
| 102 100 ? 21 python3 /opt/openclaude run | |
| 103 100 pts/8 7 bash bash | |
| """ | |
| records = process_scanner.parse_ps_output(output) | |
| assert [record.pid for record in records] == [101, 102, 103] | |
| assert records[0].tty == "pts/4" | |
| assert records[0].elapsed_seconds == 12 | |
| assert records[0].harness_hint == "codex" | |
| assert records[1].tty is None | |
| assert records[1].harness_hint == "openclaude" | |
| assert records[2].harness_hint is None | |
| def test_scan_processes_filters_to_harnesses_and_finds_terminal_parent(monkeypatch): | |
| output = """ | |
| PID PPID TT ELAPSED COMMAND COMMAND | |
| 200 1 pts/1 500 kitty kitty | |
| 201 200 pts/1 320 bash bash | |
| 202 201 pts/1 15 codex codex --cwd /work/codex | |
| 300 1 pts/2 480 gnome-terminal-server /usr/lib/gnome-terminal-server | |
| 301 300 pts/2 22 python3 /usr/local/bin/opencode dashboard | |
| 400 1 ? 40 sleep sleep 60 | |
| """ | |
| def fake_run(*args, **kwargs): | |
| return subprocess.CompletedProcess(args=args, returncode=0, stdout=output) | |
| def fake_readlink(path: str) -> str: | |
| mapping = { | |
| "/proc/202/cwd": "/work/codex", | |
| "/proc/301/cwd": "/work/opencode", | |
| } | |
| return mapping[path] | |
| monkeypatch.setattr(process_scanner.subprocess, "run", fake_run) | |
| monkeypatch.setattr(process_scanner.os, "readlink", fake_readlink) | |
| records = process_scanner.scan_processes() | |
| assert [record.pid for record in records] == [202, 301] | |
| assert records[0].cwd == "/work/codex" | |
| assert records[0].terminal_program == "kitty" | |
| assert records[0].harness_hint == "codex" | |
| assert records[1].cwd == "/work/opencode" | |
| assert records[1].terminal_program == "gnome-terminal-server" | |
| assert records[1].harness_hint == "opencode" | |
| def test_scan_processes_tolerates_proc_cwd_read_failures(monkeypatch): | |
| output = """ | |
| PID PPID TT ELAPSED COMMAND COMMAND | |
| 500 1 pts/3 80 alacritty alacritty | |
| 501 500 pts/3 60 bash bash | |
| 502 501 pts/3 9 hermes hermes serve | |
| """ | |
| def fake_run(*args, **kwargs): | |
| return subprocess.CompletedProcess(args=args, returncode=0, stdout=output) | |
| def fake_readlink(path: str) -> str: | |
| raise PermissionError(path) | |
| monkeypatch.setattr(process_scanner.subprocess, "run", fake_run) | |
| monkeypatch.setattr(process_scanner.os, "readlink", fake_readlink) | |
| records = process_scanner.scan_processes() | |
| assert len(records) == 1 | |
| assert records[0].pid == 502 | |
| assert records[0].cwd is None | |
| assert records[0].terminal_program == "alacritty" | |
| assert records[0].harness_hint == "hermes" | |