feat(discord-bridge): punte Discord -> Claude Code pe LXC 171

Implementeaza planul claude-master-plan-discord-bridge-20260830 (15 taskuri,
3 lane-uri paralele) — un bot subtire discord.py peste CLI-ul `claude`, cu
proces persistent per fir alimentat pe stdin cu --input-format stream-json.

Nucleu: runner (proces persistent + reaper 20min + respawn --resume), stream
(parser tolerant), session_store (scriere atomica, lock per fir, detectare PID
reuse, recovery), limits (max 4 procese, timeout tur, rate per user, plafon cost
pe zi), render (un loop de editare per canal, interval adaptiv).

Adaptor: allowlist guild/canal/user fail-closed cu respingerea webhook-urilor,
comenzi !new/!cd/!model/!status/!stop/!cleanup, cost si model in subsolul
fiecarui raspuns. Mesajul sosit in timpul unui tur devine steering, nu tur nou.

Securitate: hook PreToolUse fail-closed care cere confirmare in Discord pentru
operatiuni ireversibile, wrapper `infra` cu lista explicita de hosturi. Deny
rules raman strat cosmetic, nu bariera (verificat: /usr/bin/ssh trece pe langa).

Ops: alerte email pe conventia repo-ului, !cleanup pentru orfani, unit systemd
user cu KillMode=control-group si limite de memorie, install.sh idempotent.

Verificat: 275 teste fara retea/Discord/API (10.8s), identic cu si fara
discord.py instalat; e2e pe CLI real confirma steering-ul mid-tur (mesaj la 6s
intr-un tool call de 25s schimba raspunsul final).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01B29CApsP1JkSdjYaGaHpE7
This commit is contained in:
Claude Agent
2026-08-30 10:44:39 +00:00
parent d7b4007af8
commit d466f358ce
43 changed files with 7767 additions and 0 deletions

View File

@@ -0,0 +1,64 @@
"""Fixturi comune: totul in tmp, zero retea, zero Discord, zero API."""
from __future__ import annotations
import os
import pathlib
import sys
import pytest
ROOT = pathlib.Path(__file__).resolve().parent.parent
if str(ROOT) not in sys.path:
sys.path.insert(0, str(ROOT))
import config # noqa: E402
FAKE_CLAUDE = str(pathlib.Path(__file__).resolve().parent / "fake_claude.py")
@pytest.fixture(autouse=True)
def state_dir(tmp_path, monkeypatch):
"""Muta ~/.claude-discord in tmp pentru fiecare test."""
d = tmp_path / "claude-discord"
d.mkdir()
monkeypatch.setenv("CLAUDE_DISCORD_DIR", str(d))
config.reload(d)
config.ensure_dirs()
yield d
config.reload()
@pytest.fixture
def store(state_dir):
import session_store
s = session_store.SessionStore(state_dir / "state.json", alerter=lambda *a, **k: None)
s.load()
return s
@pytest.fixture
def fake_bin():
"""Comanda care inlocuieste `claude` in teste."""
return [sys.executable, FAKE_CLAUDE]
@pytest.fixture
def scenario(monkeypatch):
def _set(name: str, **env):
monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", name)
for k, v in env.items():
monkeypatch.setenv(k, str(v))
return _set
@pytest.fixture(autouse=True)
def _no_real_claude(monkeypatch, request):
"""Nicio suita implicita nu are voie sa cheme `claude --version`."""
if "e2e" in request.keywords:
return
import stream
monkeypatch.setattr(stream, "_cli_version_cache", "test-cli", raising=False)
monkeypatch.setattr(stream, "cli_version", lambda: "test-cli")

View File

@@ -0,0 +1,122 @@
#!/usr/bin/env python3
"""CLI `claude` fals pentru teste: citeste JSONL pe stdin, emite JSONL pe stdout.
Scenariul se alege din FAKE_CLAUDE_SCENARIO:
normal assistant + result pentru fiecare mesaj
unknown tip necunoscut + linie non-JSON inainte de assistant/result
tools tool_use + tool_result + text + result
eof se termina inainte de `result`
crash scrie pe stderr si iese cu cod 1
env raporteaza CLAUDE_DISCORD_THREAD_ID / _SESSION_ID din mediul propriu
slow "lucreaza" FAKE_CLAUDE_DELAY secunde, colecteaza mesajele venite intre timp
(asa se testeaza steering-ul mid-tur) si le enumera in `result`
"""
from __future__ import annotations
import json
import os
import select
import sys
import time
def emit(obj) -> None:
sys.stdout.write(json.dumps(obj, ensure_ascii=False) + "\n")
sys.stdout.flush()
def parse_args(argv):
sid, model = None, "sonnet"
for i, a in enumerate(argv):
if a == "--resume" and i + 1 < len(argv):
sid = argv[i + 1]
elif a == "--model" and i + 1 < len(argv):
model = argv[i + 1]
return sid, model
def text_of(line: str) -> str:
try:
obj = json.loads(line)
except ValueError:
return line.strip()
msg = obj.get("message") or {}
content = msg.get("content")
if isinstance(content, list):
return "".join(b.get("text", "") for b in content if isinstance(b, dict))
return str(content or "")
def assistant(sid: str, text: str) -> None:
emit({"type": "assistant", "session_id": sid,
"message": {"role": "assistant", "content": [{"type": "text", "text": text}]}})
def result(sid: str, text: str, ms: int = 10) -> None:
emit({"type": "result", "subtype": "success", "session_id": sid, "is_error": False,
"duration_ms": ms, "num_turns": 1, "result": text,
"total_cost_usd": float(os.environ.get("FAKE_CLAUDE_COST", "0.0123"))})
def drain(deadline: float) -> list[str]:
"""Citeste orice mesaj suplimentar pana la deadline (steering mid-tur)."""
extra = []
while True:
left = deadline - time.monotonic()
if left <= 0:
return extra
r, _, _ = select.select([sys.stdin], [], [], min(left, 0.05))
if r:
line = sys.stdin.readline()
if not line:
return extra
extra.append(text_of(line))
def main() -> int:
scenario = os.environ.get("FAKE_CLAUDE_SCENARIO", "normal")
sid, model = parse_args(sys.argv[1:])
sid = sid or os.environ.get("FAKE_CLAUDE_SID", "sid-fake-0001")
emit({"type": "system", "subtype": "init", "session_id": sid, "model": model,
"cwd": os.getcwd(), "tools": ["Bash", "Read"]})
if scenario == "crash":
sys.stderr.write("fake_claude: boom, ies cu 1\n")
sys.stderr.flush()
return 1
for line in sys.stdin:
prompt = text_of(line)
if scenario == "unknown":
emit({"type": "rate_limit_event", "session_id": sid, "detail": "test"})
sys.stdout.write("asta nu e JSON\n")
sys.stdout.flush()
if scenario == "tools":
emit({"type": "assistant", "session_id": sid, "message": {"role": "assistant", "content": [
{"type": "tool_use", "id": "tu_1", "name": "Bash", "input": {"command": "ls"}}]}})
emit({"type": "user", "session_id": sid, "message": {"role": "user", "content": [
{"type": "tool_result", "tool_use_id": "tu_1", "content": "fisier1\n"}]}})
if scenario == "eof":
assistant(sid, "incep si mor")
return 0
if scenario == "slow":
delay = float(os.environ.get("FAKE_CLAUDE_DELAY", "1.0"))
assistant(sid, "lucrez...")
extra = drain(time.monotonic() + delay)
msgs = [prompt] + extra
result(sid, "mesaje primite: " + " | ".join(msgs), int(delay * 1000))
continue
if scenario == "env":
assistant(sid, "raportez mediul")
result(sid, "thread={} sesiune={} marker={}".format(
os.environ.get("CLAUDE_DISCORD_THREAD_ID", "-"),
os.environ.get("CLAUDE_DISCORD_SESSION_ID", "-"),
os.environ.get("MARKER_DE_TEST", "-")))
continue
assistant(sid, f"ecou: {prompt}")
result(sid, f"ecou: {prompt}")
return 0
if __name__ == "__main__":
sys.exit(main())

View File

