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()