feat(alert-channel-integrations): integration tests using wiremock for jsm ops

- testChannel suite: create+note success (GenieKey auth, HTML body, default
  tags/source), failed-note tolerance (404 note still delivers), permanent
  422/401 on create with the note never attempted
- firing-path suite: metrics/logs default-template alerts (HTML deep links),
  create 429 retry with a single note after the successful create
- webhook verification and journal lookups accept a path regex (path_pattern)
  for paths embedding the group-hash alias

tests/fixtures/alerts.py

@@ -349,20 +349,40 @@ def verify_webhook_notification_expectation(
     notification_channel: types.TestContainerDocker,
     validation_data: dict,
 ) -> bool:
-    """Check if wiremock received a request at the given path
-    whose JSON body is a superset of the expected json_body."""
-    path = validation_data["path"]
-    json_body = validation_data["json_body"]
+    """Check that wiremock received the expected request(s) at the given path.
+
+    validation_data supports (all optional except one of path/path_pattern):
+      - path: request url path (matched as urlPath, so query strings are ignored)
+      - path_pattern: url path regex instead of path, for paths that embed a
+        dynamic segment (e.g. a group-hash alias)
+      - json_body: expected JSON subset of the request body
+      - count: exact number of requests required at the path
+      - min_count: minimum number of requests required (e.g. retries)
+    The body constraint must be satisfied by a single request; count constraints
+    apply to the total at the path."""
+    path = validation_data.get("path")
+    json_body = validation_data.get("json_body")
+    # urlPath ignores query strings; real webhook urls may carry their own (e.g. key/token).
+    matcher = {"method": "POST", "urlPath": path} if path is not None else {"method": "POST", "urlPathPattern": validation_data["path_pattern"]}
 
     url = notification_channel.host_configs["8080"].get("__admin/requests/find")
     try:
-        res = requests.post(url, json={"method": "POST", "url": path}, timeout=10)
+        res = requests.post(url, json=matcher, timeout=10)
     except requests.exceptions.RequestException:
         return False
     if res.status_code != HTTPStatus.OK:
         return False
 
-    for req in res.json()["requests"]:
+    reqs = res.json()["requests"]
+    if "count" in validation_data and len(reqs) != validation_data["count"]:
+        return False
+    if "min_count" in validation_data and len(reqs) < validation_data["min_count"]:
+        return False
+
+    if json_body is None:
+        return True
+
+    for req in reqs:
         body = json.loads(base64.b64decode(req["bodyAsBase64"]).decode("utf-8"))
         if _is_json_subset(json_body, body):
             return True
