""" 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, db, sample_db_config): """Test creating a schedule with valid parameters.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Daily Scan', config_id=sample_db_config.id, 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 = db.query(Schedule).filter(Schedule.id == schedule_id).first() assert schedule is not None assert schedule.name == 'Daily Scan' assert schedule.config_id == sample_db_config.id 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, db, sample_db_config): """Test creating a disabled schedule.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Disabled Scan', config_id=sample_db_config.id, cron_expression='0 3 * * *', enabled=False ) schedule = 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, db, sample_db_config): """Test creating a schedule with invalid cron expression.""" service = ScheduleService(db) with pytest.raises(ValueError, match="Invalid cron expression"): service.create_schedule( name='Invalid Schedule', config_id=sample_db_config.id, cron_expression='invalid cron', enabled=True ) def test_create_schedule_nonexistent_config(self, db): """Test creating a schedule with nonexistent config.""" service = ScheduleService(db) with pytest.raises(ValueError, match="not found"): service.create_schedule( name='Bad Config', config_id=99999, cron_expression='0 2 * * *', enabled=True ) def test_create_schedule_various_cron_expressions(self, db, sample_db_config): """Test creating schedules with various valid cron expressions.""" service = ScheduleService(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_id=sample_db_config.id, cron_expression=cron, enabled=True ) assert schedule_id is not None class TestScheduleServiceGet: """Tests for retrieving schedules.""" def test_get_schedule_not_found(self, db): """Test getting a nonexistent schedule.""" service = ScheduleService(db) with pytest.raises(ValueError, match="Schedule .* not found"): service.get_schedule(999) def test_get_schedule_found(self, db, sample_db_config): """Test getting an existing schedule.""" service = ScheduleService(db) # Create a schedule schedule_id = service.create_schedule( name='Test Schedule', config_id=sample_db_config.id, 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, db, sample_db_config): """Test getting schedule includes execution history.""" service = ScheduleService(db) # Create schedule schedule_id = service.create_schedule( name='Test Schedule', config_id=sample_db_config.id, 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_id=sample_db_config.id, title=f'Scan {i}', triggered_by='scheduled', schedule_id=schedule_id ) db.add(scan) 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, db): """Test listing schedules when database is empty.""" service = ScheduleService(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, db, sample_db_config): """Test listing schedules with data.""" service = ScheduleService(db) # Create multiple schedules for i in range(5): service.create_schedule( name=f'Schedule {i}', config_id=sample_db_config.id, 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, db, sample_db_config): """Test schedule pagination.""" service = ScheduleService(db) # Create 25 schedules for i in range(25): service.create_schedule( name=f'Schedule {i:02d}', config_id=sample_db_config.id, 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, db, sample_db_config): """Test filtering schedules by enabled status.""" service = ScheduleService(db) # Create enabled and disabled schedules for i in range(3): service.create_schedule( name=f'Enabled {i}', config_id=sample_db_config.id, cron_expression='0 2 * * *', enabled=True ) for i in range(2): service.create_schedule( name=f'Disabled {i}', config_id=sample_db_config.id, 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, db, sample_db_config): """Test updating schedule name.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Old Name', config_id=sample_db_config.id, 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, db, sample_db_config): """Test updating cron expression recalculates next_run.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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, db, sample_db_config): """Test updating with invalid cron expression fails.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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, db): """Test updating nonexistent schedule fails.""" service = ScheduleService(db) with pytest.raises(ValueError, match="Schedule .* not found"): service.update_schedule(999, name='New Name') def test_update_schedule_invalid_config_id(self, db, sample_db_config): """Test updating with nonexistent config ID fails.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, cron_expression='0 2 * * *', enabled=True ) with pytest.raises(ValueError, match="not found"): service.update_schedule(schedule_id, config_id=99999) class TestScheduleServiceDelete: """Tests for deleting schedules.""" def test_delete_schedule(self, db, sample_db_config): """Test deleting a schedule.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='To Delete', config_id=sample_db_config.id, cron_expression='0 2 * * *', enabled=True ) # Verify exists assert 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 db.query(Schedule).filter(Schedule.id == schedule_id).first() is None def test_delete_schedule_not_found(self, db): """Test deleting nonexistent schedule fails.""" service = ScheduleService(db) with pytest.raises(ValueError, match="Schedule .* not found"): service.delete_schedule(999) def test_delete_schedule_preserves_scans(self, db, sample_db_config): """Test that deleting schedule preserves associated scans.""" service = ScheduleService(db) # Create schedule schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, cron_expression='0 2 * * *', enabled=True ) # Create associated scan scan = Scan( timestamp=datetime.utcnow(), status='completed', config_id=sample_db_config.id, title='Test Scan', triggered_by='scheduled', schedule_id=schedule_id ) db.add(scan) db.commit() scan_id = scan.id # Delete schedule service.delete_schedule(schedule_id) # Verify scan still exists (schedule_id becomes null) remaining_scan = 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, db, sample_db_config): """Test disabling an enabled schedule.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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, db, sample_db_config): """Test enabling a disabled schedule.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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, db, sample_db_config): """Test updating last_run and next_run.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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, db): """Test updating run times for nonexistent schedule.""" service = ScheduleService(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, db): """Test validating various valid cron expressions.""" service = ScheduleService(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, db): """Test validating invalid cron expressions.""" service = ScheduleService(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, db): """Test calculating next run time.""" service = ScheduleService(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, db): """Test calculating next run from specific time.""" service = ScheduleService(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, db): """Test calculating next run with invalid cron raises error.""" service = ScheduleService(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, db, sample_db_config): """Test getting history for schedule with no executions.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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, db, sample_db_config): """Test getting history with multiple scans.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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_id=sample_db_config.id, title=f'Scan {i}', triggered_by='scheduled', schedule_id=schedule_id ) db.add(scan) 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, db, sample_db_config): """Test getting history with custom limit.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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_id=sample_db_config.id, title=f'Scan {i}', triggered_by='scheduled', schedule_id=schedule_id ) db.add(scan) 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, db, sample_db_config): """Test converting schedule to dictionary.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test Schedule', config_id=sample_db_config.id, 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_id' 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, db, sample_db_config): """Test relative time formatting in schedule dict.""" service = ScheduleService(db) schedule_id = service.create_schedule( name='Test', config_id=sample_db_config.id, 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()