672 lines
22 KiB
Python
672 lines
22 KiB
Python
"""
|
|
Unit tests for ScheduleService class.
|
|
|
|
Tests schedule lifecycle operations: create, get, list, update, delete, and
|
|
cron expression validation.
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime, timedelta
|
|
|
|
from web.models import Schedule, Scan
|
|
from web.services.schedule_service import ScheduleService
|
|
|
|
|
|
class TestScheduleServiceCreate:
|
|
"""Tests for creating schedules."""
|
|
|
|
def test_create_schedule_valid(self, test_db, sample_config_file):
|
|
"""Test creating a schedule with valid parameters."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Daily Scan',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Verify schedule created
|
|
assert schedule_id is not None
|
|
assert isinstance(schedule_id, int)
|
|
|
|
# Verify schedule in database
|
|
schedule = test_db.query(Schedule).filter(Schedule.id == schedule_id).first()
|
|
assert schedule is not None
|
|
assert schedule.name == 'Daily Scan'
|
|
assert schedule.config_file == sample_config_file
|
|
assert schedule.cron_expression == '0 2 * * *'
|
|
assert schedule.enabled is True
|
|
assert schedule.next_run is not None
|
|
assert schedule.last_run is None
|
|
|
|
def test_create_schedule_disabled(self, test_db, sample_config_file):
|
|
"""Test creating a disabled schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Disabled Scan',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 3 * * *',
|
|
enabled=False
|
|
)
|
|
|
|
schedule = test_db.query(Schedule).filter(Schedule.id == schedule_id).first()
|
|
assert schedule.enabled is False
|
|
assert schedule.next_run is None
|
|
|
|
def test_create_schedule_invalid_cron(self, test_db, sample_config_file):
|
|
"""Test creating a schedule with invalid cron expression."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Invalid cron expression"):
|
|
service.create_schedule(
|
|
name='Invalid Schedule',
|
|
config_file=sample_config_file,
|
|
cron_expression='invalid cron',
|
|
enabled=True
|
|
)
|
|
|
|
def test_create_schedule_nonexistent_config(self, test_db):
|
|
"""Test creating a schedule with nonexistent config file."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Config file not found"):
|
|
service.create_schedule(
|
|
name='Bad Config',
|
|
config_file='/nonexistent/config.yaml',
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
def test_create_schedule_various_cron_expressions(self, test_db, sample_config_file):
|
|
"""Test creating schedules with various valid cron expressions."""
|
|
service = ScheduleService(test_db)
|
|
|
|
cron_expressions = [
|
|
'0 0 * * *', # Daily at midnight
|
|
'*/15 * * * *', # Every 15 minutes
|
|
'0 2 * * 0', # Weekly on Sunday at 2 AM
|
|
'0 0 1 * *', # Monthly on the 1st at midnight
|
|
'30 14 * * 1-5', # Weekdays at 2:30 PM
|
|
]
|
|
|
|
for i, cron in enumerate(cron_expressions):
|
|
schedule_id = service.create_schedule(
|
|
name=f'Schedule {i}',
|
|
config_file=sample_config_file,
|
|
cron_expression=cron,
|
|
enabled=True
|
|
)
|
|
assert schedule_id is not None
|
|
|
|
|
|
class TestScheduleServiceGet:
|
|
"""Tests for retrieving schedules."""
|
|
|
|
def test_get_schedule_not_found(self, test_db):
|
|
"""Test getting a nonexistent schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Schedule .* not found"):
|
|
service.get_schedule(999)
|
|
|
|
def test_get_schedule_found(self, test_db, sample_config_file):
|
|
"""Test getting an existing schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Create a schedule
|
|
schedule_id = service.create_schedule(
|
|
name='Test Schedule',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Retrieve it
|
|
result = service.get_schedule(schedule_id)
|
|
|
|
assert result is not None
|
|
assert result['id'] == schedule_id
|
|
assert result['name'] == 'Test Schedule'
|
|
assert result['cron_expression'] == '0 2 * * *'
|
|
assert result['enabled'] is True
|
|
assert 'history' in result
|
|
assert isinstance(result['history'], list)
|
|
|
|
def test_get_schedule_with_history(self, test_db, sample_config_file):
|
|
"""Test getting schedule includes execution history."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Create schedule
|
|
schedule_id = service.create_schedule(
|
|
name='Test Schedule',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Create associated scans
|
|
for i in range(3):
|
|
scan = Scan(
|
|
timestamp=datetime.utcnow() - timedelta(days=i),
|
|
status='completed',
|
|
config_file=sample_config_file,
|
|
title=f'Scan {i}',
|
|
triggered_by='scheduled',
|
|
schedule_id=schedule_id
|
|
)
|
|
test_db.add(scan)
|
|
test_db.commit()
|
|
|
|
# Get schedule
|
|
result = service.get_schedule(schedule_id)
|
|
|
|
assert len(result['history']) == 3
|
|
assert result['history'][0]['title'] == 'Scan 0' # Most recent first
|
|
|
|
|
|
class TestScheduleServiceList:
|
|
"""Tests for listing schedules."""
|
|
|
|
def test_list_schedules_empty(self, test_db):
|
|
"""Test listing schedules when database is empty."""
|
|
service = ScheduleService(test_db)
|
|
|
|
result = service.list_schedules(page=1, per_page=20)
|
|
|
|
assert result['total'] == 0
|
|
assert len(result['schedules']) == 0
|
|
assert result['page'] == 1
|
|
assert result['per_page'] == 20
|
|
|
|
def test_list_schedules_populated(self, test_db, sample_config_file):
|
|
"""Test listing schedules with data."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Create multiple schedules
|
|
for i in range(5):
|
|
service.create_schedule(
|
|
name=f'Schedule {i}',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
result = service.list_schedules(page=1, per_page=20)
|
|
|
|
assert result['total'] == 5
|
|
assert len(result['schedules']) == 5
|
|
assert all('name' in s for s in result['schedules'])
|
|
|
|
def test_list_schedules_pagination(self, test_db, sample_config_file):
|
|
"""Test schedule pagination."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Create 25 schedules
|
|
for i in range(25):
|
|
service.create_schedule(
|
|
name=f'Schedule {i:02d}',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Get first page
|
|
result_page1 = service.list_schedules(page=1, per_page=10)
|
|
assert len(result_page1['schedules']) == 10
|
|
assert result_page1['total'] == 25
|
|
assert result_page1['pages'] == 3
|
|
|
|
# Get second page
|
|
result_page2 = service.list_schedules(page=2, per_page=10)
|
|
assert len(result_page2['schedules']) == 10
|
|
|
|
# Get third page
|
|
result_page3 = service.list_schedules(page=3, per_page=10)
|
|
assert len(result_page3['schedules']) == 5
|
|
|
|
def test_list_schedules_filter_enabled(self, test_db, sample_config_file):
|
|
"""Test filtering schedules by enabled status."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Create enabled and disabled schedules
|
|
for i in range(3):
|
|
service.create_schedule(
|
|
name=f'Enabled {i}',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
for i in range(2):
|
|
service.create_schedule(
|
|
name=f'Disabled {i}',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=False
|
|
)
|
|
|
|
# Filter enabled only
|
|
result_enabled = service.list_schedules(enabled_filter=True)
|
|
assert result_enabled['total'] == 3
|
|
|
|
# Filter disabled only
|
|
result_disabled = service.list_schedules(enabled_filter=False)
|
|
assert result_disabled['total'] == 2
|
|
|
|
# No filter
|
|
result_all = service.list_schedules(enabled_filter=None)
|
|
assert result_all['total'] == 5
|
|
|
|
|
|
class TestScheduleServiceUpdate:
|
|
"""Tests for updating schedules."""
|
|
|
|
def test_update_schedule_name(self, test_db, sample_config_file):
|
|
"""Test updating schedule name."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Old Name',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
result = service.update_schedule(schedule_id, name='New Name')
|
|
|
|
assert result['name'] == 'New Name'
|
|
assert result['cron_expression'] == '0 2 * * *'
|
|
|
|
def test_update_schedule_cron(self, test_db, sample_config_file):
|
|
"""Test updating cron expression recalculates next_run."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
original = service.get_schedule(schedule_id)
|
|
original_next_run = original['next_run']
|
|
|
|
# Update cron expression
|
|
result = service.update_schedule(
|
|
schedule_id,
|
|
cron_expression='0 3 * * *'
|
|
)
|
|
|
|
# Next run should be recalculated
|
|
assert result['cron_expression'] == '0 3 * * *'
|
|
assert result['next_run'] != original_next_run
|
|
|
|
def test_update_schedule_invalid_cron(self, test_db, sample_config_file):
|
|
"""Test updating with invalid cron expression fails."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Invalid cron expression"):
|
|
service.update_schedule(schedule_id, cron_expression='invalid')
|
|
|
|
def test_update_schedule_not_found(self, test_db):
|
|
"""Test updating nonexistent schedule fails."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Schedule .* not found"):
|
|
service.update_schedule(999, name='New Name')
|
|
|
|
def test_update_schedule_invalid_config_file(self, test_db, sample_config_file):
|
|
"""Test updating with nonexistent config file fails."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="Config file not found"):
|
|
service.update_schedule(schedule_id, config_file='/nonexistent.yaml')
|
|
|
|
|
|
class TestScheduleServiceDelete:
|
|
"""Tests for deleting schedules."""
|
|
|
|
def test_delete_schedule(self, test_db, sample_config_file):
|
|
"""Test deleting a schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='To Delete',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Verify exists
|
|
assert test_db.query(Schedule).filter(Schedule.id == schedule_id).first() is not None
|
|
|
|
# Delete
|
|
result = service.delete_schedule(schedule_id)
|
|
assert result is True
|
|
|
|
# Verify deleted
|
|
assert test_db.query(Schedule).filter(Schedule.id == schedule_id).first() is None
|
|
|
|
def test_delete_schedule_not_found(self, test_db):
|
|
"""Test deleting nonexistent schedule fails."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Schedule .* not found"):
|
|
service.delete_schedule(999)
|
|
|
|
def test_delete_schedule_preserves_scans(self, test_db, sample_config_file):
|
|
"""Test that deleting schedule preserves associated scans."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Create schedule
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Create associated scan
|
|
scan = Scan(
|
|
timestamp=datetime.utcnow(),
|
|
status='completed',
|
|
config_file=sample_config_file,
|
|
title='Test Scan',
|
|
triggered_by='scheduled',
|
|
schedule_id=schedule_id
|
|
)
|
|
test_db.add(scan)
|
|
test_db.commit()
|
|
scan_id = scan.id
|
|
|
|
# Delete schedule
|
|
service.delete_schedule(schedule_id)
|
|
|
|
# Verify scan still exists (schedule_id becomes null)
|
|
remaining_scan = test_db.query(Scan).filter(Scan.id == scan_id).first()
|
|
assert remaining_scan is not None
|
|
assert remaining_scan.schedule_id is None
|
|
|
|
|
|
class TestScheduleServiceToggle:
|
|
"""Tests for toggling schedule enabled status."""
|
|
|
|
def test_toggle_enabled_to_disabled(self, test_db, sample_config_file):
|
|
"""Test disabling an enabled schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
result = service.toggle_enabled(schedule_id, enabled=False)
|
|
|
|
assert result['enabled'] is False
|
|
assert result['next_run'] is None
|
|
|
|
def test_toggle_disabled_to_enabled(self, test_db, sample_config_file):
|
|
"""Test enabling a disabled schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=False
|
|
)
|
|
|
|
result = service.toggle_enabled(schedule_id, enabled=True)
|
|
|
|
assert result['enabled'] is True
|
|
assert result['next_run'] is not None
|
|
|
|
|
|
class TestScheduleServiceRunTimes:
|
|
"""Tests for updating run times."""
|
|
|
|
def test_update_run_times(self, test_db, sample_config_file):
|
|
"""Test updating last_run and next_run."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
last_run = datetime.utcnow()
|
|
next_run = datetime.utcnow() + timedelta(days=1)
|
|
|
|
result = service.update_run_times(schedule_id, last_run, next_run)
|
|
assert result is True
|
|
|
|
schedule = service.get_schedule(schedule_id)
|
|
assert schedule['last_run'] is not None
|
|
assert schedule['next_run'] is not None
|
|
|
|
def test_update_run_times_not_found(self, test_db):
|
|
"""Test updating run times for nonexistent schedule."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Schedule .* not found"):
|
|
service.update_run_times(
|
|
999,
|
|
datetime.utcnow(),
|
|
datetime.utcnow() + timedelta(days=1)
|
|
)
|
|
|
|
|
|
class TestCronValidation:
|
|
"""Tests for cron expression validation."""
|
|
|
|
def test_validate_cron_valid_expressions(self, test_db):
|
|
"""Test validating various valid cron expressions."""
|
|
service = ScheduleService(test_db)
|
|
|
|
valid_expressions = [
|
|
'0 0 * * *', # Daily at midnight
|
|
'*/15 * * * *', # Every 15 minutes
|
|
'0 2 * * 0', # Weekly on Sunday
|
|
'0 0 1 * *', # Monthly
|
|
'30 14 * * 1-5', # Weekdays
|
|
'0 */4 * * *', # Every 4 hours
|
|
]
|
|
|
|
for expr in valid_expressions:
|
|
is_valid, error = service.validate_cron_expression(expr)
|
|
assert is_valid is True, f"Expression '{expr}' should be valid"
|
|
assert error is None
|
|
|
|
def test_validate_cron_invalid_expressions(self, test_db):
|
|
"""Test validating invalid cron expressions."""
|
|
service = ScheduleService(test_db)
|
|
|
|
invalid_expressions = [
|
|
'invalid',
|
|
'60 0 * * *', # Invalid minute (0-59)
|
|
'0 24 * * *', # Invalid hour (0-23)
|
|
'0 0 32 * *', # Invalid day (1-31)
|
|
'0 0 * 13 *', # Invalid month (1-12)
|
|
'0 0 * * 7', # Invalid weekday (0-6)
|
|
]
|
|
|
|
for expr in invalid_expressions:
|
|
is_valid, error = service.validate_cron_expression(expr)
|
|
assert is_valid is False, f"Expression '{expr}' should be invalid"
|
|
assert error is not None
|
|
|
|
|
|
class TestNextRunCalculation:
|
|
"""Tests for next run time calculation."""
|
|
|
|
def test_calculate_next_run(self, test_db):
|
|
"""Test calculating next run time."""
|
|
service = ScheduleService(test_db)
|
|
|
|
# Daily at 2 AM
|
|
next_run = service.calculate_next_run('0 2 * * *')
|
|
|
|
assert next_run is not None
|
|
assert isinstance(next_run, datetime)
|
|
assert next_run > datetime.utcnow()
|
|
|
|
def test_calculate_next_run_from_time(self, test_db):
|
|
"""Test calculating next run from specific time."""
|
|
service = ScheduleService(test_db)
|
|
|
|
base_time = datetime(2025, 1, 1, 0, 0, 0)
|
|
next_run = service.calculate_next_run('0 2 * * *', from_time=base_time)
|
|
|
|
# Should be 2 AM on same day
|
|
assert next_run.hour == 2
|
|
assert next_run.minute == 0
|
|
|
|
def test_calculate_next_run_invalid_cron(self, test_db):
|
|
"""Test calculating next run with invalid cron raises error."""
|
|
service = ScheduleService(test_db)
|
|
|
|
with pytest.raises(ValueError, match="Invalid cron expression"):
|
|
service.calculate_next_run('invalid cron')
|
|
|
|
|
|
class TestScheduleHistory:
|
|
"""Tests for schedule execution history."""
|
|
|
|
def test_get_schedule_history_empty(self, test_db, sample_config_file):
|
|
"""Test getting history for schedule with no executions."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
history = service.get_schedule_history(schedule_id)
|
|
assert len(history) == 0
|
|
|
|
def test_get_schedule_history_with_scans(self, test_db, sample_config_file):
|
|
"""Test getting history with multiple scans."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Create 15 scans
|
|
for i in range(15):
|
|
scan = Scan(
|
|
timestamp=datetime.utcnow() - timedelta(days=i),
|
|
status='completed',
|
|
config_file=sample_config_file,
|
|
title=f'Scan {i}',
|
|
triggered_by='scheduled',
|
|
schedule_id=schedule_id
|
|
)
|
|
test_db.add(scan)
|
|
test_db.commit()
|
|
|
|
# Get history (default limit 10)
|
|
history = service.get_schedule_history(schedule_id, limit=10)
|
|
assert len(history) == 10
|
|
assert history[0]['title'] == 'Scan 0' # Most recent first
|
|
|
|
def test_get_schedule_history_custom_limit(self, test_db, sample_config_file):
|
|
"""Test getting history with custom limit."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
# Create 10 scans
|
|
for i in range(10):
|
|
scan = Scan(
|
|
timestamp=datetime.utcnow() - timedelta(days=i),
|
|
status='completed',
|
|
config_file=sample_config_file,
|
|
title=f'Scan {i}',
|
|
triggered_by='scheduled',
|
|
schedule_id=schedule_id
|
|
)
|
|
test_db.add(scan)
|
|
test_db.commit()
|
|
|
|
# Get only 5
|
|
history = service.get_schedule_history(schedule_id, limit=5)
|
|
assert len(history) == 5
|
|
|
|
|
|
class TestScheduleSerialization:
|
|
"""Tests for schedule serialization."""
|
|
|
|
def test_schedule_to_dict(self, test_db, sample_config_file):
|
|
"""Test converting schedule to dictionary."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test Schedule',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
result = service.get_schedule(schedule_id)
|
|
|
|
# Verify all required fields
|
|
assert 'id' in result
|
|
assert 'name' in result
|
|
assert 'config_file' in result
|
|
assert 'cron_expression' in result
|
|
assert 'enabled' in result
|
|
assert 'last_run' in result
|
|
assert 'next_run' in result
|
|
assert 'next_run_relative' in result
|
|
assert 'created_at' in result
|
|
assert 'updated_at' in result
|
|
assert 'history' in result
|
|
|
|
def test_schedule_relative_time_formatting(self, test_db, sample_config_file):
|
|
"""Test relative time formatting in schedule dict."""
|
|
service = ScheduleService(test_db)
|
|
|
|
schedule_id = service.create_schedule(
|
|
name='Test',
|
|
config_file=sample_config_file,
|
|
cron_expression='0 2 * * *',
|
|
enabled=True
|
|
)
|
|
|
|
result = service.get_schedule(schedule_id)
|
|
|
|
# Should have relative time for next_run
|
|
assert result['next_run_relative'] is not None
|
|
assert isinstance(result['next_run_relative'], str)
|
|
assert 'in' in result['next_run_relative'].lower()
|