romfast/ROMFASTSQL
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(discord-bridge): comanda /infra cu confirmare prin butoane

Browse Source 
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01KddsXCqEbKMhdFJDYbAsx8
This commit is contained in:
Claude Agent
2026-09-13 10:18:12 +00:00
parent 56027e2e6c
commit c87fa06b55
3 changed files with 398 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
proxmox/lxc171-claude-agent/discord-bridge/tests/test_commands.py
View File

@@ -2,6 +2,8 @@
from __future__ import annotations
import asyncio
import pytest
import bot
@@ -345,3 +347,156 @@ async def test_sync_fara_arbore_sau_fara_guilduri_nu_arunca():
assert await commands_slash.sync_guilds(None, ["100"]) == {}
assert await commands_slash.sync_guilds(_FakeTree(), []) == {}
# --------------------------------------------------------------------- /infra
import infra_actions # noqa: E402
def _st(**over) -> dict:
base = {
"quorum": {"quorate": True, "votes": 3, "expected": 3},
"guests": {"running": 2, "total": 3},
"ha": [], "replication": {"ok": 0, "total": 0, "last": 0, "failed": []},
"ups": None, "backup": None, "ssl": None, "vm109": None,
"dr_failover_active": None, "maint": 0, "jobs": [], "errors": {},
"nodes": [{"name": "pvemini", "ip": "10.0.20.201", "online": True,
"cpu": 0.0, "mem": 0, "maxmem": 0, "uptime": 0, "guests": 2}],
}
base.update(over)
return base
def test_slash_args_infra():
assert bot.slash_args("infra", {"actiune": "oprire", "simulare": None}) == ["oprire"]
assert bot.slash_args("infra", {"actiune": "oprire", "simulare": False}) == ["oprire", "real"]
assert bot.slash_args("infra", {"actiune": "oprire", "simulare": True}) == ["oprire", "simulare"]
async def test_infra_actiune_necunoscuta_fara_subprocess(bridge, monkeypatch):
import subprocess
monkeypatch.setattr(subprocess, "run", lambda *a, **k: pytest.fail("nu trebuia sa ruleze nimic"))
_, ch, _ = await slash(bridge, "infra", actiune="nu-exista", simulare=None)
assert "necunoscuta" in ch.all_text
async def test_infra_fara_actiune(bridge):
_, ch, _ = await slash(bridge, "infra", actiune="", simulare=None)
assert "necunoscuta" in ch.all_text or "Foloseste" in ch.all_text
async def test_infra_stare(bridge, monkeypatch):
st = _st(ups={"status": "OL", "charge": 100.0, "runtime_s": 1200})
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: st)
_, ch, _ = await slash(bridge, "infra", actiune="stare", simulare=None)
text = ch.all_text
assert "cvorum" in text and "3/3" in text and "guest-uri" in text and "pvemini" in text
assert "UPS" in text and "OL" in text
async def test_infra_sarcini_fara_joburi(bridge, monkeypatch):
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: _st(jobs=[]))
_, ch, _ = await slash(bridge, "infra", actiune="sarcini", simulare=None)
assert "Niciun job activ" in ch.all_text
async def test_infra_sarcini_cu_job_activ(bridge, monkeypatch):
jobs = [{"id": "oprire", "host": "pvemini", "active": True, "log": "/var/log/x.log"}]
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: _st(jobs=jobs))
monkeypatch.setattr(infra_actions, "log_tail", lambda aid, n=60: {"file": "/x", "lines": ["linia 1"]})
_, ch, _ = await slash(bridge, "infra", actiune="sarcini", simulare=None)
assert "oprire" in ch.all_text and "linia 1" in ch.all_text
async def test_infra_precondition_esuata(bridge, monkeypatch):
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: _st())
monkeypatch.setattr(infra_actions, "check", lambda aid, st, dry: (False, "motiv de test"))
apelat = []
monkeypatch.setattr(infra_actions, "run", lambda *a, **k: apelat.append(a) or {"ok": True, "code": 200})
_, ch, _ = await slash(bridge, "infra", actiune="wol", simulare=None)
assert "⛔" in ch.all_text and "motiv de test" in ch.all_text
assert apelat == []
async def test_infra_fara_confirmare_ruleaza_direct(bridge, monkeypatch):
"""`wol` nu cere confirmare: ruleaza imediat daca precondiția trece."""
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: _st())
monkeypatch.setattr(infra_actions, "check", lambda aid, st, dry: (True, ""))
apelat = []
def _run(aid, dry):
apelat.append((aid, dry))
return {"ok": True, "code": 200, "host": "oracle-prod-admin"}
monkeypatch.setattr(infra_actions, "run", _run)
_, ch, _ = await slash(bridge, "infra", actiune="wol", simulare=None)
assert apelat == [("wol", False)] # "wol" nu are simulare -> dry mereu False
assert "pornit" in ch.all_text
async def test_infra_cu_confirmare_arata_intrebarea_si_nu_ruleaza(bridge, monkeypatch):
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: _st())
monkeypatch.setattr(infra_actions, "check", lambda aid, st, dry: (True, ""))
apelat = []
monkeypatch.setattr(infra_actions, "run", lambda *a, **k: apelat.append(a) or {"ok": True, "code": 200})
_, ch, _ = await slash(bridge, "infra", actiune="oprire", simulare=False) # "real" -> dry=False
assert "confirmi" in ch.all_text.lower()
assert apelat == [] # butonul nu a fost apasat, comanda nu a rulat nimic
async def test_infra_actiune_de_simulare_are_dry_true(bridge, monkeypatch):
vazute = []
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: _st())
def _check(aid, st, dry):
vazute.append(dry)
return (True, "")
monkeypatch.setattr(infra_actions, "check", _check)
monkeypatch.setattr(infra_actions, "run", lambda *a, **k: {"ok": True, "code": 200})
await slash(bridge, "infra", actiune="oprire", simulare=None) # implicit: simulare
await slash(bridge, "infra", actiune="oprire", simulare=False) # explicit: real
assert vazute == [True, False]
def test_infra_view_e_none_fara_discordpy(bridge):
if bot.discord is None:
assert bridge.infra_view("oprire", True) is None
assert bridge.infra_boot_view() is None
async def test_infra_boot_check_anunta_mentenanta_apoi_se_opreste(bridge, monkeypatch):
ch = FakeChannel()
monkeypatch.setenv("DISCORD_CHANNEL_IDS", str(ch.id))
bridge.get_channel = lambda cid: ch if str(cid) == str(ch.id) else None
st = _st(maint=1)
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: st)
monkeypatch.setattr(infra_actions, "maintenance_pending", lambda s: True)
await bridge.infra_boot_check()
assert "guest-uri de pornit" in ch.all_text
async def test_infra_boot_check_iese_daca_nu_e_in_mentenanta(bridge, monkeypatch):
st = _st(maint=0)
monkeypatch.setattr(infra_actions, "status", lambda fresh=False: st)
apelat = []
monkeypatch.setattr(infra_actions, "maintenance_pending", lambda s: apelat.append(s) or False)
await bridge.infra_boot_check()
assert apelat == [] # a iesit la prima citire (maint == 0), fara sa mai verifice
async def test_infra_boot_check_nu_doboara_botul_la_eroare(bridge, monkeypatch):
"""`status()` care arunca nu are voie sa opreasca bucla sau botul."""
def _boom(fresh=False):
raise RuntimeError("retea picata")
async def _fast_sleep(_):
return None
monkeypatch.setattr(infra_actions, "status", _boom)
monkeypatch.setattr(asyncio, "sleep", _fast_sleep)
await bridge.infra_boot_check() # 20 incercari esuate, nici o exceptie in afara