test(integration): fixture for notification channel (#10089)

* chore: fixture for notification channel

* chore: return notification channel info in Create notification channel API

* fix: change scope of create channel fixture to function level

* chore: added debug message on assertion failure

* refactor: improve error handling in webhook notification channel deletion

* chore: ran py linter and fmt

* fix: silenced pylint on too broad exception

* feat: add alerts to integration CI workflow

* chore: removed clean from notification channel creation fixture + changes fixture scope of get_token

* chore: reverted fixture scopes to function

* fix: breaking alert test due to alert manager not registered for newly created org

* chore: formatted and fix lint

---------

Co-authored-by: Srikanth Chekuri <srikanth.chekuri92@gmail.com>

.github/workflows/integrationci.yaml

@@ -46,6 +46,7 @@ jobs:
           - ttl
           - preference
           - logspipelines
+          - alerts
         sqlstore-provider:
           - postgres
           - sqlite

tests/integration/conftest.py

@@ -18,6 +18,7 @@ pytest_plugins = [
     "fixtures.driver",
     "fixtures.idp",
     "fixtures.idputils",
+    "fixtures.notification_channel",
 ]
 
 

tests/integration/fixtures/notification_channel.py

@@ -0,0 +1,113 @@
+from http import HTTPStatus
+from typing import Callable
+
+import docker
+import docker.errors
+import pytest
+import requests
+from testcontainers.core.container import Network
+from wiremock.testing.testcontainer import WireMockContainer
+
+from fixtures import dev, types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+
+logger = setup_logger(__name__)
+
+
+@pytest.fixture(name="notification_channel", scope="package")
+def notification_channel(
+    network: Network,
+    request: pytest.FixtureRequest,
+    pytestconfig: pytest.Config,
+) -> types.TestContainerDocker:
+    """
+    Package-scoped fixture for WireMock container to receive notifications for Alert rules.
+    """
+
+    def create() -> types.TestContainerDocker:
+        container = WireMockContainer(image="wiremock/wiremock:2.35.1-1", secure=False)
+        container.with_network(network)
+        container.start()
+
+        return types.TestContainerDocker(
+            id=container.get_wrapped_container().id,
+            host_configs={
+                "8080": types.TestContainerUrlConfig(
+                    "http",
+                    container.get_container_host_ip(),
+                    container.get_exposed_port(8080),
+                )
+            },
+            container_configs={
+                "8080": types.TestContainerUrlConfig(
+                    "http", container.get_wrapped_container().name, 8080
+                )
+            },
+        )
+
+    def delete(container: types.TestContainerDocker):
+        client = docker.from_env()
+        try:
+            client.containers.get(container_id=container.id).stop()
+            client.containers.get(container_id=container.id).remove(v=True)
+        except docker.errors.NotFound:
+            logger.info(
+                "Skipping removal of NotificationChannel, NotificationChannel(%s) not found. Maybe it was manually removed?",
+                {"id": container.id},
+            )
+
+    def restore(cache: dict) -> types.TestContainerDocker:
+        return types.TestContainerDocker.from_cache(cache)
+
+    return dev.wrap(
+        request,
+        pytestconfig,
+        "notification_channel",
+        lambda: types.TestContainerDocker(id="", host_configs={}, container_configs={}),
+        create,
+        delete,
+        restore,
+    )
+
+
+@pytest.fixture(name="create_webhook_notification_channel", scope="function")
+def create_webhook_notification_channel(
+    signoz: types.SigNoz,
+    create_user_admin: None,  # pylint: disable=unused-argument
+    get_token: Callable[[str, str], str],
+) -> Callable[[str, str, dict, bool], str]:
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # function to create notification channel
+    def _create_webhook_notification_channel(
+        channel_name: str,
+        webhook_url: str,
+        http_config: dict = {},
+        send_resolved: bool = True,
+    ) -> str:
+        response = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/channels"),
+            json={
+                "name": channel_name,
+                "webhook_configs": [
+                    {
+                        "send_resolved": send_resolved,
+                        "url": webhook_url,
+                        "http_config": http_config,
+                    }
+                ],
+            },
+            headers={"Authorization": f"Bearer {admin_token}"},
+            timeout=5,
+        )
+        assert response.status_code == HTTPStatus.CREATED, (
+            f"Failed to create channel, "
+            f"Response: {response.text} "
+            f"Response status: {response.status_code}"
+        )
+
+        channel_id = response.json()["data"]["id"]
+        return channel_id
+
+    return _create_webhook_notification_channel

tests/integration/src/alerts/01_notification_channel.py

@@ -0,0 +1,98 @@
+import time
+import uuid
+from http import HTTPStatus
+from typing import Callable, List
+
+import requests
+from wiremock.client import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+
+
+def test_webhook_notification_channel(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, List[Mapping]], None],
+    create_webhook_notification_channel: Callable[[str, str, dict, bool], str],
+) -> None:
+    """
+    Tests the creation and delivery of test alerts on the created notification channel
+    """
+
+    # Prepare notification channel name and webhook endpoint
+    notification_channel_name = f"notification-channel-{uuid.uuid4()}"
+    webhook_endpoint_path = f"/alert/{notification_channel_name}"
+    webhook_endpoint = notification_channel.container_configs["8080"].get(
+        webhook_endpoint_path
+    )
+
+    # register the mock endpoint in notification channel
+    make_http_mocks(
+        notification_channel,
+        [
+            Mapping(
+                request=MappingRequest(
+                    method=HttpMethods.POST,
+                    url=webhook_endpoint_path,
+                ),
+                response=MappingResponse(
+                    status=200,
+                    json_body={},
+                ),
+                persistent=False,
+            )
+        ],
+    )
+
+    # Create an alert channel using the given route
+    create_webhook_notification_channel(
+        channel_name=notification_channel_name,
+        webhook_url=webhook_endpoint,
+        http_config={},
+        send_resolved=True,
+    )
+
+    # TODO: @abhishekhugetech # pylint: disable=W0511
+    # Time required for Org to be registered
+    # in the alertmanager, default 1m.
+    # this will be fixed after [https://github.com/SigNoz/engineering-pod/issues/3800]
+    time.sleep(65)
+
+    # Call test API for the notification channel
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    response = requests.post(
+        url=signoz.self.host_configs["8080"].get("/api/v1/testChannel"),
+        json={
+            "name": notification_channel_name,
+            "webhook_configs": [
+                {
+                    "send_resolved": True,
+                    "url": webhook_endpoint,
+                    "http_config": {},
+                }
+            ],
+        },
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=5,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, (
+        f"Failed to create notification channel: {response.text}"
+        f"Status code: {response.status_code}"
+    )
+
+    # Verify that the alert was sent to the notification channel
+    response = requests.post(
+        url=notification_channel.host_configs["8080"].get("/__admin/requests/count"),
+        json={"method": "POST", "url": webhook_endpoint_path},
+        timeout=5,
+    )
+    assert response.status_code == HTTPStatus.OK, (
+        f"Failed to get test notification count: {response.text}"
+        f"Status code: {response.status_code}"
+    )
+    # Verify that the test notification was sent to the notification channel
+    assert (
+        response.json()["count"] == 1
+    ), f"Expected 1 test notification, got {response.json()['count']}"