ROMFASTSQL/proxmox/lxc171-claude-agent/discord-bridge/tests/test_cleanup.py

"""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 cgroup_de_test(monkeypatch):
    """Declara cgroup-ul sesiunii curente drept "cgroup-ul serviciului".

    Procesele de test le pornim din sesiunea noastra (un `tmux-spawn-*.scope`), nu
    dintr-un serviciu; fara asta filtrul de domeniu le-ar exclude corect si niciun
    test cu procese reale n-ar mai avea ce sa gaseasca. Patch-ul spune explicit
    "pentru testul asta, serviciul e aici", si NU slabeste filtrul in productie.
    """
    propriu = cleanup._cgroup(os.getpid())
    frunza = [c for c in propriu.rsplit(":", 1)[-1].split("/") if c][-1]
    monkeypatch.setattr(cleanup, "_service_cgroup_marker", lambda: frunza)
    return frunza


@pytest.fixture()
def fake_claude(cgroup_de_test):
    """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(cgroup_de_test):
    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_familiile_sunt_sortate_dupa_rss_cumulat(fake_claude):
    """Ordonarea e pe FAMILIE (radacina + copii), nu pe proces izolat."""
    fake_claude()
    fake_claude()
    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    familii = []
    for o in orfani:
        if o["depth"] == 0:
            familii.append(o["family_rss_mb"])
    assert familii == sorted(familii, reverse=True)
    # in interiorul unei familii radacina vine prima
    for i, o in enumerate(orfani):
        if o["depth"] > 0:
            assert orfani[i - 1]["root_pid"] == o["root_pid"]


# --- 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:True" 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)
    # numarul exact depinde de bugetul de caractere, nu-l fixam; invariantul e
    # ca restul e numarat si ca mesajul incape in Discord
    assert "si inca " 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/<pid>, 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"


# ==========================================================================
# Arborele unui fir: `claude` + serverele lui MCP (npm/sh/node).
# Masurat in productie: ~440 MB pe fir, nu 300, si copiii NU se numesc `claude`.
# ==========================================================================

@pytest.fixture()
def fake_claude_cu_copii(cgroup_de_test):
    """Un `claude` fals care isi porneste doi copii, ca serverele MCP.

    Totul e `sleep`. Nu se atinge niciun proces real al sistemului.
    """
    import tempfile
    bindir = pathlib.Path(tempfile.mkdtemp(prefix="lanec-tree-"))
    binar = bindir / "claude"
    binar.write_text(
        "#!/bin/sh\n"
        "sleep 300 &\n"
        "sleep 300 &\n"
        "wait\n"
    )
    binar.chmod(0o755)

    pornite = []

    def porneste():
        proc = subprocess.Popen([str(binar)])
        pornite.append(proc)
        # asteptam sa apara amandoi copiii in /proc
        for _ in range(300):
            copii = [p for p, i in cleanup.scan_processes().items() if i["ppid"] == proc.pid]
            if len(copii) >= 2:
                return proc, sorted(copii)
            time.sleep(0.01)
        raise AssertionError("copiii nu au aparut in /proc")

    yield porneste

    for proc in pornite:
        try:
            subprocess.run(["pkill", "-P", str(proc.pid)], timeout=5)
            proc.kill()
            proc.wait(timeout=5)
        except Exception:
            pass
    shutil.rmtree(bindir, ignore_errors=True)


def test_copiii_unui_claude_orfan_sunt_gasiti(fake_claude_cu_copii):
    """Regresie: copiii nu se numesc `claude`, deci un filtru pe nume ii rata."""
    parinte, copii = fake_claude_cu_copii()
    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    gasiti = _pids(orfani)

    assert parinte.pid in gasiti, "radacina claude nu a fost gasita"
    for copil in copii:
        assert copil in gasiti, f"copilul {copil} a scapat (nu se numeste claude)"

    intrari = {o["pid"]: o for o in orfani}
    assert intrari[parinte.pid]["depth"] == 0
    for copil in copii:
        assert intrari[copil]["depth"] == 1
        assert intrari[copil]["parent_pid"] == parinte.pid
        assert intrari[copil]["root_pid"] == parinte.pid
        assert str(parinte.pid) in intrari[copil]["reason"]


