From 320de65b8f99fb20e64a4f333b9d6e1d09e0a19e Mon Sep 17 00:00:00 2001 From: derekliutao-cmd Date: Sat, 16 May 2026 11:27:09 +0800 Subject: [PATCH 1/4] feat: add Nagios XI provider --- keep/providers/nagiosxi_provider/__init__.py | 0 1 file changed, 0 insertions(+), 0 deletions(-) create mode 100644 keep/providers/nagiosxi_provider/__init__.py diff --git a/keep/providers/nagiosxi_provider/__init__.py b/keep/providers/nagiosxi_provider/__init__.py new file mode 100644 index 0000000000..e69de29bb2 From 9e342ec148e9882669a1540dd8f2ebecf5740584 Mon Sep 17 00:00:00 2001 From: derekliutao-cmd Date: Sat, 16 May 2026 11:27:19 +0800 Subject: [PATCH 2/4] feat: add Nagios XI provider implementation --- .../nagiosxi_provider/nagiosxi_provider.py | 406 ++++++++++++++++++ 1 file changed, 406 insertions(+) create mode 100644 keep/providers/nagiosxi_provider/nagiosxi_provider.py diff --git a/keep/providers/nagiosxi_provider/nagiosxi_provider.py b/keep/providers/nagiosxi_provider/nagiosxi_provider.py new file mode 100644 index 0000000000..6e21b73a23 --- /dev/null +++ b/keep/providers/nagiosxi_provider/nagiosxi_provider.py @@ -0,0 +1,406 @@ +""" +NagiosXIProvider is a class that provides a set of methods to interact with the Nagios XI REST API. + +Nagios XI is a monitoring platform. This provider polls host and service status +via the Nagios XI REST API using apikey authentication, following the same +polling pattern as the CentreonProvider. + +State mapping (Nagios convention): + - 0 = OK -> AlertStatus.RESOLVED, AlertSeverity.LOW + - 1 = WARNING -> AlertStatus.FIRING, AlertSeverity.WARNING + - 2 = CRITICAL -> AlertStatus.FIRING, AlertSeverity.CRITICAL + - 3 = UNKNOWN -> AlertStatus.FIRING, AlertSeverity.INFO + +References: + - Nagios XI REST API: https://api.nagios.org/ + - CentreonProvider (reference implementation) +""" + +import dataclasses +import datetime + +import pydantic +import requests + +from keep.api.models.alert import AlertDto, AlertSeverity, AlertStatus +from keep.contextmanager.contextmanager import ContextManager +from keep.exceptions.provider_exception import ProviderException +from keep.providers.base.base_provider import BaseProvider +from keep.providers.models.provider_config import ProviderConfig, ProviderScope + + +@pydantic.dataclasses.dataclass +class NagiosxiProviderAuthConfig: + """ + NagiosxiProviderAuthConfig holds the authentication information for the + Nagios XI provider. + """ + + host_url: pydantic.AnyHttpUrl = dataclasses.field( + metadata={ + "required": True, + "description": "Nagios XI Host URL (e.g. https://nagios.example.com/nagios)", + "sensitive": False, + "validation": "any_http_url", + }, + ) + + api_key: str = dataclasses.field( + metadata={ + "required": True, + "description": "Nagios XI API Key", + "sensitive": True, + }, + default=None, + ) + + +class NagiosxiProvider(BaseProvider): + PROVIDER_DISPLAY_NAME = "Nagios XI" + PROVIDER_TAGS = ["alert"] + PROVIDER_CATEGORY = ["Monitoring"] + PROVIDER_SCOPES = [ + ProviderScope( + name="authenticated", + description="API key is valid and user is authenticated", + ), + ] + + # Nagios state codes: + # 0 = OK, 1 = WARNING, 2 = CRITICAL, 3 = UNKNOWN + # https://assets.nagios.com/downloads/nagioscore/docs/nagioscore/3/en/statetypes.html + HOST_STATUS_MAP = { + 0: AlertStatus.RESOLVED, # UP + 1: AlertStatus.FIRING, # DOWN + 2: AlertStatus.FIRING, # UNREACHABLE + } + + HOST_SEVERITY_MAP = { + 0: AlertSeverity.LOW, # UP + 1: AlertSeverity.CRITICAL, # DOWN + 2: AlertSeverity.WARNING, # UNREACHABLE + } + + SERVICE_STATUS_MAP = { + 0: AlertStatus.RESOLVED, # OK + 1: AlertStatus.FIRING, # WARNING + 2: AlertStatus.FIRING, # CRITICAL + 3: AlertStatus.FIRING, # UNKNOWN + } + + SERVICE_SEVERITY_MAP = { + 0: AlertSeverity.LOW, # OK + 1: AlertSeverity.WARNING, # WARNING + 2: AlertSeverity.CRITICAL, # CRITICAL + 3: AlertSeverity.INFO, # UNKNOWN + } + + FINGERPRINT_FIELDS = ["id"] + + def __init__( + self, context_manager: ContextManager, provider_id: str, config: ProviderConfig + ): + super().__init__(context_manager, provider_id, config) + + def dispose(self): + pass + + def validate_config(self): + """ + Validates the configuration of the Nagios XI provider. + """ + self.authentication_config = NagiosxiProviderAuthConfig( + **self.config.authentication + ) + + def __get_url(self, endpoint: str) -> str: + """ + Build the full API URL for a given Nagios XI REST API endpoint. + + Nagios XI REST API endpoints follow the pattern: + {host_url}/api/v1/objects/{endpoint} + """ + host_url = str(self.authentication_config.host_url).rstrip("/") + return f"{host_url}/api/v1/objects/{endpoint}" + + def __get_params(self, extra: dict | None = None) -> dict: + """ + Build common query parameters including the apikey. + """ + params = {"apikey": self.authentication_config.api_key} + if extra: + params.update(extra) + return params + + def __get_headers(self) -> dict: + return { + "Content-Type": "application/json", + "Accept": "application/json", + } + + def validate_scopes(self) -> dict[str, bool | str]: + """ + Validate the scopes of the provider by making a test API call. + """ + try: + response = requests.get( + self.__get_url("hoststatus"), + params=self.__get_params(), + headers=self.__get_headers(), + timeout=10, + ) + if response.ok: + scopes = {"authenticated": True} + else: + scopes = { + "authenticated": f"Error validating scopes: {response.status_code} {response.text}" + } + except Exception as e: + scopes = { + "authenticated": f"Error validating scopes: {e}", + } + return scopes + + def __get_host_status(self) -> list[AlertDto]: + """ + Fetch host status from Nagios XI via the REST API. + + Endpoint: /api/v1/objects/hoststatus + Returns a list of AlertDto objects for each host. + """ + try: + response = requests.get( + self.__get_url("hoststatus"), + params=self.__get_params(), + headers=self.__get_headers(), + timeout=30, + ) + + if not response.ok: + self.logger.error( + "Failed to get host status from Nagios XI: %s", response.text + ) + raise ProviderException( + f"Failed to get host status from Nagios XI: {response.status_code}" + ) + + data = response.json() + host_records = data.get("hoststatus", []) + + if not isinstance(host_records, list): + # The API may return a single object if there's only one host + if isinstance(host_records, dict): + host_records = [host_records] + else: + host_records = [] + + alerts = [] + for host in host_records: + current_state = int(host.get("current_state", 3)) + last_check = host.get("last_check") + if last_check: + try: + last_received = datetime.datetime.fromtimestamp( + float(last_check) + ).isoformat() + except (ValueError, TypeError, OSError): + last_received = datetime.datetime.now().isoformat() + else: + last_received = datetime.datetime.now().isoformat() + + alert = AlertDto( + id=str(host.get("host_object_id", host.get("host_name", ""))), + name=host.get("host_name", ""), + address=host.get("address", ""), + description=host.get("output", ""), + status=self.HOST_STATUS_MAP.get( + current_state, AlertStatus.FIRING + ), + severity=self.HOST_SEVERITY_MAP.get( + current_state, AlertSeverity.INFO + ), + acknowledged=host.get("problem_has_been_acknowledged", "0") == "1", + lastReceived=last_received, + source=["nagiosxi"], + # Extra Nagios-specific fields + current_state=current_state, + host_alias=host.get("alias", ""), + host_status=host.get("status", ""), + check_command=host.get("check_command", ""), + max_check_attempts=host.get("max_check_attempts", ""), + current_check_attempt=host.get("current_check_attempt", ""), + state_type=host.get("state_type", ""), + is_flapping=host.get("is_flapping", "0") == "1", + scheduled_downtime_depth=host.get( + "scheduled_downtime_depth", "0" + ), + plugin_output=host.get("output", ""), + long_plugin_output=host.get("long_output", ""), + perf_data=host.get("perf_data", ""), + ) + alerts.append(alert) + + return alerts + + except ProviderException: + raise + except Exception as e: + self.logger.error("Error getting host status from Nagios XI: %s", e) + raise ProviderException( + f"Error getting host status from Nagios XI: {e}" + ) from e + + def __get_service_status(self) -> list[AlertDto]: + """ + Fetch service status from Nagios XI via the REST API. + + Endpoint: /api/v1/objects/servicestatus + Returns a list of AlertDto objects for each service. + """ + try: + response = requests.get( + self.__get_url("servicestatus"), + params=self.__get_params(), + headers=self.__get_headers(), + timeout=30, + ) + + if not response.ok: + self.logger.error( + "Failed to get service status from Nagios XI: %s", response.text + ) + raise ProviderException( + f"Failed to get service status from Nagios XI: {response.status_code}" + ) + + data = response.json() + service_records = data.get("servicestatus", []) + + if not isinstance(service_records, list): + if isinstance(service_records, dict): + service_records = [service_records] + else: + service_records = [] + + alerts = [] + for service in service_records: + current_state = int(service.get("current_state", 3)) + last_check = service.get("last_check") + if last_check: + try: + last_received = datetime.datetime.fromtimestamp( + float(last_check) + ).isoformat() + except (ValueError, TypeError, OSError): + last_received = datetime.datetime.now().isoformat() + else: + last_received = datetime.datetime.now().isoformat() + + host_name = service.get("host_name", "") + service_description = service.get( + "service_description", service.get("description", "") + ) + + alert = AlertDto( + id=f"{host_name}/{service_description}", + name=service_description, + host=host_name, + description=service.get("output", ""), + status=self.SERVICE_STATUS_MAP.get( + current_state, AlertStatus.FIRING + ), + severity=self.SERVICE_SEVERITY_MAP.get( + current_state, AlertSeverity.INFO + ), + acknowledged=service.get("problem_has_been_acknowledged", "0") + == "1", + lastReceived=last_received, + source=["nagiosxi"], + # Extra Nagios-specific fields + current_state=current_state, + host_id=service.get("host_object_id", ""), + service_id=service.get("service_object_id", ""), + check_command=service.get("check_command", ""), + max_check_attempts=service.get("max_check_attempts", ""), + current_check_attempt=service.get("current_check_attempt", ""), + state_type=service.get("state_type", ""), + is_flapping=service.get("is_flapping", "0") == "1", + scheduled_downtime_depth=service.get( + "scheduled_downtime_depth", "0" + ), + plugin_output=service.get("output", ""), + long_plugin_output=service.get("long_output", ""), + perf_data=service.get("perf_data", ""), + ) + alerts.append(alert) + + return alerts + + except ProviderException: + raise + except Exception as e: + self.logger.error("Error getting service status from Nagios XI: %s", e) + raise ProviderException( + f"Error getting service status from Nagios XI: {e}" + ) from e + + def _get_alerts(self) -> list[AlertDto]: + """ + Collect alerts from Nagios XI by polling both host and service status. + + This method follows the CentreonProvider polling pattern: + it calls private methods to fetch host status and service status, + then combines the results into a single list of AlertDto objects. + Errors in one method do not prevent the other from returning results. + """ + alerts = [] + try: + self.logger.info("Collecting alerts (host status) from Nagios XI") + host_status_alerts = self.__get_host_status() + alerts.extend(host_status_alerts) + except Exception as e: + self.logger.error("Error getting host status from Nagios XI: %s", e) + + try: + self.logger.info("Collecting alerts (service status) from Nagios XI") + service_status_alerts = self.__get_service_status() + alerts.extend(service_status_alerts) + except Exception as e: + self.logger.error("Error getting service status from Nagios XI: %s", e) + + return alerts + + +if __name__ == "__main__": + import logging + import os + + logging.basicConfig(level=logging.DEBUG, handlers=[logging.StreamHandler()]) + context_manager = ContextManager( + tenant_id="singletenant", + workflow_id="test", + ) + + host_url = os.environ.get("NAGIOSXI_HOST_URL") + api_key = os.environ.get("NAGIOSXI_API_KEY") + + if host_url is None: + raise ProviderException("NAGIOSXI_HOST_URL is not set") + + config = ProviderConfig( + description="Nagios XI Provider", + authentication={ + "host_url": host_url, + "api_key": api_key, + }, + ) + + provider = NagiosxiProvider( + context_manager, + provider_id="nagiosxi", + config=config, + ) + + alerts = provider._get_alerts() + for alert in alerts: + print(alert) From e4513876976321266750b669db233c1ab3bcc83e Mon Sep 17 00:00:00 2001 From: derekliutao-cmd Date: Sat, 16 May 2026 11:27:31 +0800 Subject: [PATCH 3/4] docs: add Nagios XI provider README --- keep/providers/nagiosxi_provider/README.md | 67 ++++++++++++++++++++++ 1 file changed, 67 insertions(+) create mode 100644 keep/providers/nagiosxi_provider/README.md diff --git a/keep/providers/nagiosxi_provider/README.md b/keep/providers/nagiosxi_provider/README.md new file mode 100644 index 0000000000..8c1e27a032 --- /dev/null +++ b/keep/providers/nagiosxi_provider/README.md @@ -0,0 +1,67 @@ +## Nagios XI Provider + +Nagios XI provider for Keep pulls host and service status from the Nagios XI REST API using API key authentication. + +### Supported Features + +- **Polling**: Periodically fetches host and service status from Nagios XI and maps them to Keep alerts. +- **Host Status**: Maps Nagios host states (UP/DOWN/UNREACHABLE) to Keep alert statuses and severities. +- **Service Status**: Maps Nagios service states (OK/WARNING/CRITICAL/UNKNOWN) to Keep alert statuses and severities. + +### State Mapping + +| Nagios Host State | Keep Status | Keep Severity | +|-------------------|-------------|----------------| +| 0 (UP) | RESOLVED | LOW | +| 1 (DOWN) | FIRING | CRITICAL | +| 2 (UNREACHABLE) | FIRING | WARNING | + +| Nagios Service State | Keep Status | Keep Severity | +|-----------------------|-------------|----------------| +| 0 (OK) | RESOLVED | LOW | +| 1 (WARNING) | FIRING | WARNING | +| 2 (CRITICAL) | FIRING | CRITICAL | +| 3 (UNKNOWN) | FIRING | INFO | + +### Configuration + +| Parameter | Description | Required | +|-----------|-------------|----------| +| `host_url` | Nagios XI base URL (e.g. `https://nagios.example.com/nagios`) | Yes | +| `api_key` | Nagios XI API key | Yes | + +### How to get the API Key + +1. Log in to your Nagios XI web interface. +2. Navigate to **Admin** > **Backends** > **API Keys** (or **Configure** > **API Keys** in older versions). +3. Click **Add New API Key**, provide a description, and save. +4. Copy the generated API key for use in Keep. + +### How to debug with local Nagios XI + +Start a Nagios XI Docker instance: + +```bash +docker run -d \ + --name=nagiosxi \ + -p 8080:80 \ + -p 8443:443 \ + ghcr.io/nagiosenterprises/nagiosxi:latest +``` + +Wait 2-3 minutes for initialization, then access the web UI at `https://localhost:8443/nagiosxi/`. + +Default login credentials: `nagiosadmin` / `nagiosadmin` (you will be prompted to change the password on first login). + +After logging in, generate an API key from the Admin panel and configure Keep with: + +- **host_url**: `https://localhost:8443/nagios` +- **api_key**: Your generated API key + +### Troubleshooting + +- **Connection refused**: Ensure Nagios XI is running and the host URL is correct. The URL should end with `/nagios` (not `/nagiosxi`). +- **401 Unauthorized**: Verify your API key is valid and has not expired. Regenerate the key if needed. +- **Empty results**: Make sure hosts and services are configured in Nagios XI. A fresh installation may not have any monitored objects. +- **SSL certificate errors**: If using a self-signed certificate, the provider will fail TLS verification. Consider using a valid certificate or testing with HTTP instead of HTTPS. +- **Timeout errors**: Increase the polling interval or check network connectivity between Keep and the Nagios XI server. From c7d5047feb371864124f56a0e4373a6d409f7d35 Mon Sep 17 00:00:00 2001 From: derekliutao-cmd Date: Sat, 16 May 2026 11:27:58 +0800 Subject: [PATCH 4/4] test: add Nagios XI provider unit tests --- tests/test_nagiosxi_provider.py | 553 ++++++++++++++++++++++++++++++++ 1 file changed, 553 insertions(+) create mode 100644 tests/test_nagiosxi_provider.py diff --git a/tests/test_nagiosxi_provider.py b/tests/test_nagiosxi_provider.py new file mode 100644 index 0000000000..f11d25f492 --- /dev/null +++ b/tests/test_nagiosxi_provider.py @@ -0,0 +1,553 @@ +""" +Unit tests for NagiosxiProvider with mocked HTTP responses. +""" + +import json +from unittest.mock import MagicMock, patch + +import pytest + +from keep.api.models.alert import AlertSeverity, AlertStatus +from keep.contextmanager.contextmanager import ContextManager +from keep.providers.models.provider_config import ProviderConfig + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_provider(): + """Create a NagiosxiProvider instance with a dummy config.""" + from keep.providers.nagiosxi_provider.nagiosxi_provider import NagiosxiProvider + + context_manager = ContextManager(tenant_id="test", workflow_id="test") + config = ProviderConfig( + description="Nagios XI Provider Test", + authentication={ + "host_url": "https://nagios.example.com/nagios", + "api_key": "test-api-key-123", + }, + ) + return NagiosxiProvider(context_manager, provider_id="nagiosxi", config=config) + + +# --------------------------------------------------------------------------- +# Sample API responses (based on Nagios XI REST API format) +# --------------------------------------------------------------------------- + +SAMPLE_HOST_STATUS_RESPONSE = { + "hoststatus": [ + { + "host_object_id": "123", + "host_name": "web-server-01", + "alias": "Web Server 01", + "address": "192.168.1.10", + "current_state": "0", + "output": "PING OK - Packet loss = 0%, RTA = 0.45 ms", + "long_output": "", + "perf_data": "rta=0.450000ms;3000.000;5000.000;0; pl=0%;80;100;0;", + "check_command": "check_ping!100.0,20%!500.0,60%", + "status": "up", + "last_check": "1747353600", + "max_check_attempts": "3", + "current_check_attempt": "1", + "state_type": "1", + "problem_has_been_acknowledged": "0", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + }, + { + "host_object_id": "124", + "host_name": "db-server-01", + "alias": "Database Server 01", + "address": "192.168.1.20", + "current_state": "1", + "output": "CRITICAL - Host Unreachable (192.168.1.20)", + "long_output": "", + "perf_data": "", + "check_command": "check_ping!100.0,20%!500.0,60%", + "status": "down", + "last_check": "1747353660", + "max_check_attempts": "3", + "current_check_attempt": "3", + "state_type": "1", + "problem_has_been_acknowledged": "1", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + }, + { + "host_object_id": "125", + "host_name": "mail-server-01", + "alias": "Mail Server", + "address": "192.168.1.30", + "current_state": "2", + "output": "UNREACHABLE - Host unreachable", + "long_output": "", + "perf_data": "", + "check_command": "check_ping!100.0,20%!500.0,60%", + "status": "unreachable", + "last_check": "1747353720", + "max_check_attempts": "3", + "current_check_attempt": "2", + "state_type": "0", + "problem_has_been_acknowledged": "0", + "is_flapping": "1", + "scheduled_downtime_depth": "1", + }, + ] +} + +SAMPLE_SERVICE_STATUS_RESPONSE = { + "servicestatus": [ + { + "host_object_id": "123", + "host_name": "web-server-01", + "service_object_id": "456", + "service_description": "HTTP", + "description": "HTTP", + "current_state": "0", + "output": "HTTP OK: HTTP/1.1 200 OK - 0.003s response time", + "long_output": "", + "perf_data": "time=0.003152s;;;0.000000;10.000000", + "check_command": "check_http", + "last_check": "1747353600", + "max_check_attempts": "3", + "current_check_attempt": "1", + "state_type": "1", + "problem_has_been_acknowledged": "0", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + }, + { + "host_object_id": "123", + "host_name": "web-server-01", + "service_object_id": "457", + "service_description": "Disk /", + "description": "Disk /", + "current_state": "1", + "output": "DISK WARNING - free space: / 8192 MB (15% inode=98%);", + "long_output": "", + "perf_data": "/=45056MB;;;0;524288", + "check_command": "check_disk!20%!10%!/", + "last_check": "1747353600", + "max_check_attempts": "3", + "current_check_attempt": "2", + "state_type": "1", + "problem_has_been_acknowledged": "0", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + }, + { + "host_object_id": "124", + "host_name": "db-server-01", + "service_object_id": "458", + "service_description": "MySQL", + "description": "MySQL", + "current_state": "2", + "output": "CRITICAL - Could not connect to MySQL on db-server-01", + "long_output": "Connection refused", + "perf_data": "", + "check_command": "check_mysql", + "last_check": "1747353660", + "max_check_attempts": "3", + "current_check_attempt": "3", + "state_type": "1", + "problem_has_been_acknowledged": "0", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + }, + { + "host_object_id": "125", + "host_name": "mail-server-01", + "service_object_id": "459", + "service_description": "SMTP", + "description": "SMTP", + "current_state": "3", + "output": "UNKNOWN - SMTP check not available", + "long_output": "", + "perf_data": "", + "check_command": "check_smtp", + "last_check": "1747353720", + "max_check_attempts": "3", + "current_check_attempt": "1", + "state_type": "0", + "problem_has_been_acknowledged": "0", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + }, + ] +} + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + + +class TestNagiosxiProviderHostStatus: + """Tests for __get_host_status (via _get_alerts).""" + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_host_status_ok(self, mock_get): + """Host with state 0 (UP) should map to RESOLVED/LOW.""" + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = SAMPLE_HOST_STATUS_RESPONSE + mock_get.return_value = mock_response + + # Mock service status to return empty + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = {"servicestatus": []} + mock_get.side_effect = [mock_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + host_alert = next(a for a in alerts if a.name == "web-server-01") + assert host_alert.status == AlertStatus.RESOLVED + assert host_alert.severity == AlertSeverity.LOW + assert host_alert.source == ["nagiosxi"] + assert host_alert.acknowledged is False + assert host_alert.is_flapping is False + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_host_status_down(self, mock_get): + """Host with state 1 (DOWN) should map to FIRING/CRITICAL.""" + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = SAMPLE_HOST_STATUS_RESPONSE + mock_get.return_value = mock_response + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = {"servicestatus": []} + mock_get.side_effect = [mock_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + host_alert = next(a for a in alerts if a.name == "db-server-01") + assert host_alert.status == AlertStatus.FIRING + assert host_alert.severity == AlertSeverity.CRITICAL + assert host_alert.acknowledged is True + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_host_status_unreachable(self, mock_get): + """Host with state 2 (UNREACHABLE) should map to FIRING/WARNING.""" + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = SAMPLE_HOST_STATUS_RESPONSE + mock_get.return_value = mock_response + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = {"servicestatus": []} + mock_get.side_effect = [mock_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + host_alert = next(a for a in alerts if a.name == "mail-server-01") + assert host_alert.status == AlertStatus.FIRING + assert host_alert.severity == AlertSeverity.WARNING + assert host_alert.is_flapping is True + assert host_alert.scheduled_downtime_depth == "1" + + +class TestNagiosxiProviderServiceStatus: + """Tests for __get_service_status (via _get_alerts).""" + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_service_status_ok(self, mock_get): + """Service with state 0 (OK) should map to RESOLVED/LOW.""" + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = {"hoststatus": []} + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = SAMPLE_SERVICE_STATUS_RESPONSE + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + svc_alert = next( + a for a in alerts if a.name == "HTTP" and a.host == "web-server-01" + ) + assert svc_alert.status == AlertStatus.RESOLVED + assert svc_alert.severity == AlertSeverity.LOW + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_service_status_warning(self, mock_get): + """Service with state 1 (WARNING) should map to FIRING/WARNING.""" + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = {"hoststatus": []} + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = SAMPLE_SERVICE_STATUS_RESPONSE + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + svc_alert = next( + a for a in alerts if a.name == "Disk /" and a.host == "web-server-01" + ) + assert svc_alert.status == AlertStatus.FIRING + assert svc_alert.severity == AlertSeverity.WARNING + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_service_status_critical(self, mock_get): + """Service with state 2 (CRITICAL) should map to FIRING/CRITICAL.""" + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = {"hoststatus": []} + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = SAMPLE_SERVICE_STATUS_RESPONSE + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + svc_alert = next( + a for a in alerts if a.name == "MySQL" and a.host == "db-server-01" + ) + assert svc_alert.status == AlertStatus.FIRING + assert svc_alert.severity == AlertSeverity.CRITICAL + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_service_status_unknown(self, mock_get): + """Service with state 3 (UNKNOWN) should map to FIRING/INFO.""" + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = {"hoststatus": []} + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = SAMPLE_SERVICE_STATUS_RESPONSE + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + svc_alert = next( + a for a in alerts if a.name == "SMTP" and a.host == "mail-server-01" + ) + assert svc_alert.status == AlertStatus.FIRING + assert svc_alert.severity == AlertSeverity.INFO + + +class TestNagiosxiProviderCombined: + """Tests for _get_alerts combining host and service results.""" + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_combined_alerts(self, mock_get): + """_get_alerts should return both host and service alerts.""" + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = SAMPLE_HOST_STATUS_RESPONSE + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = SAMPLE_SERVICE_STATUS_RESPONSE + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + # 3 hosts + 4 services = 7 total alerts + assert len(alerts) == 7 + + # Verify all have source=nagiosxi + for alert in alerts: + assert alert.source == ["nagiosxi"] + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_host_failure_does_not_block_service(self, mock_get): + """If host status API fails, service alerts should still be returned.""" + mock_host_response = MagicMock() + mock_host_response.ok = False + mock_host_response.status_code = 500 + mock_host_response.text = "Internal Server Error" + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = SAMPLE_SERVICE_STATUS_RESPONSE + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + # Only service alerts should be returned (4 services) + assert len(alerts) == 4 + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_service_failure_does_not_block_host(self, mock_get): + """If service status API fails, host alerts should still be returned.""" + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = SAMPLE_HOST_STATUS_RESPONSE + + mock_svc_response = MagicMock() + mock_svc_response.ok = False + mock_svc_response.status_code = 500 + mock_svc_response.text = "Internal Server Error" + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + # Only host alerts should be returned (3 hosts) + assert len(alerts) == 3 + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_single_host_response(self, mock_get): + """API may return a single dict instead of a list when there's one host.""" + single_host_response = { + "hoststatus": { + "host_object_id": "123", + "host_name": "single-host", + "alias": "Single Host", + "address": "10.0.0.1", + "current_state": "0", + "output": "PING OK", + "long_output": "", + "perf_data": "", + "check_command": "check_ping", + "status": "up", + "last_check": "1747353600", + "max_check_attempts": "3", + "current_check_attempt": "1", + "state_type": "1", + "problem_has_been_acknowledged": "0", + "is_flapping": "0", + "scheduled_downtime_depth": "0", + } + } + + mock_host_response = MagicMock() + mock_host_response.ok = True + mock_host_response.json.return_value = single_host_response + + mock_svc_response = MagicMock() + mock_svc_response.ok = True + mock_svc_response.json.return_value = {"servicestatus": []} + + mock_get.side_effect = [mock_host_response, mock_svc_response] + + provider = _make_provider() + alerts = provider._get_alerts() + + assert len(alerts) == 1 + assert alerts[0].name == "single-host" + assert alerts[0].status == AlertStatus.RESOLVED + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_empty_response(self, mock_get): + """API returns empty results.""" + mock_response = MagicMock() + mock_response.ok = True + mock_response.json.return_value = {"hoststatus": [], "servicestatus": []} + mock_get.return_value = mock_response + + provider = _make_provider() + alerts = provider._get_alerts() + + assert len(alerts) == 0 + + +class TestNagiosxiProviderConfig: + """Tests for configuration validation.""" + + def test_valid_config(self): + """Valid config should not raise.""" + provider = _make_provider() + assert provider.authentication_config.host_url == "https://nagios.example.com/nagios" + assert provider.authentication_config.api_key == "test-api-key-123" + + def test_provider_metadata(self): + """Provider metadata should be correctly set.""" + from keep.providers.nagiosxi_provider.nagiosxi_provider import NagiosxiProvider + + assert NagiosxiProvider.PROVIDER_DISPLAY_NAME == "Nagios XI" + assert NagiosxiProvider.PROVIDER_CATEGORY == ["Monitoring"] + assert NagiosxiProvider.PROVIDER_TAGS == ["alert"] + assert len(NagiosxiProvider.PROVIDER_SCOPES) == 1 + assert NagiosxiProvider.PROVIDER_SCOPES[0].name == "authenticated" + + def test_state_maps_complete(self): + """All Nagios states (0-3 for services, 0-2 for hosts) should be mapped.""" + from keep.providers.nagiosxi_provider.nagiosxi_provider import NagiosxiProvider + + # Service maps should cover all 4 states + assert 0 in NagiosxiProvider.SERVICE_STATUS_MAP + assert 1 in NagiosxiProvider.SERVICE_STATUS_MAP + assert 2 in NagiosxiProvider.SERVICE_STATUS_MAP + assert 3 in NagiosxiProvider.SERVICE_STATUS_MAP + assert 0 in NagiosxiProvider.SERVICE_SEVERITY_MAP + assert 1 in NagiosxiProvider.SERVICE_SEVERITY_MAP + assert 2 in NagiosxiProvider.SERVICE_SEVERITY_MAP + assert 3 in NagiosxiProvider.SERVICE_SEVERITY_MAP + + # Host maps should cover states 0-2 + assert 0 in NagiosxiProvider.HOST_STATUS_MAP + assert 1 in NagiosxiProvider.HOST_STATUS_MAP + assert 2 in NagiosxiProvider.HOST_STATUS_MAP + assert 0 in NagiosxiProvider.HOST_SEVERITY_MAP + assert 1 in NagiosxiProvider.HOST_SEVERITY_MAP + assert 2 in NagiosxiProvider.HOST_SEVERITY_MAP + + +class TestNagiosxiProviderScopes: + """Tests for scope validation.""" + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_validate_scopes_success(self, mock_get): + """Successful API call should return authenticated=True.""" + mock_response = MagicMock() + mock_response.ok = True + mock_get.return_value = mock_response + + provider = _make_provider() + scopes = provider.validate_scopes() + + assert scopes["authenticated"] is True + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_validate_scopes_failure(self, mock_get): + """Failed API call should return error message.""" + mock_response = MagicMock() + mock_response.ok = False + mock_response.status_code = 401 + mock_response.text = "Unauthorized" + mock_get.return_value = mock_response + + provider = _make_provider() + scopes = provider.validate_scopes() + + assert isinstance(scopes["authenticated"], str) + assert "401" in scopes["authenticated"] + + @patch("keep.providers.nagiosxi_provider.nagiosxi_provider.requests.get") + def test_validate_scopes_exception(self, mock_get): + """Network exception should return error message.""" + mock_get.side_effect = Exception("Connection refused") + + provider = _make_provider() + scopes = provider.validate_scopes() + + assert isinstance(scopes["authenticated"], str) + assert "Connection refused" in scopes["authenticated"]