From de561b88b19e36a8049fbffe57b6671f596fb57d Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 8 Jul 2026 14:14:50 +0200 Subject: [PATCH] =?UTF-8?q?tools:=20pi=5Fwake=5Fagent.py=20=E2=80=94=20Reu?= =?UTF-8?q?sable=20multi-agent=20wake=20timer=20with=20self-cron/daemon/su?= =?UTF-8?q?ccession?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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 --- test_pi_wake_agent.py | 304 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 304 insertions(+) create mode 100644 test_pi_wake_agent.py diff --git a/test_pi_wake_agent.py b/test_pi_wake_agent.py new file mode 100644 index 00000000..2d69dd27 --- /dev/null +++ b/test_pi_wake_agent.py @@ -0,0 +1,304 @@ +#!/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"])