def test_rss_ul_familiei_include_copiii(fake_claude_cu_copii):
    parinte, copii = fake_claude_cu_copii()
    intrari = {o["pid"]: o for o in cleanup.find_orphans({"version": 1, "threads": {}})}
    familie = intrari[parinte.pid]["family_rss_mb"]
    suma = intrari[parinte.pid]["rss_mb"] + sum(intrari[c]["rss_mb"] for c in copii)
    assert abs(familie - suma) < 0.2
    assert familie > intrari[parinte.pid]["rss_mb"], "familia trebuie sa fie mai grea decat radacina"


def test_copiii_unui_claude_VIU_din_state_nu_sunt_atinsi(fake_claude_cu_copii):
    """Turul care ruleaza acum: nici radacina, nici serverele lui MCP."""
    parinte, copii = fake_claude_cu_copii()
    _, start = cleanup._parse_stat(parinte.pid)
    stare = _stare_cu(pid=parinte.pid, start_time=start)
    gasiti = _pids(cleanup.find_orphans(stare))
    assert parinte.pid not in gasiti
    for copil in copii:
        assert copil not in gasiti, "am declarat orfan un server MCP al firului viu"


def test_pid_reuse_nu_protejeaza_familia(fake_claude_cu_copii):
    """state.json cu pid_start_time gresit = alt proces; familia ramane orfana."""
    parinte, copii = fake_claude_cu_copii()
    stare = _stare_cu(pid=parinte.pid, start_time=1.0)
    gasiti = _pids(cleanup.find_orphans(stare))
    assert parinte.pid in gasiti
    for copil in copii:
        assert copil in gasiti


def test_min_age_pastreaza_familia_intreaga(fake_claude_cu_copii):
    """Un copil proaspat sub o radacina veche nu se taie din familie."""
    parinte, copii = fake_claude_cu_copii()
    # radacina si copiii au aceeasi varsta aici, deci verificam ambele capete
    assert _pids(cleanup.find_orphans({"threads": {}}, min_age_s=3600)) & ({parinte.pid} | set(copii)) == set()
    gasiti = _pids(cleanup.find_orphans({"threads": {}}, min_age_s=0))
    assert parinte.pid in gasiti and set(copii) <= gasiti


# --- ordinea de omorare ----------------------------------------------------

def test_kill_order_pune_copiii_inaintea_parintilor():
    orfani = [
        {"pid": 100, "cmdline": "claude", "depth": 0},
        {"pid": 101, "cmdline": "npm exec @playwright/mcp@latest", "depth": 1},
        {"pid": 102, "cmdline": "node playwright-mcp", "depth": 2},
    ]
    assert [o["pid"] for o in cleanup._kill_order(orfani)] == [102, 101, 100]


def test_kill_order_recalculeaza_cand_lipseste_depth(fake_claude_cu_copii):
    """Intrari construite de mana, fara `depth`: ordinea vine din /proc."""
    parinte, copii = fake_claude_cu_copii()
    orfani = [{"pid": parinte.pid, "cmdline": "claude"}] + \
             [{"pid": c, "cmdline": "sleep"} for c in copii]
    ordine = [o["pid"] for o in cleanup._kill_order(orfani)]
    assert ordine[-1] == parinte.pid, "parintele trebuie omorat ultimul"
    assert set(ordine[:2]) == set(copii)


def test_kill_orphans_omoara_toata_familia(fake_claude_cu_copii):
    """Copiii intai: altfel se reparenteaza la init si scapa din trecerea asta."""
    parinte, copii = fake_claude_cu_copii()
    orfani = [o for o in cleanup.find_orphans({"version": 1, "threads": {}})
              if o["pid"] == parinte.pid or o["pid"] in copii]
    assert len(orfani) == 3

    rez = cleanup.kill_orphans(orfani, dry_run=False, grace_s=3.0)
    assert [r["pid"] for r in rez][-1] == parinte.pid, "parintele nu a fost ultimul"
    assert all(r["action"] in ("terminated", "killed", "gone") for r in rez), rez

    parinte.wait(timeout=10)
    time.sleep(0.3)
    for copil in copii:
        assert not cleanup._still_same_process({"pid": copil}), f"copilul {copil} a supravietuit"


