a13labs.infra/ansible/roles/auto_suspend/files/test_auto_suspend.py

#!/usr/bin/env python3
"""Unit tests for auto_suspend.py.

Subprocess calls are intercepted; test data comes from real command
outputs captured on gpu-01.lab.alexpires.me and stored under ``fixtures/``.
"""

import json
import os
import signal
import subprocess
import tempfile
import unittest
from pathlib import Path
from unittest.mock import patch, MagicMock

FIXTURES = Path(__file__).parent / "fixtures"


def _load_fixture(name):
    """Load a fixture file and return its content as a string."""
    return FIXTURES.joinpath(name).read_text()


def _make_mock_result(stdout="", returncode=0, stderr=""):
    return MagicMock(
        returncode=returncode,
        stdout=stdout,
        stderr=stderr,
    )


def _make_mock_timeout():
    return subprocess.TimeoutExpired(cmd=["test"], timeout=5)


# ---------------------------------------------------------------------------
# Fixtures (real samples from gpu-01)
# ---------------------------------------------------------------------------

MPSTAT_IDLE = _load_fixture("mpstat--1--1.idle.txt")
NVIDIA_IDLE = _load_fixture("nvidia-smi--query-gpu.utilization.gpu.csv.noheader.idle.txt")
NVIDIA_BUSY_4GAPS = """  0
  95
  88
  72
"""
NVIDIA_BUSY_SINGLE = """  95
"""
NVIDIA_NONE = ""
AMD_SMII_JSON = _load_fixture("amd-smi-monitor--json.txt")
AMD_SMII_GFX_0 = '[{"gpu":0,"gfx":{"value":0,"unit":"%"}}]'
AMD_SMII_GFX_80 = '[{"gpu":0,"gfx":{"value":80,"unit":"%"}}]'
WHO_WITH_USERS = _load_fixture("who.with-users.txt")
WHO_EMPTY = ""

MPSTAT_HIGH = """Linux 6.17.11-200.fc42.x86_64 (gpu-01) 	06/08/26 	_x86_64_	(12 CPU)

21:34:58     CPU    %usr   %nice    %sys %iowait    %irq   %soft  %steal  %guest  %gnice   %idle
21:34:59     all   85.30    0.00    8.20    0.50    0.00    0.10    0.00    0.00    0.00    5.90
Average:     all   85.30    0.00    8.20    0.50    0.00    0.10    0.00    0.00    0.00    5.90
"""


# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------

class TestFreshState(unittest.TestCase):

    def test_returns_defaults(self):
        from auto_suspend import fresh_state
        state = fresh_state()
        self.assertEqual(state["idle_counter"], 0)
        self.assertEqual(state["gpu_idle_counter"], 0)
        self.assertEqual(state["sample_count"], 0)
        self.assertIsNone(state["cpu_idle_n"])
        self.assertIsNone(state["cpu_idle_n1"])
        self.assertIsNone(state["cpu_avg"])


class TestLoadState(unittest.TestCase):

    def test_missing_file(self):
        from auto_suspend import load_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            state = load_state(tmpdir, logger)
        self.assertEqual(state["sample_count"], 0)

    def test_load_valid_state(self):
        from auto_suspend import load_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            state_path = Path(tmpdir) / "state.json"
            state_path.write_text(json.dumps({"idle_counter": 3, "sample_count": 5}))
            state = load_state(tmpdir, logger)
        self.assertEqual(state["idle_counter"], 3)
        self.assertEqual(state["sample_count"], 5)

    def test_load_invalid_json(self):
        from auto_suspend import load_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            state_path = Path(tmpdir) / "state.json"
            state_path.write_text("{bad json}")
            state = load_state(tmpdir, logger)
        logger.warning.assert_called()
        self.assertEqual(state["sample_count"], 0)

    def test_missing_keys_merged(self):
        from auto_suspend import load_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            state_path = Path(tmpdir) / "state.json"
            state_path.write_text(json.dumps({"idle_counter": 1}))
            state = load_state(tmpdir, logger)
        self.assertEqual(state["sample_count"], 0)
        self.assertEqual(state["cpu_avg"], None)


