diff --git a/tests/test_turbo_hardware_diag.py b/tests/test_turbo_hardware_diag.py index 37cee27a9..f71afcf27 100644 --- a/tests/test_turbo_hardware_diag.py +++ b/tests/test_turbo_hardware_diag.py @@ -1,11 +1,8 @@ """Tests for turbo_hardware_diag.py — the Python hardware diagnostic tool. - 100% line coverage target with NO real subprocess calls, GPU, or llama.cpp binaries. All platform-specific probes are mocked. Tests pass on macOS AND Linux. """ - from __future__ import annotations - import csv import io import json @@ -20,18 +17,14 @@ from pathlib import Path from unittest import mock from unittest.mock import MagicMock, PropertyMock, call, patch - import pytest - # --------------------------------------------------------------------------- # Import the module under test from scripts/ # --------------------------------------------------------------------------- SCRIPTS_DIR = str(Path(__file__).parent.parent / "scripts") if SCRIPTS_DIR not in sys.path: sys.path.insert(0, SCRIPTS_DIR) - import turbo_hardware_diag as thd # noqa: E402 - # Import the replay module directly to avoid turboquant.__init__ pulling in numpy import importlib.util _hw_replay_path = Path(__file__).parent.parent / "turboquant" / "hw_replay.py" @@ -41,20 +34,15 @@ _spec.loader.exec_module(_hw_replay) HardwareProfile = _hw_replay.HardwareProfile parse_diag_output = _hw_replay.parse_diag_output - - # --------------------------------------------------------------------------- # Shared mock data # --------------------------------------------------------------------------- MOCK_BENCH_OUTPUT_Q8 = """\ | qwen35moe 35B.A3B Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | q8_0 | q8_0 | 1 | tg128 | 85.83 ± 0.17 |""" - MOCK_BENCH_OUTPUT_TURBO3 = """\ | qwen35moe 35B.A3B Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | turbo3 | turbo3 | 1 | tg128 | 77.42 ± 0.05 |""" - MOCK_BENCH_OUTPUT_TURBO3_DEEP = """\ | qwen35moe 35B.A3B Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | turbo3 | turbo3 | 1 | tg128 @ d4096 | 70.88 ± 1.27 |""" - MOCK_CLI_OUTPUT = """\ build: 8506 (dfc109798) ggml_metal_device_init: GPU name: MTL0 @@ -67,11 +55,9 @@ print_info: arch = qwen35moe print_info: n_layer = 40 print_info: n_expert = 256""" - MOCK_PPL_OUTPUT = """\ perplexity: calculating perplexity over 8 chunks Final estimate: PPL = 6.2109 +/- 0.33250""" - MOCK_VM_STAT = """\ Mach Virtual Memory Statistics: (page size of 4096 bytes) Pages free: 1000000. @@ -85,106 +71,82 @@ Pages reactivated: 100000. Pageins: 50000. Pageouts: 1000.""" - MOCK_PMSET_THERM = """\ 2026-03-26 10:00:00 -0700 CPU_Speed_Limit = 100""" - MOCK_PMSET_BATT = """\ Now drawing from 'AC Power' -InternalBattery-0 (id=1234567) 100%; charged; 0:00 remaining present: true""" - MOCK_PROC_CPUINFO = """\ processor : 0 model name : AMD EPYC 7B13 64-Core Processor core id : 0 - processor : 1 model name : AMD EPYC 7B13 64-Core Processor core id : 1 - processor : 2 model name : AMD EPYC 7B13 64-Core Processor core id : 0 - processor : 3 model name : AMD EPYC 7B13 64-Core Processor core id : 1""" - MOCK_PROC_MEMINFO = """\ MemTotal: 131072000 kB MemFree: 50000000 kB MemAvailable: 80000000 kB SwapTotal: 16384000 kB SwapFree: 16000000 kB""" - MOCK_NVIDIA_SMI_QUERY = "NVIDIA RTX 4090, 24564 MiB, 535.183.01, 00000000:01:00.0" - MOCK_FREE_OUTPUT = """\ total used free shared buff/cache available Mem: 131072000 40000000 50000000 1000000 41072000 80000000 Swap: 16384000 384000 16000000""" - - # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_log(tmp_path: Path) -> thd.DiagLog: """Create a DiagLog writing to a temp file.""" return thd.DiagLog(str(tmp_path / "test.txt")) - - def _make_completed_process(stdout: str = "", stderr: str = "", rc: int = 0): return subprocess.CompletedProcess(args=[], returncode=rc, stdout=stdout, stderr=stderr) - - # ============================================================ # 1. TestArgParse # ============================================================ class TestArgParse: """Argument parser edge cases.""" - def test_default_args(self): with patch("sys.argv", ["prog"]): parser = thd.main.__code__ # just verify parser exists in main assert parser is not None - def test_skip_ppl_flag(self): parser = self._build_parser() args = parser.parse_args(["--skip-ppl", "/dir", "/model.gguf"]) assert args.skip_ppl is True - def test_skip_stress_flag(self): parser = self._build_parser() args = parser.parse_args(["--skip-stress", "/dir", "/model.gguf"]) assert args.skip_stress is True - def test_model_flag(self): parser = self._build_parser() args = parser.parse_args(["--model", "/my/model.gguf"]) assert args.model_flag == "/my/model.gguf" - def test_llama_dir_flag(self): parser = self._build_parser() args = parser.parse_args(["--llama-dir", "/my/llama"]) assert args.llama_dir_flag == "/my/llama" - def test_verbose_flag(self): parser = self._build_parser() args = parser.parse_args(["-v", "/dir", "/model.gguf"]) assert args.verbose is True - def test_output_dir_flag(self): parser = self._build_parser() args = parser.parse_args(["-o", "/tmp/out", "/dir", "/model.gguf"]) assert args.output_dir == "/tmp/out" - def test_help_exits_0(self): parser = self._build_parser() with pytest.raises(SystemExit) as exc: parser.parse_args(["--help"]) assert exc.value.code == 0 - @staticmethod def _build_parser(): """Rebuild the argparse parser that main() uses internally.""" @@ -203,28 +165,23 @@ def _build_parser(): parser.add_argument("--verbose", "-v", action="store_true") parser.add_argument("--output-dir", "-o", default=".") return parser - - # ============================================================ # 2. TestDiagLog # ============================================================ class TestDiagLog: """DiagLog dual-output behavior.""" - def test_write_outputs_to_file(self, tmp_path): log = _make_log(tmp_path) log.write("hello world") log.close() content = (tmp_path / "test.txt").read_text() assert "hello world" in content - def test_write_outputs_to_stdout(self, tmp_path, capsys): log = _make_log(tmp_path) log.write("stdout test") log.close() captured = capsys.readouterr() assert "stdout test" in captured.out - def test_section_format_equals(self, tmp_path): log = _make_log(tmp_path) log.section("TEST SECTION") @@ -232,14 +189,12 @@ def test_section_format_equals(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "=" * 64 in content assert "TEST SECTION" in content - def test_subsection_format(self, tmp_path): log = _make_log(tmp_path) log.subsection("My Subsection") log.close() content = (tmp_path / "test.txt").read_text() assert "--- My Subsection ---" in content - def test_file_flushed_on_each_write(self, tmp_path): log = _make_log(tmp_path) log.write("flush test") @@ -247,30 +202,25 @@ def test_file_flushed_on_each_write(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "flush test" in content log.close() - def test_concurrent_write_safety(self, tmp_path): log = _make_log(tmp_path) errors = [] - def writer(n): try: for i in range(20): log.write(f"thread-{n}-line-{i}") except Exception as e: errors.append(e) - threads = [threading.Thread(target=writer, args=(i,)) for i in range(4)] for t in threads: t.start() for t in threads: t.join() log.close() - assert len(errors) == 0 content = (tmp_path / "test.txt").read_text() # All 80 lines should be present assert content.count("\n") >= 80 - def test_write_file_only_no_stdout(self, tmp_path, capsys): log = _make_log(tmp_path) log.write_file_only("secret line") @@ -279,14 +229,11 @@ def test_write_file_only_no_stdout(self, tmp_path, capsys): assert "secret line" not in captured.out content = (tmp_path / "test.txt").read_text() assert "secret line" in content - - # ============================================================ # 3. TestBackgroundMonitor # ============================================================ class TestBackgroundMonitor: """Background system metrics monitor.""" - def test_starts_and_stops_cleanly(self, tmp_path): csv_path = str(tmp_path / "monitor.csv") mon = thd.BackgroundMonitor(csv_path) @@ -294,7 +241,6 @@ def test_starts_and_stops_cleanly(self, tmp_path): assert mon.is_alive() mon.stop() assert not mon.is_alive() - def test_csv_header_correct(self, tmp_path): csv_path = str(tmp_path / "monitor.csv") mon = thd.BackgroundMonitor(csv_path) @@ -306,7 +252,6 @@ def test_csv_header_correct(self, tmp_path): assert "gpu_temp_c" in header assert "swap_used_mb" in header # Don't call mon.stop() — thread was never started - @patch.object(thd.BackgroundMonitor, "_poll") def test_produces_samples(self, mock_poll, tmp_path): mock_poll.return_value = { @@ -328,7 +273,6 @@ def test_produces_samples(self, mock_poll, tmp_path): assert len(mon.samples) >= 1 finally: thd.MONITOR_POLL_INTERVAL = orig - @patch("subprocess.check_output") @patch("platform.system", return_value="Darwin") def test_darwin_mem_pressure(self, mock_plat, mock_subp): @@ -336,14 +280,12 @@ def test_darwin_mem_pressure(self, mock_plat, mock_subp): result = thd.BackgroundMonitor._macos_mem_pressure() # (500000 + 300000) / (500000 + 300000 + 1000000) = 44% assert result == "44" - @patch("subprocess.check_output") @patch("platform.system", return_value="Darwin") def test_darwin_cpu_speed_limit(self, mock_plat, mock_subp): mock_subp.return_value = MOCK_PMSET_THERM result = thd.BackgroundMonitor._macos_cpu_speed_limit() assert result == "100" - @patch("subprocess.check_output") @patch("platform.system", return_value="Linux") def test_linux_mem_pct(self, mock_plat, mock_subp): @@ -351,27 +293,21 @@ def test_linux_mem_pct(self, mock_plat, mock_subp): result = thd.BackgroundMonitor._linux_mem_pct() # 40000000 / 131072000 * 100 = ~31 assert int(result) == 31 - @patch("subprocess.check_output", side_effect=FileNotFoundError) def test_graceful_na_when_probes_fail(self, mock_subp): result = thd.BackgroundMonitor._nvidia_query("temperature.gpu") assert result == "N/A" - - # ============================================================ # 4. TestPlatformDetection # ============================================================ class TestPlatformDetection: """Platform detection and hardware collection.""" - @patch("platform.system", return_value="Darwin") def test_darwin_detection(self, mock_sys): assert thd.detect_platform() == "Darwin" - @patch("platform.system", return_value="Linux") def test_linux_detection(self, mock_sys): assert thd.detect_platform() == "Linux" - @patch("turbo_hardware_diag._run_cmd") @patch("turbo_hardware_diag.detect_platform", return_value="Darwin") @patch("platform.release", return_value="25.3.0") @@ -404,16 +340,13 @@ def cmd_side_effect(cmd, **kwargs): if "-g" in cmd and "therm" in cmd: return MOCK_PMSET_THERM return "" - mock_cmd.side_effect = cmd_side_effect log = _make_log(tmp_path) hw = thd.detect_hardware(log) log.close() - assert hw["cpu_brand"] == "Apple M5 Max" assert hw["ram_total_gb"] == 128 assert hw["apple_silicon"] is True - @patch("turbo_hardware_diag._run_cmd", return_value="") @patch("turbo_hardware_diag.detect_platform", return_value="Linux") @patch("platform.release", return_value="6.1.0") @@ -426,7 +359,6 @@ def test_collect_hw_linux(self, mock_mach, mock_rel, mock_plat, mock_cmd, tmp_pa mock_read.side_effect = lambda *a, **kw: { True: MOCK_PROC_CPUINFO, # first call }.get(True, MOCK_PROC_MEMINFO) - # More precise mocking for Path.read_text call_count = {"n": 0} def read_side_effect(*a, **kw): @@ -434,15 +366,11 @@ def read_side_effect(*a, **kw): if call_count["n"] == 1: return MOCK_PROC_CPUINFO return MOCK_PROC_MEMINFO - mock_read.side_effect = read_side_effect - log = _make_log(tmp_path) hw = thd.detect_hardware(log) log.close() - assert hw["platform"] == "Linux" - @patch("turbo_hardware_diag._run_cmd") @patch("turbo_hardware_diag.detect_platform", return_value="Darwin") @patch("platform.release", return_value="25.3.0") @@ -468,15 +396,12 @@ def cmd_side_effect(cmd, **kwargs): if cmd[0] == "pmset": return MOCK_PMSET_THERM return "" - mock_cmd.side_effect = cmd_side_effect log = _make_log(tmp_path) hw = thd.detect_hardware(log) log.close() - content = (tmp_path / "test.txt").read_text() assert "[HW_GPU]" in content - @patch("turbo_hardware_diag._run_cmd") @patch("turbo_hardware_diag.detect_platform", return_value="Linux") @patch("platform.release", return_value="6.1.0") @@ -487,9 +412,7 @@ def cmd_side_effect(cmd, **kwargs): if isinstance(cmd, list) and cmd[0] == "nvidia-smi": return MOCK_NVIDIA_SMI_QUERY return "" - mock_cmd.side_effect = cmd_side_effect - with patch("pathlib.Path.read_text") as mock_read, \ patch("pathlib.Path.exists", return_value=True), \ patch("pathlib.Path.is_dir", return_value=False): @@ -500,15 +423,12 @@ def read_side_effect(*a, **kw): return MOCK_PROC_CPUINFO return MOCK_PROC_MEMINFO mock_read.side_effect = read_side_effect - log = _make_log(tmp_path) hw = thd.detect_hardware(log) log.close() - content = (tmp_path / "test.txt").read_text() assert "[HW_GPU]" in content assert hw.get("gpu_backend") == "cuda" - @patch("turbo_hardware_diag._run_cmd") @patch("turbo_hardware_diag.detect_platform", return_value="Linux") @patch("platform.release", return_value="6.1.0") @@ -517,11 +437,9 @@ def read_side_effect(*a, **kw): def test_amd_rocm_detection(self, mock_which, mock_mach, mock_rel, mock_plat, mock_cmd, tmp_path): """When nvidia-smi missing but /sys/class/drm exists, detect AMD/other GPU.""" mock_cmd.return_value = "" - with patch("pathlib.Path.read_text") as mock_read, \ patch("pathlib.Path.exists", return_value=True), \ patch("pathlib.Path.is_dir") as mock_isdir: - call_count = {"n": 0} def read_side_effect(*a, **kw): call_count["n"] += 1 @@ -532,17 +450,14 @@ def read_side_effect(*a, **kw): elif "vendor" in str(a) or "vendor" in str(kw): return "0x1002" # AMD return "0x73bf" # device id - mock_read.side_effect = read_side_effect mock_isdir.return_value = True - with patch("pathlib.Path.glob") as mock_glob, \ patch("pathlib.Path.iterdir", return_value=[]): mock_glob.return_value = [] # no card dirs to iterate log = _make_log(tmp_path) hw = thd.detect_hardware(log) log.close() - @patch("turbo_hardware_diag._run_cmd", side_effect=Exception("sysctl blew up")) @patch("turbo_hardware_diag.detect_platform", return_value="Windows") @patch("platform.release", return_value="10.0") @@ -553,14 +468,11 @@ def test_graceful_failure_unsupported_platform(self, mock_mach, mock_rel, mock_p log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - - # ============================================================ # 5. TestBenchRunner # ============================================================ class TestBenchRunner: """run_bench / run_perpl subprocess wrappers.""" - def test_run_bench_emits_bench_start_end(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess", return_value=(MOCK_BENCH_OUTPUT_Q8, 0)): @@ -572,7 +484,6 @@ def test_run_bench_emits_bench_start_end(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert '[BENCH_START] label="q8_0 decode (short)"' in content assert '[BENCH_END] label="q8_0 decode (short)"' in content - def test_run_bench_passes_through_table_rows(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess", return_value=(MOCK_BENCH_OUTPUT_TURBO3, 0)): @@ -581,7 +492,6 @@ def test_run_bench_passes_through_table_rows(self, tmp_path): log, "/fake/llama-bench", "/fake/model.gguf", ) assert "77.42" in output - def test_run_bench_env_vars_for_mode2(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess") as mock_sub: @@ -595,7 +505,6 @@ def test_run_bench_env_vars_for_mode2(self, tmp_path): call_args = mock_sub.call_args assert call_args[1]["env_extra"] == {"TURBO_LAYER_ADAPTIVE": "2"} log.close() - def test_run_perpl_emits_ppl_start_end(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess", return_value=(MOCK_PPL_OUTPUT, 0)): @@ -607,7 +516,6 @@ def test_run_perpl_emits_ppl_start_end(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert '[PPL_START] label="q8_0 PPL"' in content assert '[PPL_END] label="q8_0 PPL"' in content - def test_subprocess_timeout_handled(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.Popen") as mock_popen: @@ -616,13 +524,11 @@ def test_subprocess_timeout_handled(self, tmp_path): mock_proc.wait.side_effect = subprocess.TimeoutExpired(cmd="test", timeout=600) mock_proc.kill = MagicMock() mock_popen.return_value = mock_proc - output, rc = thd._run_subprocess(["fake"], log, timeout=1) log.close() assert rc == -1 content = (tmp_path / "test.txt").read_text() assert "timed out" in content - def test_subprocess_crash_emits_failed_continues(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess", return_value=("", 1)): @@ -633,7 +539,6 @@ def test_subprocess_crash_emits_failed_continues(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "FAILED" in content - def test_correct_argument_construction(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess") as mock_sub: @@ -650,7 +555,6 @@ def test_correct_argument_construction(self, tmp_path): assert "-d" in cmd assert "4096" in cmd log.close() - def test_run_bench_wall_seconds_positive(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_subprocess", return_value=(MOCK_BENCH_OUTPUT_Q8, 0)): @@ -660,20 +564,16 @@ def test_run_bench_wall_seconds_positive(self, tmp_path): ) log.close() assert wall >= 0 - - # ============================================================ # 6. TestAnomalyDetector # ============================================================ class TestAnomalyDetector: """Real-time anomaly detection.""" - def _make_detector(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") mon = thd.BackgroundMonitor(csv_path) return thd.AnomalyDetector(log, mon), log, mon - def test_decode_ratio_drop_flagged(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_decode_ratio(4096, 0.90) @@ -681,14 +581,12 @@ def test_decode_ratio_drop_flagged(self, tmp_path): log.close() assert len(det.anomalies) == 1 assert "degradation" in det.anomalies[0].lower() - def test_thermal_throttling_flagged(self, tmp_path): det, log, mon = self._make_detector(tmp_path) mon._samples = [{"cpu_speed_limit": "80"}] det.check_thermal() log.close() assert any("thermal" in a.lower() for a in det.anomalies) - def test_swap_growth_flagged(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.set_initial_swap(100) @@ -696,21 +594,18 @@ def test_swap_growth_flagged(self, tmp_path): det.check_swap_growth() log.close() assert any("swap" in a.lower() for a in det.anomalies) - def test_normal_results_not_flagged(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_decode_ratio(4096, 0.90) det.check_decode_ratio(8192, 0.85) # 5.6% drop — fine log.close() assert len(det.anomalies) == 0 - def test_ppl_quality_regression_flagged(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.set_q8_ppl(6.0) det.check_ppl("turbo3", 7.0) # 16.7% > 10% log.close() assert any("regression" in a.lower() for a in det.anomalies) - def test_multiple_anomalies_accumulated(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_decode_ratio(4096, 0.90) @@ -719,9 +614,7 @@ def test_multiple_anomalies_accumulated(self, tmp_path): det.check_thermal() log.close() assert len(det.anomalies) == 2 - # --- Notable and Investigate detection --- - def test_turbo3_faster_than_q8_flags_investigate(self, tmp_path): """turbo3 beating q8_0 by >5% is outlandish.""" det, log, mon = self._make_detector(tmp_path) @@ -729,7 +622,6 @@ def test_turbo3_faster_than_q8_flags_investigate(self, tmp_path): log.close() assert len(det.investigations) >= 1 assert any("faster" in i.lower() for i in det.investigations) - def test_excellent_decode_ratio_flags_notable(self, tmp_path): """Near-parity at long context is notable (good).""" det, log, mon = self._make_detector(tmp_path) @@ -737,7 +629,6 @@ def test_excellent_decode_ratio_flags_notable(self, tmp_path): log.close() assert len(det.notables) >= 1 assert any("excellent" in n.lower() for n in det.notables) - def test_below_half_ratio_flags_investigate(self, tmp_path): """Below 0.5x is always a red flag.""" det, log, mon = self._make_detector(tmp_path) @@ -745,7 +636,6 @@ def test_below_half_ratio_flags_investigate(self, tmp_path): log.close() assert len(det.investigations) >= 1 assert any("0.5x" in i for i in det.investigations) - def test_decode_improving_at_depth_flags_investigate(self, tmp_path): """Decode getting faster at deeper context is suspicious.""" det, log, mon = self._make_detector(tmp_path) @@ -754,7 +644,6 @@ def test_decode_improving_at_depth_flags_investigate(self, tmp_path): log.close() assert len(det.investigations) >= 1 assert any("improved" in i.lower() for i in det.investigations) - def test_ppl_better_than_q8_flags_investigate(self, tmp_path): """turbo3 PPL better than q8_0 is outlandish.""" det, log, mon = self._make_detector(tmp_path) @@ -763,7 +652,6 @@ def test_ppl_better_than_q8_flags_investigate(self, tmp_path): log.close() assert len(det.investigations) >= 1 assert any("better" in i.lower() for i in det.investigations) - def test_ppl_near_match_flags_notable(self, tmp_path): """turbo3 PPL matching q8_0 within 0.1% is notable.""" det, log, mon = self._make_detector(tmp_path) @@ -772,28 +660,24 @@ def test_ppl_near_match_flags_notable(self, tmp_path): log.close() assert len(det.notables) >= 1 assert any("excellent" in n.lower() or "matches" in n.lower() for n in det.notables) - def test_prefill_much_faster_flags_investigate(self, tmp_path): """turbo3 prefill >10% faster than q8_0 is suspicious.""" det, log, mon = self._make_detector(tmp_path) det.check_prefill_ratio(4096, 1.15) log.close() assert len(det.investigations) >= 1 - def test_prefill_slightly_faster_flags_notable(self, tmp_path): """turbo3 prefill slightly faster at long context is notable (good).""" det, log, mon = self._make_detector(tmp_path) det.check_prefill_ratio(16384, 1.03) log.close() assert len(det.notables) >= 1 - def test_prefill_too_slow_flags_investigate(self, tmp_path): """turbo3 prefill <90% of q8_0 warrants investigation.""" det, log, mon = self._make_detector(tmp_path) det.check_prefill_ratio(4096, 0.85) log.close() assert len(det.investigations) >= 1 - def test_clean_results_no_flags(self, tmp_path): """Normal results should produce no flags at all.""" det, log, mon = self._make_detector(tmp_path) @@ -805,18 +689,14 @@ def test_clean_results_no_flags(self, tmp_path): assert len(det.anomalies) == 0 assert len(det.investigations) == 0 # notables might be 0 — that's fine - - # ============================================================ # 7. TestLiveDisplay # ============================================================ class TestLiveDisplay: """Live terminal display with rich fallback.""" - def test_ascii_fallback_when_rich_not_available(self): display = thd.LiveDisplay(use_rich=False) assert display._use_rich is False - def test_bar_chart_generation(self, capsys): display = thd.LiveDisplay(use_rich=False) display.update_decode("q8_0", 4096, 80.0) @@ -825,7 +705,6 @@ def test_bar_chart_generation(self, capsys): captured = capsys.readouterr() # Ratio is 0.90, bar should be generated assert "0.90" in captured.out - def test_colored_ratio_formatting(self): """Rich table builder produces colored ratio text when rich is available.""" if not thd.HAS_RICH: @@ -835,7 +714,6 @@ def test_colored_ratio_formatting(self): display.update_decode("turbo3", 0, 78.0) table = display._build_rich_table() assert table is not None - def test_display_updates_on_new_results(self): display = thd.LiveDisplay(use_rich=False) display.update_decode("q8_0", 0, 85.0) @@ -843,14 +721,11 @@ def test_display_updates_on_new_results(self): display.update_decode("turbo3", 0, 77.0) assert 0 in display._ratios assert abs(display._ratios[0] - 77.0 / 85.0) < 0.001 - - # ============================================================ # 8. TestSections # ============================================================ class TestSections: """Each section function produces expected tags.""" - def test_section_1_hardware_inventory(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.detect_hardware", return_value={"cpu_brand": "test"}): @@ -858,7 +733,6 @@ def test_section_1_hardware_inventory(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "HARDWARE INVENTORY" in content - def test_section_2_system_load(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.capture_load"), \ @@ -869,7 +743,6 @@ def test_section_2_system_load(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "SYSTEM LOAD" in content - def test_section_3_model_info(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -880,7 +753,6 @@ def test_section_3_model_info(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "[MODEL]" in content assert "Qwen3.5-35B-A3B" in content - def test_section_4_gpu_capabilities(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -894,7 +766,6 @@ def test_section_4_gpu_capabilities(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "[GPU]" in content assert "MTL0" in gpu_init - def test_section_5_build_validation(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -906,7 +777,6 @@ def test_section_5_build_validation(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[BUILD]" in content - def test_section_6_prefill(self, tmp_path): log = _make_log(tmp_path) display = thd.LiveDisplay(use_rich=False) @@ -916,14 +786,12 @@ def test_section_6_prefill(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "PREFILL SPEED" in content - def test_section_7_decode(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") mon = thd.BackgroundMonitor(csv_path) display = thd.LiveDisplay(use_rich=False) anomaly = thd.AnomalyDetector(log, mon) - with patch("turbo_hardware_diag.run_bench", return_value=(MOCK_BENCH_OUTPUT_Q8, 1.0)), \ patch("turbo_hardware_diag.parse_bench_tps", return_value=[{"mode": "decode", "tps": 85.0, "stddev": 0.1, "depth": 0, "ctk": "q8_0"}]), \ patch("turbo_hardware_diag.capture_load"): @@ -931,13 +799,11 @@ def test_section_7_decode(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "DECODE SPEED" in content - def test_section_10_ppl_skips_when_wiki_missing(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") mon = thd.BackgroundMonitor(csv_path) anomaly = thd.AnomalyDetector(log, mon) - with patch("os.path.isfile", return_value=False): thd.section_10_perplexity( log, "/fake/perpl", "/fake/model.gguf", "/fake/wiki.raw", @@ -946,7 +812,6 @@ def test_section_10_ppl_skips_when_wiki_missing(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "SKIPPED" in content - def test_section_10_ppl_skip_flag(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") @@ -959,27 +824,22 @@ def test_section_10_ppl_skip_flag(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "--skip-ppl" in content - def test_section_8_stress_loops_all_depths(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") mon = thd.BackgroundMonitor(csv_path) display = thd.LiveDisplay(use_rich=False) anomaly = thd.AnomalyDetector(log, mon) - call_labels = [] def fake_bench(label, *args, **kwargs): call_labels.append(label) return ("", 1.0) - with patch("turbo_hardware_diag.run_bench", side_effect=fake_bench), \ patch("turbo_hardware_diag.capture_load"): thd.section_8_stress_test(log, "/fake/bench", "/fake/model.gguf", display, anomaly) log.close() - # Should have been called for each depth in STRESS_DEPTHS x 2 (turbo3 + q8_0) assert len(call_labels) == len(thd.STRESS_DEPTHS) * 2 - def test_section_12_post_load(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.capture_load"), \ @@ -989,7 +849,6 @@ def test_section_12_post_load(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "post-benchmark" in content.lower() - def test_section_13_summary(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") @@ -999,20 +858,16 @@ def test_section_13_summary(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "TURBO_DIAG_COMPLETE=true" in content - - # ============================================================ # 9. TestTagCompatibility — CRITICAL # ============================================================ class TestTagCompatibility: """Generate a complete .txt from mocked run, verify hw_replay can parse it.""" - FULL_DIAG_OUTPUT = """\ TurboQuant Hardware Diagnostic v5 TURBO_DIAG_VERSION=5 TURBO_DIAG_TIMESTAMP=2026-03-26T13:43:09Z TURBO_DIAG_MODEL=Qwen3.5-35B-A3B-Q8_0.gguf - [HW] os=Darwin os_version=25.3.0 arch=arm64 [HW] cpu_brand=Apple M5 Max [HW] cpu_cores_physical=18 @@ -1022,65 +877,51 @@ class TestTagCompatibility: [HW] chip_model=Apple M5 Max [HW] l1_dcache=65536 [HW] l2_cache=8388608 - [GPU] ggml_metal_device_init: GPU name: MTL0 [GPU] ggml_metal_device_init: GPU family: MTLGPUFamilyApple10 (1010) [GPU] ggml_metal_device_init: has tensor = true [GPU] ggml_metal_device_init: has unified memory = true [GPU] ggml_metal_device_init: recommendedMaxWorkingSetSize = 115448.73 MB - [MODEL] print_info: general.name = Qwen3.5-35B-A3B [MODEL] print_info: arch = qwen35moe [MODEL] print_info: n_layer = 40 [MODEL] print_info: n_expert = 256 [MODEL] filename=Qwen3.5-35B-A3B-Q8_0.gguf [MODEL] filesize_bytes=36893488147 - [BUILD] dfc1097 fix: add turbo3/turbo4 cache types - [BENCH_START] label="q8_0 decode (short)" ctk=q8_0 ctv=q8_0 args="-p 0 -n 128" env="" timestamp=2026-03-26T13:45:00Z | qwen35moe 35B.A3B Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | q8_0 | q8_0 | 1 | tg128 | 85.83 ± 0.17 | [BENCH_END] label="q8_0 decode (short)" wall_sec=5 - [BENCH_START] label="turbo3 decode (short)" ctk=turbo3 ctv=turbo3 args="-p 0 -n 128" env="" timestamp=2026-03-26T13:46:00Z | qwen35moe 35B.A3B Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | turbo3 | turbo3 | 1 | tg128 | 77.42 ± 0.05 | [BENCH_END] label="turbo3 decode (short)" wall_sec=6 - [BENCH_START] label="turbo3 decode @4K" ctk=turbo3 ctv=turbo3 args="-p 0 -n 128 -d 4096" env="" timestamp=2026-03-26T13:47:00Z | qwen35moe 35B.A3B Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | turbo3 | turbo3 | 1 | tg128 @ d4096 | 70.88 ± 1.27 | [BENCH_END] label="turbo3 decode @4K" wall_sec=10 - [PPL_START] label="q8_0 PPL (8 chunks)" ctk=q8_0 ctv=q8_0 chunks=8 timestamp=2026-03-26T14:00:00Z Final estimate: PPL = 6.1109 +/- 0.32553 [PPL_END] label="q8_0 PPL (8 chunks)" - [PPL_START] label="turbo3 PPL (8 chunks)" ctk=turbo3 ctv=turbo3 chunks=8 env="" timestamp=2026-03-26T14:05:00Z Final estimate: PPL = 6.2109 +/- 0.33250 [PPL_END] label="turbo3 PPL (8 chunks)" - [LOAD_SNAPSHOT] label=pre_benchmark timestamp=2026-03-26T13:43:09Z [LOAD_SNAPSHOT] load_avg=1.5 2.0 1.8 [LOAD_SNAPSHOT] process_count=350 [LOAD_SNAPSHOT] approx_free_ram=50000 MB - [LOAD_SNAPSHOT] label=post_all_benchmarks timestamp=2026-03-26T14:10:00Z [LOAD_SNAPSHOT] load_avg=3.0 2.5 2.0 [LOAD_SNAPSHOT] process_count=355 - TURBO_DIAG_COMPLETE=true """ - def test_parse_diag_output_succeeds(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert profile.diag_version == 5 - def test_system_fields_populated(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert profile.system.platform == "Darwin" assert profile.system.cpu_brand == "Apple M5 Max" assert profile.system.ram_total_gb == 128 assert profile.system.apple_silicon is True - def test_gpu_fields_populated(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert profile.system.gpu.name == "MTL0" @@ -1088,7 +929,6 @@ def test_gpu_fields_populated(self): assert profile.system.gpu.has_tensor is True assert profile.system.gpu.has_unified_memory is True assert profile.system.gpu.recommended_max_working_set_mb == 115448.73 - def test_model_fields_populated(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert profile.model.name == "Qwen3.5-35B-A3B" @@ -1096,27 +936,23 @@ def test_model_fields_populated(self): assert profile.model.n_layer == 40 assert profile.model.n_expert == 256 assert profile.model.filename == "Qwen3.5-35B-A3B-Q8_0.gguf" - def test_benchmarks_populated(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert len(profile.benchmarks) >= 3 q8 = [b for b in profile.benchmarks if b.cache_type_k == "q8_0"] assert len(q8) >= 1 assert q8[0].tok_per_sec == 85.83 - def test_ppl_populated(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert len(profile.ppl_results) == 2 turbo_ppl = [p for p in profile.ppl_results if p.cache_type == "turbo3"] assert turbo_ppl[0].ppl == 6.2109 - def test_load_snapshots_populated(self): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) assert len(profile.load_snapshots) >= 2 labels = [s.label for s in profile.load_snapshots] assert "pre_benchmark" in labels assert "post_all_benchmarks" in labels - def test_json_roundtrip(self, tmp_path): profile = parse_diag_output(self.FULL_DIAG_OUTPUT) json_path = tmp_path / "profile.json" @@ -1126,14 +962,11 @@ def test_json_roundtrip(self, tmp_path): assert loaded.model.n_layer == profile.model.n_layer assert len(loaded.benchmarks) == len(profile.benchmarks) assert len(loaded.ppl_results) == len(profile.ppl_results) - - # ============================================================ # 10. TestJSONProfile # ============================================================ class TestJSONProfile: """build_json_profile produces valid, complete JSON.""" - def test_has_all_required_keys(self): with patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ patch("platform.release", return_value="25.3.0"), \ @@ -1149,7 +982,6 @@ def test_has_all_required_keys(self): assert "platform" in profile assert "hardware" in profile assert "model_file" in profile - def test_system_gpu_has_all_fields(self): with patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ patch("platform.release", return_value="25.3.0"), \ @@ -1166,7 +998,6 @@ def test_system_gpu_has_all_fields(self): assert "has_tensor" in hw assert hw["has_tensor"] is True assert "Apple10" in hw["gpu_family"] - def test_benchmarks_entries_complete(self): # build_json_profile doesn't include benchmark entries (that's in the .txt) # but the profile dict should still be valid JSON @@ -1176,7 +1007,6 @@ def test_benchmarks_entries_complete(self): patch("os.path.getsize", return_value=36_000_000_000): profile = thd.build_json_profile({}, "/fake/model.gguf", "", "20260326") assert profile["diag_version"] == thd.DIAG_VERSION - def test_valid_json(self): with patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ patch("platform.release", return_value="25.3.0"), \ @@ -1189,7 +1019,6 @@ def test_valid_json(self): json_str = json.dumps(profile) loaded = json.loads(json_str) assert loaded["diag_version"] == thd.DIAG_VERSION - def test_model_size_bytes_populated(self): with patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ patch("platform.release", return_value="25.3.0"), \ @@ -1199,31 +1028,23 @@ def test_model_size_bytes_populated(self): {}, "/fake/model.gguf", "", "20260326", ) assert profile["model_size_bytes"] == 36_000_000_000 - - # ============================================================ # 11. TestPackaging # ============================================================ class TestPackaging: """ZIP packaging of results.""" - def _setup_packaging(self, tmp_path): log_path = str(tmp_path / "turbo-diag-20260326.txt") log = thd.DiagLog(log_path) log.write("test log content") - csv_path = str(tmp_path / "turbo-monitor-20260326.csv") with open(csv_path, "w") as f: f.write("timestamp,load_1m\n2026-03-26,1.5\n") - mon = MagicMock() mon.csv_path = csv_path mon.sample_count = 10 - profile_json = {"diag_version": 5, "platform": "Darwin"} - return log, mon, profile_json - def test_zip_created_with_all_files(self, tmp_path): log, mon, profile_json = self._setup_packaging(tmp_path) zip_path = thd.package_results(log, mon, profile_json, "20260326", str(tmp_path)) @@ -1234,32 +1055,26 @@ def test_zip_created_with_all_files(self, tmp_path): assert any(n.endswith(".txt") for n in names) assert any(n.endswith(".json") for n in names) assert any(n.endswith(".csv") for n in names) - def test_zip_is_valid(self, tmp_path): log, mon, profile_json = self._setup_packaging(tmp_path) zip_path = thd.package_results(log, mon, profile_json, "20260326", str(tmp_path)) log.close() assert zipfile.is_zipfile(zip_path) - def test_missing_csv_doesnt_crash(self, tmp_path): log_path = str(tmp_path / "turbo-diag-20260326.txt") log = thd.DiagLog(log_path) log.write("test") - mon = MagicMock() mon.csv_path = str(tmp_path / "nonexistent.csv") # doesn't exist mon.sample_count = 0 - zip_path = thd.package_results(log, mon, {"diag_version": 5}, "20260326", str(tmp_path)) log.close() assert os.path.isfile(zip_path) - def test_filenames_follow_pattern(self, tmp_path): log, mon, profile_json = self._setup_packaging(tmp_path) zip_path = thd.package_results(log, mon, profile_json, "20260326", str(tmp_path)) log.close() assert "turbo-diag-20260326.zip" in zip_path - def test_hwprofile_json_inside_zip(self, tmp_path): log, mon, profile_json = self._setup_packaging(tmp_path) zip_path = thd.package_results(log, mon, profile_json, "20260326", str(tmp_path)) @@ -1269,19 +1084,15 @@ def test_hwprofile_json_inside_zip(self, tmp_path): assert len(json_files) == 1 content = json.loads(zf.read(json_files[0])) assert content["diag_version"] == 5 - - # ============================================================ # 12. TestGracefulDegradation # ============================================================ class TestGracefulDegradation: """System survives when probes and tools are missing.""" - - def test_missing_nvidia_smi_warning(self, tmp_path): + @patch("subprocess.check_output", side_effect=FileNotFoundError) + def test_missing_nvidia_smi_warning(self, mock_subp, tmp_path): result = thd.BackgroundMonitor._nvidia_query("temperature.gpu") - # On macOS (where these tests actually run), nvidia-smi doesn't exist assert result == "N/A" - @patch("turbo_hardware_diag._run_cmd", return_value="") @patch("turbo_hardware_diag.detect_platform", return_value="Darwin") @patch("platform.release", return_value="25.3.0") @@ -1293,7 +1104,6 @@ def test_missing_system_profiler_warning(self, mock_mach, mock_rel, mock_plat, m log.close() # Should complete without exception assert hw["platform"] == "Darwin" - @patch("turbo_hardware_diag._run_cmd", side_effect=Exception("boom")) @patch("turbo_hardware_diag.detect_platform", return_value="Darwin") @patch("platform.release", return_value="25.3.0") @@ -1304,7 +1114,6 @@ def test_sysctl_failure_warning_defaults(self, mock_mach, mock_rel, mock_plat, m log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_subprocess_timeout_produces_timeout_tag(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.Popen") as mock_popen: @@ -1313,13 +1122,11 @@ def test_subprocess_timeout_produces_timeout_tag(self, tmp_path): mock_proc.wait.side_effect = subprocess.TimeoutExpired("cmd", 5) mock_proc.kill = MagicMock() mock_popen.return_value = mock_proc - output, rc = thd._run_subprocess(["fake-cmd"], log, timeout=5) log.close() assert rc == -1 content = (tmp_path / "test.txt").read_text() assert "timed out" in content - def test_permission_denied_warning(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.Popen", side_effect=PermissionError("denied")): @@ -1328,7 +1135,6 @@ def test_permission_denied_warning(self, tmp_path): assert rc == -1 content = (tmp_path / "test.txt").read_text() assert "failed" in content.lower() - def test_empty_model_error_not_traceback(self): """main() with empty model path gives clean error, not a traceback.""" with patch("sys.argv", ["prog", "/nonexistent/dir"]), \ @@ -1339,7 +1145,6 @@ def test_empty_model_error_not_traceback(self): # Should have printed ERROR, not a traceback printed = " ".join(str(c) for c in mock_print.call_args_list) assert "ERROR" in printed - @patch("turbo_hardware_diag._run_cmd", return_value="") @patch("turbo_hardware_diag.detect_platform", return_value="Darwin") @patch("platform.release", return_value="25.3.0") @@ -1351,29 +1156,23 @@ def test_all_probes_fail_still_completes(self, mock_mach, mock_rel, mock_plat, m # hw dict should exist even if sparse assert isinstance(hw, dict) assert "platform" in hw - def test_partial_data_still_packaged(self, tmp_path): log_path = str(tmp_path / "turbo-diag-partial.txt") log = thd.DiagLog(log_path) log.write("partial data only") - mon = MagicMock() mon.csv_path = str(tmp_path / "nonexistent.csv") mon.sample_count = 0 - zip_path = thd.package_results(log, mon, {"diag_version": 5}, "partial", str(tmp_path)) log.close() assert os.path.isfile(zip_path) with zipfile.ZipFile(zip_path, "r") as zf: assert any(n.endswith(".txt") for n in zf.namelist()) - - # ============================================================ # 13. TestNoPII # ============================================================ class TestNoPII: """Output must not contain personally identifiable information.""" - def test_no_username_in_tags(self, tmp_path): username = os.environ.get("USER", "testuser") log = _make_log(tmp_path) @@ -1383,7 +1182,6 @@ def test_no_username_in_tags(self, tmp_path): patch("turbo_hardware_diag._run_cmd", return_value=""): hw = thd.detect_hardware(log) log.close() - content = (tmp_path / "test.txt").read_text() # Check that tagged lines ([HW], [GPU], etc.) don't contain username tagged_lines = [l for l in content.splitlines() @@ -1392,7 +1190,6 @@ def test_no_username_in_tags(self, tmp_path): if len(username) > 2: assert f"/Users/{username}/" not in line, f"Username in tag: {line}" assert f"/home/{username}/" not in line, f"Username in tag: {line}" - def test_no_home_dir_in_tags(self, tmp_path): home = str(Path.home()) log = _make_log(tmp_path) @@ -1402,13 +1199,11 @@ def test_no_home_dir_in_tags(self, tmp_path): patch("turbo_hardware_diag._run_cmd", return_value=""): hw = thd.detect_hardware(log) log.close() - content = (tmp_path / "test.txt").read_text() tagged_lines = [l for l in content.splitlines() if l.startswith("[HW]")] for line in tagged_lines: assert home not in line, f"Home dir in HW tag: {line}" - def test_no_email_addresses(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ @@ -1417,97 +1212,78 @@ def test_no_email_addresses(self, tmp_path): patch("turbo_hardware_diag._run_cmd", return_value=""): hw = thd.detect_hardware(log) log.close() - content = (tmp_path / "test.txt").read_text() emails = re.findall(r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b', content) assert len(emails) == 0, f"Found email addresses: {emails}" - - # ============================================================ # Bonus: parse_bench_tps and parse_ppl_final unit tests # ============================================================ class TestParsingHelpers: """Unit tests for bench output parsing.""" - def test_parse_bench_tps_decode(self): results = thd.parse_bench_tps(MOCK_BENCH_OUTPUT_Q8) assert len(results) == 1 assert results[0]["mode"] == "decode" assert results[0]["tps"] == 85.83 assert results[0]["ctk"] == "q8_0" - def test_parse_bench_tps_with_depth(self): results = thd.parse_bench_tps(MOCK_BENCH_OUTPUT_TURBO3_DEEP) assert len(results) == 1 assert results[0]["depth"] == 4096 assert results[0]["tps"] == 70.88 - def test_parse_bench_tps_prefill(self): prefill_row = "| model Q8_0 | 34.36 GiB | 34.66 B | MTL,BLAS | 6 | q8_0 | q8_0 | 1 | pp2048 | 2707.12 ± 9.17 |" results = thd.parse_bench_tps(prefill_row) assert results[0]["mode"] == "prefill" assert results[0]["depth"] == 2048 - def test_parse_ppl_final(self): ppl, stddev = thd.parse_ppl_final(MOCK_PPL_OUTPUT) assert ppl == 6.2109 assert stddev == 0.33250 - def test_parse_ppl_final_not_found(self): ppl, stddev = thd.parse_ppl_final("no ppl here") assert ppl == 0.0 assert stddev == 0.0 - def test_parse_env_string(self): result = thd._parse_env_string("TURBO_LAYER_ADAPTIVE=2 FOO=bar") assert result == {"TURBO_LAYER_ADAPTIVE": "2", "FOO": "bar"} - def test_parse_env_string_empty(self): result = thd._parse_env_string("") assert result == {} - def test_parse_bench_tps_combined(self): """Combined pp+tg row should parse as 'combined' mode.""" row = "| model Q8_0 | 34 GiB | 34 B | MTL | 6 | q8_0 | q8_0 | 1 | pp4096+tg128 | 500.00 ± 2.00 |" results = thd.parse_bench_tps(row) assert results[0]["mode"] == "combined" assert results[0]["depth"] == 4096 - def test_parse_bench_tps_no_stddev(self): """Row with only tps and no ± should still parse.""" row = "| model Q8_0 | 34 GiB | 34 B | MTL | 6 | q8_0 | q8_0 | 1 | tg128 | 85.00 |" results = thd.parse_bench_tps(row) assert results[0]["tps"] == 85.0 - def test_parse_bench_tps_short_line_skipped(self): """Lines with too few columns should be skipped.""" results = thd.parse_bench_tps("| too | few |") assert len(results) == 0 - def test_parse_bench_tps_no_test_col(self): """Lines without pp/tg pattern are skipped.""" results = thd.parse_bench_tps("| a | b | c | d | e | f | g | h | i | j |") assert len(results) == 0 - def test_safe_int_garbage(self): assert thd._safe_int("abc") == 0 assert thd._safe_int(" 42 ") == 42 assert thd._safe_int("-5 bytes") == -5 assert thd._safe_int("") == 0 - - # ============================================================ # 14. TestDetectStorageType # ============================================================ class TestDetectStorageType: """SSD/HDD detection for model files.""" - @patch("subprocess.run") def test_darwin_ssd_from_diskutil(self, mock_run): mock_run.return_value = _make_completed_process(stdout="SolidState") result = thd.detect_storage_type("/fake/model.gguf", "Darwin") assert result == "ssd" - @patch("subprocess.run") def test_darwin_apple_silicon_implies_ssd(self, mock_run): # diskutil doesn't say SolidState, but sysctl says Apple @@ -1517,11 +1293,9 @@ def side_effect(cmd, **kwargs): if "sysctl" in cmd: return _make_completed_process(stdout="Apple M5 Max") return _make_completed_process() - mock_run.side_effect = side_effect result = thd.detect_storage_type("/fake/model.gguf", "Darwin") assert result == "ssd" - @patch("subprocess.run") def test_darwin_unknown_cpu(self, mock_run): def side_effect(cmd, **kwargs): @@ -1530,11 +1304,9 @@ def side_effect(cmd, **kwargs): if "sysctl" in cmd: return _make_completed_process(stdout="Intel Core i9") return _make_completed_process() - mock_run.side_effect = side_effect result = thd.detect_storage_type("/fake/model.gguf", "Darwin") assert result == "unknown" - @patch("os.path.exists", return_value=False) @patch("os.path.realpath", return_value="/fake/model.gguf") @patch("subprocess.run") @@ -1545,7 +1317,6 @@ def test_linux_nvme_is_ssd(self, mock_run, mock_real, mock_exists): ) result = thd.detect_storage_type("/fake/model.gguf", "Linux") assert result == "ssd" - def test_linux_ssd_from_rotational(self): with patch("subprocess.run") as mock_run, \ patch("os.path.realpath", return_value="/fake/model.gguf"), \ @@ -1557,7 +1328,6 @@ def test_linux_ssd_from_rotational(self): ) result = thd.detect_storage_type("/fake/model.gguf", "Linux") assert result == "ssd" - def test_linux_hdd_from_rotational(self): with patch("subprocess.run") as mock_run, \ patch("os.path.realpath", return_value="/fake/model.gguf"), \ @@ -1569,30 +1339,24 @@ def test_linux_hdd_from_rotational(self): ) result = thd.detect_storage_type("/fake/model.gguf", "Linux") assert result == "hdd" - @patch("os.path.realpath", return_value="/fake/model.gguf") @patch("subprocess.run") def test_linux_df_failure_returns_unknown(self, mock_run, mock_real): mock_run.return_value = _make_completed_process(stdout="", rc=1) result = thd.detect_storage_type("/fake/model.gguf", "Linux") assert result == "unknown" - @patch("subprocess.run", side_effect=Exception("boom")) def test_exception_returns_unknown(self, mock_run): result = thd.detect_storage_type("/fake/model.gguf", "Darwin") assert result == "unknown" - def test_unsupported_platform_returns_unknown(self): result = thd.detect_storage_type("/fake/model.gguf", "Windows") assert result == "unknown" - - # ============================================================ # 15. TestFindModel # ============================================================ class TestFindModel: """Auto-detection of .gguf model files.""" - def test_finds_model_in_models_dir(self, tmp_path): models_dir = tmp_path / "models" models_dir.mkdir() @@ -1601,7 +1365,6 @@ def test_finds_model_in_models_dir(self, tmp_path): result = thd._find_model(str(tmp_path)) assert result is not None assert result.endswith(".gguf") - def test_finds_model_in_parent_models_dir(self, tmp_path): parent = tmp_path / "parent" child = parent / "child" @@ -1611,13 +1374,11 @@ def test_finds_model_in_parent_models_dir(self, tmp_path): (models / "found.gguf").write_text("fake") result = thd._find_model(str(child)) assert result is not None - @patch("pathlib.Path.home") def test_returns_none_when_no_model(self, mock_home, tmp_path): mock_home.return_value = tmp_path / "nonexistent_home" result = thd._find_model(str(tmp_path)) assert result is None - @patch("pathlib.Path.home") def test_checks_home_local_llms(self, mock_home, tmp_path): mock_home.return_value = tmp_path @@ -1630,128 +1391,104 @@ def test_checks_home_local_llms(self, mock_home, tmp_path): result = thd._find_model(str(empty)) assert result is not None assert "home_model.gguf" in result - - # ============================================================ # 16. TestDiagLogExtended # ============================================================ class TestDiagLogExtended: """Extended DiagLog method coverage.""" - def test_verbose_enabled(self, tmp_path): log = thd.DiagLog(str(tmp_path / "test.txt"), verbose=True) log.verbose("verbose message") log.close() content = (tmp_path / "test.txt").read_text() assert "[VERBOSE]" in content - def test_verbose_disabled(self, tmp_path): log = thd.DiagLog(str(tmp_path / "test.txt"), verbose=False) log.verbose("should not appear") log.close() content = (tmp_path / "test.txt").read_text() assert "[VERBOSE]" not in content - def test_warning_format(self, tmp_path): log = _make_log(tmp_path) log.warning("test warning") log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING] test warning" in content - def test_anomaly_format(self, tmp_path): log = _make_log(tmp_path) log.anomaly("bad thing happened") log.close() content = (tmp_path / "test.txt").read_text() assert "[ANOMALY] bad thing happened" in content - def test_notable_format(self, tmp_path): log = _make_log(tmp_path) log.notable("interesting finding") log.close() content = (tmp_path / "test.txt").read_text() assert "[NOTABLE] interesting finding" in content - def test_investigate_format(self, tmp_path): log = _make_log(tmp_path) log.investigate("needs investigation") log.close() content = (tmp_path / "test.txt").read_text() assert "[INVESTIGATE] needs investigation" in content - def test_path_property(self, tmp_path): log = _make_log(tmp_path) assert log.path == str(tmp_path / "test.txt") log.close() - - # ============================================================ # 17. TestBackgroundMonitorExtended # ============================================================ class TestBackgroundMonitorExtended: """Extended monitor coverage — static methods + _poll internals.""" - @patch("subprocess.check_output", return_value="total = 2048.00M used = 150.50M free = 1897.50M") def test_macos_swap_mb(self, mock_sub): result = thd.BackgroundMonitor._macos_swap_mb() assert result == "150.50" - @patch("subprocess.check_output", side_effect=Exception("boom")) def test_macos_swap_mb_exception(self, mock_sub): result = thd.BackgroundMonitor._macos_swap_mb() assert result == "0" - @patch("subprocess.check_output", side_effect=Exception("boom")) def test_macos_mem_pressure_exception(self, mock_sub): result = thd.BackgroundMonitor._macos_mem_pressure() assert result == "0" - @patch("subprocess.check_output", return_value="Mach Virtual Memory Statistics:\nPages active: 0.\nPages wired down: 0.\nPages free: 0.") def test_macos_mem_pressure_zero_total(self, mock_sub): result = thd.BackgroundMonitor._macos_mem_pressure() assert result == "0" - @patch("subprocess.check_output", side_effect=Exception("boom")) def test_macos_cpu_speed_limit_exception(self, mock_sub): result = thd.BackgroundMonitor._macos_cpu_speed_limit() assert result == "100" - @patch("subprocess.check_output", return_value="no CPU_Speed_Limit here") def test_macos_cpu_speed_limit_no_match(self, mock_sub): result = thd.BackgroundMonitor._macos_cpu_speed_limit() assert result == "100" - @patch("subprocess.check_output", return_value=MOCK_FREE_OUTPUT) def test_linux_swap_mb(self, mock_sub): result = thd.BackgroundMonitor._linux_swap_mb() assert result == "384000" - @patch("subprocess.check_output", side_effect=Exception("boom")) def test_linux_swap_mb_exception(self, mock_sub): result = thd.BackgroundMonitor._linux_swap_mb() assert result == "0" - @patch("subprocess.check_output", side_effect=Exception("boom")) def test_linux_mem_pct_exception(self, mock_sub): result = thd.BackgroundMonitor._linux_mem_pct() assert result == "0" - @patch("subprocess.check_output", return_value="no Mem: line here") def test_linux_mem_pct_no_mem_line(self, mock_sub): result = thd.BackgroundMonitor._linux_mem_pct() assert result == "0" - @patch("subprocess.check_output", return_value="no Swap: line here") def test_linux_swap_mb_no_swap_line(self, mock_sub): result = thd.BackgroundMonitor._linux_swap_mb() assert result == "0" - @patch("subprocess.check_output", return_value="42\n") def test_nvidia_query_success(self, mock_sub): result = thd.BackgroundMonitor._nvidia_query("temperature.gpu") assert result == "42" - @patch("platform.system", return_value="Darwin") @patch("os.getloadavg", return_value=(2.5, 1.0, 0.5)) @patch.object(thd.BackgroundMonitor, "_macos_mem_pressure", return_value="55") @@ -1765,7 +1502,6 @@ def test_poll_darwin(self, mock_cpu, mock_swap, mock_mem, mock_load, mock_plat, assert sample["mem_pressure_pct"] == "55" assert sample["swap_used_mb"] == "200" assert sample["gpu_mem_used_mb"] == "unified" - @patch("platform.system", return_value="Linux") @patch("os.getloadavg", return_value=(1.5, 1.0, 0.5)) @patch.object(thd.BackgroundMonitor, "_linux_mem_pct", return_value="30") @@ -1777,7 +1513,6 @@ def test_poll_linux(self, mock_nvidia, mock_swap, mock_mem, mock_load, mock_plat sample = mon._poll() assert sample["load_1m"] == "1.5" assert sample["mem_pressure_pct"] == "30" - @patch("platform.system", return_value="Darwin") @patch("os.getloadavg", side_effect=OSError("no loadavg")) @patch.object(thd.BackgroundMonitor, "_macos_mem_pressure", return_value="0") @@ -1788,7 +1523,6 @@ def test_poll_loadavg_failure(self, mock_cpu, mock_swap, mock_mem, mock_load, mo mon = thd.BackgroundMonitor(csv_path) sample = mon._poll() assert sample["load_1m"] == "N/A" - @patch.object(thd.BackgroundMonitor, "_poll", side_effect=Exception("poll crashed")) def test_run_error_path(self, mock_poll, tmp_path): csv_path = str(tmp_path / "mon.csv") @@ -1805,25 +1539,20 @@ def test_run_error_path(self, mock_poll, tmp_path): assert any("error" in s for s in mon.samples) finally: thd.MONITOR_POLL_INTERVAL = orig - - # ============================================================ # 18. TestLiveDisplayExtended # ============================================================ class TestLiveDisplayExtended: """Extended LiveDisplay coverage — rich paths and show_stress_summary.""" - def test_start_stop_no_rich(self): display = thd.LiveDisplay(use_rich=False) display.start() # should be no-op display.stop() # should be no-op - def test_show_section_summary_no_ratios(self, capsys): display = thd.LiveDisplay(use_rich=False) display.show_section_summary("Test") captured = capsys.readouterr() assert captured.out == "" # no ratios, no output - def test_show_section_summary_with_rich_returns_early(self): if not thd.HAS_RICH: pytest.skip("rich not installed") @@ -1831,13 +1560,11 @@ def test_show_section_summary_with_rich_returns_early(self): display._ratios = {4096: 0.90} # Should return without printing (rich Live handles it) display.show_section_summary("Test") - def test_show_stress_summary_empty(self, capsys): display = thd.LiveDisplay(use_rich=False) display.show_stress_summary("Test", {}) captured = capsys.readouterr() assert captured.out == "" # empty ratios, no output - def test_show_stress_summary_ascii(self, capsys): display = thd.LiveDisplay(use_rich=False) ratios = {4096: 0.85, 8192: 0.72} @@ -1845,7 +1572,6 @@ def test_show_stress_summary_ascii(self, capsys): captured = capsys.readouterr() assert "0.85" in captured.out assert "0.72" in captured.out - def test_show_stress_summary_rich(self): if not thd.HAS_RICH: pytest.skip("rich not installed") @@ -1853,7 +1579,6 @@ def test_show_stress_summary_rich(self): ratios = {4096: 0.95, 8192: 0.75, 16384: 0.40} # Should not raise display.show_stress_summary("Test", ratios) - def test_build_rich_table_all_ratio_bands(self): if not thd.HAS_RICH: pytest.skip("rich not installed") @@ -1866,7 +1591,6 @@ def test_build_rich_table_all_ratio_bands(self): display.update_decode("turbo3", 8192, 40.0) # < 0.7 (red) table = display._build_rich_table() assert table is not None - def test_build_rich_table_missing_counterpart(self): """q8_0 result without turbo3 should show '...' for missing columns.""" if not thd.HAS_RICH: @@ -1875,18 +1599,15 @@ def test_build_rich_table_missing_counterpart(self): display.update_decode("q8_0", 4096, 80.0) table = display._build_rich_table() assert table is not None - def test_recompute_ratios_q8_zero(self): display = thd.LiveDisplay(use_rich=False) display._decode_results["q8_0"] = {4096: 0.0} display._decode_results["turbo3"] = {4096: 50.0} display._recompute_ratios() assert 4096 not in display._ratios # q8=0, can't divide - def test_refresh_no_rich(self): display = thd.LiveDisplay(use_rich=False) display._refresh() # no-op, shouldn't crash - def test_refresh_with_rich_live(self): if not thd.HAS_RICH: pytest.skip("rich not installed") @@ -1895,7 +1616,6 @@ def test_refresh_with_rich_live(self): display.update_decode("q8_0", 0, 85.0) display.update_decode("turbo3", 0, 77.0) display.stop() - def test_depth_label_short_for_zero(self, capsys): """Depth 0 should show 'short' in ASCII summary.""" display = thd.LiveDisplay(use_rich=False) @@ -1904,58 +1624,48 @@ def test_depth_label_short_for_zero(self, capsys): display.show_section_summary("Decode") captured = capsys.readouterr() assert "short" in captured.out - - # ============================================================ # 19. TestAnomalyDetectorExtended # ============================================================ class TestAnomalyDetectorExtended: """Additional anomaly detector branches.""" - def _make_detector(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") mon = thd.BackgroundMonitor(csv_path) return thd.AnomalyDetector(log, mon), log, mon - def test_check_q8_baseline_low(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_q8_baseline(3.0, "apple_silicon") log.close() assert len(det.anomalies) == 1 assert "under load" in det.anomalies[0].lower() or "q8_0" in det.anomalies[0].lower() - def test_check_q8_baseline_ok(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_q8_baseline(50.0, "apple_silicon") log.close() assert len(det.anomalies) == 0 - def test_check_q8_baseline_zero(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_q8_baseline(0.0, "apple_silicon") log.close() assert len(det.anomalies) == 0 - def test_check_thermal_no_samples(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_thermal() # no samples log.close() assert len(det.anomalies) == 0 - def test_check_thermal_invalid_value(self, tmp_path): det, log, mon = self._make_detector(tmp_path) mon._samples = [{"cpu_speed_limit": "N/A"}] det.check_thermal() # ValueError on int("N/A") log.close() assert len(det.anomalies) == 0 - def test_check_swap_growth_no_samples(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.check_swap_growth() # no samples log.close() assert len(det.anomalies) == 0 - def test_check_swap_growth_invalid_value(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.set_initial_swap(100) @@ -1963,14 +1673,12 @@ def test_check_swap_growth_invalid_value(self, tmp_path): det.check_swap_growth() # ValueError log.close() assert len(det.anomalies) == 0 - def test_check_ppl_no_baseline(self, tmp_path): det, log, mon = self._make_detector(tmp_path) # q8_ppl is 0 — should return early det.check_ppl("turbo3", 6.5) log.close() assert len(det.anomalies) == 0 - def test_check_ppl_expected_range(self, tmp_path): det, log, mon = self._make_detector(tmp_path) det.set_q8_ppl(6.0) @@ -1978,7 +1686,6 @@ def test_check_ppl_expected_range(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[INFO]" in content - def test_check_prefill_ratio_normal(self, tmp_path): """Normal prefill ratio (0.90-1.02) should produce no flags.""" det, log, mon = self._make_detector(tmp_path) @@ -1986,14 +1693,11 @@ def test_check_prefill_ratio_normal(self, tmp_path): log.close() assert len(det.investigations) == 0 assert len(det.notables) == 0 - - # ============================================================ # 20. TestCaptureLoad # ============================================================ class TestCaptureLoad: """capture_load and platform-specific helpers.""" - @patch("turbo_hardware_diag.detect_platform", return_value="Darwin") @patch("turbo_hardware_diag._capture_load_macos") @patch("turbo_hardware_diag._run_cmd", return_value="{ 1.5 2.0 1.8 }") @@ -2004,7 +1708,6 @@ def test_capture_load_darwin(self, mock_exists, mock_cmd, mock_mac, mock_plat, t log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_SNAPSHOT] label=test_label" in content - @patch("turbo_hardware_diag.detect_platform", return_value="Linux") @patch("turbo_hardware_diag._capture_load_linux") @patch("turbo_hardware_diag._run_cmd", return_value="50\n 50 processes") @@ -2017,7 +1720,6 @@ def test_capture_load_linux(self, mock_cmd, mock_linux, mock_plat, tmp_path): with open(str(tmp_path / "test.txt")) as f: content = f.read() assert "[LOAD_SNAPSHOT]" in content - @patch("turbo_hardware_diag._run_cmd") def test_capture_load_macos_full(self, mock_cmd, tmp_path): def cmd_side(cmd, **kwargs): @@ -2033,14 +1735,12 @@ def cmd_side(cmd, **kwargs): if cmd[0] == "ioreg": return "PerformanceState = 3" return "" - mock_cmd.side_effect = cmd_side log = _make_log(tmp_path) thd._capture_load_macos(log) log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_SNAPSHOT]" in content - @patch("turbo_hardware_diag._run_cmd") def test_capture_load_macos_no_gpu_ioreg(self, mock_cmd, tmp_path): def cmd_side(cmd, **kwargs): @@ -2056,14 +1756,12 @@ def cmd_side(cmd, **kwargs): if cmd[0] == "ioreg": return "no performance state here" return "" - mock_cmd.side_effect = cmd_side log = _make_log(tmp_path) thd._capture_load_macos(log) log.close() content = (tmp_path / "test.txt").read_text() assert "no GPU metrics" in content - @patch("turbo_hardware_diag._run_cmd", return_value="") @patch("shutil.which", return_value=None) def test_capture_load_linux_full(self, mock_which, mock_cmd, tmp_path): @@ -2074,7 +1772,6 @@ def test_capture_load_linux_full(self, mock_which, mock_cmd, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_SNAPSHOT]" in content - @patch("turbo_hardware_diag._run_cmd") @patch("shutil.which", return_value="/usr/bin/nvidia-smi") def test_capture_load_linux_with_nvidia(self, mock_which, mock_cmd, tmp_path): @@ -2096,14 +1793,11 @@ def read_side(*a, **kw): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_SNAPSHOT]" in content - - # ============================================================ # 21. TestSectionsExtended # ============================================================ class TestSectionsExtended: """Extended section coverage for uncovered branches.""" - def test_section_3_model_info_exception(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run", side_effect=Exception("cli failed")), \ @@ -2115,7 +1809,6 @@ def test_section_3_model_info_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content or "filesize_bytes=0" in content - def test_section_3_model_info_ssd_detection(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -2129,7 +1822,6 @@ def test_section_3_model_info_ssd_detection(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "[STORAGE] type=hdd" in content assert "HDD" in content or "spinning disk" in content - def test_section_4_gpu_linux(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -2141,7 +1833,6 @@ def test_section_4_gpu_linux(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "CUDA" in content - def test_section_4_gpu_linux_no_nvidia(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -2152,7 +1843,6 @@ def test_section_4_gpu_linux_no_nvidia(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "nvidia-smi not found" in content - def test_section_4_darwin_no_tensor(self, tmp_path): log = _make_log(tmp_path) gpu_output = "build: 123\nhas tensor = false" @@ -2164,7 +1854,6 @@ def test_section_4_darwin_no_tensor(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "Tensor API disabled" in content - def test_section_4_exception(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run", side_effect=Exception("gpu init failed")), \ @@ -2174,7 +1863,6 @@ def test_section_4_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_section_5_turbo3_failure(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -2184,7 +1872,6 @@ def test_section_5_turbo3_failure(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "FAILED" in content - def test_section_5_metal_lib_exception(self, tmp_path): log = _make_log(tmp_path) call_count = {"n": 0} @@ -2193,14 +1880,12 @@ def side_effect(*args, **kwargs): if call_count["n"] == 1: return _make_completed_process(stdout="turbo3 OK", rc=0) raise Exception("metal lib failed") - with patch("subprocess.run", side_effect=side_effect), \ patch("turbo_hardware_diag._run_cmd", return_value="abc1234 commit"): thd.section_5_build_validation(log, "/fake/bench", "/fake/cli", "/fake/model.gguf", "/fake/llama") log.close() content = (tmp_path / "test.txt").read_text() assert "Metal library validation failed" in content - def test_section_5_no_git_repo(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run, \ @@ -2210,7 +1895,6 @@ def test_section_5_no_git_repo(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "not a git repo" in content - def test_section_9_combined(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.run_bench", return_value=("", 1.0)): @@ -2218,13 +1902,11 @@ def test_section_9_combined(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "COMBINED" in content - def test_section_10_with_wiki(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") mon = thd.BackgroundMonitor(csv_path) anomaly = thd.AnomalyDetector(log, mon) - with patch("os.path.isfile", return_value=True), \ patch("turbo_hardware_diag.run_perpl", return_value=(MOCK_PPL_OUTPUT, 60.0)), \ patch("turbo_hardware_diag.parse_ppl_final", return_value=(6.2109, 0.33)): @@ -2235,7 +1917,6 @@ def test_section_10_with_wiki(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "PERPLEXITY" in content - def test_section_11_memory(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run") as mock_run: @@ -2246,7 +1927,6 @@ def test_section_11_memory(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "MEMORY BREAKDOWN" in content - def test_section_11_memory_exception(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.run", side_effect=Exception("cli crashed")): @@ -2254,7 +1934,6 @@ def test_section_11_memory_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_section_12_post_load_linux(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.capture_load"), \ @@ -2266,7 +1945,6 @@ def test_section_12_post_load_linux(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "WARNING" in content # >90°C - def test_section_12_post_load_darwin_throttled(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.capture_load"), \ @@ -2277,7 +1955,6 @@ def test_section_12_post_load_darwin_throttled(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "85" in content assert "WARNING" in content - def test_section_13_with_all_flags(self, tmp_path): log = _make_log(tmp_path) csv_path = str(tmp_path / "mon.csv") @@ -2293,7 +1970,6 @@ def test_section_13_with_all_flags(self, tmp_path): assert "NOTABLE FINDINGS: 1" in content assert "INVESTIGATE FURTHER: 1" in content assert "ANOMALIES DETECTED: 1" in content - def test_section_2_linux_gpu_procs(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.capture_load"), \ @@ -2310,7 +1986,6 @@ def which_side(cmd): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_GPU_PROC]" in content - def test_section_2_gpu_procs_darwin_found(self, tmp_path): log = _make_log(tmp_path) with patch("turbo_hardware_diag.capture_load"), \ @@ -2321,14 +1996,11 @@ def test_section_2_gpu_procs_darwin_found(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_GPU_PROC]" in content - - # ============================================================ # 22. TestBuildJsonProfileExtended # ============================================================ class TestBuildJsonProfileExtended: """Additional build_json_profile branches.""" - def test_model_size_exception(self): with patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ patch("platform.release", return_value="25.3.0"), \ @@ -2336,20 +2008,16 @@ def test_model_size_exception(self): patch("os.path.getsize", side_effect=OSError("no file")): profile = thd.build_json_profile({}, "/fake/model.gguf", "", "20260326") assert profile["model_size_bytes"] == 0 - - # ============================================================ # 23. TestPackageResultsExtended # ============================================================ class TestPackageResultsExtended: """Extended packaging coverage.""" - def test_json_write_failure(self, tmp_path): log = thd.DiagLog(str(tmp_path / "test.txt")) log.write("test") mon = MagicMock() mon.csv_path = str(tmp_path / "nonexistent.csv") - # Make JSON write fail with patch("builtins.open", side_effect=[ mock.mock_open(read_data="")(), # zip creation @@ -2357,27 +2025,22 @@ def test_json_write_failure(self, tmp_path): # Actually, just pass a bad profile that's not JSON-serializable # Simpler: use a read-only dir pass - # Test with a non-serializable profile — the code wraps in try/except bad_profile = {"key": object()} zip_path = thd.package_results(log, mon, bad_profile, "20260326", str(tmp_path)) log.close() # Should still create zip even if JSON write failed assert os.path.isfile(zip_path) - def test_zip_creation_failure(self, tmp_path): log = thd.DiagLog(str(tmp_path / "test.txt")) log.write("test") mon = MagicMock() mon.csv_path = str(tmp_path / "nonexistent.csv") - # Use a path where zip can't be created zip_path = thd.package_results(log, mon, {"v": 5}, "20260326", str(tmp_path)) log.close() # This should succeed normally — test the happy path at least assert "turbo-diag-20260326.zip" in zip_path - - # ============================================================ # 24. TestMainFunction # ============================================================ @@ -2390,22 +2053,16 @@ def _setup_llama_dir(tmp_path): (build_bin / "llama-perplexity").write_text("fake") (build_bin / "llama-cli").write_text("fake") return llama_dir - - _MOCK_POLL_SAMPLE = { "timestamp": "T", "load_1m": "1", "mem_pressure_pct": "50", "swap_used_mb": "0", "gpu_temp_c": "N/A", "cpu_speed_limit": "100", "gpu_mem_used_mb": "N/A", "gpu_util_pct": "N/A", } - - def _run_main_with_patches(argv, extra_patches=None): """Run main() with all section functions mocked. Returns exit code. - extra_patches: list of (target, kwargs) to override defaults. """ from contextlib import ExitStack - section_patches = { "turbo_hardware_diag.section_1_hardware_inventory": {"return_value": {}}, "turbo_hardware_diag.section_2_system_load_pre": {}, @@ -2423,10 +2080,8 @@ def _run_main_with_patches(argv, extra_patches=None): "turbo_hardware_diag.package_results": {"return_value": "out.zip"}, "os.path.getsize": {"return_value": 1024}, } - if extra_patches: section_patches.update(extra_patches) - with ExitStack() as stack: stack.enter_context(patch("sys.argv", argv)) for target, kwargs in section_patches.items(): @@ -2443,17 +2098,13 @@ def _run_main_with_patches(argv, extra_patches=None): stack.enter_context(patch.object(thd.LiveDisplay, "start")) stack.enter_context(patch.object(thd.LiveDisplay, "stop")) return thd.main() - - class TestMainFunction: """main() integration path with everything mocked.""" - def test_main_model_not_found(self, tmp_path): with patch("sys.argv", ["prog", str(tmp_path), "/nonexistent/model.gguf"]): with patch("builtins.print"): rc = thd.main() assert rc == 1 - def test_main_tools_not_found(self, tmp_path): model = tmp_path / "model.gguf" model.write_text("fake") @@ -2463,7 +2114,6 @@ def test_main_tools_not_found(self, tmp_path): assert rc == 1 printed = " ".join(str(c) for c in mock_print.call_args_list) assert "not found" in printed - def test_main_full_run(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) model = tmp_path / "model.gguf" @@ -2472,7 +2122,6 @@ def test_main_full_run(self, tmp_path): "--skip-ppl", "--skip-stress", "-o", str(tmp_path / "output")] rc = _run_main_with_patches(argv) assert rc == 0 - def test_main_exception_in_sections(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) model = tmp_path / "model.gguf" @@ -2485,7 +2134,6 @@ def test_main_exception_in_sections(self, tmp_path): }, }) assert rc == 1 # Non-zero on unhandled exception - def test_main_auto_find_model(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) models_dir = llama_dir / "models" @@ -2495,7 +2143,6 @@ def test_main_auto_find_model(self, tmp_path): "--skip-ppl", "--skip-stress", "-o", str(tmp_path / "output")] rc = _run_main_with_patches(argv) assert rc == 0 - def test_main_model_size_large(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) model = tmp_path / "model.gguf" @@ -2506,7 +2153,6 @@ def test_main_model_size_large(self, tmp_path): "os.path.getsize": {"return_value": 36 * 1024**3}, }) assert rc == 0 - def test_main_model_size_mb(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) model = tmp_path / "model.gguf" @@ -2517,7 +2163,6 @@ def test_main_model_size_mb(self, tmp_path): "os.path.getsize": {"return_value": 500 * 1024**2}, }) assert rc == 0 - def test_main_model_size_small(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) model = tmp_path / "model.gguf" @@ -2528,7 +2173,6 @@ def test_main_model_size_small(self, tmp_path): "os.path.getsize": {"return_value": 500}, }) assert rc == 0 - def test_main_model_size_exception(self, tmp_path): llama_dir = _setup_llama_dir(tmp_path) model = tmp_path / "model.gguf" @@ -2539,38 +2183,30 @@ def test_main_model_size_exception(self, tmp_path): "os.path.getsize": {"side_effect": OSError("no file")}, }) assert rc == 0 - - # ============================================================ # 25. TestRunCmd # ============================================================ class TestRunCmd: """_run_cmd utility.""" - @patch("subprocess.run", side_effect=Exception("boom")) def test_run_cmd_exception_returns_empty(self, mock_run): result = thd._run_cmd(["fake"]) assert result == "" - @patch("subprocess.run") def test_run_cmd_success(self, mock_run): mock_run.return_value = _make_completed_process(stdout=" output ") result = thd._run_cmd(["fake"]) assert result == "output" - @patch("subprocess.run") def test_run_cmd_shell(self, mock_run): mock_run.return_value = _make_completed_process(stdout="shell out") result = thd._run_cmd("echo test", shell=True) assert result == "shell out" - - # ============================================================ # 26. TestSubprocessRunner # ============================================================ class TestSubprocessRunner: """_run_subprocess edge cases.""" - def test_subprocess_with_env_extra(self, tmp_path): log = _make_log(tmp_path) with patch("subprocess.Popen") as mock_popen: @@ -2579,14 +2215,12 @@ def test_subprocess_with_env_extra(self, tmp_path): mock_proc.wait.return_value = None mock_proc.returncode = 0 mock_popen.return_value = mock_proc - output, rc = thd._run_subprocess( ["cmd"], log, env_extra={"FOO": "bar"}, timeout=10 ) log.close() assert "line1" in output assert rc == 0 - def test_run_bench_rc_minus1_no_failed(self, tmp_path): """rc=-1 means timeout already handled, don't print FAILED again.""" log = _make_log(tmp_path) @@ -2595,7 +2229,6 @@ def test_run_bench_rc_minus1_no_failed(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "FAILED" not in content - def test_run_perpl_failure(self, tmp_path): """run_perpl with non-zero rc should print FAILED.""" log = _make_log(tmp_path) @@ -2604,14 +2237,11 @@ def test_run_perpl_failure(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "FAILED" in content - - # ============================================================ # 27. TestLinuxHardwareDetection # ============================================================ class TestLinuxHardwareDetection: """Linux-specific hardware detection branches.""" - @patch("turbo_hardware_diag._run_cmd", return_value="") @patch("turbo_hardware_diag.detect_platform", return_value="Linux") @patch("platform.release", return_value="6.1.0") @@ -2619,13 +2249,11 @@ class TestLinuxHardwareDetection: @patch("shutil.which", return_value=None) def test_linux_cache_hierarchy(self, mock_which, mock_mach, mock_rel, mock_plat, mock_cmd, tmp_path): log = _make_log(tmp_path) - with patch("pathlib.Path.read_text") as mock_read, \ patch("pathlib.Path.exists") as mock_exists, \ patch("pathlib.Path.is_dir") as mock_isdir, \ patch("pathlib.Path.glob", return_value=[]), \ patch("pathlib.Path.iterdir", return_value=[]): - call_n = {"n": 0} def read_side(*a, **kw): call_n["n"] += 1 @@ -2636,14 +2264,11 @@ def read_side(*a, **kw): elif call_n["n"] <= 5: return "32K" # cache sizes return "" - mock_read.side_effect = read_side mock_exists.return_value = True mock_isdir.return_value = False - hw = thd.detect_hardware(log) log.close() - @patch("turbo_hardware_diag._run_cmd", return_value="") @patch("turbo_hardware_diag.detect_platform", return_value="Linux") @patch("platform.release", return_value="6.1.0") @@ -2652,11 +2277,9 @@ def read_side(*a, **kw): def test_linux_thermal_and_power(self, mock_which, mock_mach, mock_rel, mock_plat, mock_cmd, tmp_path): """Cover thermal zone and power supply reading on Linux.""" log = _make_log(tmp_path) - with patch("pathlib.Path.read_text") as mock_read, \ patch("pathlib.Path.exists", return_value=True), \ patch("pathlib.Path.is_dir") as mock_isdir: - call_n = {"n": 0} def read_side(*a, **kw): call_n["n"] += 1 @@ -2665,15 +2288,12 @@ def read_side(*a, **kw): elif call_n["n"] == 2: return MOCK_PROC_MEMINFO return "75000" # thermal temp - mock_read.side_effect = read_side mock_isdir.return_value = True - # Mock thermal zone glob and power supply iterdir thermal_temp = MagicMock() thermal_temp.parent.name = "thermal_zone0" thermal_temp.read_text.return_value = "75000" - ps_dir = MagicMock() ps_dir.name = "BAT0" type_f = MagicMock() @@ -2683,36 +2303,29 @@ def read_side(*a, **kw): status_f.exists.return_value = True status_f.read_text.return_value = "Charging" ps_dir.__truediv__ = lambda self, key: type_f if key == "type" else status_f - with patch("pathlib.Path.glob") as mock_glob, \ patch("pathlib.Path.iterdir", return_value=[ps_dir]): mock_glob.return_value = [thermal_temp] hw = thd.detect_hardware(log) log.close() - - # ============================================================ # 28. TestExceptPaths — cover remaining except blocks # ============================================================ class TestExceptPaths: """Cover try/except branches that return defaults on failure.""" - def test_detect_macos_hw_exceptions(self, tmp_path): """Cover all except branches in _detect_macos_hw.""" log = _make_log(tmp_path) hw = {} call_n = {"n": 0} - def failing_cmd(cmd, **kwargs): call_n["n"] += 1 raise Exception(f"fail {call_n['n']}") - with patch("turbo_hardware_diag._run_cmd", side_effect=failing_cmd): thd._detect_macos_hw(log, hw) log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_detect_linux_hw_all_exceptions(self, tmp_path): """Cover all except branches in _detect_linux_hw.""" log = _make_log(tmp_path) @@ -2727,7 +2340,6 @@ def test_detect_linux_hw_all_exceptions(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content assert hw["cpu_brand"] == "unknown" - def test_capture_load_exception_paths(self, tmp_path): """capture_load with all probes failing.""" log = _make_log(tmp_path) @@ -2739,21 +2351,18 @@ def test_capture_load_exception_paths(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content or "process_count=unknown" in content - def test_capture_load_macos_vm_stat_empty(self, tmp_path): """_capture_load_macos with empty vm_stat output.""" log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_cmd", return_value=""): thd._capture_load_macos(log) log.close() - def test_capture_load_macos_all_exceptions(self, tmp_path): """_capture_load_macos with all probes throwing exceptions.""" log = _make_log(tmp_path) with patch("turbo_hardware_diag._run_cmd", side_effect=Exception("boom")): thd._capture_load_macos(log) log.close() - def test_capture_load_linux_all_exceptions(self, tmp_path): log = _make_log(tmp_path) with patch("pathlib.Path.read_text", side_effect=Exception("boom")), \ @@ -2764,7 +2373,6 @@ def test_capture_load_linux_all_exceptions(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_subprocess_kill_failure(self, tmp_path): """Cover the proc.kill() exception path in _run_subprocess.""" log = _make_log(tmp_path) @@ -2777,7 +2385,6 @@ def test_subprocess_kill_failure(self, tmp_path): output, rc = thd._run_subprocess(["fake"], log, timeout=5) log.close() assert rc == -1 - def test_section_3_storage_detection_exception(self, tmp_path): """Cover the except in section_3 storage detection.""" log = _make_log(tmp_path) @@ -2791,7 +2398,6 @@ def test_section_3_storage_detection_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[STORAGE] type=unknown" in content - def test_section_5_turbo3_exception(self, tmp_path): """Cover section 5 turbo3 validation exception path.""" log = _make_log(tmp_path) @@ -2801,7 +2407,6 @@ def test_section_5_turbo3_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "turbo3 validation error" in content - def test_section_2_disk_io_with_iostat(self, tmp_path): """Cover section 2 disk I/O with iostat available.""" log = _make_log(tmp_path) @@ -2813,7 +2418,6 @@ def test_section_2_disk_io_with_iostat(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_IO]" in content - def test_section_2_ps_exception(self, tmp_path): """Cover section 2 exception path for ps command.""" log = _make_log(tmp_path) @@ -2823,7 +2427,6 @@ def cmd_side(cmd, **kwargs): if isinstance(cmd, list) and cmd[0] == "ps": raise Exception("ps failed") return "" - with patch("turbo_hardware_diag.capture_load"), \ patch("turbo_hardware_diag._run_cmd", side_effect=cmd_side), \ patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ @@ -2832,7 +2435,6 @@ def cmd_side(cmd, **kwargs): log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_main_initial_swap_exception(self, tmp_path): """Cover the initial_swap exception path in main().""" llama_dir = _setup_llama_dir(tmp_path) @@ -2848,7 +2450,6 @@ def test_main_initial_swap_exception(self, tmp_path): # we'd need _poll to raise. But that's covered by the # _run_main_with_patches approach already. assert rc == 0 - def test_main_no_skip_stress(self, tmp_path): """Cover the non-skipped stress test branch in main().""" llama_dir = _setup_llama_dir(tmp_path) @@ -2860,7 +2461,6 @@ def test_main_no_skip_stress(self, tmp_path): "turbo_hardware_diag.section_8_stress_test": {}, }) assert rc == 0 - def test_main_verbose(self, tmp_path): """Cover verbose flag in main().""" llama_dir = _setup_llama_dir(tmp_path) @@ -2871,7 +2471,6 @@ def test_main_verbose(self, tmp_path): "-o", str(tmp_path / "output")] rc = _run_main_with_patches(argv) assert rc == 0 - def test_section_5_metal_lib_found_lines(self, tmp_path): """Cover section 5 metal library lines found.""" log = _make_log(tmp_path) @@ -2889,14 +2488,12 @@ def run_side(*args, **kwargs): if call_n["n"] == 1: return _make_completed_process(stdout="turbo3 OK", rc=0) return _make_completed_process(stdout=metal_output, rc=0) - with patch("subprocess.run", side_effect=run_side), \ patch("turbo_hardware_diag._run_cmd", return_value="abc1234 commit"): thd.section_5_build_validation(log, "/fake/bench", "/fake/cli", "/fake/model.gguf", "/fake/llama") log.close() content = (tmp_path / "test.txt").read_text() assert "metal_library" in content or "embed" in content - def test_section_8_with_actual_results(self, tmp_path): """Cover section 8 stress test with actual bench results that parse.""" log = _make_log(tmp_path) @@ -2904,26 +2501,22 @@ def test_section_8_with_actual_results(self, tmp_path): mon = thd.BackgroundMonitor(csv_path) display = thd.LiveDisplay(use_rich=False) anomaly = thd.AnomalyDetector(log, mon) - bench_out = "| model Q8_0 | 34 GiB | 34 B | MTL | 6 | turbo3 | turbo3 | 1 | tg64 @ d2048 | 70.00 ± 1.00 |" with patch("turbo_hardware_diag.run_bench", return_value=(bench_out, 1.0)), \ patch("turbo_hardware_diag.capture_load"): thd.section_8_stress_test(log, "/fake/bench", "/fake/model.gguf", display, anomaly) log.close() - def test_package_results_zip_creation_error(self, tmp_path): """Cover zip creation exception in package_results.""" log = thd.DiagLog(str(tmp_path / "test.txt")) log.write("test") mon = MagicMock() mon.csv_path = str(tmp_path / "nonexistent.csv") - with patch("zipfile.ZipFile", side_effect=OSError("disk full")): zip_path = thd.package_results(log, mon, {"v": 5}, "20260326", str(tmp_path)) log.close() content = (tmp_path / "test.txt").read_text() assert "[WARNING]" in content - def test_package_results_cleanup_exception(self, tmp_path): """Cover cleanup exception in package_results.""" log = thd.DiagLog(str(tmp_path / "test.txt")) @@ -2933,12 +2526,10 @@ def test_package_results_cleanup_exception(self, tmp_path): f.write("header\n") mon = MagicMock() mon.csv_path = csv_path - with patch("os.remove", side_effect=OSError("permission denied")): zip_path = thd.package_results(log, mon, {"v": 5}, "20260326", str(tmp_path)) log.close() assert os.path.isfile(zip_path) - def test_section_12_linux_no_thermal(self, tmp_path): """Cover section 12 Linux path when thermal zone doesn't exist.""" log = _make_log(tmp_path) @@ -2947,7 +2538,6 @@ def test_section_12_linux_no_thermal(self, tmp_path): patch("pathlib.Path.exists", return_value=False): thd.section_12_post_load(log) log.close() - def test_section_12_darwin_no_throttle(self, tmp_path): """Cover section 12 Darwin path with 100% CPU speed.""" log = _make_log(tmp_path) @@ -2959,35 +2549,28 @@ def test_section_12_darwin_no_throttle(self, tmp_path): content = (tmp_path / "test.txt").read_text() assert "final_cpu_speed_limit=100" in content assert "WARNING" not in content.split("final_cpu_speed_limit")[1] - - # ============================================================ # Coverage gap closers — exception handlers and missed paths # ============================================================ class TestCoverageGapClosers: """Tests to close the remaining 3% coverage gap.""" - # --- Line 833-835: Apple Silicon detection failure --- def test_apple_silicon_detection_exception(self, tmp_path): """Cover except branch when Apple Silicon detection raises.""" log = _make_log(tmp_path) - # Use a dict subclass that raises on .get() to trigger the except class BadDict(dict): def get(self, key, default=None): if key == "cpu_brand": raise Exception("cpu_brand read failed") return super().get(key, default) - bad_hw = BadDict() - with patch("turbo_hardware_diag._run_cmd", return_value=""): thd._detect_macos_hw(log, bad_hw) log.close() content = (tmp_path / "test.txt").read_text() assert "Could not determine Apple Silicon status" in content assert bad_hw["apple_silicon"] is False - # --- Line 868-869: Linux kernel version failure --- def test_linux_kernel_version_exception(self, tmp_path): """Cover except when platform.release() raises in _detect_linux_hw.""" @@ -3003,36 +2586,30 @@ def test_linux_kernel_version_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not get Linux kernel version" in content - # --- Line 928-929: Linux GPU detection failure --- def test_linux_gpu_detection_exception(self, tmp_path): """Cover except when GPU detection raises in _detect_linux_hw.""" log = _make_log(tmp_path) hw = {} - call_count = {"n": 0} original_read_text = Path.read_text - def patched_read_text(self, *a, **kw): # Let /proc/cpuinfo work, but raise for GPU vendor files if "cpuinfo" in str(self): return "model name : AMD EPYC\ncore id : 0\nprocessor : 0\n" raise Exception("gpu read failed") - def patched_which(cmd): if cmd == "lspci": return "/usr/bin/lspci" if cmd == "nvidia-smi": return "/usr/bin/nvidia-smi" return None - def patched_run_cmd(cmd, **kwargs): if isinstance(cmd, list) and cmd[0] == "nvidia-smi": raise Exception("nvidia-smi failed for GPU detection") if isinstance(cmd, list) and cmd[0] == "lspci": raise Exception("lspci failed") return "" - with patch("pathlib.Path.read_text", patched_read_text), \ patch("pathlib.Path.exists", return_value=False), \ patch("pathlib.Path.is_dir", return_value=False), \ @@ -3043,21 +2620,17 @@ def patched_run_cmd(cmd, **kwargs): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not detect GPU" in content - # --- Line 941-942: Linux cache hierarchy failure --- def test_linux_cache_hierarchy_exception(self, tmp_path): """Cover except when cache hierarchy read raises in _detect_linux_hw.""" log = _make_log(tmp_path) hw = {} - exists_calls = {"n": 0} - def patched_exists(self): s = str(self) if "cache/index" in s: raise Exception("cache read failed") return False - with patch("pathlib.Path.read_text", return_value="model name : AMD EPYC\ncore id : 0\nprocessor : 0\n"), \ patch("pathlib.Path.exists", patched_exists), \ patch("pathlib.Path.is_dir", return_value=False), \ @@ -3068,18 +2641,15 @@ def patched_exists(self): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not get Linux cache hierarchy" in content - # --- Line 953-954: Linux thermal zone failure --- def test_linux_thermal_zone_exception(self, tmp_path): """Cover except when thermal zone read raises in _detect_linux_hw.""" log = _make_log(tmp_path) hw = {} - def patched_is_dir(self): if "thermal" in str(self): raise Exception("thermal read failed") return False - with patch("pathlib.Path.read_text", return_value="model name : AMD EPYC\ncore id : 0\nprocessor : 0\n"), \ patch("pathlib.Path.exists", return_value=False), \ patch("pathlib.Path.is_dir", patched_is_dir), \ @@ -3090,20 +2660,17 @@ def patched_is_dir(self): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not read thermal zones" in content - # --- Line 966-967: Linux power supply failure --- def test_linux_power_supply_exception(self, tmp_path): """Cover except when power supply read raises in _detect_linux_hw.""" log = _make_log(tmp_path) hw = {} - def patched_is_dir(self): if "power_supply" in str(self): raise Exception("power supply read failed") if "thermal" in str(self): return False return False - with patch("pathlib.Path.read_text", return_value="model name : AMD EPYC\ncore id : 0\nprocessor : 0\n"), \ patch("pathlib.Path.exists", return_value=False), \ patch("pathlib.Path.is_dir", patched_is_dir), \ @@ -3114,17 +2681,14 @@ def patched_is_dir(self): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not read power supply info" in content - # --- Line 1105-1106: nvidia-smi GPU utilization failure --- def test_capture_load_linux_nvidia_smi_exception(self, tmp_path): """Cover except when nvidia-smi GPU util raises in _capture_load_linux.""" log = _make_log(tmp_path) - def patched_run_cmd(cmd, **kwargs): if isinstance(cmd, list) and cmd[0] == "nvidia-smi": raise Exception("nvidia-smi failed") return "" - with patch("pathlib.Path.read_text", return_value="MemFree: 1000 kB\nMemAvailable: 2000 kB\nSwapTotal: 500 kB\nSwapFree: 500 kB\n"), \ patch("pathlib.Path.exists", return_value=False), \ patch("shutil.which", return_value="/usr/bin/nvidia-smi"), \ @@ -3133,24 +2697,20 @@ def patched_run_cmd(cmd, **kwargs): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not query nvidia-smi for GPU utilization" in content - # --- Line 1114-1115: Linux CPU temperature failure --- def test_capture_load_linux_cpu_temp_exception(self, tmp_path): """Cover except when CPU temp read raises in _capture_load_linux.""" log = _make_log(tmp_path) - def patched_exists(self): if "thermal_zone0" in str(self): return True return False - def patched_read_text(self, *a, **kw): if "thermal_zone0" in str(self): raise Exception("temp read failed") if "meminfo" in str(self): return "MemFree: 1000 kB\nMemAvailable: 2000 kB\nSwapTotal: 500 kB\nSwapFree: 500 kB\n" return "" - with patch("pathlib.Path.read_text", patched_read_text), \ patch("pathlib.Path.exists", patched_exists), \ patch("shutil.which", return_value=None), \ @@ -3159,7 +2719,6 @@ def patched_read_text(self, *a, **kw): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not read CPU temperature" in content - # --- Line 1260: continue for non-pipe lines in parse_bench_tps --- def test_parse_bench_tps_skips_non_pipe_lines(self): """Cover the continue branch for lines not starting with |.""" @@ -3167,7 +2726,6 @@ def test_parse_bench_tps_skips_non_pipe_lines(self): results = thd.parse_bench_tps(mixed) assert len(results) == 1 assert results[0]["tps"] == 85.83 - # --- Line 1366: [LOAD_TOP] writing --- def test_section_2_load_top_writing(self, tmp_path): """Cover the [LOAD_TOP] write path in section_2.""" @@ -3181,17 +2739,14 @@ def test_section_2_load_top_writing(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_TOP]" in content - # --- Line 1377-1378: iostat failure --- def test_section_2_iostat_exception(self, tmp_path): """Cover except when iostat raises in section_2.""" log = _make_log(tmp_path) - def patched_run_cmd(cmd, **kwargs): if isinstance(cmd, list) and cmd[0] == "iostat": raise Exception("iostat failed") return "" - with patch("turbo_hardware_diag.capture_load"), \ patch("turbo_hardware_diag._run_cmd", side_effect=patched_run_cmd), \ patch("turbo_hardware_diag.detect_platform", return_value="Darwin"), \ @@ -3200,23 +2755,19 @@ def patched_run_cmd(cmd, **kwargs): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not get disk I/O stats from iostat" in content - # --- Line 1404-1406: nvidia-smi process query failure + "none detected" --- def test_section_2_linux_nvidia_smi_procs_exception(self, tmp_path): """Cover except when nvidia-smi process query raises.""" log = _make_log(tmp_path) - call_n = {"n": 0} def patched_run_cmd(cmd, **kwargs): if isinstance(cmd, list) and "query-compute-apps" in str(cmd): raise Exception("nvidia-smi query failed") return "" - def patched_which(cmd): if cmd == "nvidia-smi": return "/usr/bin/nvidia-smi" return None - with patch("turbo_hardware_diag.capture_load"), \ patch("turbo_hardware_diag._run_cmd", side_effect=patched_run_cmd), \ patch("turbo_hardware_diag.detect_platform", return_value="Linux"), \ @@ -3225,22 +2776,18 @@ def patched_which(cmd): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not query nvidia-smi for GPU processes" in content - def test_section_2_linux_nvidia_smi_procs_none_detected(self, tmp_path): """Cover the 'none detected' path when nvidia-smi returns empty.""" log = _make_log(tmp_path) - call_n = {"n": 0} def patched_run_cmd(cmd, **kwargs): if isinstance(cmd, list) and "query-compute-apps" in str(cmd): return "" # empty = no GPU processes return "" - def patched_which(cmd): if cmd == "nvidia-smi": return "/usr/bin/nvidia-smi" return None - with patch("turbo_hardware_diag.capture_load"), \ patch("turbo_hardware_diag._run_cmd", side_effect=patched_run_cmd), \ patch("turbo_hardware_diag.detect_platform", return_value="Linux"), \ @@ -3249,7 +2796,6 @@ def patched_which(cmd): log.close() content = (tmp_path / "test.txt").read_text() assert "[LOAD_GPU_PROC] none detected" in content - # --- Line 1545-1546: CUDA nvidia-smi query failure --- def test_section_4_cuda_nvidia_smi_exception(self, tmp_path): """Cover except when nvidia-smi CUDA query raises in section_4.""" @@ -3265,17 +2811,14 @@ def test_section_4_cuda_nvidia_smi_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "[CUDA] nvidia-smi query failed" in content - # --- Line 1608-1609: git repo detection failure --- def test_section_5_git_repo_exception(self, tmp_path): """Cover except when git log raises in section_5.""" log = _make_log(tmp_path) - def patched_run_cmd(cmd, **kwargs): if isinstance(cmd, list) and cmd[0] == "git": raise Exception("git failed") return "" - with patch("subprocess.run") as mock_run, \ patch("turbo_hardware_diag._run_cmd", side_effect=patched_run_cmd): mock_run.return_value = _make_completed_process(stdout="turbo3 test passed") @@ -3283,7 +2826,6 @@ def patched_run_cmd(cmd, **kwargs): log.close() content = (tmp_path / "test.txt").read_text() assert "[BUILD] not a git repo (error:" in content - # --- Line 1910-1911: macOS post-benchmark thermal failure --- def test_section_12_darwin_thermal_exception(self, tmp_path): """Cover except when macOS thermal check raises in section_12.""" @@ -3295,17 +2837,14 @@ def test_section_12_darwin_thermal_exception(self, tmp_path): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not check post-benchmark thermal state on macOS" in content - # --- Line 1920-1921: Linux post-benchmark thermal failure --- def test_section_12_linux_thermal_exception(self, tmp_path): """Cover except when Linux thermal check raises in section_12.""" log = _make_log(tmp_path) - def patched_exists(self): if "thermal_zone0" in str(self): raise Exception("thermal read failed") return False - with patch("turbo_hardware_diag.capture_load"), \ patch("turbo_hardware_diag.detect_platform", return_value="Linux"), \ patch("pathlib.Path.exists", patched_exists): @@ -3313,7 +2852,6 @@ def patched_exists(self): log.close() content = (tmp_path / "test.txt").read_text() assert "Could not check post-benchmark thermal state on Linux" in content - # --- Line 2196-2197: Model file not found in main() --- def test_main_model_file_not_found(self, tmp_path): """Cover the model-not-found early return in main().""" @@ -3325,7 +2863,6 @@ def test_main_model_file_not_found(self, tmp_path): patch("builtins.print"): rc = thd.main() assert rc == 1 - # --- Line 2265-2267: Initial swap read failure --- def test_main_initial_swap_read_failure(self, tmp_path): """Cover the except when _poll raises for initial swap.""" @@ -3334,9 +2871,7 @@ def test_main_initial_swap_read_failure(self, tmp_path): model.write_text("fake") argv = ["prog", str(llama_dir), str(model), "--skip-ppl", "--skip-stress", "-o", str(tmp_path / "output")] - from contextlib import ExitStack - section_patches = { "turbo_hardware_diag.section_1_hardware_inventory": {"return_value": {}}, "turbo_hardware_diag.section_2_system_load_pre": {}, @@ -3354,7 +2889,6 @@ def test_main_initial_swap_read_failure(self, tmp_path): "turbo_hardware_diag.package_results": {"return_value": "out.zip"}, "os.path.getsize": {"return_value": 1024}, } - with ExitStack() as stack: stack.enter_context(patch("sys.argv", argv)) for target, kwargs in section_patches.items(): @@ -3374,7 +2908,6 @@ def test_main_initial_swap_read_failure(self, tmp_path): stack.enter_context(patch.object(thd.LiveDisplay, "stop")) rc = thd.main() assert rc == 0 - # --- Line 2337-2338: Monitor summary write failure --- def test_main_monitor_summary_write_failure(self, tmp_path): """Cover except when writing monitor summary raises in main().""" @@ -3383,9 +2916,7 @@ def test_main_monitor_summary_write_failure(self, tmp_path): model.write_text("fake") argv = ["prog", str(llama_dir), str(model), "--skip-ppl", "--skip-stress", "-o", str(tmp_path / "output")] - from contextlib import ExitStack - section_patches = { "turbo_hardware_diag.section_1_hardware_inventory": {"return_value": {}}, "turbo_hardware_diag.section_2_system_load_pre": {}, @@ -3403,7 +2934,6 @@ def test_main_monitor_summary_write_failure(self, tmp_path): "turbo_hardware_diag.package_results": {"return_value": "out.zip"}, "os.path.getsize": {"return_value": 1024}, } - with ExitStack() as stack: stack.enter_context(patch("sys.argv", argv)) for target, kwargs in section_patches.items():