def test_dry_run_pe_familie_nu_omoara_nimic(fake_claude_cu_copii):
    parinte, copii = fake_claude_cu_copii()
    orfani = [o for o in cleanup.find_orphans({"version": 1, "threads": {}})
              if o["pid"] == parinte.pid or o["pid"] in copii]
    rez = cleanup.kill_orphans(orfani)          # implicit dry_run=True
    assert all(r["action"] == "dry-run" for r in rez)
    time.sleep(0.2)
    assert parinte.poll() is None
    for copil in copii:
        assert cleanup._still_same_process({"pid": copil})


def test_kill_orphans_refuza_procesele_protejate(fake_claude_cu_copii):
    """Plasa de siguranta: chiar daca cineva ne pasaza bot.py, nu-l atingem."""
    parinte, _ = fake_claude_cu_copii()
    rez = cleanup.kill_orphans(
        [{"pid": parinte.pid, "cmdline": "python3 bot.py"}], dry_run=False)
    assert rez[0]["action"] == "skipped"
    time.sleep(0.2)
    assert parinte.poll() is None


# --- instantaneul masurat in productie (clasificare pura, fara procese) ----

# 78848 python bot.py | 79423 claude | 79454 npm | 79482 sh | 79483 node
INSTANTANEU = {
    78848: {"pid": 78848, "ppid": 1, "cmdline": "/home/claude/.claude-discord/venv/bin/python bot.py",
            "age_s": 5000, "rss_mb": 41.0, "start_time": 1000.0,
            "cgroup": "0::/user.slice/.../claude-discord.service"},
    79423: {"pid": 79423, "ppid": 78848, "cmdline": "claude --resume 0a3f9f33 -p", "age_s": 900,
            "rss_mb": 300.0, "start_time": 2000.0,
            "cgroup": "0::/user.slice/.../claude-discord.service"},
    79454: {"pid": 79454, "ppid": 79423, "cmdline": "npm exec @playwright/mcp@latest", "age_s": 890,
            "rss_mb": 52.0, "start_time": 2100.0,
            "cgroup": "0::/user.slice/.../claude-discord.service"},
    79482: {"pid": 79482, "ppid": 79454, "cmdline": "sh -c playwright-mcp", "age_s": 890,
            "rss_mb": 1.0, "start_time": 2110.0,
            "cgroup": "0::/user.slice/.../claude-discord.service"},
    79483: {"pid": 79483, "ppid": 79482, "cmdline": "node /x/playwright-mcp", "age_s": 890,
            "rss_mb": 84.0, "start_time": 2120.0,
            "cgroup": "0::/user.slice/.../claude-discord.service"},
}


@pytest.fixture()
def instantaneu(monkeypatch):
    """Inlocuieste complet /proc cu instantaneul masurat. Zero procese reale."""
    def fals():
        return {k: dict(v) for k, v in INSTANTANEU.items()}

    def stat_fals(pid):
        info = INSTANTANEU.get(pid)
        return (info["ppid"], info["start_time"]) if info else None

    monkeypatch.setattr(cleanup, "scan_processes", fals)
    monkeypatch.setattr(cleanup, "_is_zombie", lambda pid: False)
    monkeypatch.setattr(cleanup, "_parse_stat", stat_fals)
    return INSTANTANEU


def test_instantaneu_firul_mort_lasa_in_urma_440MB(instantaneu):
    """state.json gol: `claude` a murit si nu si-a luat copiii cu el."""
    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    gasiti = _pids(orfani)

    assert 78848 not in gasiti, "bot.py NU are voie sa apara vreodata ca orfan"
    assert gasiti == {79423, 79454, 79482, 79483}

    intrari = {o["pid"]: o for o in orfani}
    assert intrari[79423]["depth"] == 0
    assert intrari[79454]["depth"] == 1 and intrari[79454]["parent_pid"] == 79423
    assert intrari[79483]["depth"] == 3 and intrari[79483]["root_pid"] == 79423
    # 300 + 52 + 1 + 84 = 437, nu 300
    assert intrari[79423]["family_rss_mb"] == 437.0


