diff --git a/maintenance_service_http_monitoring/tests/__init__.py b/maintenance_service_http_monitoring/tests/__init__.py new file mode 100644 index 0000000..ab5c71a --- /dev/null +++ b/maintenance_service_http_monitoring/tests/__init__.py @@ -0,0 +1 @@ +from . import test_http_monitoring diff --git a/maintenance_service_http_monitoring/tests/test_http_monitoring.py b/maintenance_service_http_monitoring/tests/test_http_monitoring.py new file mode 100644 index 0000000..0a9098f --- /dev/null +++ b/maintenance_service_http_monitoring/tests/test_http_monitoring.py @@ -0,0 +1,220 @@ +from datetime import timedelta +from unittest.mock import MagicMock, patch + +from odoo import fields +from odoo.tests.common import TransactionCase + +SERVICE_INSTANCE_REQUESTS = ( + "odoo.addons.maintenance_service_http_monitoring.models.service_instance.requests" +) +EQUIPMENT_HTTP_REQUESTS = ( + "odoo.addons.maintenance_service_http_monitoring" + ".models.maintenance_equipment.http_requests" +) + + +def _mock_response(status_code): + response = MagicMock() + response.status_code = status_code + return response + + +class TestHttpMonitoring(TransactionCase): + def setUp(self): + super().setUp() + team = self.env["maintenance.team"].search([], limit=1) + self.equipment = self.env["maintenance.equipment"].create( + { + "name": "Test Server", + "maintenance_team_id": team.id if team else False, + } + ) + self.service = self.env["service"].create({"name": "Test Service"}) + self.service_instance = self.env["service.instance"].create( + { + "equipment_id": self.equipment.id, + "service_id": self.service.id, + "service_url": "https://example.com", + } + ) + + # ------------------------------------------------------------------ + # Test 1 — HTTP 200 → service marked OK + # ------------------------------------------------------------------ + def test_http_200_sets_status_ok(self): + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(200) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + self.assertTrue(self.service_instance.http_status_ok) + self.assertEqual(self.service_instance.last_http_status_code, 200) + self.assertIsNotNone(self.service_instance.last_http_check_date) + + # ------------------------------------------------------------------ + # Test 2 — HTTP 500 → service KO + maintenance.request created + # ------------------------------------------------------------------ + def test_http_500_creates_maintenance_request(self): + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(500) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + self.assertFalse(self.service_instance.http_status_ok) + self.assertEqual(self.service_instance.last_http_status_code, 500) + + request = self.equipment.http_maintenance_request + self.assertTrue(request) + self.assertEqual(request.name, f"[HTTP KO] {self.equipment.name}") + self.assertEqual(request.priority, "2") + self.assertEqual(request.maintenance_type, "corrective") + + # ------------------------------------------------------------------ + # Test 3 — Network error → KO with code -1 + # ------------------------------------------------------------------ + def test_network_error_sets_status_ko_and_minus_one(self): + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.side_effect = Exception("connection refused") + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + self.assertFalse(self.service_instance.http_status_ok) + self.assertEqual(self.service_instance.last_http_status_code, -1) + + # ------------------------------------------------------------------ + # Test 4 — Two consecutive failures → no duplicate maintenance.request + # ------------------------------------------------------------------ + def test_no_duplicate_request_on_repeated_failure(self): + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(500) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + request_1 = self.equipment.http_maintenance_request + self.assertTrue(request_1) + + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(500) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + self.assertEqual(self.equipment.http_maintenance_request, request_1) + self.assertEqual( + self.env["maintenance.request"].search_count( + [("equipment_id", "=", self.equipment.id)] + ), + 1, + ) + + # ------------------------------------------------------------------ + # Test 5 — Equipment in maintenance mode → cron skips it + # ------------------------------------------------------------------ + def test_maintenance_mode_skips_http_check(self): + self.equipment.write({"maintenance_mode": True}) + + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(500) + mock_requests.exceptions.RequestException = Exception + self.env["service.instance"].cron_check_http_services() + + mock_requests.get.assert_not_called() + # last_http_check_date must remain unset (never checked) + self.assertFalse(self.service_instance.last_http_check_date) + + # ------------------------------------------------------------------ + # Test 6 — Expired maintenance mode → cron deactivates it + # ------------------------------------------------------------------ + def test_maintenance_mode_auto_expiry(self): + past = fields.Datetime.now() - timedelta(hours=1) + self.equipment.write( + { + "maintenance_mode": True, + "maintenance_mode_start": past - timedelta(hours=4), + "maintenance_mode_end": past, + } + ) + self.assertTrue(self.equipment.maintenance_mode) + + self.env["maintenance.equipment"].cron_deactivate_expired_maintenance_mode() + + self.assertFalse(self.equipment.maintenance_mode) + self.assertFalse(self.equipment.maintenance_mode_start) + self.assertFalse(self.equipment.maintenance_mode_end) + + # ------------------------------------------------------------------ + # Test 7 — Service without URL → ignored by cron + # ------------------------------------------------------------------ + def test_service_without_url_is_ignored(self): + self.service_instance.write({"service_url": False}) + + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(200) + mock_requests.exceptions.RequestException = Exception + self.env["service.instance"].cron_check_http_services() + + mock_requests.get.assert_not_called() + self.assertFalse(self.service_instance.last_http_check_date) + + # ------------------------------------------------------------------ + # Test 8 — HTTP 404 (non-exception) → KO with correct code + # ------------------------------------------------------------------ + def test_http_non_200_non_exception(self): + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(404) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + self.assertFalse(self.service_instance.http_status_ok) + self.assertEqual(self.service_instance.last_http_status_code, 404) + + # ------------------------------------------------------------------ + # Test 9 — Webhook called when a new maintenance.request is created + # ------------------------------------------------------------------ + def test_webhook_called_on_new_request(self): + self.env["ir.config_parameter"].sudo().set_param( + "maintenance_service_http_monitoring.webhook_url", + "https://webhook.example.com/hook", + ) + with ( + patch(SERVICE_INSTANCE_REQUESTS) as mock_requests, + patch(EQUIPMENT_HTTP_REQUESTS) as mock_http, + ): + mock_requests.get.return_value = _mock_response(500) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + mock_http.post.assert_called_once() + call_kwargs = mock_http.post.call_args + payload = call_kwargs.kwargs.get("json") or call_kwargs[1].get("json") + self.assertEqual(payload["equipment"], self.equipment.name) + + # ------------------------------------------------------------------ + # Test 10 — Webhook skipped when no URL configured + # ------------------------------------------------------------------ + def test_webhook_skipped_when_no_url(self): + self.env["ir.config_parameter"].sudo().set_param( + "maintenance_service_http_monitoring.webhook_url", "" + ) + with ( + patch(SERVICE_INSTANCE_REQUESTS) as mock_requests, + patch(EQUIPMENT_HTTP_REQUESTS) as mock_http, + ): + mock_requests.get.return_value = _mock_response(500) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + mock_http.post.assert_not_called() + + # ------------------------------------------------------------------ + # Test 11 — Service without equipment → check_http_status skips it + # ------------------------------------------------------------------ + def test_service_without_equipment_is_ignored(self): + self.service_instance.write({"equipment_id": False}) + + with patch(SERVICE_INSTANCE_REQUESTS) as mock_requests: + mock_requests.get.return_value = _mock_response(200) + mock_requests.exceptions.RequestException = Exception + self.service_instance.check_http_status() + + mock_requests.get.assert_not_called() + self.assertFalse(self.service_instance.last_http_check_date)