diff --git a/.gitignore b/.gitignore index ddca873..dfd323c 100644 --- a/.gitignore +++ b/.gitignore @@ -5,6 +5,9 @@ test/ downloads/ config/ +__pycache__/ +*.pyc +.pytest_cache/ # Editor .vscode/ diff --git a/tests/test_apply_schedule.py b/tests/test_apply_schedule.py new file mode 100644 index 0000000..49bb862 --- /dev/null +++ b/tests/test_apply_schedule.py @@ -0,0 +1,118 @@ +""" +Tests for app.apply_schedule(). + +The function rewrites the crontab file and reloads crond via SIGHUP, falling +back to spawning a fresh crond when the recorded pid is dead or absent. These +tests cover the six meaningful state combinations of (schedule, crontab present, +pid file present, pid alive). +""" + +import os +import signal +import sys +from unittest.mock import MagicMock + +import pytest + +# Make the app module importable without installing it. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import app as app_module # noqa: E402 + + +@pytest.fixture +def patched(monkeypatch, tmp_path): + """Redirect crontab/pid paths to tmp and swap os.kill + subprocess.Popen with mocks.""" + crontab = tmp_path / "crontabs" / "root" + pid_file = tmp_path / "crond.pid" + monkeypatch.setattr(app_module, "CRONTAB_FILE", str(crontab)) + monkeypatch.setattr(app_module, "CROND_PID_FILE", str(pid_file)) + + mock_kill = MagicMock() + monkeypatch.setattr(app_module.os, "kill", mock_kill) + + mock_popen = MagicMock() + mock_popen.return_value = MagicMock(pid=9999) + monkeypatch.setattr(app_module.subprocess, "Popen", mock_popen) + + return { + "crontab": crontab, + "pid_file": pid_file, + "kill": mock_kill, + "popen": mock_popen, + } + + +def test_manual_with_existing_crontab_and_live_pid(patched): + patched["crontab"].parent.mkdir(parents=True) + patched["crontab"].write_text("0 2 * * * /app/sync.sh\n") + patched["pid_file"].write_text("1234") + + app_module.apply_schedule("manual") + + assert not patched["crontab"].exists() + patched["kill"].assert_called_once_with(1234, signal.SIGHUP) + patched["popen"].assert_not_called() + + +def test_manual_with_no_crontab_and_no_pid_file(patched): + assert not patched["crontab"].exists() + assert not patched["pid_file"].exists() + + app_module.apply_schedule("manual") # should not raise + + assert not patched["crontab"].exists() + patched["kill"].assert_not_called() + patched["popen"].assert_not_called() + + +def test_real_schedule_with_live_pid(patched): + patched["pid_file"].write_text("4321") + + app_module.apply_schedule("0 4 * * *") + + expected_line = "0 4 * * * /app/sync.sh >> /var/log/youtube-sync.log 2>&1\n" + assert patched["crontab"].read_text() == expected_line + patched["kill"].assert_called_once_with(4321, signal.SIGHUP) + patched["popen"].assert_not_called() + + +def test_real_schedule_with_dead_pid(patched): + patched["pid_file"].write_text("4321") + patched["kill"].side_effect = ProcessLookupError() + + app_module.apply_schedule("0 4 * * *") + + expected_line = "0 4 * * * /app/sync.sh >> /var/log/youtube-sync.log 2>&1\n" + assert patched["crontab"].read_text() == expected_line + patched["kill"].assert_called_once_with(4321, signal.SIGHUP) + patched["popen"].assert_called_once_with( + ["crond", "-f", "-l", "2"], + stdout=app_module.subprocess.DEVNULL, + stderr=app_module.subprocess.DEVNULL, + ) + assert patched["pid_file"].read_text() == "9999" + + +def test_real_schedule_with_no_pid_file(patched): + assert not patched["pid_file"].exists() + + app_module.apply_schedule("0 4 * * *") + + expected_line = "0 4 * * * /app/sync.sh >> /var/log/youtube-sync.log 2>&1\n" + assert patched["crontab"].read_text() == expected_line + patched["kill"].assert_not_called() + patched["popen"].assert_called_once() + assert patched["pid_file"].read_text() == "9999" + + +def test_real_schedule_with_garbage_pid_file(patched): + patched["pid_file"].write_text("not-a-number") + + app_module.apply_schedule("0 4 * * *") + + expected_line = "0 4 * * * /app/sync.sh >> /var/log/youtube-sync.log 2>&1\n" + assert patched["crontab"].read_text() == expected_line + patched["kill"].assert_not_called() + patched["popen"].assert_called_once() + assert patched["pid_file"].read_text() == "9999"