@@ -425,8 +445,10 @@ def _received_notifications(
             if validation.destination_type != "webhook":
                 continue
             url = notification_channel.host_configs["8080"].get("__admin/requests/find")
+            path = validation.validation_data.get("path")
+            matcher = {"method": "POST", "urlPath": path} if path is not None else {"method": "POST", "urlPathPattern": validation.validation_data["path_pattern"]}
             try:
-                res = requests.post(url, json={"method": "POST", "url": validation.validation_data["path"]}, timeout=10)
+                res = requests.post(url, json=matcher, timeout=10)
                 webhook_bodies.extend(json.loads(base64.b64decode(req["bodyAsBase64"]).decode("utf-8")) for req in res.json()["requests"])
             except requests.exceptions.RequestException as exc:
                 webhook_bodies.append(f"<failed to fetch wiremock journal: {exc}>")
@@ -465,4 +487,9 @@ def update_raw_channel_config(
                     path = urlparse(original_url).path
                     entry[url_field] = notification_channel.container_configs["8080"].get(path)
 
+    # Google Chat validates the webhook host
+    for entry in config.get("googlechat_configs", []):
+        https = notification_channel.container_configs["443"]
+        entry["webhook_url"] = f"{https.scheme}://{https.address}{urlparse(entry['webhook_url']).path}"
+
     return config

tests/fixtures/clickhouse.py

@@ -222,7 +222,12 @@ def create_clickhouse(  # pylint: disable=too-many-arguments,too-many-positional
             remote_servers=render_remote_servers([("127.0.0.1", 9000)]),
         )
 
-        tmp_dir = tmpfs(cache_key)
+        # The mounted configs cannot live in tmpfs: pytest wipes basetemp at
+        # every session start, and clickhouse hot-reloads config.d, so a reused
+        # container would silently lose its cluster definition. Like the CA,
+        # each container gets a fresh directory in the cross-session cache.
+        tmp_dir = pytestconfig.cache.mkdir(f"{cache_key}-config") / uuid4().hex
+        tmp_dir.mkdir()
         cluster_config_file_path = os.path.join(tmp_dir, "cluster.xml")
         with open(cluster_config_file_path, "w", encoding="utf-8") as f:
             f.write(cluster_config)
@@ -406,7 +411,10 @@ def create_clickhouse_cluster(  # pylint: disable=too-many-arguments,too-many-po
                     distributed_ddl_path=distributed_ddl_path,
                 )
 
-                tmp_dir = tmpfs(f"clickhouse-{suffix}-{i:02d}")
+                # Not tmpfs: see create_clickhouse — basetemp wipes would make
+                # reused nodes lose their hot-reloaded cluster definition.
+                tmp_dir = pytestconfig.cache.mkdir(f"{cache_key}-config") / f"{suffix}-{i:02d}"
+                tmp_dir.mkdir()
                 cluster_config_file_path = os.path.join(tmp_dir, "cluster.xml")
                 with open(cluster_config_file_path, "w", encoding="utf-8") as f:
                     f.write(node_config)

tests/fixtures/notification_channel.py

@@ -1,23 +1,46 @@
 # pylint: disable=line-too-long
 import json
+import re
 import time
+import uuid
 from collections.abc import Callable
 from http import HTTPStatus
+from pathlib import Path
 
 import docker
 import docker.errors
 import pytest
 import requests
 from testcontainers.core.container import Network
+from wiremock.resources.mappings import HttpMethods, Mapping, MappingRequest, MappingResponse
 from wiremock.testing.testcontainer import WireMockContainer
 
 from fixtures import reuse, types
 from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
 from fixtures.logger import setup_logger
 from fixtures.maildev import MAILDEV_INCOMING_PASS, SMTP_TEST_FROM
+from fixtures.tls import CA_ID_LABEL, KEYSTORE_PASSWORD, ca_id, issue_server_keystore
 
 logger = setup_logger(__name__)
 
+# Google Chat validates the webhook host, so the WireMock container joins the
+# network under this alias and serves HTTPS on 443 with a certificate issued by
+# the integration CA that signoz trusts; channels point at https://<host>/...
+GOOGLE_CHAT_HOST = "chat.googleapis.com"
+# incident.io doesn't pin the host, but the same alias trick keeps channel URLs
+# identical to production ones.
+INCIDENTIO_HOST = "api.incident.io"
+# Jira validates the site host (*.atlassian.net); service accounts additionally
+# go through the api.atlassian.com gateway.
+JIRA_HOST = "signoz-test.atlassian.net"
+ATLASSIAN_API_HOST = "api.atlassian.com"
+TLS_HOSTS = [GOOGLE_CHAT_HOST, INCIDENTIO_HOST, JIRA_HOST, ATLASSIAN_API_HOST]
+
+# A reused container serving a cert without a newly added host (or missing its
+# network alias) fails TLS opaquely; this label records the hosts it was built
+# for so stale() recreates it when the list changes.
+TLS_HOSTS_LABEL = "signoz.integration.tls-hosts"
+
 
 EMAIL_TRANSPORT_KEYS = [
     "from",
@@ -124,9 +147,363 @@ email_default_config = {
 }
 
 
+def googlechat_config(space: str) -> dict:
+    """Google Chat channel config for a per-test WireMock space path. Title/text are
+    omitted so the backend applies its default templates. The host is injected at
+    runtime by update_raw_channel_config."""
+    return {
+        "googlechat_configs": [
+            {
+                "webhook_url": f"/v1/spaces/{space}/messages",  # host set on runtime
+            }
+        ],
+    }
+
+
+def googlechat_ok_mappings(path: str) -> list[Mapping]:
+    return [
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=path),
+            response=MappingResponse(status=200, json_body={"name": "spaces/x/messages/x"}),
+        )
+    ]
+
+
+def googlechat_retry_mappings(path: str) -> list[Mapping]:
+    """429 on the first call then 200, via a wiremock scenario transition."""
+    scenario = f"gc-retry-{path}"
+    return [
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=path),
+            response=MappingResponse(status=429, json_body={"error": {"code": 429, "status": "RESOURCE_EXHAUSTED"}}),
+            scenario_name=scenario,
+            required_scenario_state="Started",
+            new_scenario_state="ok",
+        ),
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=path),
+            response=MappingResponse(status=200, json_body={"name": "spaces/x/messages/x"}),
+            scenario_name=scenario,
+            required_scenario_state="ok",
+        ),
+    ]
+
+
+def googlechat_card_subset(alertname: str, buttons: list[tuple[str, str]]) -> dict:
+    """A cardsV2 subset asserting title, firing banner, rendered body, and each
+    button's text AND deep-link url (as a regex), so a broken link is caught too.
+    buttons: list of (text, url_regex)."""
+    return {
+        "text": f"[FIRING:1] {alertname}",
+        "cardsV2": [
+            {
+                "cardId": "signoz-alert",
+                "card": {
+                    "header": {"title": f"[FIRING:1] {alertname}"},
+                    "sections": [
+                        # firing banner
+                        {"widgets": [{"textParagraph": {"text": re.compile("FIRING")}}]},
+                        # rendered alert body mentions the alertname
+                        {"widgets": [{"textParagraph": {"text": re.compile(re.escape(alertname))}}]},
+                    ]
+                    + [{"widgets": [{"buttonList": {"buttons": [{"text": text, "onClick": {"openLink": {"url": re.compile(url)}}}]}}]} for text, url in buttons],
+                },
+            }
+        ],
+    }
+
+
+INCIDENTIO_TEST_TOKEN = "incidentio-test-token"  # noqa: S105
+
+
+def incidentio_path(source_id: str) -> str:
+    return f"/v2/alert_events/http/{source_id}"
+
+
+def incidentio_config(source_id: str) -> dict:
+    """incident.io channel config for a per-test alert source id. Title/description
+    are omitted so the backend applies its default templates. The URL host is the
+    wiremock network alias, so no runtime injection is needed."""
+    return {
+        "incidentio_configs": [
+            {
+                "url": f"https://{INCIDENTIO_HOST}{incidentio_path(source_id)}",
+                "token": INCIDENTIO_TEST_TOKEN,
+            }
+        ],
+    }
+
+
+# recorded incident.io Alert Events V2 responses: 202 accepted-for-processing
+# echoing the dedup key; errors are {type, status, errors: [{code, message}]}
+def incidentio_ok_mappings(path: str) -> list[Mapping]:
+    return [
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=path),
+            response=MappingResponse(status=202, json_body={"status": "accepted", "message": "Event accepted for processing", "deduplication_key": "x"}),
+        )
+    ]
+
+
+def incidentio_retry_mappings(path: str) -> list[Mapping]:
+    """429 on the first call then 202, via a wiremock scenario transition."""
+    scenario = f"incidentio-retry-{path}"
+    return [
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=path),
+            response=MappingResponse(status=429, json_body={"type": "rate_limit_error", "status": 429}),
+            scenario_name=scenario,
+            required_scenario_state="Started",
+            new_scenario_state="ok",
+        ),
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=path),
+            response=MappingResponse(status=202, json_body={"status": "accepted", "message": "Event accepted for processing", "deduplication_key": "x"}),
+            scenario_name=scenario,
+            required_scenario_state="ok",
+        ),
+    ]
+
+
+def incidentio_event_subset(alertname: str, links: list[tuple[str, str]]) -> dict:
+    """An alert-event subset asserting title, firing status, dedup key, SigNoz
+    source_url, metadata labels, and each markdown link's text AND url (as a
+    regex), so a broken link is caught too. links: (text, url_regex) pairs in
+    default-template order (View in SigNoz -> related logs -> related traces)."""
+    description = "(?s)" + re.escape(f"**Alert:** {alertname}")
+    for text, url in links:
+        description += rf".*\[{re.escape(text)}\]\([^)]*{url}"
+    return {
+        "title": f"[FIRING:1] {alertname}",
+        "status": "firing",
+        "deduplication_key": re.compile(r".+"),
+        "source_url": re.compile(r"/alerts/overview\?ruleId="),
+        "description": re.compile(description),
+        "metadata": {"alertname": alertname},
+    }
+
+
+JIRA_TEST_EMAIL = "user@acme.io"
+JIRA_SA_EMAIL = "svc@serviceaccount.atlassian.com"
+JIRA_TEST_TOKEN = "jira-test-token"  # noqa: S105
+JIRA_API_BASE = "/rest/api/3"
+
+
+def jira_config(**overrides) -> dict:
+    """Jira channel config against the wiremock atlassian.net alias, personal
+    API token auth. Summary/description are omitted so the backend applies its
+    default templates; overrides lay extra receiver fields on top."""
+    return {
+        "jira_configs": [
+            {
+                "site": f"https://{JIRA_HOST}",
+                "project": "OPS",
+                "issue_type": "Task",
+                "http_config": {"basic_auth": {"username": JIRA_TEST_EMAIL, "password": JIRA_TEST_TOKEN}},
+                **overrides,
+            }
+        ],
+    }
+
+
+def jira_search_issue(key: str, done: bool, labels: list[str]) -> dict:
+    """One issue as returned by the /search/jql stub, with the fields the
+    notifier requests (status category + labels)."""
+    return {
+        "key": key,
+        "fields": {"status": {"statusCategory": {"key": "done" if done else "indeterminate"}}, "labels": labels},
+    }
+
+
+# Jira flows span several endpoints; each mapping helper stubs one, on any base
+# (site host for personal tokens, /ex/jira/<cloud_id> gateway for service accounts).
+def jira_search_mapping(issues: list[dict], base: str = JIRA_API_BASE) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.POST, url_path=f"{base}/search/jql"),
+        response=MappingResponse(status=200, json_body={"issues": issues}),
+    )
+
+
+def jira_create_mapping(key: str = "OPS-1", base: str = JIRA_API_BASE) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.POST, url_path=f"{base}/issue"),
+        response=MappingResponse(status=201, json_body={"id": "10001", "key": key}),
+    )
+
+
+def jira_update_mapping(key: str, base: str = JIRA_API_BASE) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.PUT, url_path=f"{base}/issue/{key}"),
+        response=MappingResponse(status=204),
+    )
+
+
+def jira_transitions_mapping(key: str, transitions: list[dict], base: str = JIRA_API_BASE) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.GET, url_path=f"{base}/issue/{key}/transitions"),
+        response=MappingResponse(status=200, json_body={"transitions": transitions}),
+    )
+
+
+def jira_transition_post_mapping(key: str, base: str = JIRA_API_BASE) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.POST, url_path=f"{base}/issue/{key}/transitions"),
+        response=MappingResponse(status=204),
+    )
+
+
+def jira_comment_mapping(key: str, base: str = JIRA_API_BASE) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.POST, url_path=f"{base}/issue/{key}/comment"),
+        response=MappingResponse(status=201, json_body={"id": "1"}),
+    )
+
+
+def jira_retry_search_mappings() -> list[Mapping]:
+    """429 on the first search then 200-empty, via a wiremock scenario transition."""
+    scenario = "jira-retry-search"
+    return [
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=f"{JIRA_API_BASE}/search/jql"),
+            response=MappingResponse(status=429, json_body={"errorMessages": ["Rate limit exceeded"]}),
+            scenario_name=scenario,
+            required_scenario_state="Started",
+            new_scenario_state="ok",
+        ),
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=f"{JIRA_API_BASE}/search/jql"),
+            response=MappingResponse(status=200, json_body={"issues": []}),
+            scenario_name=scenario,
+            required_scenario_state="ok",
+        ),
+    ]
+
+
+def find_requests(notification_channel: types.TestContainerDocker, method: str, path: str | None = None, path_pattern: str | None = None) -> list[dict]:
+    """The wiremock journal entries for method+path (query strings ignored);
+    path_pattern matches the path as a regex instead, for paths that embed a
+    dynamic segment like the group-hash alias."""
+    matcher = {"method": method, "urlPath": path} if path is not None else {"method": method, "urlPathPattern": path_pattern}
+    find = requests.post(
+        notification_channel.host_configs["8080"].get("/__admin/requests/find"),
+        json=matcher,
+        timeout=10,
+    )
+    return find.json()["requests"]
+
+
+JSMOPS_TEST_API_KEY = "jsmops-test-api-key"  # noqa: S105
+# The JSM Ops gateway lives on api.atlassian.com (already aliased for Jira
+# service accounts); the notifier appends v2/alerts... to this base.
+JSMOPS_API_BASE = "/jsm/ops/integration"
+JSMOPS_NOTES_PATH_PATTERN = f"{JSMOPS_API_BASE}/v2/alerts/[a-f0-9]+/notes"
+
+
+def jsmops_config(**overrides) -> dict:
+    """JSM Ops channel config. Message/description/tags are omitted so the
+    backend applies its defaults; overrides lay extra receiver fields on top."""
+    return {
+        "jsmops_configs": [
+            {
+                "api_key": JSMOPS_TEST_API_KEY,
+                **overrides,
+            }
+        ],
+    }
+
+
+def jsmops_create_mapping() -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.POST, url_path=f"{JSMOPS_API_BASE}/v2/alerts"),
+        response=MappingResponse(status=202, json_body={"result": "Request will be processed", "took": 0.005, "requestId": "1b1f0000-0000-4000-8000-000000000001"}),
+    )
+
+
+def jsmops_notes_mapping(status: int = 202, body: dict | None = None) -> Mapping:
+    return Mapping(
+        request=MappingRequest(method=HttpMethods.POST, url_path_pattern=JSMOPS_NOTES_PATH_PATTERN),
+        response=MappingResponse(status=status, json_body=body or {"result": "Request will be processed", "took": 0.002, "requestId": "1b1f0000-0000-4000-8000-000000000002"}),
+    )
+
+
+def jsmops_retry_create_mappings() -> list[Mapping]:
+    """429 on the first create then 202, via a wiremock scenario transition."""
+    scenario = "jsmops-retry-create"
+    return [
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=f"{JSMOPS_API_BASE}/v2/alerts"),
+            response=MappingResponse(status=429, json_body={"message": "You are making too many requests!", "took": 0.001, "requestId": "x"}),
+            scenario_name=scenario,
+            required_scenario_state="Started",
+            new_scenario_state="ok",
+        ),
+        Mapping(
+            request=MappingRequest(method=HttpMethods.POST, url_path=f"{JSMOPS_API_BASE}/v2/alerts"),
+            response=MappingResponse(status=202, json_body={"result": "Request will be processed", "took": 0.005, "requestId": "x"}),
+            scenario_name=scenario,
+            required_scenario_state="ok",
+        ),
+    ]
+
+
+def jsmops_alert_subset(alertname: str, links: list[tuple[str, str]]) -> dict:
+    """A created-alert subset asserting message, alias, source, default tags,
+    details labels, and the HTML description: the rendered bold Alert run plus
+    each link's anchor (href as a regex), so a broken link is caught too.
+    links: (text, url_regex) pairs in default-template order."""
+    description = "(?s)" + re.escape("<strong>Alert:</strong>")
+    for text, url in links:
+        description += rf'.*<a href="[^"]*{url}[^"]*"[^>]*>{re.escape(text)}</a>'
+    return {
+        "alias": re.compile(r".+"),
+        "message": f"[FIRING:1] {alertname}",
+        "source": "SigNoz",
+        "tags": ["signoz"],
+        "details": {"alertname": alertname},
+        "description": re.compile(description),
+    }
+
+
+def jira_issue_subset(alertname: str, links: list[tuple[str, str]]) -> dict:
+    """A created-issue subset asserting summary, group labels, ADF status panel,
+    the rendered alert text, and each deep-link's text AND url (as a regex), so
+    a broken link is caught too. links: (text, url_regex) pairs."""
+    # the ADF renderer splits text nodes at underscores, so the alertname never
+    # sits in one node; the summary pins it exactly, the body asserts the
+    # rendered "Alert:" strong run followed by the name's first fragment
+    description_content = [
+        {"type": "panel", "attrs": {"panelType": "error"}},
+        {
+            "type": "paragraph",
+            "content": [
+                {"type": "text", "text": "Alert:", "marks": [{"type": "strong"}]},
+                {"type": "text", "text": re.compile(re.escape(alertname.split("_", maxsplit=1)[0]))},
+            ],
+        },
+    ]
+    if links:
+        description_content.append(
+            {
+                "type": "paragraph",
+                "content": [{"type": "text", "text": text, "marks": [{"type": "link", "attrs": {"href": re.compile(url)}}]} for text, url in links],
+            }
+        )
+    return {
+        "fields": {
+            "project": {"key": "OPS"},
+            "issuetype": {"name": "Task"},
+            "summary": f"[FIRING:1] {alertname}",
+            "labels": ["signoz-alert", re.compile(r"ALERT\{")],
+            "description": {"type": "doc", "version": 1, "content": description_content},
+        },
+    }
+
+
 @pytest.fixture(name="notification_channel", scope="package")