@@ -0,0 +1,281 @@
"""Teste pentru alerts.py (T12). Fara retea, fara email real."""
import os
import pathlib
import sys
import time
import pytest
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import alerts # noqa: E402
@pytest.fixture()
def sandbox(tmp_path, monkeypatch):
"""Muta starea si logurile in tmp, si inlocuieste trimiterea reala de email."""
monkeypatch.setattr(alerts, "STATE_DIR", tmp_path)
monkeypatch.setattr(alerts, "LOG_DIR", tmp_path / "logs")
monkeypatch.setattr(alerts, "DEDUP_FILE", tmp_path / "alerts-dedup.json")
monkeypatch.setattr(alerts, "LOG_FILE", tmp_path / "logs" / "alerts.log")
sent = []
real_send = alerts._send_mail
def fake_send(subject_line, body):
sent.append((subject_line, body))
return True, "trimis (fals)"
monkeypatch.setattr(alerts, "_send_mail", fake_send)
return {"dir": tmp_path, "sent": sent, "real_send": real_send}
def _log_text(sandbox):
path = sandbox["dir"] / "logs" / "alerts.log"
return path.read_text() if path.exists() else ""
# --- format ----------------------------------------------------------------
def test_subiect_are_prefixul_de_nivel(sandbox):
alerts.alert("CRITICAL", "state.json corupt", "detalii")
assert sandbox["sent"][0][0] == "[CRITICAL] state.json corupt"
def test_nivel_necunoscut_devine_warn(sandbox):
alerts.alert("chestii", "ceva", "corp")
assert sandbox["sent"][0][0].startswith("[WARN] ")
def test_nivel_case_insensitive(sandbox):
alerts.alert("info", "ceva", "corp")
assert sandbox["sent"][0][0].startswith("[INFO] ")
def test_corpul_contine_contextul_de_host_si_sursa(sandbox):
alerts.alert("WARN", "subiect", "linia mea")
body = sandbox["sent"][0][1]
assert "linia mea" in body
assert "Host:" in body and "Sursa:" in body
def test_corp_urias_e_trunchiat(sandbox):
alerts.alert("WARN", "s", "x" * (alerts.MAX_BODY_CHARS + 5000))
assert "corp trunchiat" in sandbox["sent"][0][1]
def test_subiect_multiline_e_pliat_pe_o_linie(sandbox):
alerts.alert("WARN", "prima\nlinia a doua", "corp")
assert "\n" not in sandbox["sent"][0][0]
# --- dedup -----------------------------------------------------------------
def test_dedup_blocheaza_a_doua_alerta_cu_aceeasi_cheie(sandbox):
alerts.alert("WARN", "proces mort", "a", dedup_key="proc-died:42")
alerts.alert("WARN", "proces mort", "b", dedup_key="proc-died:42")
assert len(sandbox["sent"]) == 1
assert "SKIP(dedup" in _log_text(sandbox)
def test_dedup_e_pe_cheie_nu_pe_subiect(sandbox):
alerts.alert("WARN", "proces mort", "a", dedup_key="proc-died:1")
alerts.alert("WARN", "proces mort", "b", dedup_key="proc-died:2")
assert len(sandbox["sent"]) == 2
def test_fara_dedup_key_se_trimite_de_fiecare_data(sandbox):
for _ in range(3):
alerts.alert("INFO", "acelasi subiect", "corp")
assert len(sandbox["sent"]) == 3
def test_dedup_expira_dupa_o_ora(sandbox, monkeypatch):
alerts.alert("WARN", "cost", "a", dedup_key="cost-cap")
assert len(sandbox["sent"]) == 1
real_time = time.time
monkeypatch.setattr(alerts.time, "time", lambda: real_time() + alerts.DEDUP_WINDOW_S + 10)
alerts.alert("WARN", "cost", "b", dedup_key="cost-cap")
assert len(sandbox["sent"]) == 2
def test_dedup_persistat_pe_disc(sandbox):
alerts.alert("WARN", "x", "y", dedup_key="k")
assert (sandbox["dir"] / "alerts-dedup.json").exists()
def test_dedup_curata_intrarile_expirate(sandbox):
import json
vechi = {"expirat": time.time() - 2 * alerts.DEDUP_WINDOW_S}
(sandbox["dir"] / "alerts-dedup.json").write_text(json.dumps(vechi))
alerts.alert("WARN", "x", "y", dedup_key="nou")
data = json.loads((sandbox["dir"] / "alerts-dedup.json").read_text())
assert "expirat" not in data and "nou" in data
def test_dedup_corupt_nu_opreste_alerta(sandbox):
(sandbox["dir"] / "alerts-dedup.json").write_text("{{{ nu e json")
alerts.alert("CRITICAL", "important", "corp", dedup_key="k")
assert len(sandbox["sent"]) == 1
# --- degradare si robustete ------------------------------------------------
def test_lipsa_binarului_mail_nu_arunca_si_ramane_in_log(sandbox, monkeypatch):
# aici vrem trimiterea REALA, ca sa vedem degradarea cand `mail` lipseste
monkeypatch.setattr(alerts, "_send_mail", sandbox["real_send"])
monkeypatch.setattr(alerts, "_mail_binary", lambda: None)
alerts.alert("CRITICAL", "fara mail", "corpul contine ceva important")
text = _log_text(sandbox)
assert "NESENT" in text
assert "binarul `mail` lipseste" in text
assert "corpul contine ceva important" in text
def test_esecul_mail_nu_arunca(sandbox, monkeypatch):
monkeypatch.setattr(alerts, "_send_mail", lambda s, b: (False, "cod 1"))
alerts.alert("WARN", "esec", "corp") # nu trebuie sa arunce
assert "NESENT" in _log_text(sandbox)
def test_exceptie_interna_nu_scapa_din_alert(sandbox, monkeypatch):
def explodeaza(*args, **kwargs):
raise RuntimeError("boom")
monkeypatch.setattr(alerts, "_send_mail", explodeaza)
alerts.alert("CRITICAL", "boom", "corp") # contractul: NU arunca niciodata
def test_alert_nu_arunca_nici_cand_logul_e_inaccesibil(sandbox, monkeypatch):
# LOG_DIR indica un fisier, deci mkdir si open esueaza amandoua
fisier = sandbox["dir"] / "blocaj"
fisier.write_text("x")
monkeypatch.setattr(alerts, "LOG_DIR", fisier / "logs")
monkeypatch.setattr(alerts, "LOG_FILE", fisier / "logs" / "alerts.log")
monkeypatch.setattr(alerts, "_send_mail", lambda s, b: (False, "nimic"))
alerts.alert("WARN", "x", "y")
def test_argumente_aiurea_nu_arunca(sandbox):
alerts.alert(None, None, None) # type: ignore[arg-type]
alerts.alert("", "", "", dedup_key="")
alerts.alert(123, 456, 789) # type: ignore[arg-type]
# --- destinatar ------------------------------------------------------------
def test_destinatar_implicit_root(monkeypatch):
monkeypatch.delenv("ALERT_RECIPIENT", raising=False)
monkeypatch.setattr(alerts, "_config", None)
assert alerts._recipient() == "root"
def test_destinatar_din_mediu(monkeypatch):
monkeypatch.setattr(alerts, "_config", None)
monkeypatch.setenv("ALERT_RECIPIENT", "ops@romfast.ro")
assert alerts._recipient() == "ops@romfast.ro"
def test_destinatar_din_config_daca_exista(monkeypatch):
class FakeConfig:
@staticmethod
def get(key, default=None):
return "din-config@romfast.ro" if key == "ALERT_RECIPIENT" else default
monkeypatch.setattr(alerts, "_config", FakeConfig)
assert alerts._recipient() == "din-config@romfast.ro"
def test_config_care_arunca_nu_rupe_destinatarul(monkeypatch):
class BadConfig:
@staticmethod
def get(key, default=None):
raise RuntimeError("config stricat")
monkeypatch.setattr(alerts, "_config", BadConfig)
monkeypatch.delenv("ALERT_RECIPIENT", raising=False)
assert alerts._recipient() == "root"
# --- ajutoarele pentru conditiile Lane A ----------------------------------
def test_ajutoarele_produc_alerte_cu_dedup(sandbox):
alerts.alert_process_died("123", 999, "SIGKILL")
alerts.alert_crash_loop("123", 5, 300)
alerts.alert_cost_cap(5.12, 5.00)
alerts.alert_state_corrupt("/tmp/state.json")
alerts.alert_orphans([{"pid": 1, "cmdline": "claude", "age_s": 10, "rss_mb": 406}])
assert len(sandbox["sent"]) == 5
niveluri = [s[0].split("]")[0] + "]" for s in sandbox["sent"]]
assert "[CRITICAL]" in niveluri and "[WARN]" in niveluri
# a doua oara acelasi lucru nu se retrimite
alerts.alert_cost_cap(5.20, 5.00)
assert len(sandbox["sent"]) == 5
def test_alert_orphans_cu_lista_goala_nu_arunca(sandbox):
alerts.alert_orphans([])
assert len(sandbox["sent"]) == 1
# --- integrare reala cu subprocess (fara retea) ---------------------------
def test_send_mail_foloseste_argumentele_corecte(tmp_path, monkeypatch):
"""Verifica linia de comanda construita, cu un `mail` fals scris pe disc."""
fake = tmp_path / "mail"
marker = tmp_path / "primit.txt"
fake.write_text(
"#!/bin/sh\n"
f'{{ echo "ARGS: $@"; cat; }} > "{marker}"\n'
)
fake.chmod(0o755)
monkeypatch.setattr(alerts, "_mail_binary", lambda: str(fake))
monkeypatch.setattr(alerts, "_config", None)
monkeypatch.setenv("ALERT_RECIPIENT", "test@local")
ok, detail = alerts._send_mail("[WARN] subiect de test", "corp de test")
assert ok, detail
text = marker.read_text()
assert "ARGS: -s [WARN] subiect de test test@local" in text
assert "corp de test" in text
def test_send_mail_cod_de_eroare_e_raportat(tmp_path, monkeypatch):
fake = tmp_path / "mail"
fake.write_text("#!/bin/sh\necho 'MTA indisponibil'\nexit 75\n")
fake.chmod(0o755)
monkeypatch.setattr(alerts, "_mail_binary", lambda: str(fake))
ok, detail = alerts._send_mail("[WARN] x", "y")
assert not ok and "75" in detail and "MTA indisponibil" in detail
def test_send_mail_timeout_e_abandonat(tmp_path, monkeypatch):
fake = tmp_path / "mail"
fake.write_text("#!/bin/sh\nsleep 30\n")
fake.chmod(0o755)
monkeypatch.setattr(alerts, "_mail_binary", lambda: str(fake))
monkeypatch.setattr(alerts, "MAIL_TIMEOUT_S", 0.5)
ok, detail = alerts._send_mail("[WARN] x", "y")
assert not ok and "depasit" in detail
def test_send_mail_binar_inexistent_nu_arunca(monkeypatch):
monkeypatch.setattr(alerts, "_mail_binary", lambda: "/nu/exista/mail")
ok, detail = alerts._send_mail("[WARN] x", "y")
assert not ok and "nu a putut fi lansat" in detail
def test_fisierele_de_stare_sunt_in_state_dir(sandbox):
alerts.alert("INFO", "x", "y", dedup_key="k")
nume = {p.name for p in sandbox["dir"].iterdir()}
assert "alerts-dedup.json" in nume
assert not any(n.endswith(".tmp") for n in nume), "fisier temporar ramas in urma"
assert os.path.exists(sandbox["dir"] / "logs" / "alerts.log")

View File

@@ -0,0 +1,135 @@
"""T1: filtrul de la intrare. Guild / canal / user + webhook + boti.
Regula: refuzul e TACUT (nu raspundem, nu reactionam) si fail-closed
(allowlist lipsa sau goala = nimeni nu are voie).
"""
from __future__ import annotations
import pytest
import bot
try: # rulat ca pachet `tests` sau ca module libere
from tests.test_bot import (CHANNEL, GUILD, USER, FakeChannel, FakeGuild,
FakeMessage, FakeUser, allowed, bridge)
except ImportError: # pragma: no cover
from test_bot import (CHANNEL, GUILD, USER, FakeChannel, FakeGuild,
FakeMessage, FakeUser, allowed, bridge)
__all__ = ["allowed", "bridge"] # fixturi reexportate
def ok(message) -> bool:
return bot.check_message(message, self_id="999").ok
def why(message) -> str:
return bot.check_message(message, self_id="999").reason
# ------------------------------------------------------------------ acceptare
def test_mesaj_permis(allowed):
assert ok(FakeMessage("salut"))
def test_fir_din_canalul_permis_e_acceptat(allowed):
"""Un fir Discord are alt id decat canalul; conteaza si parintele."""
thread = FakeChannel(cid="99999", parent_id=CHANNEL)
assert ok(FakeMessage("salut", channel=thread))
# -------------------------------------------------------------------- refuzuri
def test_webhook_refuzat(allowed):
"""Un webhook scurs care posteaza in canal NU comanda nimic."""
msg = FakeMessage("!cleanup --force", webhook_id="55")
assert not ok(msg)
assert why(msg) == "webhook"
def test_webhook_refuzat_chiar_daca_autorul_pare_permis(allowed):
msg = FakeMessage("salut", author=FakeUser(USER), webhook_id="55")
assert why(msg) == "webhook"
def test_bot_refuzat(allowed):
msg = FakeMessage("salut", author=FakeUser("40", is_bot=True))
assert why(msg) == "bot"
def test_propriul_mesaj_refuzat(allowed):
msg = FakeMessage("salut", author=FakeUser("999"))
assert why(msg) == "propriul mesaj"
def test_guild_strain_refuzat(allowed):
assert "guild" in why(FakeMessage("salut", guild=FakeGuild("777")))
def test_mesaj_privat_refuzat(allowed):
msg = FakeMessage("salut")
msg.guild = None
assert "guild" in why(msg)
def test_canal_strain_refuzat(allowed):
assert "canal" in why(FakeMessage("salut", channel=FakeChannel("777")))
def test_fir_cu_parinte_strain_refuzat(allowed):
thread = FakeChannel(cid="88888", parent_id="777")
assert "canal" in why(FakeMessage("salut", channel=thread))
def test_utilizator_strain_refuzat(allowed):
assert "utilizator" in why(FakeMessage("salut", author=FakeUser("777")))
# ----------------------------------------------------------------- fail-closed
@pytest.mark.parametrize("lipsa", ["DISCORD_GUILD_IDS", "DISCORD_CHANNEL_IDS", "DISCORD_USER_IDS"])
def test_allowlist_incompleta_refuza_pe_toata_lumea(allowed, monkeypatch, lipsa):
monkeypatch.setenv(lipsa, "")
assert "fail-closed" in why(FakeMessage("salut"))
def test_fara_nicio_allowlist_nimeni_nu_are_voie(monkeypatch):
for key in ("DISCORD_GUILD_IDS", "DISCORD_CHANNEL_IDS", "DISCORD_USER_IDS",
"DISCORD_GUILD_ID", "DISCORD_CHANNEL_ID", "DISCORD_USER_ID"):
monkeypatch.delenv(key, raising=False)
assert not ok(FakeMessage("salut"))
def test_se_accepta_si_forma_singulara_a_cheilor(monkeypatch):
for key in ("DISCORD_GUILD_IDS", "DISCORD_CHANNEL_IDS", "DISCORD_USER_IDS"):
monkeypatch.delenv(key, raising=False)
monkeypatch.setenv("DISCORD_GUILD_ID", GUILD)
monkeypatch.setenv("DISCORD_CHANNEL_ID", CHANNEL)
monkeypatch.setenv("DISCORD_USER_ID", USER)
assert ok(FakeMessage("salut"))
def test_liste_cu_virgula_si_spatii(monkeypatch):
monkeypatch.setenv("DISCORD_GUILD_IDS", f" 1, {GUILD} ,2")
monkeypatch.setenv("DISCORD_CHANNEL_IDS", f"{CHANNEL},3")
monkeypatch.setenv("DISCORD_USER_IDS", f"4 {USER}")
assert ok(FakeMessage("salut"))
# --------------------------------------------------- refuzul e tacut si logat
async def test_refuzul_nu_raspunde_si_nu_reactioneaza(bridge, caplog):
ch = FakeChannel()
msg = FakeMessage("!status", channel=ch, author=FakeUser("777"))
with caplog.at_level("WARNING"):
assert await bridge.handle_message(msg) == "rejected"
assert ch.sent == [] # niciun raspuns
assert msg.reactions == [] # nicio reactie
assert bridge.rejected and "utilizator" in bridge.rejected[0]
assert "respins" in caplog.text
async def test_webhookul_nu_declanseaza_comenzi(bridge):
ch = FakeChannel()
msg = FakeMessage("!cleanup --force", channel=ch, webhook_id="55")
assert await bridge.handle_message(msg) == "rejected"
assert ch.sent == []
assert bridge.runner.procs == {}

