[IMP] maintenance_service_http_monitoring: add tests
This commit is contained in:
Stéphan Sainléger
1
maintenance_service_http_monitoring/tests/__init__.py
Normal file
1
maintenance_service_http_monitoring/tests/__init__.py
Normal file
@@ -0,0 +1 @@
|
||||
from . import test_http_monitoring
|
||||
@@ -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)
|
||||
Reference in New Issue
Block a user