Add fixtures and unit tests for auto suspend functionality
- Introduced fixture files for AMD and NVIDIA GPU monitoring outputs, CPU usage statistics, and user sessions. - Created a comprehensive unit test suite for the auto_suspend module, covering state management, CPU and GPU measurement, user activity detection, and evaluation logic. - Updated Ansible tasks to install Python instead of bc, create necessary directories and log files, and ensure proper SELinux context for the auto suspend script. - Modified the systemd service and timer configurations to improve the auto suspend functionality, including command-line arguments for the Python script.
This commit is contained in:
@@ -1,61 +0,0 @@
|
||||
#!/bin/bash
|
||||
|
||||
IDLE_TIMEOUT_MINUTES=20
|
||||
|
||||
# Check 1: Any logged-in users?
|
||||
if who | grep -qv "tty"; then
|
||||
echo "Active user sessions found. Skipping suspend."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Check 2: CPU idle percentage
|
||||
CPU_IDLE=$(mpstat 1 1 | tail -n 1 | awk '/:/ {print $12}' | sed 's/,/./g')
|
||||
CPU_THRESHOLD=90
|
||||
|
||||
if (( $(echo "$CPU_IDLE < $CPU_THRESHOLD" | bc -l) )); then
|
||||
echo "CPU too active. Skipping suspend."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if ! command -v nvidia-smi &> /dev/null; then
|
||||
echo "nvidia-smi not found, skipping GPU check"
|
||||
else
|
||||
# Check 3: Is GPU idle?
|
||||
GPU_UTIL=$(nvidia-smi --query-gpu=utilization.gpu --format=csv,noheader,nounits | head -n 1)
|
||||
GPU_THRESHOLD=0
|
||||
|
||||
if (( GPU_UTIL > GPU_THRESHOLD )); then
|
||||
echo "GPU is busy (utilization=${GPU_UTIL}%). Skipping suspend."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Check 4: Is OLLAMA active?
|
||||
OLLAMA_ACTIVE=$(nvidia-smi --query-compute-apps=name,used_memory --format=csv,noheader,nounits | grep ollama | wc -l)
|
||||
if (( OLLAMA_ACTIVE > 0 )); then
|
||||
echo "OLLMAMA is active. Skipping suspend."
|
||||
exit 0
|
||||
fi
|
||||
fi
|
||||
|
||||
if | command -v amd-smi &> /dev/null; then
|
||||
# Check 5: Is AMD GPU idle?
|
||||
AMD_GPU_UTIL=$(amd-smi monitor --csv | head -n 2 | tail -n 1 | cut -d ',' -f 8)
|
||||
AMD_GPU_THRESHOLD=0
|
||||
|
||||
if (( AMD_GPU_UTIL > AMD_GPU_THRESHOLD )); then
|
||||
echo "AMD GPU is busy (utilization=${AMD_GPU_UTIL}%). Skipping suspend."
|
||||
exit 0
|
||||
fi
|
||||
|
||||
if amd-smi process --csv | head -n 2 | tail -n 1 | grep -q "No running processes detected"; then
|
||||
echo "No processes running on AMD GPU."
|
||||
else
|
||||
echo "Processes are running on AMD GPU. Skipping suspend."
|
||||
exit 0
|
||||
fi
|
||||
else
|
||||
echo "amd-smi not found, skipping AMD GPU check"
|
||||
fi
|
||||
|
||||
echo "System idle. Suspending..."
|
||||
shutdown now
|
||||
@@ -1,2 +1,10 @@
|
||||
---
|
||||
auto_suspend_grub_config_path_bios: /boot/grub2/grub.cfg
|
||||
|
||||
auto_suspend_alpha: 0.3
|
||||
auto_suspend_min_samples: 3
|
||||
auto_suspend_idle_timeout_cycles: 4
|
||||
auto_suspend_cpu_idle_threshold: 90.0
|
||||
auto_suspend_gpu_util_threshold: 5.0
|
||||
auto_suspend_state_dir: /var/lib/auto_suspend
|
||||
auto_suspend_log_file: /var/log/auto_suspend.log
|
||||
|
||||
@@ -0,0 +1,531 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
Auto-suspend daemon — checks system idleness and suspends when idle.
|
||||
|
||||
Uses an exponential moving average (EMA) of CPU idle % and GPU utilization
|
||||
to smooth out transient spikes. The system suspends only after
|
||||
idle_timeout_cycles consecutive idle checks.
|
||||
|
||||
State is persisted to /var/lib/auto_suspend/state.json and survives
|
||||
across timer invocations. State is fully reset on suspend resume.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import json
|
||||
import logging
|
||||
import logging.handlers
|
||||
import os
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
DEFAULT_STATE_DIR = "/var/lib/auto_suspend"
|
||||
DEFAULT_LOG_FILE = "/var/log/auto_suspend.log"
|
||||
DEFAULT_ALPHA = 0.3
|
||||
DEFAULT_MIN_SAMPLES = 3
|
||||
DEFAULT_IDLE_TIMEOUT_CYCLES = 4
|
||||
DEFAULT_CPU_IDLE_THRESHOLD = 90.0
|
||||
DEFAULT_GPU_UTIL_THRESHOLD = 5.0
|
||||
|
||||
STATE_FILE_NAME = "state.json"
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Logging setup
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def setup_logging(log_file, debug=False):
|
||||
"""Configure logging to both file and stderr."""
|
||||
logger = logging.getLogger("auto_suspend")
|
||||
logger.setLevel(logging.DEBUG if debug else logging.INFO)
|
||||
|
||||
formatter = logging.Formatter(
|
||||
"%(asctime)s [%(levelname)s] %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S",
|
||||
)
|
||||
|
||||
# File handler with rotation (1 MB, 3 backups)
|
||||
log_path = Path(log_file)
|
||||
log_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
file_handler = logging.handlers.RotatingFileHandler(
|
||||
str(log_path),
|
||||
maxBytes=1 * 1024 * 1024,
|
||||
backupCount=3,
|
||||
)
|
||||
file_handler.setLevel(logging.DEBUG)
|
||||
file_handler.setFormatter(formatter)
|
||||
logger.addHandler(file_handler)
|
||||
|
||||
# Stderr handler (visible in journalctl -u auto-suspend.service)
|
||||
stderr_handler = logging.StreamHandler(sys.stderr)
|
||||
stderr_handler.setLevel(logging.DEBUG)
|
||||
stderr_handler.setFormatter(formatter)
|
||||
logger.addHandler(stderr_handler)
|
||||
|
||||
return logger
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# State persistence
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def fresh_state():
|
||||
"""Return a fresh state dict with all values at defaults."""
|
||||
return {
|
||||
"cpu_idle_n": None,
|
||||
"cpu_idle_n1": None,
|
||||
"gpu_util_n": None,
|
||||
"gpu_util_n1": None,
|
||||
"cpu_avg": None,
|
||||
"gpu_avg": None,
|
||||
"idle_counter": 0,
|
||||
"sample_count": 0,
|
||||
}
|
||||
|
||||
|
||||
def load_state(state_dir, logger):
|
||||
"""Load state from disk, or return fresh state on any error."""
|
||||
state_path = Path(state_dir) / STATE_FILE_NAME
|
||||
if not state_path.exists():
|
||||
logger.debug("No state file found at %s — starting fresh", state_path)
|
||||
return fresh_state()
|
||||
|
||||
try:
|
||||
with open(state_path) as f:
|
||||
state = json.load(f)
|
||||
# Validate required keys
|
||||
for key in fresh_state():
|
||||
if key not in state:
|
||||
state[key] = fresh_state()[key]
|
||||
logger.debug("Loaded state: sample_count=%d, idle_counter=%d",
|
||||
state["sample_count"], state["idle_counter"])
|
||||
return state
|
||||
except (json.JSONDecodeError, IOError, OSError) as exc:
|
||||
logger.warning("Failed to load state (%s) — starting fresh", exc)
|
||||
return fresh_state()
|
||||
|
||||
|
||||
def save_state(state_dir, state, logger):
|
||||
"""Atomically write state to disk (tempfile + os.replace)."""
|
||||
state_path = Path(state_dir) / STATE_FILE_NAME
|
||||
state_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
fd, tmp = tempfile.mkstemp(
|
||||
dir=str(state_path.parent), suffix=".tmp", prefix=".state_"
|
||||
)
|
||||
try:
|
||||
with os.fdopen(fd, "w") as f:
|
||||
json.dump(state, f, indent=2)
|
||||
f.write("\n")
|
||||
os.replace(tmp, str(state_path))
|
||||
logger.debug("State saved: sample_count=%d, idle_counter=%d",
|
||||
state["sample_count"], state["idle_counter"])
|
||||
except BaseException:
|
||||
try:
|
||||
os.unlink(tmp)
|
||||
except OSError:
|
||||
pass
|
||||
raise
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Measurement functions
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def measure_cpu(logger):
|
||||
"""
|
||||
Measure CPU idle percentage using mpstat.
|
||||
|
||||
Runs ``mpstat 1 1`` and extracts the last field of the "Average:" line,
|
||||
which is ``%idle``. Returns a float, or None on failure.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["mpstat", "1", "1"],
|
||||
capture_output=True, text=True, timeout=15,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
logger.warning("mpstat returned non-zero: %s", result.stderr.strip())
|
||||
return None
|
||||
|
||||
for line in result.stdout.strip().splitlines():
|
||||
line = line.strip()
|
||||
if line.startswith("Average:") or line.startswith("all"):
|
||||
fields = line.split()
|
||||
# Find %idle column index from header
|
||||
idle_pct = float(fields[-1])
|
||||
logger.debug("CPU idle: %.1f%%", idle_pct)
|
||||
return idle_pct
|
||||
|
||||
return None
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("mpstat timed out")
|
||||
return None
|
||||
except (ValueError, IndexError) as exc:
|
||||
logger.warning("Failed to parse mpstat output: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def measure_gpu_nvidia(logger):
|
||||
"""
|
||||
Measure GPU utilization via nvidia-smi.
|
||||
|
||||
Returns the **maximum** utilization across all GPUs (any busy GPU means
|
||||
the system is not idle). Returns None if nvidia-smi is unavailable.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["nvidia-smi", "--query-gpu=utilization.gpu",
|
||||
"--format=csv,noheader,nounits"],
|
||||
capture_output=True, text=True, timeout=15,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
# nvidia-smi may fail if no NVIDIA GPU is present
|
||||
logger.debug("nvidia-smi returned %d — no NVIDIA GPU", result.returncode)
|
||||
return None
|
||||
|
||||
utils = []
|
||||
for line in result.stdout.strip().splitlines():
|
||||
line = line.strip()
|
||||
if line:
|
||||
utils.append(float(line))
|
||||
|
||||
if not utils:
|
||||
return None
|
||||
|
||||
max_util = max(utils)
|
||||
logger.debug("NVIDIA GPU utilization: max=%.1f%%", max_util)
|
||||
return max_util
|
||||
except FileNotFoundError:
|
||||
logger.debug("nvidia-smi not found")
|
||||
return None
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("nvidia-smi timed out")
|
||||
return None
|
||||
except (ValueError, IndexError) as exc:
|
||||
logger.warning("Failed to parse nvidia-smi output: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def measure_gpu_amd(logger):
|
||||
"""
|
||||
Measure GPU utilization via amd-smi.
|
||||
|
||||
Parses the ``gfx`` column from ``amd-smi monitor --csv`` by header name
|
||||
(not by fixed column index). Returns None if amd-smi is unavailable.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["amd-smi", "monitor", "--csv"],
|
||||
capture_output=True, text=True, timeout=15,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
logger.debug("amd-smi returned %d — no AMD GPU", result.returncode)
|
||||
return None
|
||||
|
||||
lines = result.stdout.strip().splitlines()
|
||||
if len(lines) < 2:
|
||||
logger.debug("amd-smi output too short")
|
||||
return None
|
||||
|
||||
# Filter out non-data lines (comments, annotations with wrong field count)
|
||||
csv_reader = csv.reader(lines)
|
||||
header = next(csv_reader)
|
||||
for fields in csv_reader:
|
||||
if len(fields) != len(header):
|
||||
continue
|
||||
row = dict(zip(header, fields))
|
||||
try:
|
||||
# Try both "gfx" and "gfx_util" column names
|
||||
gfx_val = row.get("gfx") or row.get("gfx_util")
|
||||
if gfx_val is None:
|
||||
logger.debug("amd-smi CSV has no 'gfx' column; columns: %s",
|
||||
list(row.keys()))
|
||||
return None
|
||||
util = float(gfx_val)
|
||||
logger.debug("AMD GPU utilization: %.1f%%", util)
|
||||
return util
|
||||
except (ValueError, TypeError):
|
||||
continue
|
||||
|
||||
return None
|
||||
except FileNotFoundError:
|
||||
logger.debug("amd-smi not found")
|
||||
return None
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("amd-smi timed out")
|
||||
return None
|
||||
except BaseException as exc:
|
||||
logger.warning("Failed to parse amd-smi output: %s", exc)
|
||||
return None
|
||||
|
||||
|
||||
def get_active_users(logger):
|
||||
"""
|
||||
Check for logged-in users.
|
||||
|
||||
Returns True if ``who`` reports any sessions (including SSH/pts).
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["who"],
|
||||
capture_output=True, text=True, timeout=5,
|
||||
)
|
||||
users = [line for line in result.stdout.strip().splitlines() if line.strip()]
|
||||
if users:
|
||||
logger.debug("Active users: %s", len(users))
|
||||
return True
|
||||
return False
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.warning("who timed out")
|
||||
return True # conservative: assume user present
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# EMA update
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def update_averages(state, cpu_n, gpu_n, alpha, logger):
|
||||
"""
|
||||
Update EMA averages using the new measurement and shift n → n-1.
|
||||
|
||||
Args:
|
||||
state: current state dict (modified in place for averages)
|
||||
cpu_n: current CPU idle %
|
||||
gpu_n: current GPU utilization %
|
||||
alpha: EMA weight for current sample
|
||||
logger: logger instance
|
||||
|
||||
Returns:
|
||||
dict with updated state (new keys: cpu_avg, gpu_avg, cpu_idle_n,
|
||||
gpu_util_n, cpu_idle_n1, gpu_util_n1, sample_count)
|
||||
"""
|
||||
new_state = dict(state)
|
||||
new_state["sample_count"] = state["sample_count"] + 1
|
||||
|
||||
# Shift current → previous
|
||||
new_state["cpu_idle_n1"] = state["cpu_idle_n"]
|
||||
new_state["gpu_util_n1"] = state["gpu_util_n"]
|
||||
|
||||
# Update current measurements
|
||||
new_state["cpu_idle_n"] = cpu_n
|
||||
new_state["gpu_util_n"] = gpu_n
|
||||
|
||||
# EMA for CPU idle average
|
||||
if state["cpu_avg"] is not None and cpu_n is not None:
|
||||
new_state["cpu_avg"] = round(alpha * cpu_n + (1 - alpha) * state["cpu_avg"], 2)
|
||||
elif cpu_n is not None:
|
||||
# First valid sample seeds the average
|
||||
new_state["cpu_avg"] = cpu_n
|
||||
|
||||
# EMA for GPU utilization average
|
||||
if state["gpu_avg"] is not None and gpu_n is not None:
|
||||
new_state["gpu_avg"] = round(alpha * gpu_n + (1 - alpha) * state["gpu_avg"], 2)
|
||||
elif gpu_n is not None:
|
||||
new_state["gpu_avg"] = gpu_n
|
||||
|
||||
logger.debug(
|
||||
"Updated averages: cpu_avg=%.1f%%, gpu_avg=%.1f%%, sample_count=%d",
|
||||
new_state["cpu_avg"] if new_state["cpu_avg"] is not None else 0,
|
||||
new_state["gpu_avg"] if new_state["gpu_avg"] is not None else 0,
|
||||
new_state["sample_count"],
|
||||
)
|
||||
return new_state
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Evaluation
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def get_why_not_idle(state, has_users, logger):
|
||||
"""Return a human-readable reason why the system is not idle."""
|
||||
reasons = []
|
||||
if has_users:
|
||||
reasons.append("users logged in")
|
||||
if state["cpu_avg"] is not None:
|
||||
if state["cpu_avg"] <= DEFAULT_CPU_IDLE_THRESHOLD:
|
||||
reasons.append(
|
||||
f"CPU idle avg {state['cpu_avg']:.1f}% <= threshold {DEFAULT_CPU_IDLE_THRESHOLD:.0f}%"
|
||||
)
|
||||
if state["gpu_avg"] is not None:
|
||||
if state["gpu_avg"] >= DEFAULT_GPU_UTIL_THRESHOLD:
|
||||
reasons.append(
|
||||
f"GPU util avg {state['gpu_avg']:.1f}% >= threshold {DEFAULT_GPU_UTIL_THRESHOLD:.0f}%"
|
||||
)
|
||||
if not reasons:
|
||||
if state["sample_count"] < DEFAULT_MIN_SAMPLES:
|
||||
reasons.append(f"warming up ({state['sample_count']}/{DEFAULT_MIN_SAMPLES} samples)")
|
||||
else:
|
||||
reasons.append("unknown")
|
||||
return "; ".join(reasons)
|
||||
|
||||
|
||||
def evaluate(state, has_users, idle_threshold, gpu_threshold,
|
||||
min_samples, logger):
|
||||
"""
|
||||
Evaluate whether the system should be considered idle.
|
||||
|
||||
Returns True if all conditions are met for one idle cycle.
|
||||
"""
|
||||
# Need enough samples before trusting the average
|
||||
if state["sample_count"] < min_samples:
|
||||
logger.debug(
|
||||
"Skipping evaluation: warmup (%d/%d samples)",
|
||||
state["sample_count"], min_samples,
|
||||
)
|
||||
return False
|
||||
|
||||
# Check active users
|
||||
if has_users:
|
||||
logger.debug("Not idle: users logged in")
|
||||
return False
|
||||
|
||||
# Check CPU idle average
|
||||
if state["cpu_avg"] is None or state["cpu_avg"] <= idle_threshold:
|
||||
logger.debug(
|
||||
"Not idle: CPU avg %.1f%% <= threshold %.0f%%",
|
||||
state["cpu_avg"] or 0, idle_threshold,
|
||||
)
|
||||
return False
|
||||
|
||||
# Check GPU utilization average (None means no GPUs = ok)
|
||||
if state["gpu_avg"] is not None and state["gpu_avg"] >= gpu_threshold:
|
||||
logger.debug(
|
||||
"Not idle: GPU avg %.1f%% >= threshold %.0f%%",
|
||||
state["gpu_avg"], gpu_threshold,
|
||||
)
|
||||
return False
|
||||
|
||||
return True
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Main
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(description="Auto-suspend based on system idle state")
|
||||
parser.add_argument("--alpha", type=float, default=DEFAULT_ALPHA,
|
||||
help="EMA weight for current sample (default: %(default)s)")
|
||||
parser.add_argument("--min-samples", type=int, default=DEFAULT_MIN_SAMPLES,
|
||||
help="Warmup samples before trusting averages (default: %(default)s)")
|
||||
parser.add_argument("--idle-timeout-cycles", type=int, default=DEFAULT_IDLE_TIMEOUT_CYCLES,
|
||||
help="Consecutive idle cycles needed (default: %(default)s)")
|
||||
parser.add_argument("--cpu-idle-threshold", type=float, default=DEFAULT_CPU_IDLE_THRESHOLD,
|
||||
help="CPU idle %% threshold (default: %(default)s)")
|
||||
parser.add_argument("--gpu-util-threshold", type=float, default=DEFAULT_GPU_UTIL_THRESHOLD,
|
||||
help="GPU utilization %% threshold (default: %(default)s)")
|
||||
parser.add_argument("--state-dir", default=DEFAULT_STATE_DIR,
|
||||
help="Directory for state file (default: %(default)s)")
|
||||
parser.add_argument("--log-file", default=DEFAULT_LOG_FILE,
|
||||
help="Log file path (default: %(default)s)")
|
||||
parser.add_argument("--debug", action="store_true",
|
||||
help="Enable debug logging")
|
||||
parser.add_argument("--dry-run", action="store_true",
|
||||
help="Run check without persisting state or suspending")
|
||||
args = parser.parse_args()
|
||||
|
||||
logger = setup_logging(args.log_file, args.debug)
|
||||
|
||||
logger.info("Auto-suspend check started")
|
||||
logger.debug(
|
||||
"Config: alpha=%.2f, min_samples=%d, idle_cycles=%d, "
|
||||
"cpu_threshold=%.0f%%, gpu_threshold=%.0f%%",
|
||||
args.alpha, args.min_samples, args.idle_timeout_cycles,
|
||||
args.cpu_idle_threshold, args.gpu_util_threshold,
|
||||
)
|
||||
|
||||
# --- Load state ---
|
||||
state = load_state(args.state_dir, logger)
|
||||
|
||||
# --- Measure ---
|
||||
cpu_idle = measure_cpu(logger)
|
||||
gpu_nvidia = measure_gpu_nvidia(logger)
|
||||
gpu_amd = measure_gpu_amd(logger)
|
||||
|
||||
# Combine GPU metrics: max across all available GPUs; None if none found
|
||||
gpu_values = [v for v in [gpu_nvidia, gpu_amd] if v is not None]
|
||||
gpu_util = max(gpu_values) if gpu_values else None
|
||||
|
||||
logger.debug(
|
||||
"Measurements: cpu_idle=%.1f%%, gpu_util=%.1f%%",
|
||||
cpu_idle if cpu_idle is not None else 0,
|
||||
gpu_util if gpu_util is not None else 0,
|
||||
)
|
||||
|
||||
# --- Update averages ---
|
||||
state = update_averages(
|
||||
state, cpu_idle, gpu_util, args.alpha, logger,
|
||||
)
|
||||
|
||||
# --- Evaluate ---
|
||||
has_users = get_active_users(logger)
|
||||
|
||||
is_idle = evaluate(
|
||||
state, has_users,
|
||||
idle_threshold=args.cpu_idle_threshold,
|
||||
gpu_threshold=args.gpu_util_threshold,
|
||||
min_samples=args.min_samples,
|
||||
logger=logger,
|
||||
)
|
||||
|
||||
if is_idle:
|
||||
state["idle_counter"] = state.get("idle_counter", 0) + 1
|
||||
logger.info(
|
||||
"System idle (cpu_avg=%.1f%%, gpu_avg=%.1f%%, "
|
||||
"idle_counter=%d/%d)",
|
||||
state["cpu_avg"] or 0,
|
||||
state["gpu_avg"] or 0,
|
||||
state["idle_counter"],
|
||||
args.idle_timeout_cycles,
|
||||
)
|
||||
|
||||
if state["idle_counter"] >= args.idle_timeout_cycles:
|
||||
if args.dry_run:
|
||||
print("Would suspend: idle timeout reached (idle_counter=%d/%d)" % (
|
||||
state["idle_counter"], args.idle_timeout_cycles))
|
||||
else:
|
||||
logger.info("Idle timeout reached — suspending...")
|
||||
save_state(args.state_dir, state, logger)
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["systemctl", "suspend"],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
)
|
||||
if result.returncode != 0:
|
||||
logger.error("systemctl suspend failed: %s", result.stderr.strip())
|
||||
else:
|
||||
logger.info("systemctl suspend succeeded")
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error("systemctl suspend timed out")
|
||||
except FileNotFoundError:
|
||||
logger.error("systemctl not found — cannot suspend")
|
||||
except BaseException as exc:
|
||||
logger.error("Failed to suspend: %s", exc)
|
||||
|
||||
# Reset state on resume
|
||||
save_state(args.state_dir, fresh_state(), logger)
|
||||
return
|
||||
else:
|
||||
if not args.dry_run:
|
||||
state["idle_counter"] = 0
|
||||
reason = get_why_not_idle(state, has_users, logger)
|
||||
logger.info("Not idle (%s) — reset counter", reason)
|
||||
|
||||
# --- Save and exit ---
|
||||
if not args.dry_run:
|
||||
save_state(args.state_dir, state, logger)
|
||||
logger.info("Auto-suspend check completed")
|
||||
|
||||
|
||||
def signal_handler(signum, frame):
|
||||
"""Handle SIGTERM/SIGINT gracefully."""
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
signal.signal(signal.SIGTERM, signal_handler)
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
main()
|
||||
@@ -0,0 +1,3 @@
|
||||
gpu,power_usage,hotspot_temperature,memory_temperature,gfx,gfx_clock,mem,mem_clock,encoder,decoder,vclock,dclock,single_bit_ecc,double_bit_ecc,pcie_replay,vram_used,vram_total,pcie_bw,pviol,tviol,phot_tviol,vr_tviol,hbm_tviol
|
||||
0,20,52,60,100,1200,0,96,N/A,0.0,25,25,0,0,N/A,11913,16334,N/A,N/A,N/A,N/A,N/A,N/A
|
||||
|
||||
@@ -0,0 +1,5 @@
|
||||
Linux 6.17.11-200.fc42.x86_64 (gpu-01) 06/08/26 _x86_64_ (12 CPU)
|
||||
|
||||
21:37:19 CPU %usr %nice %sys %iowait %irq %soft %steal %guest %gnice %idle
|
||||
21:37:20 all 0.00 0.00 0.17 0.00 0.17 0.08 0.00 0.00 0.00 99.58
|
||||
Average: all 0.00 0.00 0.17 0.00 0.17 0.08 0.00 0.00 0.00 99.58
|
||||
+1
@@ -0,0 +1 @@
|
||||
0
|
||||
@@ -0,0 +1,2 @@
|
||||
provision sshd 2026-06-08 21:37 (10.19.4.203)
|
||||
provision sshd pts/0 2026-06-06 19:05 (10.19.4.203)
|
||||
@@ -0,0 +1,503 @@
|
||||
#!/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_GFX = _load_fixture("amd-smi-monitor--csv.txt")
|
||||
AMD_SMII_GFX_0 = """gpu,power_usage,hotspot_temperature,memory_temperature,gfx,gfx_clock,mem,mem_clock,encoder,decoder,vclock,dclock,single_bit_ecc,double_bit_ecc,pcie_replay,vram_used,vram_total,pcie_bw,pviol,tviol,phot_tviol,vr_tviol,hbm_tviol
|
||||
0,20,52,62,0,1200,0,96,N/A,0.0,25,25,0,0,N/A,11928,16334,N/A,N/A,N/A,N/A,N/A,N/A
|
||||
"""
|
||||
AMD_SMII_GFX_80 = """gpu,power_usage,hotspot_temperature,memory_temperature,gfx,gfx_clock,mem,mem_clock,encoder,decoder,vclock,dclock,single_bit_ecc,double_bit_ecc,pcie_replay,vram_used,vram_total,pcie_bw,pviol,tviol,phot_tviol,vr_tviol,hbm_tviol
|
||||
0,180,78,82,80,1800,90,200,N/A,45.0,25,25,0,0,N/A,14000,16334,N/A,N/A,N/A,N/A,N/A,N/A
|
||||
"""
|
||||
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
|
||||
"""
|
||||
|
||||
AMD_SMII_ONLY_HEADER = """gpu,power_usage,hotspot_temperature,memory_temperature,gfx
|
||||
0,20,52,62,0
|
||||
"""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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["sample_count"], 0)
|
||||
self.assertIsNone(state["cpu_idle_n"])
|
||||
self.assertIsNone(state["cpu_idle_n1"])
|
||||
self.assertIsNone(state["gpu_util_n"])
|
||||
self.assertIsNone(state["gpu_util_n1"])
|
||||
self.assertIsNone(state["cpu_avg"])
|
||||
self.assertIsNone(state["gpu_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")
|
||||
|
||||
|
||||
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_GFX)
|
||||
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_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, alpha=0.3, logger=logger)
|
||||
self.assertEqual(new_state["sample_count"], 1)
|
||||
self.assertAlmostEqual(new_state["cpu_avg"], 95.0)
|
||||
self.assertAlmostEqual(new_state["gpu_avg"], 2.0)
|
||||
self.assertEqual(new_state["cpu_idle_n1"], None)
|
||||
self.assertEqual(new_state["gpu_util_n1"], None)
|
||||
|
||||
def test_ema_update(self):
|
||||
from auto_suspend import fresh_state, update_averages
|
||||
logger = MagicMock()
|
||||
state = fresh_state()
|
||||
state = update_averages(state, cpu_n=90.0, gpu_n=10.0, alpha=0.3, logger=logger)
|
||||
state = update_averages(state, cpu_n=95.0, gpu_n=5.0, alpha=0.3, logger=logger)
|
||||
# EMA = 0.3 * 95 + 0.7 * 90 = 91.5
|
||||
self.assertAlmostEqual(state["cpu_avg"], 91.5)
|
||||
# EMA = 0.3 * 5 + 0.7 * 10 = 8.5
|
||||
self.assertAlmostEqual(state["gpu_avg"], 8.5)
|
||||
self.assertEqual(state["sample_count"], 2)
|
||||
|
||||
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, alpha=0.3, logger=logger)
|
||||
state = update_averages(state, cpu_n=85.0, gpu_n=30.0, alpha=0.3, logger=logger)
|
||||
self.assertEqual(state["cpu_idle_n"], 85.0)
|
||||
self.assertEqual(state["cpu_idle_n1"], 80.0)
|
||||
self.assertEqual(state["gpu_util_n"], 30.0)
|
||||
self.assertEqual(state["gpu_util_n1"], 50.0)
|
||||
|
||||
def test_none_measurements_preserve_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, alpha=0.3, logger=logger)
|
||||
state = update_averages(state, cpu_n=None, gpu_n=None, alpha=0.3, logger=logger)
|
||||
self.assertAlmostEqual(state["cpu_avg"], 90.0)
|
||||
self.assertAlmostEqual(state["gpu_avg"], 5.0)
|
||||
self.assertEqual(state["sample_count"], 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, alpha=0.3, logger=logger)
|
||||
self.assertTrue(evaluate(state, has_users=False, 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, 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, alpha=0.3, logger=logger)
|
||||
self.assertFalse(evaluate(state, has_users=True, 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, alpha=0.3, logger=logger)
|
||||
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
|
||||
|
||||
def test_gpu_busy(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=50.0, alpha=0.3, logger=logger)
|
||||
self.assertFalse(evaluate(state, has_users=False, 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, alpha=0.3, logger=logger)
|
||||
self.assertTrue(evaluate(state, has_users=False, 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, alpha=0.3, logger=logger)
|
||||
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
|
||||
|
||||
def test_at_threshold_gpu_above(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=5.0, alpha=0.3, logger=logger)
|
||||
self.assertFalse(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, 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, alpha=0.3, logger=logger)
|
||||
self.assertTrue(evaluate(state, has_users=False, idle_threshold=90.0, gpu_threshold=5.0, min_samples=3, logger=logger))
|
||||
self.assertFalse(evaluate(state, has_users=False, idle_threshold=95.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_avg": 1.0}
|
||||
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_avg": 1.0}
|
||||
reason = get_why_not_idle(state, has_users=False, logger=MagicMock())
|
||||
self.assertIn("CPU idle avg", reason)
|
||||
|
||||
def test_gpu_high(self):
|
||||
from auto_suspend import get_why_not_idle
|
||||
state = {"sample_count": 5, "cpu_avg": 95.0, "gpu_avg": 50.0}
|
||||
reason = get_why_not_idle(state, has_users=False, logger=MagicMock())
|
||||
self.assertIn("GPU util avg", reason)
|
||||
|
||||
def test_warming_up(self):
|
||||
from auto_suspend import get_why_not_idle
|
||||
state = {"sample_count": 2, "cpu_avg": None, "gpu_avg": None}
|
||||
reason = get_why_not_idle(state, has_users=False, 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_avg": 50.0}
|
||||
reason = get_why_not_idle(state, has_users=True, logger=MagicMock())
|
||||
self.assertIn("users logged in", reason)
|
||||
self.assertIn("CPU idle avg", reason)
|
||||
self.assertIn("GPU util avg", 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()
|
||||
@@ -6,7 +6,7 @@
|
||||
state: present
|
||||
loop:
|
||||
- sysstat
|
||||
- bc
|
||||
- python3
|
||||
|
||||
- name: Enable sysstat service
|
||||
become: true
|
||||
@@ -25,18 +25,13 @@
|
||||
ansible.builtin.set_fact:
|
||||
existing_cmdline: "{{ grub_line.stdout }}"
|
||||
|
||||
- name: Set fedora grub file locations
|
||||
ansible.builtin.set_fact:
|
||||
grub_config_path_bios: /boot/grub2/grub.cfg
|
||||
when: ansible_distribution == "Fedora" and ansible_os_family == "RedHat"
|
||||
|
||||
- name: Show current GRUB_CMDLINE_LINUX
|
||||
ansible.builtin.debug:
|
||||
msg: "Current GRUB_CMDLINE_LINUX: {{ existing_cmdline }}"
|
||||
|
||||
- name: Updated GRUB_CMDLINE_LINUX
|
||||
ansible.builtin.debug:
|
||||
msg: "Current GRUB_CMDLINE_LINUX: {{ existing_cmdline | regex_replace(' *mem_sleep_default=\\S+', '') }} mem_sleep_default=deep"
|
||||
msg: "Updated GRUB_CMDLINE_LINUX: {{ existing_cmdline | regex_replace(' *mem_sleep_default=\\S+', '') }} mem_sleep_default=deep"
|
||||
|
||||
- name: Ensure mem_sleep_default=deep is set in GRUB_CMDLINE_LINUX
|
||||
become: true
|
||||
@@ -45,24 +40,57 @@
|
||||
GRUB_CMDLINE_LINUX="{{ existing_cmdline | regex_replace(' *mem_sleep_default=\\S+', '') }} mem_sleep_default=deep"
|
||||
ansible.builtin.lineinfile:
|
||||
path: /etc/default/grub
|
||||
regexp: '^GRUB_CMDLINE_LINUX='
|
||||
regexp: "^GRUB_CMDLINE_LINUX="
|
||||
line: "{{ grub_cmdline_linux_line }}"
|
||||
|
||||
- name: Generate grub config for BIOS
|
||||
- name: Generate GRUB config
|
||||
become: true
|
||||
changed_when: false
|
||||
ansible.builtin.command: grub2-mkconfig -o {{ grub_config_path_bios }}
|
||||
ansible.builtin.command: grub2-mkconfig -o {{ auto_suspend_grub_config_path_bios }}
|
||||
|
||||
- name: Create idle check and suspend script
|
||||
- name: Create state directory
|
||||
become: true
|
||||
ansible.builtin.copy:
|
||||
dest: /usr/local/bin/auto_suspend_script.sh
|
||||
mode: '0755'
|
||||
ansible.builtin.file:
|
||||
path: "{{ auto_suspend_state_dir }}"
|
||||
state: directory
|
||||
owner: root
|
||||
group: root
|
||||
content: "{{ lookup('file', 'scripts/auto_suspend.sh') }}"
|
||||
mode: "0755"
|
||||
|
||||
- name: Create systemd service for auto suspend
|
||||
- name: Create log file
|
||||
become: true
|
||||
ansible.builtin.file:
|
||||
path: "{{ auto_suspend_log_file }}"
|
||||
state: touch
|
||||
owner: root
|
||||
group: root
|
||||
mode: "0644"
|
||||
|
||||
- name: Create auto-suspend Python script
|
||||
become: true
|
||||
ansible.builtin.copy:
|
||||
src: auto_suspend.py
|
||||
dest: /usr/local/bin/auto_suspend.py
|
||||
mode: "0755"
|
||||
owner: root
|
||||
group: root
|
||||
|
||||
- name: Set SELinux file context on script
|
||||
become: true
|
||||
community.general.sefcontext:
|
||||
path: /usr/local/bin/auto_suspend.py
|
||||
setype: bin_t
|
||||
state: present
|
||||
ignore_errors: true
|
||||
|
||||
- name: Restore SELinux context on script
|
||||
become: true
|
||||
ansible.builtin.command: restorecon -v /usr/local/bin/auto_suspend.py
|
||||
changed_when: true
|
||||
failed_when: false
|
||||
ignore_errors: true
|
||||
|
||||
- name: Create systemd service for auto-suspend
|
||||
become: true
|
||||
ansible.builtin.copy:
|
||||
dest: /etc/systemd/system/auto-suspend.service
|
||||
@@ -73,10 +101,22 @@
|
||||
|
||||
[Service]
|
||||
Type=oneshot
|
||||
ExecStart=/usr/local/bin/auto_suspend_script.sh
|
||||
ExecStart=/usr/local/bin/auto_suspend.py \
|
||||
--alpha {{ auto_suspend_alpha }} \
|
||||
--min-samples {{ auto_suspend_min_samples }} \
|
||||
--idle-timeout-cycles {{ auto_suspend_idle_timeout_cycles }} \
|
||||
--cpu-idle-threshold {{ auto_suspend_cpu_idle_threshold }} \
|
||||
--gpu-util-threshold {{ auto_suspend_gpu_util_threshold }} \
|
||||
--state-dir {{ auto_suspend_state_dir }} \
|
||||
--log-file {{ auto_suspend_log_file }}
|
||||
Restart=on-failure
|
||||
RestartSec=30
|
||||
|
||||
[Install]
|
||||
WantedBy=multi-user.target
|
||||
owner: root
|
||||
group: root
|
||||
mode: '0644'
|
||||
mode: "0644"
|
||||
|
||||
- name: Create sleep.conf.d folder if it doesn't exist
|
||||
become: true
|
||||
@@ -85,7 +125,7 @@
|
||||
state: directory
|
||||
owner: root
|
||||
group: root
|
||||
mode: '0755'
|
||||
mode: "0755"
|
||||
|
||||
- name: Create sleep.conf.d for deep sleep
|
||||
become: true
|
||||
@@ -96,26 +136,26 @@
|
||||
MemorySleepMode=deep
|
||||
owner: root
|
||||
group: root
|
||||
mode: '0644'
|
||||
mode: "0644"
|
||||
|
||||
- name: Create systemd timer for auto suspend
|
||||
- name: Create systemd timer for auto-suspend
|
||||
become: true
|
||||
ansible.builtin.copy:
|
||||
dest: /etc/systemd/system/auto-suspend.timer
|
||||
content: |
|
||||
[Unit]
|
||||
Description=Run idle suspend check every 10 minutes
|
||||
Description=Run idle suspend check every 5 minutes
|
||||
|
||||
[Timer]
|
||||
OnBootSec=10min
|
||||
OnUnitActiveSec=10min
|
||||
OnBootSec=5min
|
||||
OnUnitActiveSec=5min
|
||||
Unit=auto-suspend.service
|
||||
|
||||
[Install]
|
||||
WantedBy=timers.target
|
||||
owner: root
|
||||
group: root
|
||||
mode: '0644'
|
||||
mode: "0644"
|
||||
|
||||
- name: Reload systemd daemon
|
||||
become: true
|
||||
@@ -128,3 +168,8 @@
|
||||
name: auto-suspend.timer
|
||||
enabled: true
|
||||
state: started
|
||||
|
||||
- name: Verify script is executable
|
||||
become: true
|
||||
ansible.builtin.command: /usr/local/bin/auto_suspend.py --dry-run
|
||||
changed_when: false
|
||||
|
||||
Reference in New Issue
Block a user