def test_instantaneu_firul_viu_nu_e_atins(instantaneu):
    """Acelasi instantaneu, dar `claude` e inregistrat viu si valid in state.json."""
    stare = {"version": 1, "threads": {"111": {"pid": 79423, "pid_start_time": 2000.0}}}
    # pid_start_time e verificat din /proc; il facem sa se potriveasca
    orfani = cleanup.find_orphans(stare)
    assert _pids(orfani) == set(), f"am declarat orfan ceva dintr-un fir viu: {orfani}"


def test_instantaneu_copil_reparentat_la_init(instantaneu, monkeypatch):
    """npm a ramas fara parinte (ppid 1) — arborele nu-l mai leaga, cgroup-ul da."""
    def fals():
        snap = {k: dict(v) for k, v in INSTANTANEU.items()}
        del snap[79423]                 # claude a murit
        snap[79454]["ppid"] = 1         # npm reparentat la init
        return snap
    monkeypatch.setattr(cleanup, "scan_processes", fals)

    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    gasiti = _pids(orfani)
    assert 78848 not in gasiti
    assert gasiti == {79454, 79482, 79483}
    intrari = {o["pid"]: o for o in orfani}
    assert "reparentat la init" in intrari[79454]["reason"]
    assert intrari[79483]["root_pid"] == 79454
    assert intrari[79454]["family_rss_mb"] == 137.0


def test_instantaneu_raportul_arata_ierarhia(instantaneu):
    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    text = cleanup.format_report(orfani)

    assert "4 procese orfane" in text
    assert "437" in text, "totalul familiei trebuie sa se vada, nu doar 300"
    assert "`--" in text, "copiii trebuie indentati sub radacina"
    assert len(text) < 2000

    linii = text.splitlines()
    i_radacina = next(i for i, l in enumerate(linii) if "pid=79423" in l)
    i_copil = next(i for i, l in enumerate(linii) if "pid=79454" in l)
    assert i_radacina < i_copil, "radacina trebuie sa apara inaintea copiilor"


def test_state_json_gol_nu_declara_bot_py_orfan(instantaneu):
    """Cea mai periculoasa cale: cleanup rulat cu state.json gol sau corupt."""
    for stare in ({}, {"threads": {}}, {"threads": None}, {"version": 1}):
        assert 78848 not in _pids(cleanup.find_orphans(stare))


def test_raportul_respecta_limita_discord_si_cu_rezultate_lungi():
    """Regresie: cu rezultatele de omorare atasate raportul depasea 2000 de caractere."""
    orfani = []
    for i in range(1, 41):
        orfani.append({"pid": 1000 + i, "cmdline": "claude " + "x" * 200, "age_s": 8000,
                       "rss_mb": 600.0, "depth": 0, "root_pid": 1000 + i,
                       "family_rss_mb": 900.0})
        orfani.append({"pid": 2000 + i, "cmdline": "node " + "y" * 200, "age_s": 8000,
                       "rss_mb": 300.0, "depth": 1, "root_pid": 1000 + i,
                       "family_rss_mb": 900.0})
    rez = [{"pid": o["pid"], "action": "killed",
            "detail": "nu a raspuns la SIGTERM in 3s, SIGKILL"} for o in orfani]

    seac = cleanup.format_report(orfani)
    plin = cleanup.format_report(orfani, rez)
    assert len(seac) < 2000, len(seac)
    assert len(plin) < 2000, len(plin)
    assert "si inca" in plin
    assert plin.rstrip().endswith("```")


# ==========================================================================
# DOMENIUL: doar cgroup-ul serviciului.
#
# Pe LXC 171 ruleaza permanent sesiuni Claude interactive care nu au nicio
# legatura cu puntea. Prima versiune a acestui modul le raporta pe toate ca
# orfane — `/cleanup force:True` din Discord si-ar fi omorat propria sesiune
# plus tot ce rula omul in tmux. Testele de mai jos exista ca sa pice imediat
# daca cineva scoate vreodata filtrul pe cgroup.
# ==========================================================================

CG_TMUX_1 = ("0::/user.slice/user-1000.slice/user@1000.service/"
             "tmux-spawn-d3c24a89-1111-2222-3333-444444444444.scope")
