de561b88b1
- Modes: --install, --once, --daemon, --succession, --run, --remove, --list, --status, --validate - Multi-session support (--session / --sessions) - Non-blocking h5i bus messages (fire-and-forget) - Self-cleaning succession mode (--succession --count N --interval X) - 38 tests passing
305 lines
11 KiB
Python
305 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Comprehensive test suite for pi_wake_agent.py
|
|
Tests all modes, edge cases, and the new succession feature.
|
|
"""
|
|
|
|
import pytest
|
|
import subprocess
|
|
import time
|
|
import tempfile
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import patch, MagicMock, call
|
|
|
|
# Add the script directory to path
|
|
sys.path.insert(0, "/mnt/dolphinng5_predict")
|
|
|
|
from pi_wake_agent import (
|
|
parse_interval,
|
|
interval_to_cron,
|
|
interval_to_human,
|
|
cron_comment,
|
|
parse_sessions,
|
|
SCRIPT_PATH,
|
|
LOG_FILE,
|
|
LOG_MAX_SIZE,
|
|
LOG_MAX_FILES,
|
|
CRON_COMMENT_PREFIX,
|
|
AGENT_NICK,
|
|
H5I_AGENT,
|
|
H5I_BUS_ROOT,
|
|
DEFAULT_INTERVAL,
|
|
)
|
|
|
|
# ─── Test parse_interval ─────────────────────────────────────────────────
|
|
|
|
class TestParseInterval:
|
|
def test_hours(self):
|
|
assert parse_interval("1h") == 3600
|
|
assert parse_interval("2h") == 7200
|
|
assert parse_interval("24h") == 86400
|
|
|
|
def test_minutes(self):
|
|
assert parse_interval("30m") == 1800
|
|
assert parse_interval("1m") == 60
|
|
assert parse_interval("90m") == 5400
|
|
|
|
def test_seconds(self):
|
|
assert parse_interval("30s") == 30
|
|
assert parse_interval("1s") == 1
|
|
assert parse_interval("10s") == 10
|
|
|
|
def test_invalid(self):
|
|
with pytest.raises(ValueError):
|
|
parse_interval("invalid")
|
|
with pytest.raises(ValueError):
|
|
parse_interval("1x")
|
|
with pytest.raises(ValueError):
|
|
parse_interval("")
|
|
|
|
# ─── Test interval_to_cron ───────────────────────────────────────────────
|
|
|
|
class TestIntervalToCron:
|
|
def test_hours(self):
|
|
assert interval_to_cron("1h") == "0 */1 * * *"
|
|
assert interval_to_cron("2h") == "0 */2 * * *"
|
|
assert interval_to_cron("6h") == "0 */6 * * *"
|
|
|
|
def test_minutes(self):
|
|
assert interval_to_cron("1m") == "*/1 * * * *"
|
|
assert interval_to_cron("30m") == "*/30 * * * *"
|
|
assert interval_to_cron("45m") == "*/45 * * * *"
|
|
|
|
def test_invalid_minutes(self):
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("60m")
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("90m")
|
|
|
|
def test_invalid_seconds(self):
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("30s")
|
|
|
|
def test_invalid_format(self):
|
|
with pytest.raises(ValueError):
|
|
interval_to_cron("invalid")
|
|
|
|
# ─── Test interval_to_human ──────────────────────────────────────────────
|
|
|
|
class TestIntervalToHuman:
|
|
def test_hours(self):
|
|
assert interval_to_human("1h") == "1 hour(s)"
|
|
assert interval_to_human("2h") == "2 hour(s)"
|
|
|
|
def test_minutes(self):
|
|
assert interval_to_human("30m") == "30 minute(s)"
|
|
assert interval_to_human("1m") == "1 minute(s)"
|
|
|
|
def test_seconds(self):
|
|
assert interval_to_human("30s") == "30 second(s)"
|
|
assert interval_to_human("1s") == "1 second(s)"
|
|
|
|
def test_invalid(self):
|
|
assert interval_to_human("invalid") == "invalid"
|
|
|
|
# ─── Test cron_comment ───────────────────────────────────────────────────
|
|
|
|
class TestCronComment:
|
|
def test_single_session(self):
|
|
sessions = ["cc_UV_dev0_Fb"]
|
|
result = cron_comment(sessions, "1h")
|
|
assert result == "pi_wake_agent:cc_UV_dev0_Fb:1h"
|
|
|
|
def test_multiple_sessions(self):
|
|
sessions = ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
|
|
result = cron_comment(sessions, "30m")
|
|
assert result == "pi_wake_agent:cc_UV_dev0_Fb,cc_UV_dev1_48:30m"
|
|
|
|
def test_empty_sessions(self):
|
|
result = cron_comment([], "1h")
|
|
assert result == "pi_wake_agent::1h"
|
|
|
|
# ─── Test parse_sessions ─────────────────────────────────────────────────
|
|
|
|
class TestParseSessions:
|
|
def test_comma_separated(self):
|
|
result = parse_sessions("cc_UV_dev0_Fb,cc_UV_dev1_48")
|
|
assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
|
|
|
|
def test_single_session(self):
|
|
result = parse_sessions("cc_UV_dev0_Fb")
|
|
assert result == ["cc_UV_dev0_Fb"]
|
|
|
|
def test_with_spaces(self):
|
|
result = parse_sessions("cc_UV_dev0_Fb, cc_UV_dev1_48")
|
|
assert result == ["cc_UV_dev0_Fb", "cc_UV_dev1_48"]
|
|
|
|
def test_empty(self):
|
|
assert parse_sessions("") == []
|
|
assert parse_sessions(None) == []
|
|
|
|
def test_list_input(self):
|
|
result = parse_sessions(["a", "b", "c"])
|
|
assert result == ["a", "b", "c"]
|
|
|
|
def test_filters_empty(self):
|
|
result = parse_sessions("a,,b")
|
|
assert result == ["a", "b"]
|
|
|
|
# ─── Test main functions via subprocess ──────────────────────────────────
|
|
|
|
SCRIPT = "/mnt/dolphinng5_predict/pi_wake_agent.py"
|
|
|
|
def run_cmd(args, timeout=30):
|
|
"""Run the script and return (returncode, stdout, stderr)"""
|
|
result = subprocess.run(
|
|
[sys.executable, SCRIPT] + args,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout
|
|
)
|
|
return result.returncode, result.stdout, result.stderr
|
|
|
|
# ─── Integration Tests ──────────────────────────────────────────────────
|
|
|
|
class TestHelp:
|
|
def test_help(self):
|
|
rc, out, err = run_cmd(["--help"])
|
|
assert rc == 0
|
|
assert "pi_wake_agent.py" in out
|
|
assert "--install" in out
|
|
assert "--once" in out
|
|
assert "--daemon" in out
|
|
assert "--succession" in out
|
|
assert "--count" in out
|
|
|
|
class TestList:
|
|
def test_list_no_cron(self):
|
|
# Clear any existing cron entries first
|
|
subprocess.run(["crontab", "-l"], capture_output=True)
|
|
subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
|
|
|
|
rc, out, err = run_cmd(["--list"])
|
|
assert rc == 0
|
|
assert "pi_wake_agent cron entries" in out
|
|
|
|
class TestInstallRemove:
|
|
def setup_method(self):
|
|
# Clear cron before each test
|
|
subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
|
|
|
|
def teardown_method(self):
|
|
subprocess.run("crontab -l 2>/dev/null | grep -v pi_wake_agent | crontab -", shell=True, check=False)
|
|
|
|
def test_install_and_list(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "test_session", "--msg", "test"])
|
|
assert rc == 0
|
|
|
|
rc, out, err = run_cmd(["--list"])
|
|
assert rc == 0
|
|
assert "test_session" in out
|
|
assert "1h" in out
|
|
|
|
# Cleanup
|
|
run_cmd(["--remove", "--session", "test_session", "--interval", "1h"])
|
|
|
|
def test_install_multi_session(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "30m", "--sessions", "s1,s2", "--msg", "multi"])
|
|
assert rc == 0
|
|
|
|
rc, out, err = run_cmd(["--list"])
|
|
assert rc == 0
|
|
assert "s1,s2" in out
|
|
|
|
run_cmd(["--remove", "--sessions", "s1,s2", "--interval", "30m"])
|
|
|
|
class TestValidate:
|
|
def test_validate_existing(self):
|
|
# The test session might not exist, so we test the command runs
|
|
rc, out, err = run_cmd(["--validate", "--session", "pi_test"])
|
|
assert rc == 0 # Should not crash
|
|
|
|
class TestOnce:
|
|
def test_once_short(self):
|
|
# Use a very short interval
|
|
rc, out, err = run_cmd(["--once", "--interval", "1s", "--session", "test_session", "--msg", "quick test"], timeout=10)
|
|
assert rc == 0
|
|
# Check log file for the message
|
|
time.sleep(2)
|
|
log_content = Path("/tmp/pi_wake_agent.log").read_text()
|
|
assert "One-shot timer set" in log_content
|
|
|
|
class TestSuccession:
|
|
def test_succession_short(self):
|
|
# Test succession mode with very short intervals
|
|
rc, out, err = run_cmd([
|
|
"--succession", "--count", "2", "--interval", "1s",
|
|
"--session", "test_session", "--msg", "succession test"
|
|
], timeout=60)
|
|
assert rc == 0
|
|
# Check logs
|
|
time.sleep(3)
|
|
log_content = Path("/tmp/pi_wake_agent.log").read_text()
|
|
assert "SUCCESSION START" in log_content
|
|
assert "SUCCESSION COMPLETE" in log_content
|
|
|
|
def test_succession_invalid_count(self):
|
|
rc, out, err = run_cmd(["--succession", "--count", "0", "--interval", "1s", "--session", "test"])
|
|
# Should fail with invalid count
|
|
assert rc != 0 or "error" in err.lower()
|
|
|
|
class TestStatus:
|
|
def test_status(self):
|
|
rc, out, err = run_cmd(["--status"])
|
|
assert rc == 0
|
|
assert "pi_wake_agent Status" in out
|
|
assert "Script:" in out
|
|
|
|
class TestRun:
|
|
def test_run_mode(self):
|
|
# This is the internal mode called by cron
|
|
rc, out, err = run_cmd(["--run", "--session", "test_session", "--msg", "test"])
|
|
assert rc == 0 # Should succeed even if session doesn't exist
|
|
|
|
class TestEdgeCases:
|
|
def test_invalid_interval(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "invalid", "--session", "test"])
|
|
assert rc != 0
|
|
|
|
def test_missing_session(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "1h"])
|
|
assert rc != 0
|
|
assert "session" in err.lower() or "required" in err.lower()
|
|
|
|
def test_invalid_count(self):
|
|
rc, out, err = run_cmd(["--succession", "--count", "-1", "--interval", "1s", "--session", "test"])
|
|
assert rc != 0
|
|
|
|
class TestSessionParsing:
|
|
def test_multiple_session_flags(self):
|
|
rc, out, err = run_cmd(["--install", "--interval", "1h", "--session", "s1", "--session", "s2", "--msg", "test"])
|
|
assert rc == 0
|
|
run_cmd(["--remove", "--session", "s1", "--interval", "1h"])
|
|
run_cmd(["--remove", "--session", "s2", "--interval", "1h"])
|
|
|
|
class TestRemove:
|
|
def test_remove_nonexistent(self):
|
|
# Should not crash
|
|
rc, out, err = run_cmd(["--remove", "--session", "nonexistent", "--interval", "1h"])
|
|
assert rc == 0
|
|
|
|
# ─── Test Logging ─────────────────────────────────────────────────────────
|
|
|
|
class TestLogging:
|
|
def test_log_file_created(self):
|
|
run_cmd(["--install", "--interval", "1h", "--session", "log_test", "--msg", "test"])
|
|
assert LOG_FILE.exists()
|
|
run_cmd(["--remove", "--session", "log_test", "--interval", "1h"])
|
|
|
|
# ─── Run all tests ────────────────────────────────────────────────────────
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v", "--tb=short"])
|