Gilles Soulier
325b5dc281
- disk_type : rota booléen (False/True) en plus des int (0/1) — lsblk récent - df --local : évite le timeout sur montages réseau/NFS (Proxmox) - 6 nouveaux tests disk_type Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
256 lines
10 KiB
Python
256 lines
10 KiB
Python
import sys, os, json
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
|
||
|
||
from unittest.mock import patch, MagicMock, mock_open, call
|
||
import inventaire
|
||
|
||
|
||
# ── parse_args ────────────────────────────────────────────────────────────────
|
||
|
||
def test_args_défauts(monkeypatch):
|
||
monkeypatch.delenv("MES_HDD_HOST", raising=False)
|
||
monkeypatch.delenv("MES_HDD_PORT", raising=False)
|
||
with patch("sys.argv", ["inventaire.py"]):
|
||
args = inventaire.parse_args()
|
||
assert args.host == "10.0.0.50"
|
||
assert args.port == 8088
|
||
assert args.dry_run is False
|
||
assert args.debug is False
|
||
assert args.output is None
|
||
|
||
def test_args_host_port():
|
||
with patch("sys.argv", ["inventaire.py", "--host", "192.168.1.10", "--port", "9000"]):
|
||
args = inventaire.parse_args()
|
||
assert args.host == "192.168.1.10"
|
||
assert args.port == 9000
|
||
|
||
def test_args_dry_run():
|
||
with patch("sys.argv", ["inventaire.py", "-n"]):
|
||
args = inventaire.parse_args()
|
||
assert args.dry_run is True
|
||
|
||
def test_args_debug():
|
||
with patch("sys.argv", ["inventaire.py", "--debug"]):
|
||
args = inventaire.parse_args()
|
||
assert args.debug is True
|
||
|
||
def test_args_output():
|
||
with patch("sys.argv", ["inventaire.py", "--output", "/tmp/inv.json"]):
|
||
args = inventaire.parse_args()
|
||
assert args.output == "/tmp/inv.json"
|
||
|
||
def test_args_depuis_env(monkeypatch):
|
||
monkeypatch.setenv("MES_HDD_HOST", "10.0.0.99")
|
||
monkeypatch.setenv("MES_HDD_PORT", "9999")
|
||
with patch("sys.argv", ["inventaire.py"]):
|
||
args = inventaire.parse_args()
|
||
assert args.host == "10.0.0.99"
|
||
assert args.port == 9999
|
||
|
||
|
||
# ── detect_os ─────────────────────────────────────────────────────────────────
|
||
|
||
def test_detect_proxmox_via_variant():
|
||
content = 'ID=debian\nVARIANT_ID=proxmox\nVERSION_ID="8.2"\n'
|
||
with patch("builtins.open", mock_open(read_data=content)):
|
||
with patch("os.path.isdir", return_value=False):
|
||
os_type, version = inventaire.detect_os()
|
||
assert os_type == "proxmox"
|
||
assert version == "8.2"
|
||
|
||
def test_detect_proxmox_via_pve_dir():
|
||
content = 'ID=debian\nVERSION_ID="8.2"\n'
|
||
with patch("builtins.open", mock_open(read_data=content)):
|
||
with patch("os.path.isdir", side_effect=lambda p: "/etc/pve" in p):
|
||
os_type, _ = inventaire.detect_os()
|
||
assert os_type == "proxmox"
|
||
|
||
def test_detect_ubuntu():
|
||
content = 'ID=ubuntu\nVERSION_ID="22.04"\n'
|
||
with patch("builtins.open", mock_open(read_data=content)):
|
||
with patch("os.path.isdir", return_value=False):
|
||
os_type, version = inventaire.detect_os()
|
||
assert os_type == "ubuntu"
|
||
assert version == "22.04"
|
||
|
||
def test_detect_debian():
|
||
content = 'ID=debian\nVERSION_ID="12"\n'
|
||
with patch("builtins.open", mock_open(read_data=content)):
|
||
with patch("os.path.isdir", return_value=False):
|
||
os_type, version = inventaire.detect_os()
|
||
assert os_type == "debian"
|
||
|
||
def test_detect_os_fichier_absent():
|
||
with patch("builtins.open", side_effect=FileNotFoundError):
|
||
with patch("os.path.isdir", return_value=False):
|
||
os_type, version = inventaire.detect_os()
|
||
assert os_type == "debian"
|
||
assert version == ""
|
||
|
||
|
||
# ── bytes_human ───────────────────────────────────────────────────────────────
|
||
|
||
def test_bytes_human_to():
|
||
assert inventaire.bytes_human(1_099_511_627_776) == "1.0 To"
|
||
|
||
def test_bytes_human_go():
|
||
assert inventaire.bytes_human(1_073_741_824) == "1.0 Go"
|
||
|
||
def test_bytes_human_mo():
|
||
assert inventaire.bytes_human(500 * 1024 * 1024) == "500.0 Mo"
|
||
|
||
def test_bytes_human_none():
|
||
assert inventaire.bytes_human(None) == "?"
|
||
|
||
|
||
# -- disk_type ----------------------------------------------------------------
|
||
|
||
def test_disk_type_nvme():
|
||
assert inventaire.disk_type("nvme0n1", None) == "NVMe"
|
||
|
||
def test_disk_type_hdd_int():
|
||
assert inventaire.disk_type("sda", 1) == "HDD"
|
||
|
||
def test_disk_type_ssd_int():
|
||
assert inventaire.disk_type("sdb", 0) == "SSD"
|
||
|
||
def test_disk_type_hdd_bool():
|
||
assert inventaire.disk_type("sda", True) == "HDD"
|
||
|
||
def test_disk_type_ssd_bool():
|
||
assert inventaire.disk_type("sdb", False) == "SSD"
|
||
|
||
def test_disk_type_inconnu():
|
||
assert inventaire.disk_type("sdc", None) == "inconnu"
|
||
|
||
|
||
# ── get_smart ─────────────────────────────────────────────────────────────────
|
||
|
||
SMART_PASSED = """
|
||
SMART overall-health self-assessment test result: PASSED
|
||
190 Airflow_Temperature_Cel 0x0022 073 060 045 Old_age Always - 27
|
||
9 Power_On_Hours 0x0032 099 099 000 Old_age Always - 2847
|
||
5 Reallocated_Sector_Ct 0x0033 100 100 036 Pre-fail Always - 0
|
||
197 Current_Pending_Sector 0x0012 100 100 000 Old_age Always - 0
|
||
198 Offline_Uncorrectable 0x0010 100 100 000 Old_age Offline - 0
|
||
"""
|
||
|
||
SMART_FAILED = "SMART overall-health self-assessment test result: FAILED!"
|
||
|
||
SMART_WARN = """
|
||
SMART overall-health self-assessment test result: PASSED
|
||
5 Reallocated_Sector_Ct 0x0033 100 100 036 Pre-fail Always - 3
|
||
197 Current_Pending_Sector 0x0012 100 100 000 Old_age Always - 0
|
||
198 Offline_Uncorrectable 0x0010 100 100 000 Old_age Offline - 0
|
||
"""
|
||
|
||
def test_smart_ok():
|
||
with patch.object(inventaire, "run", return_value=SMART_PASSED):
|
||
r = inventaire.get_smart("/dev/sda")
|
||
assert r["status"] == "ok"
|
||
assert r["label"] == "Bon état"
|
||
assert r["power_on_hours"] == 2847
|
||
assert r["temperature_c"] == 27
|
||
assert r["reallocated_sectors"] == 0
|
||
assert "2 847h" in r["detail"].replace(" ", " ")
|
||
assert "27°C" in r["detail"]
|
||
assert "aucun secteur" in r["detail"]
|
||
|
||
def test_smart_fail():
|
||
with patch.object(inventaire, "run", return_value=SMART_FAILED):
|
||
r = inventaire.get_smart("/dev/sda")
|
||
assert r["status"] == "fail"
|
||
assert r["label"] == "Défaillance probable"
|
||
assert "remplacement" in r["detail"]
|
||
|
||
def test_smart_warn_secteurs_réalloués():
|
||
with patch.object(inventaire, "run", return_value=SMART_WARN):
|
||
r = inventaire.get_smart("/dev/sda")
|
||
assert r["status"] == "warn"
|
||
assert r["label"] == "Attention"
|
||
assert "réalloué" in r["detail"]
|
||
assert r["reallocated_sectors"] == 3
|
||
|
||
def test_smart_unavailable_commande_absente():
|
||
with patch.object(inventaire, "run", return_value=None):
|
||
r = inventaire.get_smart("/dev/sda")
|
||
assert r["status"] == "unavailable"
|
||
assert r["label"] == "SMART indisponible"
|
||
|
||
def test_smart_unavailable_résultat_inconnu():
|
||
with patch.object(inventaire, "run", return_value="Device: some output without health"):
|
||
r = inventaire.get_smart("/dev/sda")
|
||
assert r["status"] == "unavailable"
|
||
|
||
|
||
# ── get_home_users ────────────────────────────────────────────────────────────
|
||
|
||
def test_home_users_triés_par_taille():
|
||
du_output = "100000\t/home/alice\n400000\t/home/gilles\n850000000\t/home\n"
|
||
with patch.object(inventaire, "run", return_value=du_output):
|
||
with patch("os.path.isdir", return_value=True):
|
||
users = inventaire.get_home_users()
|
||
assert users[0]["user"] == "gilles"
|
||
assert users[0]["size_bytes"] == 400000
|
||
assert users[1]["user"] == "alice"
|
||
assert len(users) == 2 # /home lui-même est filtré
|
||
|
||
def test_home_users_home_absent():
|
||
with patch("os.path.isdir", return_value=False):
|
||
users = inventaire.get_home_users()
|
||
assert users == []
|
||
|
||
def test_home_users_du_échoue():
|
||
with patch("os.path.isdir", return_value=True):
|
||
with patch.object(inventaire, "run", return_value=None):
|
||
users = inventaire.get_home_users()
|
||
assert users is None
|
||
|
||
|
||
# ── post_to_api ───────────────────────────────────────────────────────────────
|
||
|
||
def test_post_to_api_succès():
|
||
payload = {"hostname": "pve1", "disks": []}
|
||
response_body = json.dumps({"accepted": 0, "hostname": "pve1"}).encode()
|
||
mock_resp = MagicMock()
|
||
mock_resp.__enter__ = MagicMock(return_value=mock_resp)
|
||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||
mock_resp.read.return_value = response_body
|
||
|
||
with patch("urllib.request.urlopen", return_value=mock_resp):
|
||
inventaire.post_to_api(payload, "http://10.0.0.50:8088") # pas d'exception = succès
|
||
|
||
def test_post_to_api_http_error():
|
||
import urllib.error
|
||
payload = {"hostname": "pve1", "disks": []}
|
||
with patch("urllib.request.urlopen",
|
||
side_effect=urllib.error.HTTPError("url", 500, "Server Error", {}, None)):
|
||
with patch("sys.exit") as mock_exit:
|
||
inventaire.post_to_api(payload, "http://10.0.0.50:8088")
|
||
mock_exit.assert_called_once_with(1)
|
||
|
||
def test_post_to_api_connexion_refusée():
|
||
import urllib.error
|
||
payload = {"hostname": "pve1", "disks": []}
|
||
with patch("urllib.request.urlopen",
|
||
side_effect=urllib.error.URLError("Connection refused")):
|
||
with patch("sys.exit") as mock_exit:
|
||
inventaire.post_to_api(payload, "http://10.0.0.50:8088")
|
||
mock_exit.assert_called_once_with(1)
|
||
|
||
|
||
# ── dry_run / debug output ────────────────────────────────────────────────────
|
||
|
||
def test_dry_run_naffiche_que_json(capsys):
|
||
payload = {"hostname": "pve1", "disks": [], "os": "debian"}
|
||
inventaire.print_json(payload)
|
||
captured = capsys.readouterr()
|
||
parsed = json.loads(captured.out)
|
||
assert parsed["hostname"] == "pve1"
|
||
|
||
def test_dry_run_nenvoie_pas(capsys):
|
||
payload = {"hostname": "pve1", "disks": []}
|
||
with patch.object(inventaire, "post_to_api") as mock_post:
|
||
inventaire.print_json(payload)
|
||
mock_post.assert_not_called()
|