"""Teste pentru cleanup.py (T13). Testele ating /proc-ul real, dar NUMAI cu procese pe care le pornesc ele insele: copii ai lui `sleep` redenumiti `claude`. Nu se atinge niciodata un proces al sistemului si nu se apeleaza `kill_orphans(dry_run=False)` decat pe acesti copii. """ import os import pathlib import shutil import subprocess import sys import time import pytest sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent)) import cleanup # noqa: E402 # --- ajutoare: procese inofensive ------------------------------------------ @pytest.fixture() def fake_claude(): """Un binar `claude` fals (copie de sleep) + pornirea/oprirea proceselor. NU folosim `tmp_path`: calea lui contine "pytest", care e in NEVER_KILL, si procesul de test ar iesi protejat din greseala. """ import tempfile bindir = pathlib.Path(tempfile.mkdtemp(prefix="lanec-bin-")) binar = bindir / "claude" shutil.copy(shutil.which("sleep") or "/bin/sleep", binar) binar.chmod(0o755) pornite = [] def porneste(secunde="60"): proc = subprocess.Popen([str(binar), secunde]) pornite.append(proc) # asteptam sa apara in /proc cu cmdline complet for _ in range(50): if cleanup._cmdline(proc.pid): break time.sleep(0.01) return proc yield porneste for proc in pornite: try: proc.kill() proc.wait(timeout=5) except Exception: pass shutil.rmtree(bindir, ignore_errors=True) def _stare_cu(pid=None, start_time=None): thread = {"sid": "x", "cwd": "/workspace", "model": "sonnet"} if pid is not None: thread["pid"] = pid if start_time is not None: thread["pid_start_time"] = start_time return {"version": 1, "threads": {"111": thread}, "cost": {"day": "2026-08-30", "usd": 0.0}} def _pids(orphans): return {o["pid"] for o in orphans} # --- citirea /proc --------------------------------------------------------- def test_parse_stat_pe_procesul_curent(): ppid, starttime = cleanup._parse_stat(os.getpid()) assert ppid == os.getppid() assert starttime > 0 def test_parse_stat_pid_inexistent(): assert cleanup._parse_stat(4_000_000) is None def test_parse_stat_suporta_comm_cu_spatii(tmp_path, monkeypatch): """comm-ul e intre paranteze si poate contine spatii/paranteze.""" fals = tmp_path / "777" fals.mkdir() # campurile 3..24: state, ppid, apoi umplutura pana la campul 22 = starttime campuri = ["S", "42"] + ["0"] * 17 + ["987654"] + ["0", "0"] (fals / "stat").write_text("777 (nume ciudat (x)) " + " ".join(campuri) + "\n") monkeypatch.setattr(cleanup, "PROC", tmp_path) assert cleanup._parse_stat(777) == (42, 987654.0) def test_rss_si_cmdline_pentru_procesul_curent(): assert cleanup._rss_mb(os.getpid()) > 0 assert "python" in cleanup._cmdline(os.getpid()).lower() def test_scan_processes_contine_procesul_curent(): procs = cleanup.scan_processes() assert os.getpid() in procs info = procs[os.getpid()] assert set(info) >= {"pid", "ppid", "cmdline", "age_s", "rss_mb", "start_time", "cgroup"} def test_scan_processes_doar_uid_ul_curent(): procs = cleanup.scan_processes() # pid 1 apartine altui utilizator in acest container assert all(p > 0 for p in procs) assert cleanup._uid(os.getpid()) == os.getuid() # --- clasificare ----------------------------------------------------------- def test_is_claude_recunoaste_variantele(): assert cleanup._is_claude("claude -p --resume abc") assert cleanup._is_claude("/home/claude/.nvm/versions/node/v20.19.6/bin/claude -p") assert cleanup._is_claude("node /home/x/node_modules/@anthropic-ai/claude-code/bin/claude") def test_is_claude_nu_confunda_home_ul_utilizatorului(): # utilizatorul se numeste `claude`, deci caile lui contin cuvantul assert not cleanup._is_claude("/usr/bin/python3 /home/claude/script.py") assert not cleanup._is_claude("sleep 300") def test_procese_protejate_nu_sunt_orfane(): assert cleanup._is_protected("/usr/lib/systemd/systemd --user") assert cleanup._is_protected("sshd: claude@pts/0") assert cleanup._is_protected("python3 bot.py") def test_gaseste_proces_claude_neinregistrat(fake_claude): proc = fake_claude() orfani = cleanup.find_orphans({"version": 1, "threads": {}}) assert proc.pid in _pids(orfani) entry = next(o for o in orfani if o["pid"] == proc.pid) assert set(entry) >= {"pid", "cmdline", "age_s", "rss_mb"} assert entry["age_s"] >= 0 assert "neinregistrat" in entry["reason"] def test_procesul_din_state_json_nu_e_orfan(fake_claude): proc = fake_claude() stare = _stare_cu(pid=proc.pid) assert proc.pid not in _pids(cleanup.find_orphans(stare)) def test_pid_start_time_care_nu_se_potriveste_nu_protejeaza(fake_claude): """PID reuse: state.json crede ca stie pid-ul, dar e alt proces acum.""" proc = fake_claude() stare = _stare_cu(pid=proc.pid, start_time=1.0) # start_time vechi, gresit assert proc.pid in _pids(cleanup.find_orphans(stare)) def test_pid_start_time_corect_protejeaza(fake_claude): proc = fake_claude() _, start = cleanup._parse_stat(proc.pid) stare = _stare_cu(pid=proc.pid, start_time=start) assert proc.pid not in _pids(cleanup.find_orphans(stare)) def test_descendentii_unui_proces_cunoscut_sunt_protejati(fake_claude): """Copiii turului care ruleaza acum nu au voie sa fie declarati orfani.""" proc = fake_claude() # declaram procesul curent (pytest) ca fiind procesul firului; copilul lui # `proc` e descendentul lui, deci protejat stare = _stare_cu(pid=os.getpid()) assert proc.pid not in _pids(cleanup.find_orphans(stare)) def test_procesul_curent_nu_e_niciodata_orfan(): orfani = cleanup.find_orphans({"version": 1, "threads": {}}) assert os.getpid() not in _pids(orfani) def test_min_age_filtreaza_procesele_proaspete(fake_claude): proc = fake_claude() orfani = cleanup.find_orphans({"version": 1, "threads": {}}, min_age_s=3600) assert proc.pid not in _pids(orfani) def test_state_aiurea_nu_arunca(): for stare in ({}, {"threads": None}, {"threads": {"a": None}}, {"threads": {"a": {"pid": "nu-i numar"}}}, {"threads": {"a": {"pid": -5}}}): assert isinstance(cleanup.find_orphans(stare), list) def test_orfanii_sunt_sortati_dupa_rss(fake_claude): fake_claude() fake_claude() orfani = cleanup.find_orphans({"version": 1, "threads": {}}) rss = [o["rss_mb"] for o in orfani] assert rss == sorted(rss, reverse=True) # --- oprire ---------------------------------------------------------------- def test_dry_run_e_implicit_si_nu_omoara_nimic(fake_claude): proc = fake_claude() orfani = [o for o in cleanup.find_orphans({"version": 1, "threads": {}}) if o["pid"] == proc.pid] rez = cleanup.kill_orphans(orfani) # fara dry_run explicit assert rez[0]["action"] == "dry-run" time.sleep(0.2) assert proc.poll() is None, "procesul a fost omorat desi era rulare seaca" def test_kill_orphans_opreste_efectiv_cu_force(fake_claude): proc = fake_claude() orfani = [o for o in cleanup.find_orphans({"version": 1, "threads": {}}) if o["pid"] == proc.pid] assert orfani, "procesul de test nu a fost gasit ca orfan" rez = cleanup.kill_orphans(orfani, dry_run=False, grace_s=3.0) assert rez[0]["action"] in ("terminated", "killed") assert proc.wait(timeout=5) is not None def test_proces_deja_disparut_e_raportat_gone(fake_claude): proc = fake_claude() _, start = cleanup._parse_stat(proc.pid) orfan = {"pid": proc.pid, "cmdline": "claude", "age_s": 1, "rss_mb": 1, "start_time": start} proc.kill() proc.wait(timeout=5) rez = cleanup.kill_orphans([orfan], dry_run=False) assert rez[0]["action"] == "gone" def test_pid_reuse_impiedica_omorarea_gresita(fake_claude): """start_time nepotrivit => refuzam sa omoram, chiar cu dry_run=False.""" proc = fake_claude() orfan = {"pid": proc.pid, "cmdline": "claude", "age_s": 1, "rss_mb": 1, "start_time": 1.0} # alt proces, evident rez = cleanup.kill_orphans([orfan], dry_run=False) assert rez[0]["action"] == "gone" time.sleep(0.2) assert proc.poll() is None, "am omorat un proces cu start_time nepotrivit" def test_kill_orphans_refuza_procesul_curent(): _, start = cleanup._parse_stat(os.getpid()) orfan = {"pid": os.getpid(), "cmdline": "pytest", "age_s": 1, "rss_mb": 1, "start_time": start} rez = cleanup.kill_orphans([orfan], dry_run=False) assert rez[0]["action"] == "skipped" def test_kill_orphans_pe_lista_goala(): assert cleanup.kill_orphans([]) == [] assert cleanup.kill_orphans(None) == [] def test_kill_orphans_intrare_aiurea_nu_arunca(): rez = cleanup.kill_orphans([{"cmdline": "fara pid"}, {"pid": None}], dry_run=False) assert all(r["action"] == "gone" for r in rez) # --- raport ---------------------------------------------------------------- def test_raport_gol(): assert "Niciun proces orfan" in cleanup.format_report([]) def test_raport_cu_orfani(): orfani = [{"pid": 1234, "cmdline": "claude -p --resume abc", "age_s": 900, "rss_mb": 406.0}] text = cleanup.format_report(orfani) assert "1 procese orfane" in text assert "1234" in text and "406" in text assert "--force" in text def test_raport_cu_rezultate_si_sub_limita_discord(): orfani = [{"pid": i, "cmdline": "claude -p " + "x" * 200, "age_s": i, "rss_mb": 406.0} for i in range(1, 41)] rez = [{"pid": o["pid"], "action": "killed", "detail": "SIGKILL"} for o in orfani] text = cleanup.format_report(orfani, rez) assert "si inca 25" in text assert len(text) < 2000, "raportul depaseste limita de mesaj Discord" assert "killed" in text def test_zombie_nu_e_orfan_si_nu_se_omoara(fake_claude): """Un copil terminat dar nereaped are inca /proc/, dar e mort.""" proc = fake_claude() proc.terminate() for _ in range(200): if cleanup._is_zombie(proc.pid): break time.sleep(0.01) assert cleanup._is_zombie(proc.pid), "nu am reusit sa produc un zombi" assert proc.pid not in _pids(cleanup.find_orphans({"version": 1, "threads": {}})) _, start = cleanup._parse_stat(proc.pid) rez = cleanup.kill_orphans( [{"pid": proc.pid, "cmdline": "claude", "age_s": 1, "rss_mb": 0, "start_time": start}], dry_run=False, ) assert rez[0]["action"] == "gone" def test_varsta_e_calculata_corect_in_container(fake_claude): """Regresie: /proc/uptime e virtualizat de lxcfs, `starttime` nu. Cu scaderea naiva, un proces pornit acum iesea cu varsta negativa (deci 0) si filtrul `min_age_s` devenea inutil. Referinta corecta e `btime`. """ proc = fake_claude() info = cleanup.scan_processes()[proc.pid] assert 0 <= info["age_s"] < 60, f"varsta absurda: {info['age_s']}s" eu = cleanup.scan_processes()[os.getpid()] assert eu["age_s"] < 3600, "procesul de test pare mai vechi de o ora"