View File

@@ -0,0 +1,269 @@
"""Adaptorul Discord: falsuri pentru discord.py, zero retea, zero API.
Fisierul contine si falsurile folosite de test_allowlist.py si test_commands.py.
"""
from __future__ import annotations
import asyncio
import pytest
import bot
import limits as limits_mod
import runner as runner_mod
GUILD, CHANNEL, USER = "100", "200", "300"
# ------------------------------------------------------------------ falsuri
class FakeUser:
def __init__(self, uid=USER, is_bot=False):
self.id = uid
self.bot = is_bot
self.display_name = f"user-{uid}"
class FakeSent:
"""Mesajul returnat de channel.send(), editabil ca in discord.py."""
def __init__(self, channel, content, **kw):
self.channel = channel
self.content = content
self.kwargs = kw
self.edits: list[str] = []
self.id = f"sent-{id(self)}"
async def edit(self, content=None, **kw):
if content is not None:
self.content = content
self.edits.append(content)
return self
class FakeChannel:
def __init__(self, cid=CHANNEL, parent_id=None):
self.id = cid
self.parent_id = parent_id
self.sent: list[FakeSent] = []
async def send(self, content=None, **kw):
msg = FakeSent(self, content, **kw)
self.sent.append(msg)
return msg
@property
def texts(self) -> list[str]:
return [m.content or "" for m in self.sent]
@property
def all_text(self) -> str:
return "\n".join(self.texts + [e for m in self.sent for e in m.edits])
class FakeGuild:
def __init__(self, gid=GUILD):
self.id = gid
class FakeMessage:
def __init__(self, content="salut", *, author=None, channel=None, guild=None,
webhook_id=None, mid="m1"):
self.content = content
self.author = author if author is not None else FakeUser()
self.channel = channel if channel is not None else FakeChannel()
self.guild = guild if guild is not None else FakeGuild()
self.webhook_id = webhook_id
self.id = mid
self.reactions: list[str] = []
async def add_reaction(self, emoji):
self.reactions.append(emoji)
# ------------------------------------------------------------------ fixturi
@pytest.fixture
def allowed(monkeypatch):
"""Allowlist completa in mediu (config.get cade pe os.environ)."""
monkeypatch.setenv("DISCORD_GUILD_IDS", GUILD)
monkeypatch.setenv("DISCORD_CHANNEL_IDS", CHANNEL)
monkeypatch.setenv("DISCORD_USER_IDS", USER)
return {"guild": GUILD, "channel": CHANNEL, "user": USER}
@pytest.fixture
async def bridge(store, fake_bin, allowed, monkeypatch):
monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "normal")
monkeypatch.setenv("FAKE_CLAUDE_COST", "0.0123")
mgr = runner_mod.RunnerManager(
store, claude_bin=fake_bin, is_inflight=store.is_inflight, poll_s=3600
)
lim = limits_mod.Limits(
store, cost_cap=10.0, rate_per_min=100, max_procs=2,
alerter=lambda *a, **k: None,
)
br = bot.Bridge(store, mgr, lim, self_id="999")
br.render.kw = {"min_interval": 0.05, "max_interval": 0.2} # teste rapide
yield br
await br.shutdown()
# -------------------------------------------------------------------- teste
async def test_tur_normal_are_subsol_cu_model_durata_si_cost(bridge):
msg = FakeMessage("cat fac 2+2?")
assert await bridge.handle_message(msg) == "ok"
final = msg.channel.sent[0].content
assert "ecou: cat fac 2+2?" in final
assert "sonnet" in final and "$0.0123 tur" in final and "$0.0123 fir" in final
async def test_costul_se_acumuleaza_pe_fir_si_pe_zi(bridge, store):
ch = FakeChannel()
for i in range(2):
await bridge.handle_message(FakeMessage(f"mesaj {i}", channel=ch, mid=f"m{i}"))
tid = str(ch.id)
assert store.thread(tid)["cost_usd_total"] == pytest.approx(0.0246)
assert store.cost_today() == pytest.approx(0.0246)
assert "$0.0246 fir" in ch.sent[1].content
async def test_inflight_curatat_si_sid_persistat(bridge, store):
msg = FakeMessage("salut")
await bridge.handle_message(msg)
tid = str(msg.channel.id)
assert store.is_inflight(tid) is False
assert store.thread(tid)["sid"] == "sid-fake-0001"
async def test_mesaj_in_timpul_turului_e_steering_nu_tur_nou(bridge, monkeypatch):
monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "slow")
monkeypatch.setenv("FAKE_CLAUDE_DELAY", "1.5")
ch = FakeChannel()
first = FakeMessage("prima", channel=ch, mid="m1")
task = asyncio.create_task(bridge.handle_message(first))
tid = str(ch.id)
for _ in range(200): # asteptam sa intre turul in zbor
proc = bridge.runner.procs.get(tid)
if proc is not None and proc.alive and proc.inflight:
break
await asyncio.sleep(0.02)
else: # pragma: no cover
pytest.fail("turul nu a pornit")
await asyncio.sleep(0.25) # lasam CLI-ul fals sa consume primul mesaj
second = FakeMessage("steering", channel=ch, mid="m2")
assert await bridge.handle_message(second) == "steered"
assert second.reactions == ["➡️"]
assert bridge.steered == 1
assert await task == "ok"
# mesajul de steering a ajuns in acelasi tur, nu a deschis unul nou
assert "prima | steering" in bridge.last_result.text
assert len(bridge.runner.procs) == 1
async def test_plafonul_de_cost_opreste_botul_si_o_spune_in_fir(bridge, store):
bridge.limits.cost_cap = 0.001
store.add_cost(None, 0.5)
msg = FakeMessage("mai fa ceva")
assert await bridge.handle_message(msg) == "cost-cap"
assert "plafon" in msg.channel.all_text.lower()
assert bridge.runner.procs == {}
async def test_plafonul_atins_dupa_tur_e_anuntat(bridge):
bridge.limits.cost_cap = 0.005 # sub costul unui tur fals (0.0123)
msg = FakeMessage("un tur scump")
assert await bridge.handle_message(msg) == "ok"
assert "Ma opresc" in msg.channel.all_text
async def test_rate_limit_per_utilizator(bridge):
bridge.limits.rate_per_min = 1
ch = FakeChannel()
assert await bridge.handle_message(FakeMessage("unu", channel=ch, mid="1")) == "ok"
assert await bridge.handle_message(FakeMessage("doi", channel=ch, mid="2")) == "rate-limited"
assert "prea multe mesaje" in ch.all_text
async def test_tur_esuat_raspunde_in_fir(bridge, monkeypatch):
monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "eof")
msg = FakeMessage("ceva")
assert await bridge.handle_message(msg) == "failed"
assert "esuat" in msg.channel.all_text
assert bridge.store.is_inflight(str(msg.channel.id)) is False
async def test_sweep_la_pornire_anunta_turul_pierdut_fara_reluare(bridge, store, monkeypatch):
ch = FakeChannel(cid="200")
store.update_thread("200", pid=999999, pid_start_time=1.0)
store.set_inflight("200", "t1", USER, "m1")
bridge.get_channel = lambda cid: ch if str(cid) == "200" else None
lost = await bridge.startup()
assert [x["thread_id"] for x in lost] == ["200"]
assert "pierdut" in ch.all_text
assert store.is_inflight("200") is False
assert bridge.runner.procs == {} # niciun tur nu a fost repornit
async def test_raspuns_lung_devine_atasament(bridge, store, monkeypatch):
lung = "x" * 7000
async def fake_turn(prompt, on_event=None, timeout=None):
await on_event(bot.stream_mod.AssistantText(text=lung))
return runner_mod.TurnOutcome(
result=bot.stream_mod.Result(0.01, 100, False, 1, text=lung)
)
proc = bridge.runner.get("200")
monkeypatch.setattr(proc, "run_turn", fake_turn)
msg = FakeMessage("da-mi mult text")
assert await bridge.handle_message(msg) == "ok"
assert "raspuns lung" in msg.channel.sent[0].content # previzualizarea
if bot.discord is not None: # atasamentul propriu-zis
assert any(m.kwargs.get("file") is not None for m in msg.channel.sent)
# ------------------------------------------------------------------ aprobari
def test_decizia_cere_allowlist(bridge, allowed):
assert "allowlist" in bridge.decide("777", "req-1", "allow")
def test_decizia_ajunge_la_lane_b(bridge, allowed, monkeypatch):
calls = []
monkeypatch.setattr(
bot.approvals, "submit_decision",
lambda rid, dec: calls.append((rid, dec)) or True,
)
out = bridge.decide(USER, "req-1", "allow")
assert calls == [("req-1", "allow")]
assert "Permis" in out
def test_cerere_inexistenta_nu_arunca(bridge, allowed):
assert "nu mai exista" in bridge.decide(USER, "req-inexistent", "deny")
async def test_botul_porneste_si_fara_modulele_lui_b_si_c(bridge, monkeypatch):
"""Lane B/C absente: pornire normala, doar fara aprobari si fara !cleanup."""
monkeypatch.setattr(bot, "approvals", None)
monkeypatch.setattr(bot, "cleanup", None)
assert bridge.wire_approvals() is False
ch = FakeChannel()
await bridge.handle_message(FakeMessage("!cleanup", channel=ch))
assert "nu e disponibil" in ch.all_text
assert bridge.approval_view("req-1") is None
assert await bridge.handle_message(FakeMessage("salut", channel=ch)) == "ok"
async def test_cererea_de_aprobare_posteaza_butoane_in_fir(bridge, allowed):
ch = FakeChannel()
bridge.get_channel = lambda cid: ch
await bridge.on_approval_request(
{"request_id": "r1", "thread_id": "200", "tool_name": "Bash", "command": "rm -rf /tmp/x"}
)
assert "Confirmare ceruta" in ch.texts[0]
assert "rm -rf /tmp/x" in ch.texts[0]
if bot.discord is not None:
assert ch.sent[0].kwargs.get("view") is not None

