240 lines
7.6 KiB
Python
240 lines
7.6 KiB
Python
"""Tests for the sync daemon."""
|
|
|
|
import pytest
|
|
import os
|
|
import tempfile
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch, AsyncMock
|
|
|
|
from src.cli.sync_daemon import (
|
|
SyncDaemon,
|
|
create_daemon_config,
|
|
)
|
|
|
|
|
|
class TestCreateDaemonConfig:
|
|
"""Tests for create_daemon_config function."""
|
|
|
|
def test_default_config(self):
|
|
"""Test default configuration values."""
|
|
config = create_daemon_config()
|
|
|
|
assert config["dry_run"] is False
|
|
assert config["vdir"] == "~/Calendar"
|
|
assert config["icsfile"] is None
|
|
assert config["org"] == "corteva"
|
|
assert config["days_back"] == 1
|
|
assert config["days_forward"] == 30
|
|
assert config["continue_iteration"] is False
|
|
assert config["download_attachments"] is False
|
|
assert config["two_way_calendar"] is False
|
|
assert config["notify"] is False
|
|
assert config["sync_interval"] == 300
|
|
assert config["check_interval"] == 10
|
|
|
|
def test_custom_config(self):
|
|
"""Test custom configuration values."""
|
|
config = create_daemon_config(
|
|
dry_run=True,
|
|
org="mycompany",
|
|
vdir="~/MyCalendar",
|
|
notify=True,
|
|
sync_interval=600,
|
|
)
|
|
|
|
assert config["dry_run"] is True
|
|
assert config["org"] == "mycompany"
|
|
assert config["vdir"] == "~/MyCalendar"
|
|
assert config["notify"] is True
|
|
assert config["sync_interval"] == 600
|
|
|
|
def test_pid_file_default(self):
|
|
"""Test default PID file path."""
|
|
config = create_daemon_config()
|
|
|
|
assert "luk.pid" in config["pid_file"]
|
|
|
|
def test_log_file_default(self):
|
|
"""Test default log file path."""
|
|
config = create_daemon_config()
|
|
|
|
assert "luk.log" in config["log_file"]
|
|
|
|
|
|
class TestSyncDaemon:
|
|
"""Tests for the SyncDaemon class."""
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Create a temporary directory for tests."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
@pytest.fixture
|
|
def daemon_config(self, temp_dir):
|
|
"""Create a daemon config with temp paths."""
|
|
return create_daemon_config(
|
|
pid_file=str(temp_dir / "test_daemon.pid"),
|
|
log_file=str(temp_dir / "test_daemon.log"),
|
|
)
|
|
|
|
def test_init(self, daemon_config):
|
|
"""Test daemon initialization."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
assert daemon.config == daemon_config
|
|
assert daemon.running is False
|
|
assert daemon.sync_interval == 300
|
|
assert daemon.check_interval == 10
|
|
|
|
def test_init_custom_intervals(self, temp_dir):
|
|
"""Test daemon with custom intervals."""
|
|
config = create_daemon_config(
|
|
pid_file=str(temp_dir / "test_daemon.pid"),
|
|
log_file=str(temp_dir / "test_daemon.log"),
|
|
sync_interval=600,
|
|
check_interval=30,
|
|
)
|
|
|
|
daemon = SyncDaemon(config)
|
|
|
|
assert daemon.sync_interval == 600
|
|
assert daemon.check_interval == 30
|
|
|
|
def test_setup_logging(self, daemon_config, temp_dir):
|
|
"""Test logging setup."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
# Logger should be configured
|
|
assert daemon.logger is not None
|
|
assert daemon.logger.level == logging.INFO
|
|
|
|
# Log directory should be created
|
|
log_file = temp_dir / "test_daemon.log"
|
|
assert log_file.parent.exists()
|
|
|
|
def test_is_running_no_pid_file(self, daemon_config):
|
|
"""Test is_running when no PID file exists."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
assert daemon.is_running() is False
|
|
|
|
def test_is_running_stale_pid_file(self, daemon_config, temp_dir):
|
|
"""Test is_running with stale PID file."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
# Create a PID file with non-existent process
|
|
pid_file = temp_dir / "test_daemon.pid"
|
|
pid_file.write_text("99999999") # Very unlikely to exist
|
|
|
|
# Should detect stale PID and clean up
|
|
assert daemon.is_running() is False
|
|
assert not pid_file.exists()
|
|
|
|
def test_get_pid(self, daemon_config, temp_dir):
|
|
"""Test get_pid reads PID correctly."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
# Create a PID file
|
|
pid_file = temp_dir / "test_daemon.pid"
|
|
pid_file.write_text("12345")
|
|
|
|
assert daemon.get_pid() == 12345
|
|
|
|
def test_stop_not_running(self, daemon_config, capsys):
|
|
"""Test stop when daemon is not running."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
daemon.stop()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "not running" in captured.out.lower()
|
|
|
|
def test_status_not_running(self, daemon_config, capsys):
|
|
"""Test status when daemon is not running."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
daemon.status()
|
|
|
|
captured = capsys.readouterr()
|
|
assert "not running" in captured.out.lower()
|
|
|
|
|
|
class TestSyncDaemonAsync:
|
|
"""Tests for async daemon methods."""
|
|
|
|
@pytest.fixture
|
|
def temp_dir(self):
|
|
"""Create a temporary directory for tests."""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|
|
@pytest.fixture
|
|
def daemon_config(self, temp_dir):
|
|
"""Create a daemon config with temp paths."""
|
|
return create_daemon_config(
|
|
pid_file=str(temp_dir / "test_daemon.pid"),
|
|
log_file=str(temp_dir / "test_daemon.log"),
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_for_changes_no_changes(self, daemon_config):
|
|
"""Test _check_for_changes when no changes."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
with patch("src.cli.sync_daemon.should_run_godspeed_sync", return_value=False):
|
|
with patch("src.cli.sync_daemon.should_run_sweep", return_value=False):
|
|
result = await daemon._check_for_changes()
|
|
|
|
assert result is False
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_for_changes_godspeed_due(self, daemon_config):
|
|
"""Test _check_for_changes when godspeed sync is due."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
with patch("src.cli.sync_daemon.should_run_godspeed_sync", return_value=True):
|
|
with patch("src.cli.sync_daemon.should_run_sweep", return_value=False):
|
|
result = await daemon._check_for_changes()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_check_for_changes_sweep_due(self, daemon_config):
|
|
"""Test _check_for_changes when sweep is due."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
with patch("src.cli.sync_daemon.should_run_godspeed_sync", return_value=False):
|
|
with patch("src.cli.sync_daemon.should_run_sweep", return_value=True):
|
|
result = await daemon._check_for_changes()
|
|
|
|
assert result is True
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_sync_success(self, daemon_config):
|
|
"""Test _perform_sync success."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
with patch(
|
|
"src.cli.sync_daemon._sync_outlook_data", new_callable=AsyncMock
|
|
) as mock_sync:
|
|
await daemon._perform_sync()
|
|
|
|
mock_sync.assert_called_once()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_perform_sync_failure(self, daemon_config):
|
|
"""Test _perform_sync handles failure."""
|
|
daemon = SyncDaemon(daemon_config)
|
|
|
|
with patch(
|
|
"src.cli.sync_daemon._sync_outlook_data", new_callable=AsyncMock
|
|
) as mock_sync:
|
|
mock_sync.side_effect = Exception("Sync failed")
|
|
|
|
# Should not raise
|
|
await daemon._perform_sync()
|
|
|
|
# Error should be logged (we'd need to check the log file)
|