| from pathlib import Path |
| import os |
|
|
| from agent_process_dashboard.procfs import parse_stat, scan_processes |
|
|
|
|
| def stat_line( |
| pid: int, |
| comm: str, |
| *, |
| ppid: int = 1, |
| utime: int = 7, |
| stime: int = 8, |
| start: int = 900, |
| rss: int = 15, |
| ) -> str: |
| return ( |
| f"{pid} ({comm}) S {ppid} {pid} {pid} 0 0 0 0 0 0 0 " |
| f"{utime} {stime} 0 0 20 0 4 0 {start} 4096 {rss}" |
| ) |
|
|
|
|
| def write_proc(proc: Path, pid: int, comm: str, *, ppid: int = 1, start: int = 900) -> None: |
| path = proc / str(pid) |
| path.mkdir() |
| (path / "stat").write_text(stat_line(pid, comm, ppid=ppid, start=start)) |
| (path / "status").write_text(f"Name:\t{comm}\nUid:\t{os.getuid()}\t{os.getuid()}\t{os.getuid()}\t{os.getuid()}\nThreads:\t4\n") |
| (path / "cmdline").write_bytes(f"{comm}\0--api-key\0secret-value\0".encode()) |
| (path / "statm").write_text("20 10 0 0 0 0 0\n") |
| (path / "cgroup").write_text("0::/user.slice\n") |
|
|
|
|
| def test_parse_stat_handles_spaces_and_parentheses(): |
| parsed = parse_stat(stat_line(123, "name with ) parens", ppid=9, start=12345)) |
|
|
| assert parsed.pid == 123 |
| assert parsed.comm == "name with ) parens" |
| assert parsed.ppid == 9 |
| assert parsed.start_time_ticks == 12345 |
| assert parsed.utime_ticks == 7 |
| assert parsed.stime_ticks == 8 |
|
|
|
|
| def test_scan_processes_builds_identity_and_masks_cmdline(tmp_path): |
| proc = tmp_path |
| (proc / "sys/kernel/random").mkdir(parents=True) |
| (proc / "sys/kernel/random/boot_id").write_text("boot-1\n") |
| (proc / "uptime").write_text("1000.00 0.00\n") |
| write_proc(proc, 101, "codex", start=500) |
|
|
| samples = scan_processes(proc, current_uid_only=True) |
|
|
| assert len(samples) == 1 |
| sample = samples[0] |
| assert sample.process_key == f"boot-1:101:500:{os.getuid()}" |
| assert sample.argv == ("codex", "--api-key", "<redacted>") |
| assert sample.argv_masked_count == 1 |
|
|
|
|
| def test_pid_reuse_changes_process_key(tmp_path): |
| proc = tmp_path |
| (proc / "sys/kernel/random").mkdir(parents=True) |
| (proc / "sys/kernel/random/boot_id").write_text("boot-1\n") |
| (proc / "uptime").write_text("1000.00 0.00\n") |
| write_proc(proc, 101, "codex", start=500) |
| first = scan_processes(proc)[0] |
| (proc / "101/stat").write_text(stat_line(101, "codex", start=700)) |
|
|
| second = scan_processes(proc)[0] |
|
|
| assert first.process_key != second.process_key |
|
|
|
|
| def test_missing_stat_is_ignored(tmp_path): |
| proc = tmp_path |
| (proc / "sys/kernel/random").mkdir(parents=True) |
| (proc / "sys/kernel/random/boot_id").write_text("boot-1\n") |
| (proc / "uptime").write_text("1000.00 0.00\n") |
| (proc / "999").mkdir() |
|
|
| assert scan_processes(proc) == [] |
|
|
|
|