class TestSaveState(unittest.TestCase):

    def test_atomic_write(self):
        from auto_suspend import save_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            save_state(tmpdir, {"sample_count": 5, "idle_counter": 2}, logger)
            state_path = Path(tmpdir) / "state.json"
            self.assertTrue(state_path.exists())
            data = json.load(open(state_path))
            self.assertEqual(data["sample_count"], 5)

    def test_creates_parent(self):
        from auto_suspend import save_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            nested = Path(tmpdir) / "a" / "b"
            save_state(str(nested), {"sample_count": 1, "idle_counter": 0}, logger)
            self.assertTrue((nested / "state.json").exists())

    def test_no_tmp_after_save(self):
        from auto_suspend import save_state
        logger = MagicMock()
        with tempfile.TemporaryDirectory() as tmpdir:
            save_state(tmpdir, {"sample_count": 1, "idle_counter": 0}, logger)
            tmpfiles = [f for f in os.listdir(tmpdir) if f.startswith(".state_")]
            self.assertEqual(len(tmpfiles), 0)


class TestMeasureCpu(unittest.TestCase):

    def test_idle(self):
        from auto_suspend import measure_cpu
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(MPSTAT_IDLE)
            result = measure_cpu(logger)
        self.assertAlmostEqual(result, 99.58, places=1)

    def test_high_cpu(self):
        from auto_suspend import measure_cpu
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(MPSTAT_HIGH)
            result = measure_cpu(logger)
        self.assertAlmostEqual(result, 5.9, places=1)

    def test_returncode_nonzero(self):
        from auto_suspend import measure_cpu
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(stdout="", returncode=1, stderr="fail")
            result = measure_cpu(logger)
        self.assertIsNone(result)
        logger.warning.assert_called()

    def test_timeout(self):
        from auto_suspend import measure_cpu
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.side_effect = subprocess.TimeoutExpired(cmd=["mpstat", "1", "1"], timeout=15)
            result = measure_cpu(logger)
        self.assertIsNone(result)
        logger.warning.assert_called_with("mpstat timed out")

    def test_portuguese_locale(self):
        """Verify mpstat with C locale (LC_ALL=C) parses correctly.

        The measure_cpu function forces LC_ALL=C so mpstat always outputs
        in English regardless of system locale.
        """
        from auto_suspend import measure_cpu
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(MPSTAT_IDLE)
            result = measure_cpu(logger)
            # Verify subprocess was called with LC_ALL=C
            call_kwargs = run.call_args
            self.assertEqual(call_kwargs[1]["env"]["LC_ALL"], "C")
        self.assertAlmostEqual(result, 99.58, places=1)


class TestMeasureGpuNvidia(unittest.TestCase):

    def test_idle(self):
        from auto_suspend import measure_gpu_nvidia
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(NVIDIA_IDLE)
            result = measure_gpu_nvidia(logger)
        self.assertAlmostEqual(result, 0.0, places=1)

    def test_busy_single(self):
        from auto_suspend import measure_gpu_nvidia
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(NVIDIA_BUSY_SINGLE)
            result = measure_gpu_nvidia(logger)
        self.assertAlmostEqual(result, 95.0, places=1)

    def test_multi_gpu_max(self):
        from auto_suspend import measure_gpu_nvidia
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(NVIDIA_BUSY_4GAPS)
            result = measure_gpu_nvidia(logger)
        self.assertAlmostEqual(result, 95.0, places=1)

    def test_file_not_found(self):
        from auto_suspend import measure_gpu_nvidia
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.side_effect = FileNotFoundError()
            result = measure_gpu_nvidia(logger)
        self.assertIsNone(result)

    def test_timeout(self):
        from auto_suspend import measure_gpu_nvidia
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.side_effect = subprocess.TimeoutExpired(cmd=["nvidia-smi"], timeout=15)
            result = measure_gpu_nvidia(logger)
        self.assertIsNone(result)
        logger.warning.assert_called_with("nvidia-smi timed out")

    def test_returncode_nonzero(self):
        from auto_suspend import measure_gpu_nvidia
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(returncode=1)
            result = measure_gpu_nvidia(logger)
        self.assertIsNone(result)