-def notification_channel(
+def notification_channel(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     network: Network,
+    tls: types.TLS,
+    tmpfs: Callable[[str], Path],
     request: pytest.FixtureRequest,
     pytestconfig: pytest.Config,
 ) -> types.TestContainerDocker:
@@ -135,9 +512,25 @@ def notification_channel(
     """
 
     def create() -> types.TestContainerDocker:
+        # http:8080 for admin API + plain webhook delivery; https:443 aliased as
+        # chat.googleapis.com with a CA-issued cert so Google Chat's validated
+        # webhook host routes here over real TLS (signoz trusts the integration CA).
+        keystore_path = issue_server_keystore(tls, tmpfs("notification-channel-certs"), *TLS_HOSTS)
+
         container = WireMockContainer(image="wiremock/wiremock:2.35.1-1", secure=False)
+        container.with_volume_mapping(str(keystore_path.parent), "/certs", "ro")
         container.with_network(network)
-        container.start()
+        container.with_network_aliases(*TLS_HOSTS)
+        container.with_kwargs(labels={CA_ID_LABEL: ca_id(tls), TLS_HOSTS_LABEL: ",".join(TLS_HOSTS)})
+
+        try:
+            container.start(f"--port 8080 --https-port 443 --https-keystore /certs/keystore.p12 --keystore-type PKCS12 --keystore-password {KEYSTORE_PASSWORD}")
+        except Exception:
+            # Ryuk is disabled: a started-but-unready container would survive and
+            # keep squatting on the chat.googleapis.com alias, poisoning DNS for
+            # any replacement on the shared network.
+            container.stop()
+            raise
 
         return types.TestContainerDocker(
             id=container.get_wrapped_container().id,
@@ -148,7 +541,11 @@ def notification_channel(
                     container.get_exposed_port(8080),
                 )
             },
-            container_configs={"8080": types.TestContainerUrlConfig("http", container.get_wrapped_container().name, 8080)},
+            container_configs={
+                "8080": types.TestContainerUrlConfig("http", container.get_wrapped_container().name, 8080),
+                # Google Chat delivery: https to the validated host via the network alias.
+                "443": types.TestContainerUrlConfig("https", GOOGLE_CHAT_HOST, 443),
+            },
         )
 
     def delete(container: types.TestContainerDocker):
@@ -165,6 +562,16 @@ def notification_channel(
     def restore(cache: dict) -> types.TestContainerDocker:
         return types.TestContainerDocker.from_cache(cache)
 
+    def stale(container: types.TestContainerDocker) -> bool:
+        # A container built against a rotated/absent CA can't serve a cert signoz
+        # trusts; recreate it instead of failing TLS opaquely.
+        client = docker.from_env()
+        try:
+            labels = client.containers.get(container_id=container.id).attrs["Config"]["Labels"]
+        except docker.errors.NotFound:
+            return True
+        return labels.get(CA_ID_LABEL) != ca_id(tls) or labels.get(TLS_HOSTS_LABEL) != ",".join(TLS_HOSTS)
+
     return reuse.wrap(
         request,
         pytestconfig,
@@ -173,6 +580,7 @@ def notification_channel(
         create,
         delete,
         restore,
+        stale=stale,
     )
 
 
@@ -248,6 +656,31 @@ def create_webhook_notification_channel(
     return _create_webhook_notification_channel
 
 
+def wait_for_org_registration(signoz: types.SigNoz, token: str, notification_channel: types.TestContainerDocker, wait_seconds: int = 60) -> None:
+    """Polls until the org's alertmanager server is registered (one poll tick).
+
+    channels/test 404s until then, before reaching any notifier. The sentinel
+    receiver posts to its own unstubbed wiremock path, so request journals
+    asserted by tests stay clean."""
+    sentinel = {
+        "name": str(uuid.uuid4()),
+        "webhook_configs": [{"url": notification_channel.container_configs["8080"].get("/org-registration-sentinel")}],
+    }
+    deadline = time.time() + wait_seconds
+    last = None
+    while time.time() < deadline:
+        last = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+            json=sentinel,
+            headers={"Authorization": f"Bearer {token}"},
+            timeout=30,
+        )
+        if last.status_code != HTTPStatus.NOT_FOUND:
+            return
+        time.sleep(2)
+    raise AssertionError(f"org alertmanager did not register within {wait_seconds}s, last response: {last.status_code} {last.text}")
+
+
 def send_test_notification(signoz: types.SigNoz, token: str, receiver: dict, wait_seconds: int = 90) -> None:
     deadline = time.time() + wait_seconds
     last = None

tests/fixtures/tls.py

@@ -107,10 +107,11 @@ def tls(
     )
 
 
-def issue_server_keystore(tls: types.TLS, directory: Path, hostname: str) -> Path:
+def issue_server_keystore(tls: types.TLS, directory: Path, *hostnames: str) -> Path:
     """Write a PKCS12 keystore (keystore.p12, password KEYSTORE_PASSWORD) into
-    directory, holding a certificate for hostname issued by the integration CA.
-    Mount it into a mock container that must serve TLS as hostname."""
+    directory, holding a certificate for the hostnames (SANs, CN is the first)
+    issued by the integration CA. Mount it into a mock container that must
+    serve TLS as those hostnames."""
     ca_cert = x509.load_pem_x509_certificate(Path(tls.ca_cert_path).read_bytes())
     ca_key = serialization.load_pem_private_key(Path(tls.ca_key_path).read_bytes(), password=None)
 
@@ -118,13 +119,13 @@ def issue_server_keystore(tls: types.TLS, directory: Path, hostname: str) -> Pat
     leaf_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
     leaf_cert = (
         x509.CertificateBuilder()
-        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostname)]))
+        .subject_name(x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, hostnames[0])]))
         .issuer_name(ca_cert.subject)
         .public_key(leaf_key.public_key())
         .serial_number(x509.random_serial_number())
         .not_valid_before(now - datetime.timedelta(days=1))
         .not_valid_after(now + datetime.timedelta(days=3650))
