""" Tests pour les routes de gestion des schedules. Couvre: - GET /api/schedules - GET /api/schedules/{schedule_id} - POST /api/schedules - PUT /api/schedules/{schedule_id} - DELETE /api/schedules/{schedule_id} - POST /api/schedules/{schedule_id}/pause - POST /api/schedules/{schedule_id}/resume - POST /api/schedules/{schedule_id}/run - GET /api/schedules/stats - POST /api/schedules/validate-cron """ import pytest from unittest.mock import patch, MagicMock, AsyncMock from httpx import AsyncClient pytestmark = [pytest.mark.unit, pytest.mark.asyncio] class TestGetSchedules: """Tests pour GET /api/schedules.""" async def test_list_schedules_empty(self, client: AsyncClient): """Liste vide sans schedules.""" with patch("app.services.scheduler_service.scheduler_service") as mock_sched: mock_sched.get_all_schedules.return_value = [] response = await client.get("/api/schedules") assert response.status_code == 200 data = response.json() assert data["schedules"] == [] async def test_list_schedules_with_filter(self, client: AsyncClient): """Liste avec filtre enabled.""" response = await client.get("/api/schedules?enabled=true") assert response.status_code == 200 # Response should have schedules key assert "schedules" in response.json() class TestGetSchedule: """Tests pour GET /api/schedules/{schedule_id}.""" async def test_get_schedule_not_found(self, client: AsyncClient): """404 si schedule non trouvé.""" with patch("app.services.scheduler_service.scheduler_service") as mock_sched: mock_sched.get_schedule.return_value = None response = await client.get("/api/schedules/nonexistent") assert response.status_code == 404 class TestCreateSchedule: """Tests pour POST /api/schedules.""" @pytest.mark.integration async def test_create_schedule_success( self, client: AsyncClient, db_session ): """Création de schedule réussie (integration test).""" with patch("app.services.websocket_service.ws_manager") as mock_ws: mock_ws.broadcast = AsyncMock() response = await client.post( "/api/schedules", json={ "name": "Daily Backup Test", "playbook": "backup.yml", "target": "all", "schedule_type": "recurring", "recurrence": { "type": "daily", "time": "02:00" } } ) # This test requires proper auth setup - marked as integration assert response.status_code in [200, 400, 401] async def test_create_schedule_invalid_payload(self, client: AsyncClient): """Création échoue avec payload invalide.""" response = await client.post( "/api/schedules", json={ "name": "", # Empty name "playbook": "backup.yml" } ) # Pydantic validation should fail or auth error assert response.status_code in [422, 401] class TestScheduleActions: """Tests pour les actions pause/resume/run.""" async def test_pause_schedule(self, client: AsyncClient, db_session, schedule_factory): """Pause un schedule.""" # Create a real schedule in DB first schedule = await schedule_factory.create(db_session, id="test-schedule", enabled=True) with patch("app.services.scheduler_service.scheduler_service") as mock_sched: mock_sched.pause_schedule.return_value = True response = await client.post("/api/schedules/test-schedule/pause") assert response.status_code == 200 async def test_pause_schedule_not_found(self, client: AsyncClient): """Pause échoue si schedule non trouvé.""" response = await client.post("/api/schedules/nonexistent/pause") assert response.status_code == 404 async def test_resume_schedule(self, client: AsyncClient, db_session, schedule_factory): """Resume un schedule.""" schedule = await schedule_factory.create(db_session, id="test-resume", enabled=False) with patch("app.services.scheduler_service.scheduler_service") as mock_sched: mock_sched.resume_schedule.return_value = True response = await client.post("/api/schedules/test-resume/resume") assert response.status_code == 200 class TestScheduleStats: """Tests pour GET /api/schedules/stats.""" async def test_get_stats(self, client: AsyncClient): """Récupère les statistiques.""" response = await client.get("/api/schedules/stats") assert response.status_code == 200 data = response.json() # Verify structure assert "stats" in data assert "total" in data["stats"] assert "active" in data["stats"] assert "paused" in data["stats"] class TestCronValidation: """Tests pour GET /api/schedules/validate-cron.""" async def test_validate_cron_valid(self, client: AsyncClient): """Validation expression cron valide.""" with patch("app.services.scheduler_service.scheduler_service") as mock_sched: mock_sched.validate_cron_expression.return_value = { "valid": True, "expression": "0 2 * * *", "next_runs": ["2024-01-15T02:00:00"], "error": None } response = await client.get("/api/schedules/validate-cron?expression=0%202%20*%20*%20*") assert response.status_code == 200 data = response.json() assert data["valid"] is True async def test_validate_cron_invalid(self, client: AsyncClient): """Validation expression cron invalide.""" with patch("app.services.scheduler_service.scheduler_service") as mock_sched: mock_sched.validate_cron_expression.return_value = { "valid": False, "expression": "invalid", "next_runs": None, "error": "Invalid cron expression" } response = await client.get("/api/schedules/validate-cron?expression=invalid") assert response.status_code == 200 data = response.json() assert data["valid"] is False assert data["error"] is not None class TestGetUpcoming: """Tests pour GET /api/schedules/upcoming.""" async def test_get_upcoming(self, client: AsyncClient): """Récupère les prochaines exécutions.""" response = await client.get("/api/schedules/upcoming") assert response.status_code == 200 data = response.json() assert "upcoming" in data assert "count" in data async def test_get_upcoming_with_limit(self, client: AsyncClient): """Récupère les prochaines exécutions avec limite.""" response = await client.get("/api/schedules/upcoming?limit=5") assert response.status_code == 200 class TestGetScheduleById: """Tests pour GET /api/schedules/{schedule_id}.""" async def test_get_schedule_by_id_found(self, client: AsyncClient, db_session, schedule_factory): """Récupère un schedule par ID.""" schedule = await schedule_factory.create(db_session, id="sched-test-123", name="Test Schedule") response = await client.get("/api/schedules/sched-test-123") assert response.status_code == 200 data = response.json() assert data["id"] == "sched-test-123" assert data["name"] == "Test Schedule" async def test_get_schedule_by_id_not_found(self, client: AsyncClient): """404 si schedule non trouvé.""" response = await client.get("/api/schedules/nonexistent-id") assert response.status_code == 404 class TestDeleteSchedule: """Tests pour DELETE /api/schedules/{schedule_id}.""" async def test_delete_schedule_not_found(self, client: AsyncClient): """404 si schedule non trouvé.""" response = await client.delete("/api/schedules/nonexistent") # May return 200 with error message or 404 assert response.status_code in [200, 404] async def test_delete_schedule_success(self, client: AsyncClient, db_session, schedule_factory): """Suppression réussie.""" schedule = await schedule_factory.create(db_session, id="sched-delete-test", name="To Delete") with patch("app.routes.schedules.ws_manager") as mock_ws: mock_ws.broadcast = AsyncMock() response = await client.delete("/api/schedules/sched-delete-test") assert response.status_code == 200 class TestUpdateSchedule: """Tests pour PUT /api/schedules/{schedule_id}.""" async def test_update_schedule_not_found(self, client: AsyncClient): """404 si schedule non trouvé.""" response = await client.put( "/api/schedules/nonexistent", json={"name": "Updated Name"} ) assert response.status_code == 404 async def test_update_schedule_success(self, client: AsyncClient, db_session, schedule_factory): """Mise à jour réussie.""" schedule = await schedule_factory.create(db_session, id="sched-update-test", name="Original") with patch("app.routes.schedules.ws_manager") as mock_ws: mock_ws.broadcast = AsyncMock() response = await client.put( "/api/schedules/sched-update-test", json={"name": "Updated Name"} ) assert response.status_code == 200 class TestResumeScheduleNotFound: """Tests supplémentaires pour resume.""" async def test_resume_schedule_not_found(self, client: AsyncClient): """Resume échoue si schedule non trouvé.""" response = await client.post("/api/schedules/nonexistent/resume") assert response.status_code == 404 class TestScheduleRuns: """Tests pour GET /api/schedules/{schedule_id}/runs.""" async def test_get_schedule_runs(self, client: AsyncClient, db_session, schedule_factory): """Récupère les exécutions d'un schedule.""" schedule = await schedule_factory.create(db_session, id="sched-runs-test") response = await client.get("/api/schedules/sched-runs-test/runs") assert response.status_code == 200 async def test_get_schedule_runs_not_found(self, client: AsyncClient): """404 si schedule non trouvé pour runs.""" response = await client.get("/api/schedules/nonexistent/runs") assert response.status_code == 404 class TestRunScheduleNow: """Tests pour POST /api/schedules/{schedule_id}/run.""" async def test_run_schedule_not_found(self, client: AsyncClient): """404 si schedule non trouvé pour run.""" response = await client.post("/api/schedules/nonexistent/run") assert response.status_code == 404 async def test_run_schedule_success(self, client: AsyncClient, db_session, schedule_factory): """Exécution manuelle réussie.""" schedule = await schedule_factory.create( db_session, id="sched-run-test", playbook="test.yml" ) with patch("app.routes.schedules.ws_manager") as mock_ws, \ patch("app.routes.schedules.ansible_service") as mock_ansible: mock_ws.broadcast = AsyncMock() mock_ansible.execute_playbook = AsyncMock(return_value={ "success": True, "stdout": "OK", "stderr": "" }) response = await client.post("/api/schedules/sched-run-test/run") assert response.status_code == 200 class TestCreateScheduleValidation: """Tests de validation pour la création de schedules.""" async def test_create_schedule_playbook_not_found(self, client: AsyncClient): """Création échoue si playbook non trouvé.""" with patch("app.routes.schedules.ansible_service") as mock_ansible: mock_ansible.get_playbooks.return_value = [] response = await client.post( "/api/schedules", json={ "name": "Test Schedule", "playbook": "nonexistent.yml", "target": "all", "target_type": "group", "schedule_type": "recurring", "recurrence": {"type": "daily", "time": "02:00"} } ) assert response.status_code == 400 assert "non trouvé" in response.json()["detail"] async def test_create_schedule_group_not_found(self, client: AsyncClient): """Création échoue si groupe non trouvé.""" with patch("app.routes.schedules.ansible_service") as mock_ansible: mock_ansible.get_playbooks.return_value = [ {"filename": "test.yml", "name": "test"} ] mock_ansible.get_groups.return_value = ["env_prod"] response = await client.post( "/api/schedules", json={ "name": "Test Schedule", "playbook": "test.yml", "target": "nonexistent_group", "target_type": "group", "schedule_type": "recurring", "recurrence": {"type": "daily", "time": "02:00"} } ) assert response.status_code == 400 assert "non trouvé" in response.json()["detail"] async def test_create_schedule_host_not_found(self, client: AsyncClient): """Création échoue si hôte non trouvé.""" with patch("app.routes.schedules.ansible_service") as mock_ansible: mock_ansible.get_playbooks.return_value = [ {"filename": "test.yml", "name": "test"} ] mock_ansible.host_exists.return_value = False response = await client.post( "/api/schedules", json={ "name": "Test Schedule", "playbook": "test.yml", "target": "nonexistent_host", "target_type": "host", "schedule_type": "recurring", "recurrence": {"type": "daily", "time": "02:00"} } ) assert response.status_code == 400 assert "non trouvé" in response.json()["detail"] async def test_create_schedule_missing_recurrence(self, client: AsyncClient): """Création échoue si récurrence manquante pour type recurring.""" with patch("app.routes.schedules.ansible_service") as mock_ansible: mock_ansible.get_playbooks.return_value = [ {"filename": "test.yml", "name": "test"} ] mock_ansible.get_groups.return_value = ["all"] response = await client.post( "/api/schedules", json={ "name": "Test Schedule", "playbook": "test.yml", "target": "all", "target_type": "group", "schedule_type": "recurring" # Missing recurrence } ) assert response.status_code == 400 assert "récurrence" in response.json()["detail"].lower() async def test_create_schedule_invalid_cron(self, client: AsyncClient): """Création échoue si expression cron invalide.""" with patch("app.routes.schedules.ansible_service") as mock_ansible, \ patch("app.routes.schedules.scheduler_service") as mock_sched: mock_ansible.get_playbooks.return_value = [ {"filename": "test.yml", "name": "test", "hosts": "all"} ] mock_ansible.get_groups.return_value = ["all"] mock_ansible.is_target_compatible_with_playbook.return_value = True mock_sched.validate_cron_expression.return_value = { "valid": False, "error": "Invalid expression" } response = await client.post( "/api/schedules", json={ "name": "Test Schedule", "playbook": "test.yml", "target": "all", "target_type": "group", "schedule_type": "recurring", "recurrence": { "type": "custom", "cron_expression": "invalid" } } ) assert response.status_code == 400 assert "cron" in response.json()["detail"].lower() class TestScheduleFilters: """Tests pour les filtres de schedules.""" async def test_list_schedules_with_playbook_filter(self, client: AsyncClient): """Liste avec filtre playbook.""" response = await client.get("/api/schedules?playbook=backup.yml") assert response.status_code == 200 assert "schedules" in response.json() async def test_list_schedules_with_tag_filter(self, client: AsyncClient): """Liste avec filtre tag.""" response = await client.get("/api/schedules?tag=backup") assert response.status_code == 200 assert "schedules" in response.json() async def test_list_schedules_with_pagination(self, client: AsyncClient): """Liste avec pagination.""" response = await client.get("/api/schedules?limit=5&offset=0") assert response.status_code == 200 assert "schedules" in response.json() class TestCreateScheduleSuccess: """Tests pour création réussie de schedules.""" async def test_create_schedule_with_valid_data(self, client: AsyncClient, db_session): """Création avec données valides.""" with patch("app.routes.schedules.ansible_service") as mock_ansible, \ patch("app.routes.schedules.scheduler_service") as mock_sched, \ patch("app.routes.schedules.ws_manager") as mock_ws: mock_ansible.get_playbooks.return_value = [ {"filename": "backup.yml", "name": "backup", "hosts": "all"} ] mock_ansible.get_groups.return_value = ["all", "env_prod"] mock_ansible.is_target_compatible_with_playbook.return_value = True mock_sched.add_schedule_to_cache = MagicMock() mock_ws.broadcast = AsyncMock() response = await client.post( "/api/schedules", json={ "name": "Daily Backup", "playbook": "backup.yml", "target": "all", "target_type": "group", "schedule_type": "recurring", "recurrence": {"type": "daily", "time": "02:00"} } ) assert response.status_code == 200 data = response.json() assert data["success"] is True async def test_create_schedule_with_custom_cron(self, client: AsyncClient, db_session): """Création avec expression cron custom.""" with patch("app.routes.schedules.ansible_service") as mock_ansible, \ patch("app.routes.schedules.scheduler_service") as mock_sched, \ patch("app.routes.schedules.ws_manager") as mock_ws: mock_ansible.get_playbooks.return_value = [ {"filename": "test.yml", "name": "test", "hosts": "all"} ] mock_ansible.get_groups.return_value = ["all"] mock_ansible.is_target_compatible_with_playbook.return_value = True mock_sched.validate_cron_expression.return_value = {"valid": True} mock_sched.add_schedule_to_cache = MagicMock() mock_ws.broadcast = AsyncMock() response = await client.post( "/api/schedules", json={ "name": "Custom Cron", "playbook": "test.yml", "target": "all", "target_type": "group", "schedule_type": "recurring", "recurrence": { "type": "custom", "cron_expression": "0 2 * * *" } } ) assert response.status_code == 200 async def test_create_schedule_with_host_target(self, client: AsyncClient, db_session): """Création avec cible hôte.""" with patch("app.routes.schedules.ansible_service") as mock_ansible, \ patch("app.routes.schedules.scheduler_service") as mock_sched, \ patch("app.routes.schedules.ws_manager") as mock_ws: mock_ansible.get_playbooks.return_value = [ {"filename": "test.yml", "name": "test", "hosts": "all"} ] mock_ansible.host_exists.return_value = True mock_ansible.is_target_compatible_with_playbook.return_value = True mock_sched.add_schedule_to_cache = MagicMock() mock_ws.broadcast = AsyncMock() response = await client.post( "/api/schedules", json={ "name": "Host Schedule", "playbook": "test.yml", "target": "server1.local", "target_type": "host", "schedule_type": "recurring", "recurrence": {"type": "daily", "time": "03:00"} } ) assert response.status_code == 200 class TestUpdateScheduleValidation: """Tests de validation pour mise à jour de schedules.""" async def test_update_schedule_invalid_playbook(self, client: AsyncClient, db_session, schedule_factory): """Mise à jour échoue si playbook invalide.""" schedule = await schedule_factory.create(db_session, id="sched-update-pb") with patch("app.routes.schedules.ansible_service") as mock_ansible: mock_ansible.get_playbooks.return_value = [] response = await client.put( "/api/schedules/sched-update-pb", json={"playbook": "nonexistent.yml"} ) assert response.status_code == 400 async def test_update_schedule_invalid_cron(self, client: AsyncClient, db_session, schedule_factory): """Mise à jour échoue si cron invalide.""" schedule = await schedule_factory.create(db_session, id="sched-update-cron") with patch("app.routes.schedules.scheduler_service") as mock_sched: mock_sched.get_schedule.return_value = MagicMock(name="Test") mock_sched.validate_cron_expression.return_value = { "valid": False, "error": "Invalid" } response = await client.put( "/api/schedules/sched-update-cron", json={ "recurrence": { "type": "custom", "cron_expression": "invalid" } } ) assert response.status_code == 400 class TestScheduleRunsWithData: """Tests pour runs avec données.""" async def test_get_schedule_runs_with_pagination(self, client: AsyncClient, db_session, schedule_factory): """Récupère les runs avec pagination.""" schedule = await schedule_factory.create(db_session, id="sched-runs-pag") response = await client.get("/api/schedules/sched-runs-pag/runs?limit=10&offset=0") assert response.status_code == 200 data = response.json() assert "runs" in data assert "count" in data class TestDeleteScheduleAlreadyDeleted: """Tests pour suppression de schedule déjà supprimé.""" async def test_delete_schedule_already_deleted(self, client: AsyncClient): """Suppression d'un schedule déjà supprimé retourne succès.""" with patch("app.routes.schedules.scheduler_service") as mock_sched: mock_sched.delete_schedule.side_effect = Exception("Not found") response = await client.delete("/api/schedules/already-deleted") # Should return success even if not found assert response.status_code == 200 assert "supprimé" in response.json()["message"].lower() or "inexistant" in response.json()["message"].lower()