CG_TMUX_2 = ("0::/user.slice/user-1000.slice/user@1000.service/"
             "tmux-spawn-bdd8316f-5555-6666-7777-888888888888.scope")
CG_SERVICIU = ("0::/user.slice/user-1000.slice/user@1000.service/"
               "app.slice/claude-discord.service")

# Instantaneul real de pe container: doua sesiuni de lucru + puntea.
INSTANTANEU_MIXT = {
    22280: {"pid": 22280, "ppid": 1, "cmdline": "claude --dangerously-skip-permissions",
            "age_s": 8199, "rss_mb": 586.2, "start_time": 100.0, "cgroup": CG_TMUX_1},
    22417: {"pid": 22417, "ppid": 22280, "cmdline": "npm exec @playwright/mcp@latest",
            "age_s": 8198, "rss_mb": 51.6, "start_time": 110.0, "cgroup": CG_TMUX_1},
    5501: {"pid": 5501, "ppid": 1, "cmdline": "claude --dangerously-skip-permissions",
           "age_s": 21413, "rss_mb": 637.1, "start_time": 200.0, "cgroup": CG_TMUX_2},
    78848: {"pid": 78848, "ppid": 1,
            "cmdline": "/home/claude/.claude-discord/venv/bin/python bot.py",
            "age_s": 5000, "rss_mb": 41.0, "start_time": 300.0, "cgroup": CG_SERVICIU},
    79423: {"pid": 79423, "ppid": 78848, "cmdline": "claude --resume 0a3f9f33 -p",
            "age_s": 900, "rss_mb": 300.0, "start_time": 400.0, "cgroup": CG_SERVICIU},
}


@pytest.fixture()
def instantaneu_mixt(monkeypatch):
    def fals():
        return {k: dict(v) for k, v in INSTANTANEU_MIXT.items()}

    def stat_fals(pid):
        info = INSTANTANEU_MIXT.get(pid)
        return (info["ppid"], info["start_time"]) if info else None

    monkeypatch.setattr(cleanup, "scan_processes", fals)
    monkeypatch.setattr(cleanup, "_parse_stat", stat_fals)
    monkeypatch.setattr(cleanup, "_is_zombie", lambda pid: False)
    monkeypatch.setattr(cleanup, "_service_cgroup_marker", lambda: "claude-discord.service")
    return INSTANTANEU_MIXT


def test_sesiunile_de_lucru_din_tmux_nu_sunt_niciodata_raportate(instantaneu_mixt):
    """Regresie dura: doar `claude`-ul puntii, niciodata cele din tmux-spawn."""
    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    gasiti = _pids(orfani)

    assert gasiti == {79423}, f"domeniul e gresit, s-a raportat: {gasiti}"
    assert 22280 not in gasiti, "sesiune de lucru din tmux raportata ca orfana"
    assert 5501 not in gasiti, "sesiunea principala a utilizatorului raportata ca orfana"
    assert 22417 not in gasiti, "copil al unei sesiuni de lucru raportat ca orfan"
    assert 78848 not in gasiti, "bot.py raportat ca orfan"


def test_dry_run_ul_nu_arata_nici_macar_sesiunile_de_lucru(instantaneu_mixt):
    """Nici in raport nu au voie sa apara: omul n-ar trebui sa fie tentat."""
    orfani = cleanup.find_orphans({"version": 1, "threads": {}})
    text = cleanup.format_report(orfani, cleanup.kill_orphans(orfani))
    for pid in (22280, 5501, 22417, 78848):
        assert str(pid) not in text, f"pid {pid} nu are ce cauta in raport"


def test_cgroup_necitibil_e_tratat_ca_neeligibil(instantaneu_mixt, monkeypatch):
    """Fail-closed: mai bine ratam un orfan decat sa omoram sesiunea cuiva."""
    def fals():
        snap = {k: dict(v) for k, v in INSTANTANEU_MIXT.items()}
        snap[79423]["cgroup"] = ""       # /proc/<pid>/cgroup ilizibil
        return snap
    monkeypatch.setattr(cleanup, "scan_processes", fals)
    assert cleanup.find_orphans({"version": 1, "threads": {}}) == []