-        .add_extension(x509.SubjectAlternativeName([x509.DNSName(hostname)]), critical=False)
+        .add_extension(x509.SubjectAlternativeName([x509.DNSName(hostname) for hostname in hostnames]), critical=False)
         .add_extension(x509.ExtendedKeyUsage([x509.oid.ExtendedKeyUsageOID.SERVER_AUTH]), critical=False)
         .sign(ca_key, hashes.SHA256())
     )
@@ -132,7 +133,7 @@ def issue_server_keystore(tls: types.TLS, directory: Path, hostname: str) -> Pat
     keystore_path = directory / "keystore.p12"
     keystore_path.write_bytes(
         pkcs12.serialize_key_and_certificates(
-            name=hostname.encode(),
+            name=hostnames[0].encode(),
             key=leaf_key,
             cert=leaf_cert,
             cas=[ca_cert],

tests/integration/tests/alertmanager/04_googlechat.py

@@ -0,0 +1,187 @@
+import json
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from wiremock.resources.mappings import Mapping
+
+from fixtures import types
+from fixtures.alerts import (
+    get_testdata_file_path,
+    update_raw_channel_config,
+    update_rule_channel_name,
+    verify_notification_expectation,
+)
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import (
+    googlechat_card_subset,
+    googlechat_config,
+    googlechat_ok_mappings,
+    googlechat_retry_mappings,
+    wait_for_org_registration,
+)
+
+logger = setup_logger(__name__)
+
+METRICS_DATA = "alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
+METRICS_RULE = "alerts/test_scenarios/threshold_above_at_least_once/rule.json"
+LOGS_DATA = "alerts/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"
+LOGS_RULE = "alerts/test_scenarios/threshold_below_at_least_once/rule.json"
+TRACES_DATA = "alerts/test_scenarios/threshold_above_average/alert_data.jsonl"
+TRACES_RULE = "alerts/test_scenarios/threshold_above_average/rule.json"
+
+
+GOOGLECHAT_CASES = [
+    types.AlertManagerNotificationTestCase(
+        name="googlechat_default_metrics_firing",
+        rule_path=METRICS_RULE,
+        alert_data=[types.AlertData(type="metrics", data_path=METRICS_DATA)],
+        channel_config=googlechat_config("gc-metrics"),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v1/spaces/gc-metrics/messages",
+                        "count": 1,
+                        "json_body": googlechat_card_subset("threshold_above_at_least_once", [("Open in SigNoz", r"/alerts/overview\?ruleId=")]),
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="googlechat_rich_card_logs",
+        rule_path=LOGS_RULE,
+        alert_data=[types.AlertData(type="logs", data_path=LOGS_DATA)],
+        channel_config=googlechat_config("gc-logs"),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v1/spaces/gc-logs/messages",
+                        "count": 1,
+                        "json_body": googlechat_card_subset(
+                            "threshold_below_at_least_once",
+                            [("View Related Logs", r"/logs/logs-explorer\?"), ("Open in SigNoz", r"/alerts/overview\?ruleId=")],
+                        ),
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="googlechat_rich_card_traces",
+        rule_path=TRACES_RULE,
+        alert_data=[types.AlertData(type="traces", data_path=TRACES_DATA)],
+        channel_config=googlechat_config("gc-traces"),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": "/v1/spaces/gc-traces/messages",
+                        "count": 1,
+                        "json_body": googlechat_card_subset(
+                            "threshold_above_average",
+                            [("View Related Traces", r"traces-explorer\?"), ("Open in SigNoz", r"/alerts/overview\?ruleId=")],
+                        ),
+                    },
+                ),
+            ],
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "gc_test_case",
+    GOOGLECHAT_CASES,
+    ids=lambda c: c.name,
+)
+def test_googlechat_notifier(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+    gc_test_case: types.AlertManagerNotificationTestCase,
+) -> None:
+    channel_name = str(uuid.uuid4())
+    path = gc_test_case.notification_expectation.notification_validations[0].validation_data["path"]
+
+    channel_config = update_raw_channel_config(gc_test_case.channel_config, channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, googlechat_ok_mappings(path))
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data(gc_test_case.alert_data, base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(gc_test_case.rule_path), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(notification_channel, maildev, gc_test_case.notification_expectation)
+
+
+def test_googlechat_retry_429_then_200(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+) -> None:
+    channel_name = str(uuid.uuid4())
+    path = "/v1/spaces/gc-retry/messages"
+
+    channel_config = update_raw_channel_config(googlechat_config("gc-retry"), channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, googlechat_retry_mappings(path))
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data([types.AlertData(type="metrics", data_path=METRICS_DATA)], base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(METRICS_RULE), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(
+        notification_channel,
+        maildev,
+        types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        # a retryable 429 is followed by a successful re-POST => >=2 hits
+                        "path": path,
+                        "min_count": 2,
+                        "json_body": {"cardsV2": [{"cardId": "signoz-alert"}]},
+                    },
+                ),
+            ],
+        ),
+    )

tests/integration/tests/alertmanager/05_incidentio.py

@@ -0,0 +1,163 @@
+import json
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from wiremock.resources.mappings import Mapping
+
+from fixtures import types
+from fixtures.alerts import (
+    get_testdata_file_path,
+    update_raw_channel_config,
+    update_rule_channel_name,
+    verify_notification_expectation,
+)
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import (
+    incidentio_config,
+    incidentio_event_subset,
+    incidentio_ok_mappings,
+    incidentio_path,
+    incidentio_retry_mappings,
+    wait_for_org_registration,
+)
+
+logger = setup_logger(__name__)
+
+METRICS_DATA = "alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
+METRICS_RULE = "alerts/test_scenarios/threshold_above_at_least_once/rule.json"
+LOGS_DATA = "alerts/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"
+LOGS_RULE = "alerts/test_scenarios/threshold_below_at_least_once/rule.json"
+
+
+INCIDENTIO_CASES = [
+    types.AlertManagerNotificationTestCase(
+        name="incidentio_default_metrics_firing",
+        rule_path=METRICS_RULE,
+        alert_data=[types.AlertData(type="metrics", data_path=METRICS_DATA)],
+        channel_config=incidentio_config("inc-metrics"),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": incidentio_path("inc-metrics"),
+                        "count": 1,
+                        "json_body": incidentio_event_subset("threshold_above_at_least_once", [("View in SigNoz", r"/alerts/overview\?ruleId=")]),
+                    },
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="incidentio_rich_event_logs",
+        rule_path=LOGS_RULE,
+        alert_data=[types.AlertData(type="logs", data_path=LOGS_DATA)],
+        channel_config=incidentio_config("inc-logs"),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": incidentio_path("inc-logs"),
+                        "count": 1,
+                        "json_body": incidentio_event_subset(
+                            "threshold_below_at_least_once",
+                            [("View in SigNoz", r"/alerts/overview\?ruleId="), ("View related logs", r"/logs/logs-explorer\?")],
+                        ),
+                    },
+                ),
+            ],
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "incidentio_test_case",
+    INCIDENTIO_CASES,
+    ids=lambda c: c.name,
+)
+def test_incidentio_notifier(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+    incidentio_test_case: types.AlertManagerNotificationTestCase,
+) -> None:
+    channel_name = str(uuid.uuid4())
+    path = incidentio_test_case.notification_expectation.notification_validations[0].validation_data["path"]
+
+    channel_config = update_raw_channel_config(incidentio_test_case.channel_config, channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, incidentio_ok_mappings(path))
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data(incidentio_test_case.alert_data, base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(incidentio_test_case.rule_path), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(notification_channel, maildev, incidentio_test_case.notification_expectation)
+
+
+def test_incidentio_retry_429_then_202(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+) -> None:
+    channel_name = str(uuid.uuid4())
+    path = incidentio_path("inc-retry")
+
+    channel_config = update_raw_channel_config(incidentio_config("inc-retry"), channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, incidentio_retry_mappings(path))
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data([types.AlertData(type="metrics", data_path=METRICS_DATA)], base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(METRICS_RULE), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(
+        notification_channel,
+        maildev,
+        types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        # a retryable 429 is followed by a successful re-POST => >=2 hits
+                        "path": path,
+                        "min_count": 2,
+                        "json_body": {"status": "firing"},
+                    },
+                ),
+            ],
+        ),
+    )