class TestMeasureGpuAmd(unittest.TestCase):

    def test_busy(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(AMD_SMII_JSON)
            result = measure_gpu_amd(logger)
        self.assertAlmostEqual(result, 100.0, places=1)

    def test_idle_gfx0(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(AMD_SMII_GFX_0)
            result = measure_gpu_amd(logger)
        self.assertAlmostEqual(result, 0.0, places=1)

    def test_gfx80(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(AMD_SMII_GFX_80)
            result = measure_gpu_amd(logger)
        self.assertAlmostEqual(result, 80.0, places=1)

    def test_file_not_found(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.side_effect = FileNotFoundError()
            result = measure_gpu_amd(logger)
        self.assertIsNone(result)

    def test_short_output(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result("# header only\n")
            result = measure_gpu_amd(logger)
        self.assertIsNone(result)

    def test_empty_array(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result("[]")
            result = measure_gpu_amd(logger)
        self.assertIsNone(result)

    def test_returncode_nonzero(self):
        from auto_suspend import measure_gpu_amd
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(returncode=1)
            result = measure_gpu_amd(logger)
        self.assertIsNone(result)


class TestGetActiveUsers(unittest.TestCase):

    def test_no_users(self):
        from auto_suspend import get_active_users
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(WHO_EMPTY)
            result = get_active_users(logger)
        self.assertFalse(result)

    def test_with_users(self):
        from auto_suspend import get_active_users
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.return_value = _make_mock_result(WHO_WITH_USERS)
            result = get_active_users(logger)
        self.assertTrue(result)

    def test_timeout(self):
        from auto_suspend import get_active_users
        logger = MagicMock()
        with patch("auto_suspend.subprocess.run") as run:
            run.side_effect = subprocess.TimeoutExpired(cmd=["who"], timeout=5)
            result = get_active_users(logger)
        self.assertTrue(result)  # conservative: assume user present


class TestUpdateAverages(unittest.TestCase):

    def test_first_sample(self):
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        new_state = update_averages(state, cpu_n=95.0, gpu_n=2.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(new_state["sample_count"], 1)
        self.assertAlmostEqual(new_state["cpu_avg"], 95.0)
        self.assertEqual(new_state["gpu_idle_counter"], 1)  # idle → +1
        self.assertEqual(new_state["cpu_idle_n1"], None)

    def test_ema_update_cpu_only(self):
        """CPU EMA updates, GPU idle counter increments."""
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        state = update_averages(state, cpu_n=95.0, gpu_n=2.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        # EMA = 0.3 * 95 + 0.7 * 90 = 91.5
        self.assertAlmostEqual(state["cpu_avg"], 91.5)
        # GPU idle both cycles → counter = 2
        self.assertEqual(state["gpu_idle_counter"], 2)
        self.assertEqual(state["sample_count"], 2)

    def test_gpu_busy_resets_counter(self):
        """GPU above threshold resets idle counter to 0."""
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        # Cycle 1: idle GPU → counter = 1
        state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 1)
        # Cycle 2: busy GPU → counter resets to 0
        state = update_averages(state, cpu_n=90.0, gpu_n=50.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 0)
        # Cycle 3: idle GPU → counter = 1
        state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 1)

    def test_gpu_at_threshold_is_idle(self):
        """GPU exactly at threshold counts as idle (strictly greater = busy)."""
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        # GPU at exactly 5.0 with threshold 5.0 → idle (not > threshold)
        state = update_averages(state, cpu_n=90.0, gpu_n=5.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 1)
        # GPU at 5.1 → busy
        state = update_averages(state, cpu_n=90.0, gpu_n=5.1, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 0)

    def test_gpu_none_is_idle(self):
        """None GPU (no GPU) counts as idle."""
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        state = update_averages(state, cpu_n=90.0, gpu_n=None, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 1)

    def test_shift_n_to_n1(self):
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        state = update_averages(state, cpu_n=80.0, gpu_n=50.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        state = update_averages(state, cpu_n=85.0, gpu_n=30.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["cpu_idle_n"], 85.0)
        self.assertEqual(state["cpu_idle_n1"], 80.0)

    def test_none_cpu_preserves_avg(self):
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        state = update_averages(state, cpu_n=90.0, gpu_n=5.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        state = update_averages(state, cpu_n=None, gpu_n=None, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertAlmostEqual(state["cpu_avg"], 90.0)
        self.assertEqual(state["sample_count"], 2)

    def test_gpu_idle_counter_survives_cpu_none(self):
        """GPU counter increments even when CPU measurement is None."""
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()
        state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        state = update_averages(state, cpu_n=None, gpu_n=0.5, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 2)
        self.assertAlmostEqual(state["cpu_avg"], 90.0)

    def test_consecutive_idle_then_spike(self):
        """
        Simulate the real data pattern: idle GPU with periodic 100% spikes.

        Each spike resets the counter. This is the key scenario that the
        old EMA approach handled poorly (spikes lingered in the average).
        """
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()

        # 3 idle cycles → counter = 3
        for _ in range(3):
            state = update_averages(state, cpu_n=99.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 3)

        # 1 spike → counter resets to 0
        state = update_averages(state, cpu_n=99.0, gpu_n=100.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 0)

        # 3 idle cycles after spike → counter = 3
        for _ in range(3):
            state = update_averages(state, cpu_n=99.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertEqual(state["gpu_idle_counter"], 3)

    def test_ema_cpu_convergence(self):
        """
        Verify CPU EMA converges to the steady-state value.

        With alpha=0.3 and 100 identical samples, the average should equal
        the sample value (since EMA_new = 0.3*x + 0.7*x = x).
        """
        from auto_suspend import fresh_state, update_averages
        logger = MagicMock()
        state = fresh_state()

        cpu_idle = 97.0
        for _ in range(50):
            state = update_averages(state, cpu_n=cpu_idle, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)

        self.assertAlmostEqual(state["cpu_avg"], cpu_idle, places=2)


class TestEvaluate(unittest.TestCase):

    def test_idle_system(self):
        from auto_suspend import fresh_state, update_averages, evaluate
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=95.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_warmup_skips(self):
        from auto_suspend import fresh_state, evaluate
        logger = MagicMock()
        state = fresh_state()
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_users_logged_in(self):
        from auto_suspend import fresh_state, evaluate, update_averages
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=95.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertFalse(evaluate(state, has_users=True, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_cpu_not_idle(self):
        from auto_suspend import fresh_state, evaluate, update_averages
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=40.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_gpu_busy_resets_evaluation(self):
        """GPU busy this cycle (counter=0) blocks evaluation even if CPU is idle."""
        from auto_suspend import fresh_state, update_averages, evaluate
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=95.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        # Force GPU busy by resetting counter
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_gpu_none_is_idle(self):
        from auto_suspend import fresh_state, evaluate, update_averages
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=95.0, gpu_n=None, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_at_threshold_cpu_below(self):
        from auto_suspend import fresh_state, update_averages, evaluate
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=90.0, gpu_n=1.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_gpu_at_threshold_is_idle(self):
        """GPU at exactly threshold should NOT block evaluation."""
        from auto_suspend import fresh_state, update_averages, evaluate
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=95.0, gpu_n=5.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_gpu_above_threshold_blocks(self):
        """GPU above threshold should block evaluation."""
        from auto_suspend import fresh_state, update_averages, evaluate
        logger = MagicMock()
        state = fresh_state()
        state = update_averages(state, cpu_n=95.0, gpu_n=50.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=1, logger=logger))

    def test_custom_thresholds(self):
        from auto_suspend import fresh_state, update_averages, evaluate
        logger = MagicMock()
        state = fresh_state()
        for _ in range(3):
            state = update_averages(state, cpu_n=95.0, gpu_n=3.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertTrue(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=3, idle_threshold=95.0, gpu_threshold=5.0, min_samples=3, logger=logger))

    def test_gpu_busy_during_warmup(self):
        """GPU busy during warmup: counter=0 but still in warmup."""
        from auto_suspend import fresh_state, evaluate, update_averages
        logger = MagicMock()
        state = fresh_state()
        # Only 1 sample, counter=0 because GPU was busy
        state = update_averages(state, cpu_n=95.0, gpu_n=100.0, gpu_threshold=5.0, alpha=0.3, logger=logger)
        self.assertFalse(evaluate(state, has_users=False, gpu_idle_counter=0, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))


class TestGetWhyNotIdle(unittest.TestCase):

    def test_users(self):
        from auto_suspend import get_why_not_idle
        state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_idle_counter": 3}
        reason = get_why_not_idle(state, has_users=True, logger=MagicMock())
        self.assertIn("users logged in", reason)

    def test_cpu_low(self):
        from auto_suspend import get_why_not_idle
        state = {"sample_count": 5, "cpu_avg": 40.0, "gpu_idle_counter": 3}
        reason = get_why_not_idle(state, has_users=False, logger=MagicMock())
        self.assertIn("CPU idle avg", reason)

    def test_gpu_busy(self):
        from auto_suspend import get_why_not_idle
        state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_idle_counter": 0}
        reason = get_why_not_idle(state, has_users=False, gpu_idle_counter=0, logger=MagicMock())
        self.assertIn("GPU busy", reason)

    def test_gpu_idle_no_reason(self):
        """GPU idle counter > 0 should not add a reason."""
        from auto_suspend import get_why_not_idle
        state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_idle_counter": 3}
        reason = get_why_not_idle(state, has_users=False, gpu_idle_counter=3, logger=MagicMock())
        self.assertNotIn("GPU busy", reason)

    def test_warming_up(self):
        from auto_suspend import get_why_not_idle
        state = {"sample_count": 2, "cpu_avg": None, "gpu_idle_counter": 0}
        reason = get_why_not_idle(state, has_users=False, gpu_idle_counter=0, logger=MagicMock())
        self.assertIn("warming up", reason)

    def test_multiple_reasons(self):
        from auto_suspend import get_why_not_idle
        state = {"sample_count": 5, "cpu_avg": 40.0, "gpu_idle_counter": 0}
        reason = get_why_not_idle(state, has_users=True, gpu_idle_counter=0, logger=MagicMock())
        self.assertIn("users logged in", reason)
        self.assertIn("CPU idle avg", reason)
        self.assertIn("GPU busy", reason)


class TestSetupLogging(unittest.TestCase):

    def test_returns_logger(self):
        from auto_suspend import setup_logging
        logger = setup_logging("/tmp/_test_suspend.log", debug=False)
        self.assertEqual(logger.name, "auto_suspend")
        self.assertEqual(len(logger.handlers), 2)  # file + stderr


class TestSignalHandler(unittest.TestCase):

    def test_returns_zero(self):
        from auto_suspend import signal_handler
        with self.assertRaises(SystemExit) as ctx:
            signal_handler(signal.SIGTERM, None)
        self.assertEqual(ctx.exception.code, 0)


if __name__ == "__main__":
    unittest.main()