fdff0a791a
apply_schedule() was the actual behavioral fix in the prior commit but shipped without tests. Six new pytest cases now cover its five+ state combinations: - manual mode with an existing crontab + live crond pid -> crontab removed, SIGHUP sent - manual mode with no crontab and no pid file -> no-op (function returns cleanly) - real schedule with a live pid -> crontab written, SIGHUP sent - real schedule with a dead pid (ProcessLookupError) -> crontab written, fresh crond spawned, new pid file - real schedule with no pid file (e.g. container started in manual mode, user enables a schedule via UI) -> crontab written, fresh crond spawned - real schedule with a garbage pid file (non-integer contents) -> ValueError caught, fresh crond spawned os.kill and subprocess.Popen are mocked so no real signals fire and no real processes spawn during tests. CRONTAB_FILE / CROND_PID_FILE are redirected to tmp paths via monkeypatch. .gitignore: add __pycache__/, *.pyc, and .pytest_cache/ to prevent future contributors from accidentally committing test artifacts.
119 lines
3.8 KiB
Python
119 lines
3.8 KiB
Python
"""
|
|
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"
|