tests/integration/tests/alertmanager/06_jira.py

@@ -0,0 +1,167 @@
+import json
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from wiremock.resources.mappings import Mapping
+
+from fixtures import types
+from fixtures.alerts import (
+    get_testdata_file_path,
+    update_raw_channel_config,
+    update_rule_channel_name,
+    verify_notification_expectation,
+)
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import (
+    JIRA_API_BASE,
+    jira_config,
+    jira_create_mapping,
+    jira_issue_subset,
+    jira_retry_search_mappings,
+    jira_search_mapping,
+    wait_for_org_registration,
+)
+
+logger = setup_logger(__name__)
+
+METRICS_DATA = "alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
+METRICS_RULE = "alerts/test_scenarios/threshold_above_at_least_once/rule.json"
+LOGS_DATA = "alerts/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"
+LOGS_RULE = "alerts/test_scenarios/threshold_below_at_least_once/rule.json"
+
+
+JIRA_CASES = [
+    types.AlertManagerNotificationTestCase(
+        name="jira_default_metrics_firing",
+        rule_path=METRICS_RULE,
+        alert_data=[types.AlertData(type="metrics", data_path=METRICS_DATA)],
+        channel_config=jira_config(),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": f"{JIRA_API_BASE}/issue",
+                        "count": 1,
+                        "json_body": jira_issue_subset("threshold_above_at_least_once", [("Open in SigNoz", r"/alerts/overview\?ruleId=")]),
+                    },
+                ),
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={"path": f"{JIRA_API_BASE}/search/jql", "count": 1},
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="jira_rich_issue_logs",
+        rule_path=LOGS_RULE,
+        alert_data=[types.AlertData(type="logs", data_path=LOGS_DATA)],
+        channel_config=jira_config(),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": f"{JIRA_API_BASE}/issue",
+                        "count": 1,
+                        "json_body": jira_issue_subset(
+                            "threshold_below_at_least_once",
+                            [("Open in SigNoz", r"/alerts/overview\?ruleId="), ("View Related Logs", r"/logs/logs-explorer\?")],
+                        ),
+                    },
+                ),
+            ],
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "jira_test_case",
+    JIRA_CASES,
+    ids=lambda c: c.name,
+)
+def test_jira_notifier(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+    jira_test_case: types.AlertManagerNotificationTestCase,
+) -> None:
+    channel_name = str(uuid.uuid4())
+
+    channel_config = update_raw_channel_config(jira_test_case.channel_config, channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, [jira_search_mapping([]), jira_create_mapping()])
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data(jira_test_case.alert_data, base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(jira_test_case.rule_path), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(notification_channel, maildev, jira_test_case.notification_expectation)
+
+
+def test_jira_retry_429_then_200(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+) -> None:
+    channel_name = str(uuid.uuid4())
+
+    channel_config = update_raw_channel_config(jira_config(), channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, [*jira_retry_search_mappings(), jira_create_mapping()])
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data([types.AlertData(type="metrics", data_path=METRICS_DATA)], base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(METRICS_RULE), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(
+        notification_channel,
+        maildev,
+        types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    # a retryable 429 on the search re-runs the whole notify => >=2 searches
+                    validation_data={"path": f"{JIRA_API_BASE}/search/jql", "min_count": 2},
+                ),
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    # but the issue is still only created once
+                    validation_data={"path": f"{JIRA_API_BASE}/issue", "count": 1},
+                ),
+            ],
+        ),
+    )

tests/integration/tests/alertmanager/07_jsmops.py

@@ -0,0 +1,169 @@
+import json
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from wiremock.resources.mappings import Mapping
+
+from fixtures import types
+from fixtures.alerts import (
+    get_testdata_file_path,
+    update_raw_channel_config,
+    update_rule_channel_name,
+    verify_notification_expectation,
+)
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import (
+    JSMOPS_API_BASE,
+    JSMOPS_NOTES_PATH_PATTERN,
+    jsmops_alert_subset,
+    jsmops_config,
+    jsmops_create_mapping,
+    jsmops_notes_mapping,
+    jsmops_retry_create_mappings,
+    wait_for_org_registration,
+)
+
+logger = setup_logger(__name__)
+
+METRICS_DATA = "alerts/test_scenarios/threshold_above_at_least_once/alert_data.jsonl"
+METRICS_RULE = "alerts/test_scenarios/threshold_above_at_least_once/rule.json"
+LOGS_DATA = "alerts/test_scenarios/threshold_below_at_least_once/alert_data.jsonl"
+LOGS_RULE = "alerts/test_scenarios/threshold_below_at_least_once/rule.json"
+
+
+JSMOPS_CASES = [
+    types.AlertManagerNotificationTestCase(
+        name="jsmops_default_metrics_firing",
+        rule_path=METRICS_RULE,
+        alert_data=[types.AlertData(type="metrics", data_path=METRICS_DATA)],
+        channel_config=jsmops_config(),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": f"{JSMOPS_API_BASE}/v2/alerts",
+                        "count": 1,
+                        "json_body": jsmops_alert_subset("threshold_above_at_least_once", [("View in SigNoz", r"/alerts/overview\?ruleId=")]),
+                    },
+                ),
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    # every fire appends a timeline note
+                    validation_data={"path_pattern": JSMOPS_NOTES_PATH_PATTERN, "count": 1},
+                ),
+            ],
+        ),
+    ),
+    types.AlertManagerNotificationTestCase(
+        name="jsmops_rich_alert_logs",
+        rule_path=LOGS_RULE,
+        alert_data=[types.AlertData(type="logs", data_path=LOGS_DATA)],
+        channel_config=jsmops_config(),
+        notification_expectation=types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    validation_data={
+                        "path": f"{JSMOPS_API_BASE}/v2/alerts",
+                        "count": 1,
+                        "json_body": jsmops_alert_subset(
+                            "threshold_below_at_least_once",
+                            [("View in SigNoz", r"/alerts/overview\?ruleId="), ("View related logs", r"/logs/logs-explorer\?")],
+                        ),
+                    },
+                ),
+            ],
+        ),
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "jsmops_test_case",
+    JSMOPS_CASES,
+    ids=lambda c: c.name,
+)
+def test_jsmops_notifier(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+    jsmops_test_case: types.AlertManagerNotificationTestCase,
+) -> None:
+    channel_name = str(uuid.uuid4())
+
+    channel_config = update_raw_channel_config(jsmops_test_case.channel_config, channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, [jsmops_create_mapping(), jsmops_notes_mapping()])
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data(jsmops_test_case.alert_data, base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(jsmops_test_case.rule_path), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(notification_channel, maildev, jsmops_test_case.notification_expectation)
+
+
+def test_jsmops_retry_429_then_202(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    create_notification_channel: Callable[[dict], str],
+    create_alert_rule: Callable[[dict], str],
+    insert_alert_data: Callable[[list[types.AlertData], datetime], None],
+    maildev: types.TestContainerDocker,
+) -> None:
+    channel_name = str(uuid.uuid4())
+
+    channel_config = update_raw_channel_config(jsmops_config(), channel_name, notification_channel)
+
+    make_http_mocks(notification_channel, [*jsmops_retry_create_mappings(), jsmops_notes_mapping()])
+
+    create_notification_channel(channel_config)
+    wait_for_org_registration(signoz, get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD), notification_channel)
+
+    insert_alert_data([types.AlertData(type="metrics", data_path=METRICS_DATA)], base_time=datetime.now(tz=UTC) - timedelta(minutes=5))
+
+    with open(get_testdata_file_path(METRICS_RULE), encoding="utf-8") as f:
+        rule_data = json.loads(f.read())
+    update_rule_channel_name(rule_data, channel_name)
+    create_alert_rule(rule_data)
+
+    verify_notification_expectation(
+        notification_channel,
+        maildev,
+        types.AMNotificationExpectation(
+            should_notify=True,
+            wait_time_seconds=60,
+            notification_validations=[
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    # a retryable 429 on the create re-runs the whole notify => >=2 creates
+                    validation_data={"path": f"{JSMOPS_API_BASE}/v2/alerts", "min_count": 2},
+                ),
+                types.NotificationValidation(
+                    destination_type="webhook",
+                    # the note only goes out after the create succeeded
+                    validation_data={"path_pattern": JSMOPS_NOTES_PATH_PATTERN, "count": 1},
+                ),
+            ],
+        ),
+    )