def test_cgroup_lipsa_din_dictionar_e_tot_neeligibil(instantaneu_mixt, monkeypatch):
    def fals():
        snap = {k: dict(v) for k, v in INSTANTANEU_MIXT.items()}
        del snap[79423]["cgroup"]
        return snap
    monkeypatch.setattr(cleanup, "scan_processes", fals)
    assert cleanup.find_orphans({"version": 1, "threads": {}}) == []


# --- derivarea numelui de unitate -----------------------------------------

def test_marker_derivat_cand_rulam_ca_serviciu(monkeypatch):
    """Redenumirea unitului se propaga singura."""
    monkeypatch.setattr(cleanup, "_cgroup", lambda pid:
                        "0::/user.slice/user-1000.slice/user@1000.service/"
                        "app.slice/punte-discord.service")
    assert cleanup._service_cgroup_marker() == "punte-discord.service"


def test_marker_cade_pe_constanta_intr_o_sesiune_interactiva(monkeypatch):
    """Din tmux NU avem voie sa derivam: am tinti chiar sesiunea omului."""
    monkeypatch.setattr(cleanup, "_cgroup", lambda pid: CG_TMUX_1)
    assert cleanup._service_cgroup_marker() == "claude-discord.service"


def test_marker_nu_urca_niciodata_la_user_service(monkeypatch):
    """user@1000.service ar cuprinde TOATA sesiunea de utilizator."""
    monkeypatch.setattr(cleanup, "_cgroup", lambda pid:
                        "0::/user.slice/user-1000.slice/user@1000.service")
    assert cleanup._service_cgroup_marker() == "claude-discord.service"


def test_marker_pe_cgroup_gol_sau_ilizibil(monkeypatch):
    monkeypatch.setattr(cleanup, "_cgroup", lambda pid: "")
    assert cleanup._service_cgroup_marker() == "claude-discord.service"

    def explodeaza(pid):
        raise OSError("proc restrictionat")
    monkeypatch.setattr(cleanup, "_cgroup", explodeaza)
    assert cleanup._service_cgroup_marker() == "claude-discord.service"


# --- pid_start_time: ticks vs secunde -------------------------------------

def test_pid_start_time_acceptat_in_ambele_conventii():
    """Masurat pe firul viu 89112: /proc da 6906587 ticks, state.json 69065.87 s.

    INTERFACES.md cere ticks; Lane A scrie secunde. Comparatia asta PROTEJEAZA un
    proces, deci in caz de dubiu acceptam ambele — altfel declaram orfan un fir viu
    si ii omoram serverele MCP la mijlocul turului.
    """
    assert cleanup._start_time_matches(6906587.0, 6906587.0)    # ticks
    assert cleanup._start_time_matches(6906587.0, 69065.87)     # secunde
    assert not cleanup._start_time_matches(6906587.0, 1.0)      # chiar alt proces
    assert not cleanup._start_time_matches(6906587.0, None)
    assert not cleanup._start_time_matches(6906587.0, "aiurea")


def test_firul_viu_e_protejat_desi_state_json_scrie_secunde(instantaneu_mixt, monkeypatch):
    """Exact cazul de pe container: fara toleranta, tot arborele firului viu cadea."""
    def fals():
        snap = {k: dict(v) for k, v in INSTANTANEU_MIXT.items()}
        snap[79454] = {"pid": 79454, "ppid": 79423,
                       "cmdline": "npm exec @playwright/mcp@latest", "age_s": 890,
                       "rss_mb": 52.0, "start_time": 410.0, "cgroup": CG_SERVICIU}
        return snap
    monkeypatch.setattr(cleanup, "scan_processes", fals)

    # state.json noteaza start_time-ul in SECUNDE (400 ticks / 100)
    stare = {"version": 1, "threads": {"1": {"pid": 79423, "pid_start_time": 4.0}}}
    assert cleanup.find_orphans(stare) == [], "am declarat orfan un fir viu"

    # acelasi lucru scris in ticks trebuie sa mearga la fel
    stare_ticks = {"version": 1, "threads": {"1": {"pid": 79423, "pid_start_time": 400.0}}}
    assert cleanup.find_orphans(stare_ticks) == []