View File

@@ -0,0 +1,313 @@
"""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/<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"

View File

@@ -0,0 +1,231 @@
"""Comenzile puntii: !new (+ --fork), !cd, !model, !status, !stop, !cleanup, !help."""
from __future__ import annotations
import pytest
import bot
try:
from tests.test_bot import (CHANNEL, USER, FakeChannel, FakeMessage, allowed, bridge)
except ImportError: # pragma: no cover
from test_bot import (CHANNEL, USER, FakeChannel, FakeMessage, allowed, bridge)
__all__ = ["allowed", "bridge"]
async def run(bridge, text, ch=None):
ch = ch or FakeChannel()
msg = FakeMessage(text, channel=ch)
label = await bridge.handle_message(msg)
return label, ch
# ------------------------------------------------------------------- parser
def test_parse_command():
cmd = bot.parse_command("!cd /workspace/romfastsql")
assert cmd.name == "cd" and cmd.rest == "/workspace/romfastsql"
assert bot.parse_command("!MODEL opus").name == "model"
assert bot.parse_command("salut") is None
assert bot.parse_command("!") is None
async def test_comanda_necunoscuta(bridge):
label, ch = await run(bridge, "!inexistenta")
assert label == "unknown-command"
assert "nu exista" in ch.all_text
async def test_help(bridge):
label, ch = await run(bridge, "!help")
assert label == "cmd:help"
for c in ("!new", "!cd", "!model", "!status", "!stop", "!cleanup"):
assert c in ch.all_text
# ---------------------------------------------------------------------- !new
async def test_new_sterge_sesiunea(bridge, store):
ch = FakeChannel()
await run(bridge, "salut", ch) # creeaza sesiunea
assert store.thread(str(ch.id))["sid"] == "sid-fake-0001"
label, _ = await run(bridge, "!new", ch)
assert label == "cmd:new"
assert store.thread(str(ch.id))["sid"] is None
assert bridge.runner.procs[str(ch.id)].alive is False
async def test_new_fork_pastreaza_contextul(bridge, store):
ch = FakeChannel()
await run(bridge, "salut", ch)
await run(bridge, "!new --fork", ch)
assert store.thread(str(ch.id))["sid"] == "sid-fake-0001"
assert "fork" in ch.all_text
assert bridge.runner.procs[str(ch.id)].sid == "sid-fake-0001"
# ----------------------------------------------------------------------- !cd
async def test_cd_oriunde_in_workspace(bridge, store, tmp_path):
target = tmp_path / "proiect-nou"
target.mkdir()
label, ch = await run(bridge, f"!cd {target}")
assert label == "cmd:cd"
assert store.thread(str(ch.id))["cwd"] == str(target)
assert str(target) in ch.all_text
async def test_cd_refuza_ce_nu_e_director(bridge, store):
label, ch = await run(bridge, "!cd /nu/exista/nicaieri")
assert "nu e un director" in ch.all_text
assert store.thread(str(ch.id))["cwd"] != "/nu/exista/nicaieri"
async def test_cd_fara_argument(bridge):
_, ch = await run(bridge, "!cd")
assert "Foloseste" in ch.all_text
async def test_cd_se_aplica_procesului_urmator(bridge, tmp_path):
ch = FakeChannel()
await run(bridge, "salut", ch)
target = tmp_path / "alt-proiect"
target.mkdir()
await run(bridge, f"!cd {target}", ch)
assert bridge.runner.procs[str(ch.id)].cwd == str(target)
assert bridge.runner.procs[str(ch.id)].alive is False # se reia cu --resume
# -------------------------------------------------------------------- !model
async def test_model_implicit_e_sonnet(bridge, store):
_, ch = await run(bridge, "salut")
assert store.thread(str(ch.id))["model"] == "sonnet"
assert "sonnet" in ch.all_text
async def test_model_opus_per_fir(bridge, store):
ch = FakeChannel()
label, _ = await run(bridge, "!model opus", ch)
assert label == "cmd:model"
assert store.thread(str(ch.id))["model"] == "opus"
# firul vecin (alt thread, acelasi canal-parinte) ramane pe sonnet
ch2 = FakeChannel("201", parent_id=CHANNEL)
await run(bridge, "!status", ch2)
assert "model : sonnet" in ch2.all_text
async def test_modelul_ales_ajunge_in_subsol(bridge):
ch = FakeChannel()
await run(bridge, "!model opus", ch)
await run(bridge, "salut", ch)
assert "opus" in ch.sent[-1].content
async def test_model_necunoscut_refuzat(bridge, store):
ch = FakeChannel()
await run(bridge, "!model gpt", ch)
assert "necunoscut" in ch.all_text
assert store.thread(str(ch.id))["model"] == "sonnet"
async def test_model_fara_argument_arata_curentul(bridge):
_, ch = await run(bridge, "!model")
assert "Model curent" in ch.all_text
# ------------------------------------------------------------------- !status
async def test_status_arata_campurile_cerute(bridge):
ch = FakeChannel()
await run(bridge, "salut", ch)
_, _ = await run(bridge, "!status", ch)
text = ch.sent[-1].content
for camp in ("fir", "sesiune", "director", "model", "proces", "tur in zbor",
"in coada", "cost fir", "cost azi"):
assert camp in text
assert "sid-fake-0001" in text
assert "$0.0123" in text
async def test_status_arata_stderr(bridge, monkeypatch):
ch = FakeChannel()
monkeypatch.setenv("FAKE_CLAUDE_SCENARIO", "crash")
await run(bridge, "salut", ch) # esueaza si lasa stderr in buffer
proc = bridge.runner.procs[str(ch.id)]
proc.stderr_buf.append("fake_claude: boom")
_, _ = await run(bridge, "!status", ch)
assert "boom" in ch.sent[-1].content
async def test_status_pe_fir_fara_proces(bridge):
_, ch = await run(bridge, "!status")
assert "proces : oprit" in ch.all_text
assert "(noua)" in ch.all_text
# --------------------------------------------------------------------- !stop
async def test_stop_opreste_procesul(bridge, store):
ch = FakeChannel()
await run(bridge, "salut", ch)
proc = bridge.runner.procs[str(ch.id)]
assert proc.alive is True # procesul ramane viu intre tururi
await run(bridge, "!stop", ch)
assert "oprit" in ch.all_text
assert proc.alive is False
async def test_stop_fara_proces(bridge):
_, ch = await run(bridge, "!stop")
assert "Nu ruleaza nimic" in ch.all_text
async def test_stop_curata_inflight(bridge, store, monkeypatch):
ch = FakeChannel()
store.set_inflight(str(ch.id), "t1", USER, "m1")
await run(bridge, "!stop", ch)
assert store.is_inflight(str(ch.id)) is False
# ------------------------------------------------------------------ !cleanup
async def test_cleanup_deleaga_in_lane_c(bridge, monkeypatch):
calls = {}
class FakeCleanup:
@staticmethod
def find_orphans(state):
calls["state"] = state
return [{"pid": 1234, "cmdline": "claude -p", "age_s": 10, "rss_mb": 406.0}]
@staticmethod
def kill_orphans(orphans, dry_run=True):
calls["dry_run"] = dry_run
return [{"pid": 1234, "action": "dry-run", "detail": "s-ar trimite SIGTERM"}]
@staticmethod
def format_report(orphans, results=None):
calls["results"] = results
return f"{len(orphans)} orfani"
monkeypatch.setattr(bot, "cleanup", FakeCleanup)
_, ch = await run(bridge, "!cleanup")
assert "1 orfani" in ch.all_text
assert calls["dry_run"] is True and calls["results"] is None
assert "threads" in calls["state"]
_, ch2 = await run(bridge, "!cleanup --force")
assert calls["dry_run"] is False and calls["results"] is not None
async def test_cleanup_care_crapa_nu_doboara_botul(bridge, monkeypatch):
class Boom:
@staticmethod
def find_orphans(state):
raise RuntimeError("proc ilizibil")
monkeypatch.setattr(bot, "cleanup", Boom)
_, ch = await run(bridge, "!cleanup")
assert "a esuat" in ch.all_text
async def test_cleanup_real_ruleaza_in_dry_run(bridge):
"""Modulul real al lui Lane C: nu omoara nimic implicit."""
cleanup = pytest.importorskip("cleanup")
_, ch = await run(bridge, "!cleanup")
assert ch.all_text.strip() != ""

View File

@@ -0,0 +1,408 @@
"""Teste pentru hook-ul PreToolUse si canalul de aprobari (Lane B).
Fara retea, fara Discord, fara CLI real. Toate cererile merg intr-un tmp_path.
"""
from __future__ import annotations
import asyncio
import importlib
import io
import json
import pathlib
import sys
import subprocess
import threading
import time
import pytest
SEC_DIR = pathlib.Path(__file__).resolve().parents[1] / "security"
if str(SEC_DIR) not in sys.path:
sys.path.insert(0, str(SEC_DIR))
import approvals # noqa: E402
import confirm_hook as hook # noqa: E402
@pytest.fixture()
def home(tmp_path, monkeypatch):
"""Un ~/.claude-discord fals, izolat pe test."""
base = tmp_path / ".claude-discord"
(base / "approvals" / "done").mkdir(parents=True)
(base / "logs").mkdir(parents=True)
monkeypatch.setenv("CLAUDE_DISCORD_DIR", str(base))
monkeypatch.setenv("CLAUDE_DISCORD_APPROVAL_TIMEOUT", "1")
importlib.reload(approvals)
importlib.reload(hook)
return base
def payload(cmd: str, tool: str = "Bash") -> str:
return json.dumps(
{
"session_id": "s-1",
"cwd": "/workspace/romfastsql",
"hook_event_name": "PreToolUse",
"tool_name": tool,
"tool_input": {"command": cmd},
}
)
# --------------------------------------------------------------- clasificator
PERICULOASE = [
"rm -rf /tmp/ceva",
"rm -rf /tmp/ceva", # spatii multiple
"sudo rm -rf /tmp/ceva", # sudo in fata
"/bin/rm -rf /tmp/ceva", # cale absoluta
"rm -r -f /tmp/ceva", # optiuni separate
"rm -fr /tmp/ceva", # ordine inversa
'bash -c "rm -rf /tmp/ceva"', # shell imbricat
'sh -c "sudo /bin/rm --recursive --force /tmp/x"',
"echo hai && rm -rf /tmp/x", # al doilea segment
"ls /tmp; rm -rf /tmp/x",
"TERM=dumb rm -rf /tmp/x", # atribuire in fata
"timeout 30 rm -rf /tmp/x",
"dd if=/dev/zero of=/dev/sdb",
"mkfs.ext4 /dev/sdb1",
"wipefs -a /dev/sdb",
"cat imagine.img > /dev/sdb",
"shutdown -h now",
"reboot",
"systemctl poweroff",
"pct destroy 171",
"qm destroy 201",
"zfs destroy rpool/data/vm-201-disk-0",
"zpool destroy rpool",
"pvesm remove backup-nfs",
"lvremove /dev/pve/data",
"sqlplus / as sysdba @drop.sql; echo DROP TABLE ROMFAST.FACTURI",
"echo 'truncate table facturi;' | sqlplus -s user/parola@xe",
"git push --force origin master",
"git push -f",
"git clean -fdx",
"git reset --hard origin/master",
"systemctl stop oracle-xe",
"systemctl stop pveproxy",
"ssh root@10.0.20.36 uptime",
"ssh 10.0.20.201 pct list",
"find /tmp -name '*.log' -delete",
"docker volume rm date-oracle",
"shred -u secret.key",
"pvesh delete /nodes/pvemini/lxc/171",
]
NEPERICULOASE = [
"ls -la /workspace",
"git status",
"git push origin feat/discord-bridge",
"rm /tmp/un-fisier",
"rm -f /tmp/un-fisier",
"grep -rn 'rm -rf' docs/", # doar mentioneaza, nu ruleaza
"pytest tests/ -q",
"echo 'nu sterge nimic' > /tmp/nota.txt",
"cat /dev/null > /tmp/gol",
"python3 -c \"print('drop')\"",
"systemctl status oracle-xe",
"systemctl --user restart claude-discord",
"pct list",
"docker ps",
"df -h",
"echo salut > /dev/null",
]
@pytest.mark.parametrize("cmd", PERICULOASE)
def test_clasificator_prinde_periculoasele(cmd):
verdict = hook.classify("Bash", {"command": cmd})
assert verdict is not None, f"nedetectat: {cmd}"
assert verdict[0] and verdict[1]
@pytest.mark.parametrize("cmd", NEPERICULOASE)
def test_clasificator_lasa_nepericuloasele(cmd):
assert hook.classify("Bash", {"command": cmd}) is None, f"fals pozitiv: {cmd}"
def test_alte_tooluri_trec():
assert hook.classify("Read", {"file_path": "/etc/passwd"}) is None
assert hook.classify("Bash", {}) is None
def test_ghilimele_neinchise_nu_arunca():
# lexerul cade, dar clasificatorul trebuie sa raspunda, nu sa crape
hook.classify("Bash", {"command": 'echo "neinchis'})
# --------------------------------------------------------------- fail-closed
def test_json_corupt_da_deny(home, capsys):
rc = hook.run("{asta nu e json")
out = json.loads(capsys.readouterr().out)
assert rc == 0
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
assert "invalida" in out["hookSpecificOutput"]["permissionDecisionReason"]
def test_payload_care_nu_e_obiect_da_deny(home, capsys):
hook.run("[1, 2, 3]")
out = json.loads(capsys.readouterr().out)
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
def test_director_lipsa_da_deny(tmp_path, monkeypatch, capsys):
monkeypatch.setenv("CLAUDE_DISCORD_DIR", str(tmp_path / "nu-exista"))
monkeypatch.setenv("CLAUDE_DISCORD_APPROVAL_TIMEOUT", "1")
importlib.reload(approvals)
importlib.reload(hook)
hook.run(payload("rm -rf /tmp/x"))
out = json.loads(capsys.readouterr().out)
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
assert "nu pot cere aprobarea" in out["hookSpecificOutput"]["permissionDecisionReason"]
def test_timeout_fara_raspuns_da_deny(home, capsys):
t0 = time.monotonic()
hook.run(payload("rm -rf /tmp/x"))
dur = time.monotonic() - t0
out = json.loads(capsys.readouterr().out)
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
assert "neaprobat in Discord" in out["hookSpecificOutput"]["permissionDecisionReason"]
assert 0.8 < dur < 5.0 # a asteptat secunda configurata, apoi a refuzat
def test_cerere_corupta_pe_disc_da_deny(home, capsys):
"""Daca cineva strica fisierul cererii intre timp, decizia e deny."""
def strica():
for _ in range(40):
for p in (home / "approvals").glob("*.json"):
p.write_text("}{ corupt")
return
time.sleep(0.05)
th = threading.Thread(target=strica)
th.start()
hook.run(payload("pct destroy 171"))
th.join()
out = json.loads(capsys.readouterr().out)
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
def test_exceptie_neasteptata_in_main_da_deny(home, monkeypatch, capsys):
"""Plasa finala din main(): orice exceptie devine deny, nu crash."""
def crapa(_text):
raise RuntimeError("boom")
monkeypatch.setattr(hook, "run", crapa)
monkeypatch.setattr(sys, "stdin", io.StringIO(payload("ls")))
rc = hook.main()
out = json.loads(capsys.readouterr().out)
assert rc == 0
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
assert "eroare interna" in out["hookSpecificOutput"]["permissionDecisionReason"]
# ------------------------------------------------------- fluxul de aprobare
def test_aprobare_externa_permite_executia(home, capsys):
"""Botul aproba dupa ~0.5s; hook-ul trebuie sa raspunda allow."""
def aproba():
for _ in range(60):
reqs = asyncio.run(approvals.pending_requests())
if reqs:
assert approvals.submit_decision(reqs[0]["request_id"], "allow") is True
return
time.sleep(0.05)
th = threading.Thread(target=aproba)
th.start()
hook.run(payload("rm -rf /tmp/directorul-meu"))
th.join()
out = json.loads(capsys.readouterr().out)
assert out["hookSpecificOutput"]["permissionDecision"] == "allow"
assert "aprobat in Discord" in out["hookSpecificOutput"]["permissionDecisionReason"]
# cererea a fost mutata in done/, nu mai apare ca pending
assert asyncio.run(approvals.pending_requests()) == []
assert list((home / "approvals" / "done").glob("*.json"))
def test_refuz_explicit_din_discord(home, capsys):
def refuza():
for _ in range(60):
reqs = asyncio.run(approvals.pending_requests())
if reqs:
approvals.submit_decision(reqs[0]["request_id"], "deny")
return
time.sleep(0.05)
th = threading.Thread(target=refuza)
th.start()
hook.run(payload("qm destroy 201"))
th.join()
out = json.loads(capsys.readouterr().out)
assert out["hookSpecificOutput"]["permissionDecision"] == "deny"
def test_formatul_cererii(home):
"""Formatul fisierului de cerere e contract cu bot.py."""
req = approvals.create_request(
tool_name="Bash",
command="rm -rf /tmp/x",
reason="stergere recursiva",
rule="rm_recursiv",
thread_id="123",
session_id="s-1",
cwd="/workspace",
timeout=300,
)
path = home / "approvals" / f"{req['request_id']}.json"
disc = json.loads(path.read_text())
for k in ("request_id", "thread_id", "tool_name", "command", "created_at",
"expires_at", "status", "rule", "reason", "cwd", "session_id"):
assert k in disc, k
assert disc["status"] == "pending"
pending = asyncio.run(approvals.pending_requests())
assert len(pending) == 1
assert set(pending[0]) >= {"request_id", "thread_id", "tool_name", "command", "created_at"}
def test_comanda_nepericuloasa_nu_scrie_cerere(home, capsys):
rc = hook.run(payload("ls -la"))
assert rc == 0
assert capsys.readouterr().out == "" # fara iesire = flux normal
assert list((home / "approvals").glob("*.json")) == []
# ------------------------------------------------------------- approvals API
def test_submit_pe_cerere_inexistenta(home):
assert approvals.submit_decision("nu-exista", "allow") is False
def test_submit_cu_decizie_invalida_devine_deny(home):
req = approvals.create_request(tool_name="Bash", command="rm -rf /x", timeout=5)
assert approvals.submit_decision(req["request_id"], "poate") is False
assert approvals.read_decision(req["request_id"]) == "deny"
def test_request_id_cu_traversare_e_respins(home):
assert approvals.submit_decision("../../etc/passwd", "allow") is False
assert approvals.read_decision("../evadare") == "deny"
def test_wait_for_decision_timeout_da_deny(home):
req = approvals.create_request(tool_name="Bash", command="rm -rf /x", timeout=5)
assert asyncio.run(approvals.wait_for_decision(req["request_id"], 0.3)) == "deny"
def test_wait_for_decision_allow(home):
req = approvals.create_request(tool_name="Bash", command="rm -rf /x", timeout=5)
approvals.submit_decision(req["request_id"], "allow")
assert asyncio.run(approvals.wait_for_decision(req["request_id"], 1)) == "allow"
def test_scrierea_e_atomica(home):
"""Nu trebuie sa ramana fisiere temporare vizibile dupa scriere."""
approvals.create_request(tool_name="Bash", command="rm -rf /x", timeout=5)
assert [p.name for p in (home / "approvals").glob(".*tmp")] == []
def test_set_on_request_anunta_botul(home):
"""Callback-ul lui Lane A primeste cererea scrisa de hook (alt proces)."""
primite: list[dict] = []
async def scenariu():
async def cb(req):
primite.append(req)
approvals.set_on_request(cb)
approvals.create_request(tool_name="Bash", command="pct destroy 171", timeout=5)
for _ in range(40):
if primite:
break
await asyncio.sleep(0.05)
approvals.set_on_request(None)
asyncio.run(scenariu())
assert primite and primite[0]["command"] == "pct destroy 171"
def test_callback_care_crapa_nu_opreste_urmarirea(home):
ok: list[str] = []
async def scenariu():
async def cb(req):
if not ok:
ok.append("prima")
raise RuntimeError("boom")
ok.append(req["request_id"])
approvals.set_on_request(cb)
approvals.create_request(tool_name="Bash", command="rm -rf /a", timeout=5)
await asyncio.sleep(0.8)
approvals.create_request(tool_name="Bash", command="rm -rf /b", timeout=5)
for _ in range(40):
if len(ok) >= 2:
break
await asyncio.sleep(0.05)
approvals.set_on_request(None)
asyncio.run(scenariu())
assert len(ok) >= 2
# ------------------------------------------------- hook rulat ca alt proces
def test_hook_ca_proces_separat_asteapta_si_primeste_allow(home):
"""Cazul real: hook-ul e alt proces, botul aproba prin directorul de pe disc."""
env = {
"PATH": "/usr/bin:/bin",
"HOME": str(home.parent),
"CLAUDE_DISCORD_DIR": str(home),
"CLAUDE_DISCORD_APPROVAL_TIMEOUT": "20",
"CLAUDE_DISCORD_THREAD_ID": "999",
}
proc = subprocess.Popen(
[sys.executable, str(SEC_DIR / "confirm_hook.py")],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
text=True, env=env,
)
proc.stdin.write(payload("rm -rf /tmp/lane-b-test"))
proc.stdin.close()
rid = None
for _ in range(100):
reqs = asyncio.run(approvals.pending_requests())
if reqs:
rid = reqs[0]["request_id"]
assert reqs[0]["thread_id"] == "999"
approvals.submit_decision(rid, "allow")
break
time.sleep(0.05)
assert rid, "hook-ul nu a scris cererea pe disc"
out = proc.stdout.read()
err = proc.stderr.read()
proc.wait(timeout=30)
assert proc.returncode == 0, err
decision = json.loads(out)["hookSpecificOutput"]
assert decision["permissionDecision"] == "allow"
assert rid in decision["permissionDecisionReason"]
def test_hook_ca_proces_separat_comanda_banala_tace(home):
env = {"PATH": "/usr/bin:/bin", "HOME": str(home.parent),
"CLAUDE_DISCORD_DIR": str(home), "CLAUDE_DISCORD_APPROVAL_TIMEOUT": "5"}
res = subprocess.run(
[sys.executable, str(SEC_DIR / "confirm_hook.py")],
input=payload("ls -la /workspace"), capture_output=True, text=True,
env=env, timeout=30,
)
assert res.returncode == 0
assert res.stdout.strip() == ""

View File

@@ -0,0 +1,62 @@
"""E2E cu CLI-ul `claude` real. Exclus implicit (`addopts = -m "not e2e"`).
Ruleaza-l explicit: pytest -m e2e tests/test_e2e_steering.py
Dureaza ~40s si consuma cota reala; verifica exact ipoteza pe care sta runner-ul:
un mesaj trimis pe stdin IN TIMPUL unui tur ajunge la model si schimba raspunsul final.
"""
from __future__ import annotations
import asyncio
import os
import shutil
import pytest
import runner
pytestmark = [
pytest.mark.e2e,
pytest.mark.skipif(shutil.which("claude") is None, reason="CLI-ul claude nu e instalat"),
]
async def test_steering_mid_tur_pe_cli_real(tmp_path):
p = runner.ClaudeProcess("e2e", str(tmp_path), "sonnet")
try:
prompt = (
"Foloseste unealta Bash cu run_in_background=false si timeout=60000 pentru "
"comanda: python3 -c \"import time; time.sleep(25); print('terminat')\" . "
"Asteapta iesirea in acelasi apel, nu o porni in fundal. "
"Dupa ce ai iesirea, raspunde-mi cu un singur cuvant: ALFA"
)
async def steer():
await asyncio.sleep(6)
await p.send("Schimbare de plan: la final raspunde cu cuvantul BETA, nu ALFA.")
task = asyncio.create_task(steer())
out = await p.run_turn(prompt, timeout=180)
await task
assert out.result is not None and not out.result.is_error
text = (out.result.text or "").upper()
assert "BETA" in text, f"steering-ul nu a ajuns; raspuns: {out.result.text!r}"
assert out.result.total_cost_usd > 0
assert p.sid # session_id-ul vine din system/init
finally:
await p.stop()
async def test_tur_simplu_pe_cli_real(tmp_path):
"""Verifica formatul stream-json real: init -> ... -> result cu cost."""
p = runner.ClaudeProcess("e2e2", str(tmp_path), "sonnet")
try:
out = await p.run_turn("Raspunde cu un singur cuvant: MERGE", timeout=120)
assert out.result is not None and "MERGE" in (out.result.text or "").upper()
assert out.result.duration_ms > 0 and p.sid
sid = p.sid
out2 = await p.run_turn("Repeta ultimul cuvant.", timeout=120)
assert p.sid == sid # acelasi proces, aceeasi sesiune
assert out2.result is not None
finally:
await p.stop()

View File

@@ -0,0 +1,171 @@
"""Teste pentru wrapper-ul `infra` (Lane B).
Nu se conecteaza nicaieri: totul ruleaza cu INFRA_DRY_RUN=1, care doar tipareste
comanda ssh pe care ar fi rulat-o.
"""
from __future__ import annotations
import importlib.machinery
import importlib.util
import json
import pathlib
import subprocess
import sys
import pytest
INFRA = pathlib.Path(__file__).resolve().parents[1] / "security" / "infra"
def load_infra():
"""Incarca `infra` ca modul, desi fisierul nu are extensia .py."""
spec = importlib.util.spec_from_loader(
"infra_mod", importlib.machinery.SourceFileLoader("infra_mod", str(INFRA))
)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
@pytest.fixture()
def home(tmp_path, monkeypatch):
base = tmp_path / ".claude-discord"
(base / "logs").mkdir(parents=True)
monkeypatch.setenv("CLAUDE_DISCORD_DIR", str(base))
return base
def run_cli(args, home_dir, dry=True):
env = {
"PATH": "/usr/bin:/bin",
"HOME": str(home_dir.parent),
"CLAUDE_DISCORD_DIR": str(home_dir),
}
if dry:
env["INFRA_DRY_RUN"] = "1"
return subprocess.run(
[sys.executable, str(INFRA), *args],
capture_output=True, text=True, env=env, timeout=30,
)
# ------------------------------------------------------------------ interfata
def test_este_executabil():
assert INFRA.stat().st_mode & 0o111, "infra trebuie sa fie executabil"
def test_lista_hosturilor(home):
res = run_cli(["--list"], home)
assert res.returncode == 0
for host in ("pvemini", "oracle", "oracle-prod", "gitea"):
assert host in res.stdout
assert "10.0.20.201" in res.stdout
assert "[PRODUCTIE]" in res.stdout
def test_fara_argumente_e_utilizare_gresita(home):
res = run_cli([], home)
assert res.returncode == 2
assert "infra <host>" in res.stdout
def test_host_permis_construieste_ssh(home):
res = run_cli(["pvemini", "pct", "list"], home)
assert res.returncode == 0
assert "root@10.0.20.201" in res.stdout
assert res.stdout.rstrip().endswith("pct list")
assert "BatchMode=yes" in res.stdout
def test_host_necunoscut_refuzat(home):
res = run_cli(["10.0.20.99", "uptime"], home)
assert res.returncode == 3
assert "host necunoscut" in res.stderr
assert "10.0.20.99" not in res.stdout # nu a construit nicio comanda ssh
def test_host_necunoscut_nu_incearca_dns(home):
res = run_cli(["router.local", "reboot"], home)
assert res.returncode == 3
assert "Hosturi permise" in res.stderr
def test_comanda_lipsa_refuzata(home):
res = run_cli(["pvemini"], home)
assert res.returncode == 2
assert "lipseste comanda" in res.stderr
# ----------------------------------------------------------------- jurnalizare
def test_apelul_e_jurnalizat(home):
run_cli(["oracle", "docker", "ps", "-a"], home)
log = (home / "logs" / "infra.log").read_text()
assert "host=oracle" in log
assert "target=root@10.0.20.121" in log
assert "docker ps -a" in log
assert "rc=dry-run" in log
def test_refuzul_e_jurnalizat(home):
run_cli(["host-strain", "rm", "-rf", "/"], home)
log = (home / "logs" / "infra.log").read_text()
assert "host=host-strain" in log
assert "rc=refuzat" in log
assert "host in afara listei" in log
assert "rm -rf /" in log
def test_comanda_completa_in_jurnal_cu_ghilimele(home):
run_cli(["docker", "sh", "-c", "echo unu doi"], home)
log = (home / "logs" / "infra.log").read_text()
assert "'echo unu doi'" in log # shlex.join pastreaza argumentul intreg
# ------------------------------------------------------- lista configurabila
def test_fisier_de_hosturi_propriu_inlocuieste_lista(home):
(home / "infra-hosts.json").write_text(
json.dumps({"labo": {"addr": "10.9.9.9", "user": "test", "desc": "laborator"}})
)
res = run_cli(["labo", "uptime"], home)
assert res.returncode == 0
assert "test@10.9.9.9" in res.stdout
# hosturile implicite nu mai sunt valabile daca fisierul exista
res2 = run_cli(["pvemini", "uptime"], home)
assert res2.returncode == 3
def test_fisier_de_hosturi_corupt_opreste_totul(home):
(home / "infra-hosts.json").write_text("{ nu e json")
res = run_cli(["pvemini", "uptime"], home)
assert res.returncode == 4
assert "nu e JSON valid" in res.stderr
def test_intrare_invalida_in_fisierul_de_hosturi(home):
(home / "infra-hosts.json").write_text(json.dumps({"x": {"user": "root"}}))
res = run_cli(["x", "uptime"], home)
assert res.returncode == 4
assert "intrare invalida" in res.stderr
def test_forma_scurta_addr_ca_string(home):
(home / "infra-hosts.json").write_text(json.dumps({"scurt": "10.1.2.3"}))
res = run_cli(["scurt", "uptime"], home)
assert res.returncode == 0
assert "root@10.1.2.3" in res.stdout
# ------------------------------------------------------------- lista implicita
def test_toate_hosturile_implicite_au_adresa():
mod = load_infra()
for name, spec in mod.DEFAULT_HOSTS.items():
assert spec["addr"].count(".") == 3, name
assert spec["user"], name
# hosturile de productie sunt marcate ca atare
assert mod.DEFAULT_HOSTS["oracle-prod"]["prod"] is True
assert mod.DEFAULT_HOSTS["pvemini"]["prod"] is True

View File

@@ -0,0 +1,118 @@
"""T8: 4 procese, coada per fir, rate limit, plafon de cost."""
from __future__ import annotations
import asyncio
import pytest
import limits as L
def mkl(store=None, **kw):
kw.setdefault("alerter", lambda *a, **k: None)
return L.Limits(store, **kw)
async def test_maxim_4_procese_al_cincilea_asteapta():
lim = mkl(max_procs=4)
held = []
ev = asyncio.Event()
async def hold(i):
async with lim.process_slot():
held.append(i)
await ev.wait()
tasks = [asyncio.create_task(hold(i)) for i in range(5)]
await asyncio.sleep(0.05)
assert len(held) == 4 and lim.free_slots == 0
ev.set()
await asyncio.gather(*tasks)
assert len(held) == 5 and lim.free_slots == 4
async def test_coada_per_fir_serializeaza_acelasi_fir():
lim = mkl()
ordine = []
async def tur(tag, delay):
async with lim.thread_lock("1"):
ordine.append(f"start-{tag}")
await asyncio.sleep(delay)
ordine.append(f"stop-{tag}")
await asyncio.gather(tur("a", 0.02), tur("b", 0.0))
assert ordine == ["start-a", "stop-a", "start-b", "stop-b"]
async def test_fire_diferite_merg_in_paralel():
lim = mkl()
async with lim.thread_lock("1"):
assert lim.queued("1") and not lim.queued("2")
async with lim.thread_lock("2"):
pass
def test_rate_limit_per_user():
t = [1000.0]
lim = mkl(rate_per_min=3, clock=lambda: t[0])
for _ in range(3):
lim.admit("u1")
with pytest.raises(L.RateLimited) as exc:
lim.admit("u1")
assert exc.value.retry_after == pytest.approx(60.0)
lim.admit("u2") # alt utilizator nu e afectat
t[0] += 61.0
lim.admit("u1") # fereastra a trecut
def test_plafon_de_cost_opreste_botul(store):
lim = mkl(store, cost_cap=1.0)
assert not lim.stopped() and lim.cost_remaining() == pytest.approx(1.0)
lim.record_cost(0.4, "1")
lim.admit("u1")
assert lim.record_cost(0.7, "1") == pytest.approx(1.1)
assert lim.stopped() and lim.cost_remaining() == 0.0
with pytest.raises(L.CostCapReached):
lim.admit("u1")
def test_plafon_alerteaza_o_singura_data(store):
alerte = []
lim = mkl(store, cost_cap=0.5, alerter=lambda *a, **k: alerte.append(a))
lim.record_cost(0.6, "1")
lim.record_cost(0.1, "1")
assert len(alerte) == 1 and alerte[0][0] == "CRITICAL"
def test_plafonul_se_ridica_a_doua_zi(store):
lim = mkl(store, cost_cap=1.0)
lim.record_cost(2.0, "1")
assert lim.stopped()
store.state["cost"]["day"] = "2000-01-01" # ziua se schimba
assert not lim.stopped()
def test_costul_fara_store_e_local():
lim = mkl(None, cost_cap=1.0)
lim.record_cost(0.5)
lim.record_cost("gunoi")
assert lim.cost_today() == pytest.approx(0.5)
def test_valorile_implicite_vin_din_config(state_dir):
import config
(state_dir / "env").write_text("COST_CAP_USD_DAY=2.5\nMAX_PROCS=2 # comentariu\n")
config.reload(state_dir)
lim = mkl()
assert lim.cost_cap == 2.5 and lim.max_procs == 2
assert lim.turn_timeout == 900.0
async def test_contextul_turn_verifica_tot(store):
lim = mkl(store, max_procs=1, cost_cap=10.0, turn_timeout=42.0)
async with lim.turn("1", "u1") as timeout:
assert timeout == 42.0
assert lim.free_slots == 0
assert lim.free_slots == 1

View File

@@ -0,0 +1,137 @@
"""T10: chunker si loop de editare per canal."""
from __future__ import annotations
import asyncio
import pytest
import render
def test_text_scurt_ramane_intr_o_bucata():
r = render.split_message("salut")
assert r.kind == "chunks" and r.parts == ["salut"]
def test_sparge_la_2000_de_caractere():
text = "\n".join(f"linia {i} " + "x" * 50 for i in range(60))
r = render.split_message(text)
assert r.kind == "chunks" and len(r.parts) > 1
assert all(len(p) <= render.MAX_MSG for p in r.parts)
assert "linia 0" in r.parts[0] and "linia 59" in r.parts[-1]
def test_fence_nu_se_rupe_peste_granita():
corp = "\n".join(f"cod linia {i}" for i in range(300))
r = render.split_message(f"intro\n```python\n{corp}\n```\ngata")
assert len(r.parts) >= 2
for p in r.parts:
assert p.count("```") % 2 == 0 # fiecare bucata are fence-uri echilibrate
assert r.parts[0].endswith("```")
assert r.parts[1].startswith("```python")
assert all(len(p) <= render.MAX_MSG for p in r.parts)
def test_fence_fara_limbaj():
corp = "\n".join(f"linia {i}" for i in range(400))
r = render.split_message(f"```\n{corp}\n```")
assert r.parts[1].startswith("```\n")
def test_linie_unica_uriasa_se_taie():
r = render.split_message("y" * 5000)
assert r.kind == "chunks" and all(len(p) <= render.MAX_MSG for p in r.parts)
def test_peste_6000_devine_atasament():
r = render.split_message("z" * 6001)
assert r.kind == "attachment"
assert isinstance(r, render.Attachment)
assert r.content == "z" * 6001 and r.filename.endswith(".md")
assert len(r.preview) <= render.MAX_MSG
def test_text_gol():
assert render.split_message("").parts == [""]
def test_footer():
s = render.footer("sonnet", 34500, 0.1547, 1.2)
assert "sonnet" in s and "34.5s" in s and "0.1547" in s and "1.2000" in s
# ------------------------------------------------------------- edit loop
async def test_loop_coalesceaza_actualizarile():
edits = []
async def edit(target, text):
edits.append((target, text))
loop = render.ChannelEditLoop("c1", edit, min_interval=0.05, max_interval=0.2)
loop.start()
for i in range(20):
loop.queue("msg1", f"text {i}")
await asyncio.sleep(0.15)
await loop.stop()
assert len(edits) < 20 # nu s-a trimis fiecare actualizare
assert edits[-1] == ("msg1", "text 19") # ultima versiune ajunge
async def test_flush_trimite_tot_ce_a_ramas():
edits = []
loop = render.ChannelEditLoop("c1", lambda t, x: _append(edits, t, x),
min_interval=10, max_interval=10)
loop.queue("m1", "a")
loop.queue("m2", "b")
await loop.flush()
assert sorted(edits) == [("m1", "a"), ("m2", "b")]
async def _append(lst, t, x):
lst.append((t, x))
async def test_intervalul_creste_sub_presiune_si_scade_la_liniste():
loop = render.ChannelEditLoop("c1", lambda t, x: _append([], t, x),
min_interval=0.01, max_interval=0.2)
loop.start()
for i in range(10):
loop.queue(f"m{i}", "x")
await asyncio.sleep(0.1)
crescut = loop.interval
assert crescut > 0.01
loop.note_rate_limited()
assert loop.interval == 0.2
await asyncio.sleep(0.5)
await loop.stop()
assert loop.interval < 0.2 # linistea coboara intervalul
async def test_editarea_esuata_nu_doboara_loop_ul():
async def edit(target, text):
raise RuntimeError("429 sau altceva")
loop = render.ChannelEditLoop("c1", edit, min_interval=0.01, max_interval=0.05)
loop.start()
loop.queue("m1", "a")
await asyncio.sleep(0.1)
assert loop.errors >= 1
loop.queue("m1", "b")
await asyncio.sleep(0.1)
assert loop.errors >= 2 # loop-ul inca traieste
await loop.stop()
async def test_un_singur_loop_per_canal():
edits = []
mgr = render.RenderManager(lambda t, x: _append(edits, t, x),
min_interval=0.01, max_interval=0.05)
l1 = mgr.loop_for("canal-1")
assert mgr.loop_for("canal-1") is l1
assert mgr.loop_for("canal-2") is not l1
mgr.queue("canal-1", "m1", "salut")
await asyncio.sleep(0.08)
assert ("m1", "salut") in edits
await mgr.stop_all()
assert mgr.loops == {}

View File

@@ -0,0 +1,222 @@
"""T4 + T5: proces persistent, steering, respawn, reaper, EOF, timeout."""
from __future__ import annotations
import asyncio
import os
import pytest
import runner
import stream
def mk(fake_bin, store=None, **kw):
return runner.ClaudeProcess("1", os.getcwd(), "sonnet", claude_bin=fake_bin,
on_pid=(store.set_pid if store else None), **kw)
def test_build_cmd_are_toate_flagurile():
cmd = runner.build_cmd("claude", "opus", None, "/x/settings.json")
assert cmd[:2] == ["claude", "-p"]
for flag in ("--input-format", "--output-format", "--verbose", "--permission-mode",
"--settings", "--model", "--autocompact"):
assert flag in cmd
assert cmd[cmd.index("--permission-mode") + 1] == "bypassPermissions"
assert "--resume" not in cmd
assert "--resume" in runner.build_cmd(["python", "fake"], "sonnet", "sid-1")
async def test_tur_normal_si_proces_persistent(fake_bin, scenario):
scenario("normal")
p = mk(fake_bin)
out = await p.run_turn("salut")
assert out.result.text == "ecou: salut" and out.result.total_cost_usd == pytest.approx(0.0123)
assert p.sid == "sid-fake-0001" and not out.restarted
pid = p.pid
out2 = await p.run_turn("inca unul")
assert out2.result.text == "ecou: inca unul"
assert p.pid == pid and p.alive # ACELASI proces pentru turul urmator
await p.stop()
assert not p.alive
async def test_evenimentele_ajung_la_callback(fake_bin, scenario):
scenario("tools")
p = mk(fake_bin)
seen = []
await p.run_turn("fa ceva", on_event=lambda ev: (seen.append(ev), asyncio.sleep(0))[1])
tipuri = [type(e) for e in seen]
assert stream.SystemInit in tipuri and stream.ToolUse in tipuri
assert stream.ToolResult in tipuri and tipuri[-1] is stream.Result
await p.stop()
async def test_stream_tolerant_in_tur_real(fake_bin, scenario):
scenario("unknown")
p = mk(fake_bin)
out = await p.run_turn("salut") # tip necunoscut + linie non-JSON pe mijloc
assert out.result.text == "ecou: salut"
await p.stop()
async def test_steering_mid_tur(fake_bin, scenario):
"""Mesaj trimis in timp ce turul ruleaza ajunge la proces inainte de result."""
scenario("slow", FAKE_CLAUDE_DELAY=0.6)
p = mk(fake_bin)
async def steer():
await asyncio.sleep(0.15)
await p.send("de fapt, opreste-te")
task = asyncio.create_task(steer())
out = await p.run_turn("porneste ceva lung")
await task
assert "porneste ceva lung" in out.result.text
assert "de fapt, opreste-te" in out.result.text
await p.stop()
async def test_eof_inainte_de_result_da_turn_failed(fake_bin, scenario):
scenario("eof")
p = mk(fake_bin)
with pytest.raises(runner.TurnFailed):
await p.run_turn("salut")
assert not p.alive
async def test_crash_pastreaza_stderr_pentru_status(fake_bin, scenario):
scenario("crash")
p = mk(fake_bin)
with pytest.raises(runner.TurnFailed) as exc:
await p.run_turn("salut")
assert "boom" in str(exc.value)
async def test_stderr_buffer_circular(fake_bin, scenario):
scenario("crash")
p = mk(fake_bin)
with pytest.raises(runner.TurnFailed):
await p.run_turn("x")
assert p.stderr_buf.maxlen == runner.STDERR_TAIL
async def test_respawn_transparent_cu_resume(fake_bin, scenario, store):
scenario("normal")
p = mk(fake_bin, store=store)
await p.run_turn("primul")
sid = p.sid
p.proc.kill() # OOM simulat
await p.proc.wait()
p.proc = None
out = await p.run_turn("al doilea")
assert out.restarted is True
restart_ev = [e for e in out.events if isinstance(e, runner.SessionRestarted)]
assert restart_ev and "sesiune repornita" in restart_ev[0].text
assert p.sid == sid # --resume a pastrat sesiunea
assert p.restarts == 1
await p.stop()
async def test_timeout_de_tur_omoara_procesul(fake_bin, scenario):
scenario("slow", FAKE_CLAUDE_DELAY=5)
p = mk(fake_bin)
with pytest.raises(runner.TurnTimeout):
await p.run_turn("lung", timeout=0.3)
assert not p.alive
async def test_send_pe_proces_mort_da_eroare(fake_bin, scenario):
scenario("normal")
p = mk(fake_bin)
with pytest.raises(runner.TurnFailed):
await p.send("nimeni nu asculta")
async def test_pid_ul_ajunge_in_state(fake_bin, scenario, store):
scenario("normal")
m = runner.RunnerManager(store, claude_bin=fake_bin)
p = m.get("42", os.getcwd(), "sonnet")
await p.run_turn("salut")
assert store.thread("42")["pid"] == p.pid
assert store.thread_process_alive("42")
await m.stop_all()
assert store.thread("42")["pid"] is None
async def test_reaper_omoara_inactivii_dar_nu_turul_in_zbor(fake_bin, scenario, store):
scenario("normal")
m = runner.RunnerManager(store, claude_bin=fake_bin, idle_s=0.0)
a = m.get("a", os.getcwd(), "sonnet")
b = m.get("b", os.getcwd(), "sonnet")
await a.run_turn("x")
await b.run_turn("y")
store.set_inflight("b", "t", "u", "m") # firul b are tur in zbor
killed = await m.reap_once()
assert killed == ["a"]
assert not a.alive and b.alive
store.clear_inflight("b")
assert await m.reap_once() == ["b"]
await m.stop_all()
async def test_reaper_nu_taie_daca_procesul_e_activ(fake_bin, scenario):
scenario("normal")
m = runner.RunnerManager(None, claude_bin=fake_bin, idle_s=60.0)
p = m.get("a", os.getcwd(), "sonnet")
await p.run_turn("x")
assert await m.reap_once() == [] and p.alive
await m.stop_all()
async def test_set_options_opreste_procesul_iar_resume_pastreaza_sesiunea(fake_bin, scenario):
scenario("normal")
m = runner.RunnerManager(None, claude_bin=fake_bin)
p = m.get("a", os.getcwd(), "sonnet")
await p.run_turn("x")
sid = p.sid
assert await m.set_options("a", model="opus") is True
assert not p.alive and p.model == "opus"
out = await p.run_turn("y")
assert out.restarted and p.sid == sid
await m.stop_all()
async def test_reset_new_sterge_sesiunea(fake_bin, scenario):
scenario("normal")
m = runner.RunnerManager(None, claude_bin=fake_bin)
p = m.get("a", os.getcwd(), "sonnet")
await p.run_turn("x")
await m.reset("a")
assert p.sid is None and not p.alive
await m.stop_all()
async def test_live_count_si_stop_all(fake_bin, scenario):
scenario("normal")
m = runner.RunnerManager(None, claude_bin=fake_bin)
for tid in ("a", "b"):
await m.get(tid, os.getcwd(), "sonnet").run_turn("x")
assert m.live_count() == 2
await m.stop_all()
assert m.live_count() == 0
async def test_thread_id_ajunge_in_mediul_procesului(fake_bin, scenario, monkeypatch):
"""Lane B: hook-ul PreToolUse citeste firul din CLAUDE_DISCORD_THREAD_ID."""
scenario("env")
monkeypatch.setenv("MARKER_DE_TEST", "pastrat")
p = runner.ClaudeProcess("fir-777", os.getcwd(), "sonnet", claude_bin=fake_bin)
out = await p.run_turn("x")
assert "thread=fir-777" in out.result.text
assert "sesiune=-" in out.result.text # inca nu avem sid la prima pornire
assert "marker=pastrat" in out.result.text # restul mediului ramane intact
await p.stop()
async def test_session_id_ajunge_in_mediu_la_respawn(fake_bin, scenario):
scenario("env")
p = runner.ClaudeProcess("fir-888", os.getcwd(), "sonnet", sid="sid-vechi", claude_bin=fake_bin)
out = await p.run_turn("x")
assert "thread=fir-888" in out.result.text and "sesiune=sid-vechi" in out.result.text
await p.stop()

View File

@@ -0,0 +1,127 @@
"""T7 + T5: scriere atomica, lock per fir, PID reuse, recovery, sweep."""
from __future__ import annotations
import json
import os
import time
import pytest
import session_store as ss
def test_roundtrip_si_scriere_atomica(store, state_dir):
store.update_thread("111", sid="s1", cwd="/workspace/x", model="opus")
raw = json.loads((state_dir / "state.json").read_text())
assert raw["threads"]["111"]["sid"] == "s1"
assert raw["version"] == ss.VERSION and "cost" in raw
# nu raman fisiere temporare in urma
assert not [p for p in state_dir.iterdir() if p.name.startswith(".state.")]
alt = ss.SessionStore(state_dir / "state.json")
alt.load()
assert alt.thread("111")["model"] == "opus"
def test_fisier_corupt_e_pus_deoparte_si_alerteaza(state_dir):
(state_dir / "state.json").write_text('{"threads": {"a": ')
alerts = []
s = ss.SessionStore(state_dir / "state.json", alerter=lambda *a, **k: alerts.append(a))
st = s.load()
assert st["threads"] == {}
assert alerts and alerts[0][0] == "CRITICAL"
assert [p for p in state_dir.iterdir() if ".corrupt-" in p.name]
# fisierul nou e valid
assert json.loads((state_dir / "state.json").read_text())["threads"] == {}
def test_json_valid_dar_structura_gresita(state_dir):
(state_dir / "state.json").write_text('["nu", "e", "obiect"]')
s = ss.SessionStore(state_dir / "state.json", alerter=lambda *a, **k: None)
assert s.load()["threads"] == {}
def test_campuri_necunoscute_sunt_ignorate_la_incarcare(state_dir):
(state_dir / "state.json").write_text(json.dumps(
{"version": 1, "threads": {"7": {"sid": "s", "ceva_nou": 1}}, "cost": {"day": "2026-01-01", "usd": 3.0}}))
s = ss.SessionStore(state_dir / "state.json")
st = s.load()
assert st["threads"]["7"]["sid"] == "s" and "ceva_nou" not in st["threads"]["7"]
assert st["threads"]["7"]["inflight"] is None
def test_pid_start_time_si_pid_reuse():
me = os.getpid()
start = ss.pid_start_time(me)
assert start and start > 0
assert ss.pid_alive(me, start) is True
# acelasi PID cu alt starttime = PID reciclat, NU e procesul nostru
assert ss.pid_alive(me, start + 500.0) is False
assert ss.pid_alive(None) is False
assert ss.pid_alive(4194303, 1.0) is False
def test_thread_process_alive(store):
me = os.getpid()
store.set_pid("1", me)
assert store.thread_process_alive("1") is True
store.update_thread("1", pid_start_time=ss.pid_start_time(me) + 100)
assert store.thread_process_alive("1") is False
def test_sweep_marcheaza_turul_pierdut_fara_reluare(store):
store.update_thread("1", pid=4194303, pid_start_time=1.0)
store.set_inflight("1", "t1", "u1", "m1")
lost = store.sweep_lost_turns()
assert len(lost) == 1
assert lost[0]["thread_id"] == "1" and lost[0]["message_id"] == "m1"
assert "NU il reiau automat" in lost[0]["warning"]
assert store.thread("1")["inflight"] is None
assert store.sweep_lost_turns() == [] # idempotent
def test_sweep_nu_atinge_un_proces_viu(store):
store.set_pid("1", os.getpid())
store.set_inflight("1", "t1", "u1", "m1")
assert store.sweep_lost_turns() == []
assert store.is_inflight("1")
def test_cost_pe_fir_si_pe_zi(store):
store.add_cost("1", 0.1547)
store.add_cost("1", 0.05)
store.add_cost("2", 1.0)
assert store.thread("1")["cost_usd_total"] == pytest.approx(0.2047)
assert store.cost_today() == pytest.approx(1.2047)
store.add_cost("1", "nu-i numar")
assert store.cost_today() == pytest.approx(1.2047)
def test_cost_se_reseteaza_la_zi_noua(store):
store.add_cost("1", 5.0)
store.state["cost"]["day"] = "2000-01-01"
assert store.cost_today() == 0.0
assert store.state["cost"]["day"] == ss.today()
async def test_lock_per_fir_e_distinct(store):
a, b = store.lock_for("1"), store.lock_for("2")
assert a is store.lock_for("1") and a is not b
async with a:
assert a.locked() and not b.locked()
def test_lock_de_fisier_intre_procese(state_dir):
s = ss.SessionStore(state_dir / "state.json")
s.load()
s.save()
assert (state_dir / "state.json.lock").exists()
def test_inflight_set_si_clear(store):
store.set_inflight("9", "t", "u", "m")
assert store.is_inflight("9")
assert store.thread("9")["inflight"]["started_at"] <= time.time()
store.clear_inflight("9")
assert not store.is_inflight("9")

View File

@@ -0,0 +1,122 @@
"""T6: parser tolerant."""
from __future__ import annotations
import json
import pytest
import stream
def P(**kw):
return stream.StreamParser(**kw)
def test_system_init_aduce_session_id():
p = P()
(ev,) = p.feed_line(json.dumps({"type": "system", "subtype": "init",
"session_id": "s1", "model": "sonnet", "cwd": "/workspace",
"tools": ["Bash"]}))
assert isinstance(ev, stream.SystemInit)
assert ev.session_id == "s1" and ev.model == "sonnet" and ev.tools == ("Bash",)
def test_assistant_text_si_tool_use():
p = P()
evs = p.feed_line(json.dumps({"type": "assistant", "session_id": "s1", "message": {"content": [
{"type": "text", "text": "salut"},
{"type": "tool_use", "id": "tu1", "name": "Bash", "input": {"command": "ls"}}]}}))
assert [type(e) for e in evs] == [stream.AssistantText, stream.ToolUse]
assert evs[0].text == "salut"
assert evs[1].name == "Bash" and evs[1].input["command"] == "ls"
def test_tool_result_din_user():
p = P()
(ev,) = p.feed_line(json.dumps({"type": "user", "message": {"content": [
{"type": "tool_result", "tool_use_id": "tu1", "content": [{"type": "text", "text": "out"}]}]}}))
assert isinstance(ev, stream.ToolResult) and ev.text == "out" and ev.is_error is False
def test_result_aduce_cost_si_durata():
p = P()
(ev,) = p.feed_line(json.dumps({"type": "result", "subtype": "success", "is_error": False,
"total_cost_usd": 0.1547, "duration_ms": 34500,
"num_turns": 3, "result": "gata"}))
assert (ev.total_cost_usd, ev.duration_ms, ev.num_turns, ev.text) == (0.1547, 34500, 3, "gata")
assert ev.is_error is False and p.saw_result
def test_result_cu_campuri_stricate_nu_crapa():
p = P()
(ev,) = p.feed_line(json.dumps({"type": "result", "total_cost_usd": "nu-i numar",
"duration_ms": None, "num_turns": "x"}))
assert ev.total_cost_usd == 0.0 and ev.duration_ms == 0 and ev.num_turns == 0
def test_result_de_eroare_marcat():
p = P()
(ev,) = p.feed_line(json.dumps({"type": "result", "subtype": "error_during_execution"}))
assert ev.is_error is True
def test_tip_necunoscut_logat_o_singura_data():
calls = []
p = P(version_fn=lambda: (calls.append(1), "2.1.251")[1])
line = json.dumps({"type": "rate_limit_event", "detail": "x"})
assert p.feed_line(line) == []
assert p.feed_line(line) == []
assert p.feed_line(json.dumps({"type": "alt_tip_nou"})) == []
assert p.unknown_types == {"rate_limit_event", "alt_tip_nou"}
assert len(calls) == 2 # o data per tip, nu per linie
def test_linie_non_json_ignorata(caplog):
p = P()
with caplog.at_level("WARNING"):
assert p.feed_line("Error: something went wrong") == []
assert p.feed_line("[1, 2, 3]") == []
assert p.bad_lines == 2
assert any("non-JSON" in r.message or "nu e obiect" in r.message for r in caplog.records)
def test_linii_goale_ignorate():
p = P()
assert p.feed_line("") == [] and p.feed_line(" \n") == []
async def _aiter(lines):
for x in lines:
yield x
async def test_eof_inainte_de_result_da_eroare_explicita():
p = P()
got = []
with pytest.raises(stream.StreamEOFError):
async for ev in p.aiter_events(_aiter([
json.dumps({"type": "system", "subtype": "init", "session_id": "s1"}),
json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "hm"}]}}),
])):
got.append(ev)
assert len(got) == 2
async def test_aiter_se_opreste_la_result():
p = P()
evs = [e async for e in p.aiter_events(_aiter([
json.dumps({"type": "system", "subtype": "init", "session_id": "s1"}),
json.dumps({"type": "result", "total_cost_usd": 1.0}),
json.dumps({"type": "assistant", "message": {"content": [{"type": "text", "text": "dupa"}]}}),
]))]
assert [type(e) for e in evs] == [stream.SystemInit, stream.Result]
async def test_aiter_accepta_bytes_si_gunoi():
p = P()
evs = [e async for e in p.aiter_events(_aiter([
b"gunoi binar\n",
b'{"type":"result","total_cost_usd":0.5}\n',
]))]
assert len(evs) == 1 and evs[0].total_cost_usd == 0.5