tests/integration/tests/alertmanager/conftest.py

@@ -13,6 +13,7 @@ def signoz(  # pylint: disable=too-many-arguments,too-many-positional-arguments
     gateway: types.TestContainerDocker,
     sqlstore: types.TestContainerSQL,
     clickhouse: types.TestContainerClickhouse,
+    tls: types.TLS,
     request: pytest.FixtureRequest,
     pytestconfig: pytest.Config,
     maildev: types.TestContainerDocker,
@@ -24,6 +25,7 @@ def signoz(  # pylint: disable=too-many-arguments,too-many-positional-arguments
         gateway=gateway,
         sqlstore=sqlstore,
         clickhouse=clickhouse,
+        tls=tls,
         request=request,
         pytestconfig=pytestconfig,
         cache_key="signoz_alertmanager",

tests/integration/tests/alerts/02_googlechat_test_channel.py

@@ -0,0 +1,114 @@
+import base64
+import json
+import re
+import time
+import uuid
+from collections.abc import Callable
+from http import HTTPStatus
+from typing import NamedTuple
+
+import pytest
+import requests
+from wiremock.resources.mappings import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.alerts import update_raw_channel_config
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import googlechat_config
+
+logger = setup_logger(__name__)
+
+
+# channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
+# with a hardcoded test alert and no retry — the deterministic place to assert
+# permanent-failure behaviour. Rich cards + retry are covered in alertmanager/04_googlechat.py.
+class TestChannelCase(NamedTuple):
+    __test__ = False
+    name: str
+    space: str
+    status: int  # stub status
+    body: dict  # stub body
+    expect_delivered: bool  # expect channels/test 204
+
+
+TEST_CHANNEL_CASES = [
+    TestChannelCase("success", "gc-tc-ok", 200, {"name": "spaces/x/messages/x"}, True),
+    TestChannelCase("permanent_400", "gc-tc-400", 400, {"error": {"code": 400, "status": "INVALID_ARGUMENT", "message": "Message cannot be empty."}}, False),
+    TestChannelCase("permission_403", "gc-tc-403", 403, {"error": {"code": 403, "status": "PERMISSION_DENIED", "message": "Method doesn't allow unregistered callers"}}, False),
+]
+
+
+@pytest.mark.parametrize(
+    "case",
+    TEST_CHANNEL_CASES,
+    ids=lambda c: c.name,
+)
+def test_googlechat_test_channel(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    case: TestChannelCase,
+) -> None:
+    path = f"/v1/spaces/{case.space}/messages"
+    make_http_mocks(
+        notification_channel,
+        [
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url_path=path),
+                response=MappingResponse(status=case.status, json_body=case.body),
+            )
+        ],
+    )
+
+    channel_name = str(uuid.uuid4())
+    receiver = update_raw_channel_config(googlechat_config(case.space), channel_name, notification_channel)
+
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # channels/test 404s until the org's alertmanager registers (one poll tick),
+    # without reaching the notifier — so the first non-404 response is the single
+    # authoritative delivery attempt and the count == 1 assertion below holds
+    deadline = time.time() + 60
+    while True:
+        response = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+            json=receiver,
+            headers={"Authorization": f"Bearer {admin_token}"},
+            timeout=30,
+        )
+        if response.status_code != HTTPStatus.NOT_FOUND or time.time() > deadline:
+            break
+        time.sleep(2)
+
+    if case.expect_delivered:
+        assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+    else:
+        # a downstream 400/403 surfaces as a 500 (untyped notify error) whose body
+        # carries the real downstream status code; pin it to distinguish 400 vs 403
+        assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR, f"expected 500, got {response.status_code}: {response.text}"
+        assert f"unexpected status code {case.status}" in response.text, f"expected downstream {case.status} in error body: {response.text}"
+
+    # exactly one delivery attempt either way (testChannel never retries)
+    count = requests.post(
+        notification_channel.host_configs["8080"].get("/__admin/requests/count"),
+        json={"method": "POST", "urlPath": path},
+        timeout=10,
+    )
+    assert count.json()["count"] == 1, f"expected exactly 1 request (no retry), got {count.text}"
+
+    if case.expect_delivered:
+        find = requests.post(
+            notification_channel.host_configs["8080"].get("/__admin/requests/find"),
+            json={"method": "POST", "urlPath": path},
+            timeout=10,
+        )
+        req = find.json()["requests"][0]
+        # the configured webhook url is posted verbatim, nothing appended
+        assert req["url"] == path, f"expected webhook url {path} posted verbatim, got {req['url']}"
+        # cardsV2 shape with the hardcoded test alert
+        card = json.loads(base64.b64decode(req["bodyAsBase64"]).decode("utf-8"))
+        assert card["cardsV2"][0]["cardId"] == "signoz-alert"
+        assert re.search(r"Test Alert \(", card["cardsV2"][0]["card"]["header"]["title"])

tests/integration/tests/alerts/03_incidentio_test_channel.py

@@ -0,0 +1,120 @@
+import base64
+import json
+import re
+import time
+import uuid
+from collections.abc import Callable
+from http import HTTPStatus
+from typing import NamedTuple
+
+import pytest
+import requests
+from wiremock.resources.mappings import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.alerts import update_raw_channel_config
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import INCIDENTIO_TEST_TOKEN, incidentio_config, incidentio_path
+
+logger = setup_logger(__name__)
+
+
+# channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
+# with a hardcoded test alert and no retry — the deterministic place to assert
+# permanent-failure behaviour. Rich events + retry are covered in alertmanager/05_incidentio.py.
+# Stub bodies are the recorded incident.io Alert Events V2 responses.
+class TestChannelCase(NamedTuple):
+    __test__ = False
+    name: str
+    source: str
+    status: int  # stub status
+    body: dict  # stub body
+    expect_delivered: bool  # expect channels/test 204
+
+
+TEST_CHANNEL_CASES = [
+    TestChannelCase("success", "inc-tc-ok", 202, {"status": "accepted", "message": "Event accepted for processing", "deduplication_key": "x"}, True),
+    TestChannelCase("permanent_401", "inc-tc-401", 401, {"type": "authentication_error", "status": 401, "errors": [{"code": "invalid_authentication_material", "message": "Secret token not valid"}]}, False),
+    TestChannelCase("permanent_422", "inc-tc-422", 422, {"type": "validation_error", "status": 422, "errors": [{"code": "missing_field", "message": '"title" is missing from body'}]}, False),
+]
+
+
+@pytest.mark.parametrize(
+    "case",
+    TEST_CHANNEL_CASES,
+    ids=lambda c: c.name,
+)
+def test_incidentio_test_channel(  # pylint: disable=too-many-arguments,too-many-positional-arguments,too-many-locals
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    case: TestChannelCase,
+) -> None:
+    path = incidentio_path(case.source)
+    make_http_mocks(
+        notification_channel,
+        [
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url_path=path),
+                response=MappingResponse(status=case.status, json_body=case.body),
+            )
+        ],
+    )
+
+    channel_name = str(uuid.uuid4())
+    receiver = update_raw_channel_config(incidentio_config(case.source), channel_name, notification_channel)
+
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+
+    # channels/test 404s until the org's alertmanager registers (one poll tick),
+    # without reaching the notifier — so the first non-404 response is the single
+    # authoritative delivery attempt and the count == 1 assertion below holds
+    deadline = time.time() + 60
+    while True:
+        response = requests.post(
+            signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+            json=receiver,
+            headers={"Authorization": f"Bearer {admin_token}"},
+            timeout=30,
+        )
+        if response.status_code != HTTPStatus.NOT_FOUND or time.time() > deadline:
+            break
+        time.sleep(2)
+
+    if case.expect_delivered:
+        assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+    else:
+        # a downstream 401/422 surfaces as a 500 (untyped notify error) whose body
+        # carries the real downstream status code; pin it to distinguish 401 vs 422
+        assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR, f"expected 500, got {response.status_code}: {response.text}"
+        assert f"unexpected status code {case.status}" in response.text, f"expected downstream {case.status} in error body: {response.text}"
+
+    # exactly one delivery attempt either way (testChannel never retries)
+    count = requests.post(
+        notification_channel.host_configs["8080"].get("/__admin/requests/count"),
+        json={"method": "POST", "urlPath": path},
+        timeout=10,
+    )
+    assert count.json()["count"] == 1, f"expected exactly 1 request (no retry), got {count.text}"
+
+    find = requests.post(
+        notification_channel.host_configs["8080"].get("/__admin/requests/find"),
+        json={"method": "POST", "urlPath": path},
+        timeout=10,
+    )
+    req = find.json()["requests"][0]
+    # the configured url is posted verbatim, nothing appended, and the token is
+    # sent with a single Bearer prefix (header name lowercased on the wire by h2)
+    assert req["url"] == path, f"expected alert events url {path} posted verbatim, got {req['url']}"
+    headers = {name.lower(): value for name, value in req["headers"].items()}
+    assert headers.get("authorization") == f"Bearer {INCIDENTIO_TEST_TOKEN}", f"expected single Bearer prefix, got {headers.get('authorization')}"
+
+    if case.expect_delivered:
+        # alert event shape with the hardcoded test alert
+        event = json.loads(base64.b64decode(req["bodyAsBase64"]).decode("utf-8"))
+        assert re.search(r"\[FIRING:1\] Test Alert \(", event["title"]), f"unexpected title: {event['title']}"
+        assert event["status"] == "firing"
+        assert event["deduplication_key"], "expected a non-empty deduplication_key"

