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

"""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()
    # oracle-prod-admin e un alias din ~/.ssh/config (portul/cheia vin de acolo),
    # nu o adresa IP -- restul hosturilor raman IP-uri clasice.
    for name, spec in mod.DEFAULT_HOSTS.items():
        if name == "oracle-prod-admin":
            assert spec["addr"], name
        else:
            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