tests/integration/tests/alerts/04_jira_test_channel.py

@@ -0,0 +1,324 @@
+import base64
+import json
+import re
+import uuid
+from collections.abc import Callable
+from http import HTTPStatus
+from typing import NamedTuple
+
+import pytest
+import requests
+from wiremock.resources.mappings import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.alerts import update_raw_channel_config
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import (
+    JIRA_API_BASE,
+    JIRA_SA_EMAIL,
+    JIRA_TEST_EMAIL,
+    JIRA_TEST_TOKEN,
+    find_requests,
+    jira_comment_mapping,
+    jira_config,
+    jira_create_mapping,
+    jira_search_issue,
+    jira_search_mapping,
+    jira_transition_post_mapping,
+    jira_transitions_mapping,
+    jira_update_mapping,
+    wait_for_org_registration,
+)
+
+logger = setup_logger(__name__)
+
+# channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
+# with a hardcoded firing test alert and no retry. The search stub decides which
+# branch runs (create / update / reopen), so the whole issue lifecycle is
+# deterministic here; default-template events + retry are in alertmanager/06_jira.py.
+
+BASIC_AUTH = "Basic " + base64.b64encode(f"{JIRA_TEST_EMAIL}:{JIRA_TEST_TOKEN}".encode()).decode()
+
+
+def test_jira_create_issue(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    make_http_mocks(notification_channel, [jira_search_mapping([]), jira_create_mapping()])
+
+    receiver = update_raw_channel_config(jira_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+
+    searches = find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/search/jql")
+    assert len(searches) == 1
+    # basic auth on every call (header name lowercased on the wire by h2)
+    headers = {name.lower(): value for name, value in searches[0]["headers"].items()}
+    assert headers.get("authorization") == BASIC_AUTH, f"expected basic auth, got {headers.get('authorization')}"
+    jql = json.loads(base64.b64decode(searches[0]["bodyAsBase64"]).decode("utf-8"))["jql"]
+    assert 'project="OPS"' in jql, jql
+    assert 'labels="ALERT{' in jql, jql
+    # default reopen_duration (72h) becomes the firing reopen window
+    assert "resolutiondate >= -4320m" in jql, jql
+
+    creates = find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue")
+    assert len(creates) == 1
+    fields = json.loads(base64.b64decode(creates[0]["bodyAsBase64"]).decode("utf-8"))["fields"]
+    assert fields["project"] == {"key": "OPS"}
+    assert fields["issuetype"] == {"name": "Task"}
+    assert re.search(r"\[FIRING:1\] Test Alert \(", fields["summary"]), fields["summary"]
+    assert "signoz-alert" in fields["labels"]
+    assert any(label.startswith("ALERT{") for label in fields["labels"]), fields["labels"]
+    # ADF body leads with the firing status panel
+    panel = fields["description"]["content"][0]
+    assert panel["attrs"] == {"panelType": "error"}
+    assert panel["content"][0]["content"][0]["text"] == "🔴 FIRING"
+
+
+def test_jira_wont_fix_resolution_in_search_jql(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    make_http_mocks(notification_channel, [jira_search_mapping([]), jira_create_mapping()])
+
+    receiver = update_raw_channel_config(jira_config(wont_fix_resolution="Won't Do"), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+
+    searches = find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/search/jql")
+    assert len(searches) == 1
+    jql = json.loads(base64.b64decode(searches[0]["bodyAsBase64"]).decode("utf-8"))["jql"]
+    # issues resolved as won't-fix stay closed: the search skips them so a
+    # refire creates a fresh issue instead of reopening
+    assert '(resolution is EMPTY or resolution != "Won\'t Do")' in jql, jql
+
+
+def test_jira_updates_existing_open_issue(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    make_http_mocks(
+        notification_channel,
+        [
+            jira_search_mapping([jira_search_issue("OPS-7", done=False, labels=["user-added", "signoz-alert"])]),
+            jira_update_mapping("OPS-7"),
+            jira_comment_mapping("OPS-7"),
+        ],
+    )
+
+    receiver = update_raw_channel_config(jira_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+
+    # still-firing group with an open issue: refresh + comment, no create, no transition
+    updates = find_requests(notification_channel, "PUT", f"{JIRA_API_BASE}/issue/OPS-7")
+    assert len(updates) == 1
+    fields = json.loads(base64.b64decode(updates[0]["bodyAsBase64"]).decode("utf-8"))["fields"]
+    assert "user-added" in fields["labels"], f"user-added labels must survive the update: {fields['labels']}"
+    assert "signoz-alert" in fields["labels"]
+    assert "project" not in fields and "issuetype" not in fields, "create-only fields must not be sent on update"
+
+    assert len(find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue")) == 0
+    assert len(find_requests(notification_channel, "GET", f"{JIRA_API_BASE}/issue/OPS-7/transitions")) == 0
+
+    comments = find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue/OPS-7/comment")
+    assert len(comments) == 1
+    body = json.loads(base64.b64decode(comments[0]["bodyAsBase64"]).decode("utf-8"))["body"]
+    assert body["content"][0]["attrs"] == {"panelType": "error"}, "comment carries the same ADF snapshot"
+
+
+def test_jira_reopens_done_issue(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    make_http_mocks(
+        notification_channel,
+        [
+            jira_search_mapping([jira_search_issue("OPS-7", done=True, labels=["signoz-alert"])]),
+            jira_update_mapping("OPS-7"),
+            jira_transitions_mapping(
+                "OPS-7",
+                [
+                    {"id": "31", "name": "Done", "to": {"statusCategory": {"key": "done"}}},
+                    {"id": "11", "name": "To Do", "to": {"statusCategory": {"key": "new"}}},
+                ],
+            ),
+            jira_transition_post_mapping("OPS-7"),
+            jira_comment_mapping("OPS-7"),
+        ],
+    )
+
+    receiver = update_raw_channel_config(jira_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+
+    # firing group whose issue is done: update, then transition out of done, then comment
+    assert len(find_requests(notification_channel, "PUT", f"{JIRA_API_BASE}/issue/OPS-7")) == 1
+    transitions = find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue/OPS-7/transitions")
+    assert len(transitions) == 1
+    body = json.loads(base64.b64decode(transitions[0]["bodyAsBase64"]).decode("utf-8"))
+    assert body == {"transition": {"id": "11"}}, f"expected the not-done transition to be applied: {body}"
+    assert len(find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue/OPS-7/comment")) == 1
+    assert len(find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue")) == 0
+
+
+class PermanentErrorCase(NamedTuple):
+    __test__ = False
+    name: str
+    mappings: list[Mapping]
+    downstream_status: int
+    search_count: int
+    create_count: int
+
+
+PERMANENT_ERROR_CASES = [
+    PermanentErrorCase(
+        name="create_400",
+        mappings=[
+            jira_search_mapping([]),
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url_path=f"{JIRA_API_BASE}/issue"),
+                response=MappingResponse(status=400, json_body={"errorMessages": [], "errors": {"issuetype": "The issue type selected is invalid."}}),
+            ),
+        ],
+        downstream_status=400,
+        search_count=1,
+        create_count=1,
+    ),
+    PermanentErrorCase(
+        name="search_401",
+        mappings=[
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url_path=f"{JIRA_API_BASE}/search/jql"),
+                response=MappingResponse(status=401, json_body={"errorMessages": ["Client must be authenticated to access this resource."]}),
+            ),
+        ],
+        downstream_status=401,
+        search_count=1,
+        create_count=0,
+    ),
+]
+
+
+@pytest.mark.parametrize(
+    "case",
+    PERMANENT_ERROR_CASES,
+    ids=lambda c: c.name,
+)
+def test_jira_permanent_error(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    case: PermanentErrorCase,
+) -> None:
+    make_http_mocks(notification_channel, case.mappings)
+
+    receiver = update_raw_channel_config(jira_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    # a downstream 4xx surfaces as a 500 (untyped notify error) whose body
+    # carries the real downstream status code; testChannel never retries
+    assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR, f"expected 500, got {response.status_code}: {response.text}"
+    assert f"unexpected status code {case.downstream_status}" in response.text, response.text
+
+    assert len(find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/search/jql")) == case.search_count
+    assert len(find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/issue")) == case.create_count
+
+
+def test_jira_service_account_uses_gateway(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    cloud_id = "b8e7c297-4c56-4d39-9e1a-000000000001"
+    gateway_base = f"/ex/jira/{cloud_id}/rest/api/3"
+    make_http_mocks(
+        notification_channel,
+        [
+            Mapping(
+                request=MappingRequest(method=HttpMethods.GET, url_path="/_edge/tenant_info"),
+                response=MappingResponse(status=200, json_body={"cloudId": cloud_id}),
+            ),
+            jira_search_mapping([], base=gateway_base),
+            jira_create_mapping(base=gateway_base),
+        ],
+    )
+
+    receiver = update_raw_channel_config(
+        jira_config(http_config={"basic_auth": {"username": JIRA_SA_EMAIL, "password": JIRA_TEST_TOKEN}}),
+        str(uuid.uuid4()),
+        notification_channel,
+    )
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+
+    # cloud id resolved from the site's tenant_info, then every API call goes
+    # through the api.atlassian.com gateway instead of the site host
+    assert len(find_requests(notification_channel, "GET", "/_edge/tenant_info")) == 1
+    assert len(find_requests(notification_channel, "POST", f"{gateway_base}/search/jql")) == 1
+    assert len(find_requests(notification_channel, "POST", f"{gateway_base}/issue")) == 1
+    assert len(find_requests(notification_channel, "POST", f"{JIRA_API_BASE}/search/jql")) == 0

tests/integration/tests/alerts/05_jsmops_test_channel.py

@@ -0,0 +1,165 @@
+import base64
+import json
+import re
+import uuid
+from collections.abc import Callable
+from http import HTTPStatus
+from typing import NamedTuple
+
+import pytest
+import requests
+from wiremock.resources.mappings import HttpMethods, Mapping, MappingRequest, MappingResponse
+
+from fixtures import types
+from fixtures.alerts import update_raw_channel_config
+from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
+from fixtures.logger import setup_logger
+from fixtures.notification_channel import (
+    JSMOPS_API_BASE,
+    JSMOPS_NOTES_PATH_PATTERN,
+    JSMOPS_TEST_API_KEY,
+    find_requests,
+    jsmops_config,
+    jsmops_create_mapping,
+    jsmops_notes_mapping,
+    wait_for_org_registration,
+)
+
+logger = setup_logger(__name__)
+
+# channel test (POST /api/v1/channels/test) drives the notifier once, synchronously,
+# with a hardcoded firing test alert and no retry: create alert on the JSM Ops
+# gateway, then append a timeline note. Default-template events + retry are in
+# alertmanager/07_jsmops.py.
+
+
+def test_jsmops_create_alert_with_note(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    make_http_mocks(notification_channel, [jsmops_create_mapping(), jsmops_notes_mapping()])
+
+    receiver = update_raw_channel_config(jsmops_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204, got {response.status_code}: {response.text}"
+
+    creates = find_requests(notification_channel, "POST", f"{JSMOPS_API_BASE}/v2/alerts")
+    assert len(creates) == 1
+    # GenieKey auth on every call (header name lowercased on the wire by h2)
+    headers = {name.lower(): value for name, value in creates[0]["headers"].items()}
+    assert headers.get("authorization") == f"GenieKey {JSMOPS_TEST_API_KEY}", f"expected GenieKey auth, got {headers.get('authorization')}"
+    alert = json.loads(base64.b64decode(creates[0]["bodyAsBase64"]).decode("utf-8"))
+    assert alert["alias"], "alias carries the group hash for dedup/close"
+    assert re.search(r"\[FIRING:1\] Test Alert \(", alert["message"]), alert["message"]
+    assert alert["source"] == "SigNoz"
+    assert alert["tags"] == ["signoz"]
+    # advanced treatment renders the default body as HTML
+    assert "<div>" in alert["description"], alert["description"]
+
+    notes = find_requests(notification_channel, "POST", path_pattern=JSMOPS_NOTES_PATH_PATTERN)
+    assert len(notes) == 1
+    assert notes[0]["queryParams"]["identifierType"]["values"] == ["alias"]
+    note = json.loads(base64.b64decode(notes[0]["bodyAsBase64"]).decode("utf-8"))
+    assert note["source"] == "SigNoz"
+    assert note["note"].strip(), "the timeline note carries the plain-text snapshot"
+
+
+def test_jsmops_failed_note_does_not_fail_delivery(
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+) -> None:
+    # notes are enrichment: a permanent note failure (e.g. the first-fire note
+    # racing JSM's async alert create) is dropped and the delivery still succeeds
+    make_http_mocks(
+        notification_channel,
+        [
+            jsmops_create_mapping(),
+            jsmops_notes_mapping(status=404, body={"message": "Alert with id/alias does not exist", "took": 0.001, "requestId": "x"}),
+        ],
+    )
+
+    receiver = update_raw_channel_config(jsmops_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    assert response.status_code == HTTPStatus.NO_CONTENT, f"expected 204 despite the failed note, got {response.status_code}: {response.text}"
+
+    assert len(find_requests(notification_channel, "POST", f"{JSMOPS_API_BASE}/v2/alerts")) == 1
+    assert len(find_requests(notification_channel, "POST", path_pattern=JSMOPS_NOTES_PATH_PATTERN)) == 1
+
+
+class PermanentErrorCase(NamedTuple):
+    __test__ = False
+    name: str
+    status: int
+    body: dict
+
+
+PERMANENT_ERROR_CASES = [
+    PermanentErrorCase("create_422", 422, {"message": "Message can not be empty.", "took": 0.001, "requestId": "x"}),
+    PermanentErrorCase("create_401", 401, {"message": "Could not authenticate.", "took": 0.001, "requestId": "x"}),
+]
+
+
+@pytest.mark.parametrize(
+    "case",
+    PERMANENT_ERROR_CASES,
+    ids=lambda c: c.name,
+)
+def test_jsmops_permanent_error(  # pylint: disable=too-many-arguments,too-many-positional-arguments
+    signoz: types.SigNoz,
+    get_token: Callable[[str, str], str],
+    create_user_admin: None,  # pylint: disable=unused-argument
+    notification_channel: types.TestContainerDocker,
+    make_http_mocks: Callable[[types.TestContainerDocker, list[Mapping]], None],
+    case: PermanentErrorCase,
+) -> None:
+    make_http_mocks(
+        notification_channel,
+        [
+            Mapping(
+                request=MappingRequest(method=HttpMethods.POST, url_path=f"{JSMOPS_API_BASE}/v2/alerts"),
+                response=MappingResponse(status=case.status, json_body=case.body),
+            ),
+            jsmops_notes_mapping(),
+        ],
+    )
+
+    receiver = update_raw_channel_config(jsmops_config(), str(uuid.uuid4()), notification_channel)
+    admin_token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
+    wait_for_org_registration(signoz, admin_token, notification_channel)
+
+    response = requests.post(
+        signoz.self.host_configs["8080"].get("/api/v1/channels/test"),
+        json=receiver,
+        headers={"Authorization": f"Bearer {admin_token}"},
+        timeout=30,
+    )
+    # a downstream 4xx on the create surfaces as a 500 (untyped notify error)
+    # whose body carries the real downstream status code; testChannel never retries
+    assert response.status_code == HTTPStatus.INTERNAL_SERVER_ERROR, f"expected 500, got {response.status_code}: {response.text}"
+    assert f"unexpected status code {case.status}" in response.text, response.text
+
+    assert len(find_requests(notification_channel, "POST", f"{JSMOPS_API_BASE}/v2/alerts")) == 1
+    # the request loop stops at the failed create, so the note is never attempted
+    assert len(find_requests(notification_channel, "POST", path_pattern=JSMOPS_NOTES_PATH_PATTERN)) == 0