From 960eb7d2284755c0dc78c3c0e2844b75d214b421 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Sun, 23 Aug 2026 03:27:57 -0700 Subject: [PATCH 1/8] feat(breakfix): verify tenant notification delivery Signed-off-by: Hasan Khan --- .../providers/aws/config/bare_metal.yaml | 32 ++ .../breakfix/query_failure_notifications.py | 4 + .../breakfix/query_planned_notifications.py | 4 + .../breakfix/query_tenant_notification.py | 427 ++++++++++++++++++ isvctl/configs/suites/README.md | 4 +- .../test_notification_delivery_provider.py | 245 ++++++++++ isvtest/src/isvtest/validations/breakfix.py | 61 ++- isvtest/tests/test_breakfix.py | 114 ++++- 8 files changed, 867 insertions(+), 24 deletions(-) create mode 100644 isvctl/configs/providers/shared/breakfix/query_tenant_notification.py create mode 100644 isvctl/tests/test_notification_delivery_provider.py diff --git a/isvctl/configs/providers/aws/config/bare_metal.yaml b/isvctl/configs/providers/aws/config/bare_metal.yaml index 94c5bb374..a07e6e990 100644 --- a/isvctl/configs/providers/aws/config/bare_metal.yaml +++ b/isvctl/configs/providers/aws/config/bare_metal.yaml @@ -265,6 +265,38 @@ commands: - "{{ '--skip' if steps.deploy_nim.skipped | default(false) else '' }}" timeout: 300 + # BFX05/BFX06: publish through an ephemeral SNS topic, prove delivery + # through its SQS subscriber, and remove both resources. + - name: query_planned_notifications + phase: test + continue_on_failure: true + command: "python3 ../../shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "aws" + - "--event-type" + - "planned_maintenance" + - "--message" + - "Planned node maintenance notification validation" + - "--region" + - "{{region}}" + timeout: 120 + + - name: query_failure_notifications + phase: test + continue_on_failure: true + command: "python3 ../../shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "aws" + - "--event-type" + - "node_failure" + - "--message" + - "Immediate node failure notification validation" + - "--region" + - "{{region}}" + timeout: 120 + # Teardown bare-metal instance (boto3) # Set AWS_BM_SKIP_TEARDOWN=true to keep the instance alive - name: teardown diff --git a/isvctl/configs/providers/my-isv/scripts/breakfix/query_failure_notifications.py b/isvctl/configs/providers/my-isv/scripts/breakfix/query_failure_notifications.py index b5f8951e7..f69bd26b5 100644 --- a/isvctl/configs/providers/my-isv/scripts/breakfix/query_failure_notifications.py +++ b/isvctl/configs/providers/my-isv/scripts/breakfix/query_failure_notifications.py @@ -31,6 +31,10 @@ def main() -> int: "type": "node_failure", "message": "Node became unreachable; GPU fault detected", "notified_at": "2026-06-24T12:00:00Z", + "failed_at": "2026-06-24T11:59:30Z", + "channel": "webhook", + "delivery_status": "delivered", + "delivery_id": "demo-failure-delivery-001", } ], ) diff --git a/isvctl/configs/providers/my-isv/scripts/breakfix/query_planned_notifications.py b/isvctl/configs/providers/my-isv/scripts/breakfix/query_planned_notifications.py index e93ba67b4..9c7e2cd91 100644 --- a/isvctl/configs/providers/my-isv/scripts/breakfix/query_planned_notifications.py +++ b/isvctl/configs/providers/my-isv/scripts/breakfix/query_planned_notifications.py @@ -31,6 +31,10 @@ def main() -> int: "type": "planned_maintenance", "message": "Scheduled firmware update (demo)", "notified_at": "2026-06-24T12:00:00Z", + "scheduled_at": "2026-06-25T12:00:00Z", + "channel": "webhook", + "delivery_status": "delivered", + "delivery_id": "demo-planned-delivery-001", } ], ) diff --git a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py new file mode 100644 index 000000000..7c84462fe --- /dev/null +++ b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py @@ -0,0 +1,427 @@ +#!/usr/bin/env python3 +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Deliver and verify a tenant notification for BFX05-01 or BFX06-01. + +The script emits only the provider-neutral JSON consumed by the break-fix +validators. It supports three delivery transports: + +* ``aws`` creates an ephemeral SNS topic and SQS subscription, publishes the + notification, proves that the subscriber received the same delivery ID, and + removes both resources. +* ``kubernetes`` creates an ephemeral in-cluster HTTP receiver, posts from a + short-lived pod, verifies its acknowledgement, and removes the namespace. +* ``webhook`` posts to a configured Slack, Teams, or generic HTTP endpoint and + treats a successful HTTP response as the delivery acknowledgement. + +Resource identifiers, webhook URLs, response bodies, and credentials are never +included in the output contract. +""" + +from __future__ import annotations + +import argparse +import json +import os +import shlex +import subprocess +import sys +import uuid +from datetime import UTC, datetime, timedelta +from typing import Any +from urllib import error, request + +_RECEIVER_CODE = r""" +import json +from http.server import BaseHTTPRequestHandler, HTTPServer + +class Receiver(BaseHTTPRequestHandler): + def do_POST(self): + try: + size = int(self.headers.get("Content-Length", "0")) + payload = json.loads(self.rfile.read(size)) + delivery_id = payload["delivery_id"] + if not isinstance(delivery_id, str) or not delivery_id: + raise ValueError("missing delivery_id") + body = json.dumps({"delivery_id": delivery_id, "status": "delivered"}).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + except Exception: + self.send_response(400) + self.end_headers() + + def log_message(self, format, *args): + return + +HTTPServer(("0.0.0.0", 8080), Receiver).serve_forever() +""" + + +class DeliveryError(RuntimeError): + """Raised when a notification cannot be proved delivered.""" + + +def _timestamp(value: datetime) -> str: + """Render a UTC timestamp in the provider-neutral ISO 8601 format.""" + return value.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") + + +def _emit(payload: dict[str, Any]) -> int: + """Write one JSON document and return a process exit code.""" + print(json.dumps(payload, separators=(",", ":"))) + return 0 if payload.get("success") else 1 + + +def _payload(args: argparse.Namespace, delivery_id: str, started_at: datetime) -> dict[str, str]: + """Build the event delivered to the selected communication system.""" + common = { + "delivery_id": delivery_id, + "machine_id": args.machine_id, + "type": args.event_type, + "message": args.message, + } + if args.event_type == "planned_maintenance": + common["scheduled_at"] = _timestamp(started_at + timedelta(hours=args.schedule_hours)) + else: + common["failed_at"] = _timestamp(started_at) + return common + + +def _deliver_aws(payload: dict[str, str], region: str) -> str: + """Publish through SNS and prove receipt through an ephemeral SQS subscriber.""" + try: + import boto3 + except ImportError as exc: # pragma: no cover - dependency is present in the workspace + raise DeliveryError("AWS notification backend requires boto3") from exc + + suffix = uuid.uuid4().hex[:12] + session = boto3.Session(region_name=region) + sns = session.client("sns") + sqs = session.client("sqs") + topic_arn = "" + queue_url = "" + try: + topic_arn = sns.create_topic(Name=f"isvtest-notification-{suffix}")["TopicArn"] + queue_url = sqs.create_queue( + QueueName=f"isvtest-notification-{suffix}", + Attributes={"MessageRetentionPeriod": "300", "ReceiveMessageWaitTimeSeconds": "10"}, + )["QueueUrl"] + queue_arn = sqs.get_queue_attributes(QueueUrl=queue_url, AttributeNames=["QueueArn"])["Attributes"]["QueueArn"] + policy = { + "Version": "2012-10-17", + "Statement": [ + { + "Effect": "Allow", + "Principal": {"Service": "sns.amazonaws.com"}, + "Action": "sqs:SendMessage", + "Resource": queue_arn, + "Condition": {"ArnEquals": {"aws:SourceArn": topic_arn}}, + } + ], + } + sqs.set_queue_attributes(QueueUrl=queue_url, Attributes={"Policy": json.dumps(policy)}) + sns.subscribe( + TopicArn=topic_arn, + Protocol="sqs", + Endpoint=queue_arn, + Attributes={"RawMessageDelivery": "true"}, + ReturnSubscriptionArn=True, + ) + sns.publish(TopicArn=topic_arn, Message=json.dumps(payload, separators=(",", ":"))) + + for _ in range(3): + response = sqs.receive_message(QueueUrl=queue_url, MaxNumberOfMessages=1, WaitTimeSeconds=10) + for message in response.get("Messages", []): + try: + received = json.loads(message["Body"]) + except (KeyError, json.JSONDecodeError): + continue + if received.get("delivery_id") == payload["delivery_id"]: + sqs.delete_message(QueueUrl=queue_url, ReceiptHandle=message["ReceiptHandle"]) + return "aws_sns" + raise DeliveryError("AWS notification subscriber did not receive the published event") + except DeliveryError: + raise + except Exception as exc: + raise DeliveryError(f"AWS notification delivery failed: {type(exc).__name__}") from exc + finally: + cleanup_failed = False + if topic_arn: + try: + sns.delete_topic(TopicArn=topic_arn) + except Exception: + cleanup_failed = True + if queue_url: + try: + sqs.delete_queue(QueueUrl=queue_url) + except Exception: + cleanup_failed = True + if cleanup_failed: + raise DeliveryError("AWS notification cleanup failed") + + +def _kubectl_base() -> list[str]: + """Return the configured kubectl command without invoking a shell.""" + command = shlex.split(os.environ.get("KUBECTL", "kubectl")) + if not command: + raise DeliveryError("KUBECTL command is empty") + return command + + +def _run(command: list[str], *, stdin: str | None = None, timeout: int = 120) -> subprocess.CompletedProcess[str]: + """Run a subprocess and convert diagnostics to a sanitized delivery error.""" + try: + result = subprocess.run(command, input=stdin, text=True, capture_output=True, timeout=timeout, check=False) + except (OSError, subprocess.TimeoutExpired) as exc: + raise DeliveryError(f"Kubernetes notification command failed: {type(exc).__name__}") from exc + if result.returncode != 0: + raise DeliveryError("Kubernetes notification command returned a non-zero status") + return result + + +def _kubernetes_objects(namespace: str, receiver_name: str) -> dict[str, Any]: + """Build the ephemeral HTTP receiver Pod and Service manifests.""" + labels = {"app.kubernetes.io/name": "isvtest-notification-receiver", "isvtest.nvidia.com/run": receiver_name} + return { + "apiVersion": "v1", + "kind": "List", + "items": [ + { + "apiVersion": "v1", + "kind": "Pod", + "metadata": {"name": receiver_name, "namespace": namespace, "labels": labels}, + "spec": { + "restartPolicy": "Never", + "containers": [ + { + "name": "receiver", + "image": "python:3.12-alpine", + "command": ["python", "-c", _RECEIVER_CODE], + "ports": [{"containerPort": 8080}], + "readinessProbe": {"tcpSocket": {"port": 8080}, "initialDelaySeconds": 1}, + "resources": { + "requests": {"cpu": "10m", "memory": "16Mi"}, + "limits": {"cpu": "100m", "memory": "64Mi"}, + }, + "securityContext": { + "allowPrivilegeEscalation": False, + "capabilities": {"drop": ["ALL"]}, + }, + } + ], + "securityContext": { + "runAsNonRoot": True, + "runAsUser": 65532, + "seccompProfile": {"type": "RuntimeDefault"}, + }, + }, + }, + { + "apiVersion": "v1", + "kind": "Service", + "metadata": {"name": "receiver", "namespace": namespace, "labels": labels}, + "spec": {"selector": labels, "ports": [{"port": 8080, "targetPort": 8080}]}, + }, + ], + } + + +def _extract_acknowledgement(output: str, delivery_id: str) -> bool: + """Return whether command output contains the receiver acknowledgement.""" + for line in reversed(output.splitlines()): + try: + value = json.loads(line.strip()) + except json.JSONDecodeError: + continue + if value.get("delivery_id") == delivery_id and value.get("status") == "delivered": + return True + return False + + +def _deliver_kubernetes(payload: dict[str, str]) -> str: + """Post from one pod to an ephemeral in-cluster HTTP receiver.""" + kubectl = _kubectl_base() + suffix = uuid.uuid4().hex[:10] + namespace = f"isvtest-notification-{suffix}" + receiver_name = f"receiver-{suffix}" + sender_name = f"sender-{suffix}" + created_namespace = False + try: + _run([*kubectl, "create", "namespace", namespace]) + created_namespace = True + objects = json.dumps(_kubernetes_objects(namespace, receiver_name)) + _run([*kubectl, "apply", "-f", "-"], stdin=objects) + _run([*kubectl, "wait", "--for=condition=Ready", f"pod/{receiver_name}", "-n", namespace, "--timeout=120s"]) + result = _run( + [ + *kubectl, + "run", + sender_name, + "-n", + namespace, + "--image=curlimages/curl:8.10.1", + "--restart=Never", + "--attach", + "--rm", + "--quiet", + "--command", + "--", + "curl", + "--fail-with-body", + "--silent", + "--show-error", + "--max-time", + "30", + "-H", + "Content-Type: application/json", + "--data-binary", + json.dumps(payload, separators=(",", ":")), + "http://receiver:8080/notify", + ], + timeout=180, + ) + if not _extract_acknowledgement(result.stdout, payload["delivery_id"]): + raise DeliveryError("Kubernetes webhook receiver did not acknowledge the notification") + return "kubernetes_webhook" + finally: + if created_namespace: + try: + cleanup = subprocess.run( + [ + *kubectl, + "delete", + "namespace", + namespace, + "--wait=true", + "--timeout=60s", + "--ignore-not-found", + ], + text=True, + capture_output=True, + timeout=70, + check=False, + ) + except (OSError, subprocess.TimeoutExpired) as exc: + raise DeliveryError("Kubernetes notification cleanup failed") from exc + if cleanup.returncode != 0: + raise DeliveryError("Kubernetes notification cleanup failed") + + +def _deliver_webhook(payload: dict[str, str], url: str, channel: str) -> str: + """Post to a configured Slack, Teams, or generic HTTP webhook.""" + if not url.startswith("https://") and not url.startswith("http://127.0.0.1:"): + raise DeliveryError("Webhook URL must use HTTPS (loopback HTTP is allowed for local testing)") + outgoing: dict[str, Any] = payload + text = ( + f"{payload['message']}\nNode: {payload['machine_id']}\n" + f"Event: {payload['type']}\nDelivery ID: {payload['delivery_id']}" + ) + if channel == "slack": + outgoing = {"text": text} + elif channel == "teams": + outgoing = { + "type": "message", + "attachments": [ + { + "contentType": "application/vnd.microsoft.card.adaptive", + "contentUrl": None, + "content": { + "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", + "type": "AdaptiveCard", + "version": "1.4", + "body": [{"type": "TextBlock", "text": text, "wrap": True}], + }, + } + ], + } + body = json.dumps(outgoing, separators=(",", ":")).encode() + req = request.Request(url, data=body, headers={"Content-Type": "application/json"}, method="POST") + try: + with request.urlopen(req, timeout=30) as response: + if not 200 <= response.status < 300: + raise DeliveryError("Webhook endpoint did not accept the notification") + except DeliveryError: + raise + except (error.URLError, TimeoutError) as exc: + raise DeliveryError(f"Webhook notification delivery failed: {type(exc).__name__}") from exc + return channel + + +def _parse_args() -> argparse.Namespace: + """Parse the delivery probe arguments.""" + parser = argparse.ArgumentParser(description="Deliver and verify one tenant notification") + parser.add_argument("--backend", choices=("aws", "kubernetes", "webhook"), required=True) + parser.add_argument("--event-type", choices=("planned_maintenance", "node_failure"), required=True) + parser.add_argument("--machine-id", default="notification-probe-node") + parser.add_argument("--message", required=True) + parser.add_argument("--region", default=os.environ.get("AWS_REGION", "us-west-2")) + parser.add_argument("--schedule-hours", type=int, default=24) + parser.add_argument("--webhook-url", default=os.environ.get("ISVTEST_NOTIFICATION_WEBHOOK_URL", "")) + parser.add_argument( + "--webhook-channel", + choices=("slack", "teams", "webhook"), + default=os.environ.get("ISVTEST_NOTIFICATION_CHANNEL", "webhook"), + ) + args = parser.parse_args() + if not args.machine_id.strip() or not args.message.strip(): + parser.error("--machine-id and --message must be non-empty") + if args.event_type == "planned_maintenance" and args.schedule_hours <= 0: + parser.error("--schedule-hours must be positive") + if args.backend == "webhook" and not args.webhook_url: + parser.error("--webhook-url or ISVTEST_NOTIFICATION_WEBHOOK_URL is required") + return args + + +def main() -> int: + """Deliver a notification and emit normalized proof of receipt.""" + args = _parse_args() + metadata = { + "platform": args.backend, + "test_name": ( + "query_planned_notifications" if args.event_type == "planned_maintenance" else "query_failure_notifications" + ), + } + started_at = datetime.now(UTC) + delivery_id = str(uuid.uuid4()) + payload = _payload(args, delivery_id, started_at) + try: + if args.backend == "aws": + channel = _deliver_aws(payload, args.region) + elif args.backend == "kubernetes": + channel = _deliver_kubernetes(payload) + else: + channel = _deliver_webhook(payload, args.webhook_url, args.webhook_channel) + except DeliveryError as exc: + return _emit( + { + "success": False, + **metadata, + "notification_channel_observable": False, + "notifications": [], + "error": str(exc), + } + ) + + notified_at = datetime.now(UTC) + record = { + **payload, + "notified_at": _timestamp(notified_at), + "channel": channel, + "delivery_status": "delivered", + } + return _emit( + { + "success": True, + **metadata, + "notification_channel_observable": True, + "notifications": [record], + } + ) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/isvctl/configs/suites/README.md b/isvctl/configs/suites/README.md index 4c71c3bba..214cb58ac 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -269,8 +269,8 @@ its plan item is not platform-scoped. | `return_rack_maintenance` | test | `providers/my-isv/scripts/breakfix/return_rack_maintenance.py` | `operation.{requested,accepted,rack_id}` (BFX01-03) | | `request_host_replacement` | test | `providers/my-isv/scripts/breakfix/request_host_replacement.py` | `operation.{requested,node_removed_from_pool,machine_id}` (BFX01-05) | | `query_node_health_agents` | test | `providers/my-isv/scripts/breakfix/query_node_health_agents.py` | `agents_observable`, `agents[].{node_id,agent_name,running}` (BFX04-01) | -| `query_planned_notifications` | test | `providers/my-isv/scripts/breakfix/query_planned_notifications.py` | `notification_channel_observable`, `notifications[].{machine_id,type,message,notified_at}` (BFX05-01) | -| `query_failure_notifications` | test | `providers/my-isv/scripts/breakfix/query_failure_notifications.py` | `notification_channel_observable`, `notifications[].{machine_id,type,message,notified_at}` (BFX06-01) | +| `query_planned_notifications` | test | provider implementation or `providers/shared/breakfix/query_tenant_notification.py` | `notification_channel_observable`, `notifications[].{machine_id,type,message,notified_at,scheduled_at,channel,delivery_status,delivery_id}`; PASS requires acknowledged delivery and a schedule after notification (BFX05-01) | +| `query_failure_notifications` | test | provider implementation or `providers/shared/breakfix/query_tenant_notification.py` | `notification_channel_observable`, `notifications[].{machine_id,type,message,failed_at,notified_at,channel,delivery_status,delivery_id}`; PASS requires acknowledged delivery within five minutes of failure (BFX06-01) | ### Storage (`storage.yaml`) diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py new file mode 100644 index 000000000..52f52eb6d --- /dev/null +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -0,0 +1,245 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +"""Tests for the shared BFX05/BFX06 notification-delivery provider.""" + +from __future__ import annotations + +import importlib.util +import json +import threading +from argparse import Namespace +from datetime import UTC, datetime +from http.server import BaseHTTPRequestHandler, HTTPServer +from pathlib import Path +from types import ModuleType, SimpleNamespace +from typing import Any, ClassVar + +import pytest + +SCRIPT = Path(__file__).parents[1] / "configs" / "providers" / "shared" / "breakfix" / "query_tenant_notification.py" + + +def _load_module() -> ModuleType: + """Load the provider script as a module.""" + spec = importlib.util.spec_from_file_location("query_tenant_notification", SCRIPT) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture +def provider() -> ModuleType: + """Return a freshly loaded provider module.""" + return _load_module() + + +def _args(event_type: str) -> Namespace: + """Build the arguments needed by the normalized payload helper.""" + return Namespace( + machine_id="node-1", + event_type=event_type, + message="Tenant notification", + schedule_hours=24, + ) + + +def test_planned_payload_has_future_schedule(provider: ModuleType) -> None: + """The planned payload carries a future, timezone-aware schedule.""" + started_at = datetime(2026, 8, 23, 10, 0, tzinfo=UTC) + payload = provider._payload(_args("planned_maintenance"), "delivery-1", started_at) + assert payload == { + "delivery_id": "delivery-1", + "machine_id": "node-1", + "type": "planned_maintenance", + "message": "Tenant notification", + "scheduled_at": "2026-08-24T10:00:00Z", + } + + +def test_failure_payload_records_detection_time(provider: ModuleType) -> None: + """The immediate-failure payload carries the original failure time.""" + started_at = datetime(2026, 8, 23, 10, 0, tzinfo=UTC) + payload = provider._payload(_args("node_failure"), "delivery-2", started_at) + assert payload["failed_at"] == "2026-08-23T10:00:00Z" + assert payload["type"] == "node_failure" + + +class _CaptureHandler(BaseHTTPRequestHandler): + """Capture one test webhook request and acknowledge it.""" + + payload: ClassVar[dict[str, Any]] = {} + + def do_POST(self) -> None: + """Capture the JSON body and return HTTP 204.""" + length = int(self.headers["Content-Length"]) + type(self).payload = json.loads(self.rfile.read(length)) + self.send_response(204) + self.end_headers() + + def log_message(self, fmt: str, *args: object) -> None: + """Suppress HTTP server logs in the unit test.""" + + +@pytest.mark.parametrize("channel", ["webhook", "slack", "teams"]) +def test_webhook_backends_accept_acknowledged_delivery(provider: ModuleType, channel: str) -> None: + """Generic, Slack, and Teams webhook formats accept a 2xx acknowledgement.""" + server = HTTPServer(("127.0.0.1", 0), _CaptureHandler) + thread = threading.Thread(target=server.handle_request) + thread.start() + try: + payload = { + "delivery_id": "delivery-3", + "machine_id": "node-1", + "type": "node_failure", + "message": "Node failed", + } + result = provider._deliver_webhook(payload, f"http://127.0.0.1:{server.server_port}/notify", channel) + finally: + thread.join(timeout=5) + server.server_close() + assert result == channel + if channel == "webhook": + assert _CaptureHandler.payload["delivery_id"] == "delivery-3" + elif channel == "slack": + assert "delivery-3" in _CaptureHandler.payload["text"] + else: + assert _CaptureHandler.payload["type"] == "message" + + +def test_webhook_rejects_cleartext_non_loopback_url(provider: ModuleType) -> None: + """Webhook credentials cannot be sent over cleartext remote HTTP.""" + with pytest.raises(provider.DeliveryError, match="must use HTTPS"): + provider._deliver_webhook({}, "http://example.com/hook", "webhook") + + +class _FakeSns: + """Minimal SNS client that forwards publications to the fake SQS client.""" + + def __init__(self, sqs: _FakeSqs) -> None: + self.sqs = sqs + self.deleted = False + + def create_topic(self, **kwargs: Any) -> dict[str, str]: + """Return an ephemeral topic ARN.""" + return {"TopicArn": "arn:aws:sns:us-west-2:123456789012:test"} + + def subscribe(self, **kwargs: Any) -> dict[str, str]: + """Return a confirmed subscription ARN.""" + return {"SubscriptionArn": "arn:aws:sns:subscription"} + + def publish(self, **kwargs: Any) -> dict[str, str]: + """Forward the message to the fake queue.""" + self.sqs.body = kwargs["Message"] + return {"MessageId": "message-1"} + + def delete_topic(self, **kwargs: Any) -> None: + """Record cleanup.""" + self.deleted = True + + +class _FakeSqs: + """Minimal SQS client that returns one published message.""" + + body = "" + + def __init__(self) -> None: + self.deleted = False + + def create_queue(self, **kwargs: Any) -> dict[str, str]: + """Return an ephemeral queue URL.""" + return {"QueueUrl": "https://sqs.us-west-2.amazonaws.com/123456789012/test"} + + def get_queue_attributes(self, **kwargs: Any) -> dict[str, dict[str, str]]: + """Return the queue ARN.""" + return {"Attributes": {"QueueArn": "arn:aws:sqs:us-west-2:123456789012:test"}} + + def set_queue_attributes(self, **kwargs: Any) -> None: + """Accept the SNS delivery policy.""" + + def receive_message(self, **kwargs: Any) -> dict[str, list[dict[str, str]]]: + """Return the published message.""" + return {"Messages": [{"Body": self.body, "ReceiptHandle": "receipt-1"}]} + + def delete_message(self, **kwargs: Any) -> None: + """Accept message deletion.""" + + def delete_queue(self, **kwargs: Any) -> None: + """Record cleanup.""" + self.deleted = True + + +def test_aws_backend_proves_receipt_and_cleans_up(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: + """SNS PASS requires the exact delivery ID to arrive and cleans temporary resources.""" + sqs = _FakeSqs() + sns = _FakeSns(sqs) + session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) + fake_boto3 = SimpleNamespace(Session=lambda **kwargs: session) + monkeypatch.setitem(provider.sys.modules, "boto3", fake_boto3) + payload = {"delivery_id": "delivery-4", "machine_id": "node-1", "type": "node_failure"} + assert provider._deliver_aws(payload, "us-west-2") == "aws_sns" + assert sns.deleted + assert sqs.deleted + + +def test_aws_backend_does_not_pass_when_cleanup_fails(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: + """A delivered AWS message cannot PASS while its probe resources remain.""" + sqs = _FakeSqs() + sns = _FakeSns(sqs) + + def fail_delete(**_kwargs: Any) -> None: + raise RuntimeError("denied") + + sns.delete_topic = fail_delete + session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) + monkeypatch.setitem(provider.sys.modules, "boto3", SimpleNamespace(Session=lambda **kwargs: session)) + with pytest.raises(provider.DeliveryError, match="cleanup failed"): + provider._deliver_aws({"delivery_id": "delivery-4b"}, "us-west-2") + assert sqs.deleted + + +def test_kubernetes_backend_requires_receiver_ack(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: + """Kubernetes PASS requires the in-cluster receiver to echo the delivery ID.""" + commands: list[list[str]] = [] + + def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: + commands.append(command) + if "run" in command: + return SimpleNamespace(stdout='{"delivery_id":"delivery-5","status":"delivered"}\n') + return SimpleNamespace(stdout="") + + monkeypatch.setattr(provider, "_run", fake_run) + monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) + monkeypatch.setenv("KUBECTL", "kubectl --context test-cluster") + payload = {"delivery_id": "delivery-5", "machine_id": "node-1", "type": "planned_maintenance"} + assert provider._deliver_kubernetes(payload) == "kubernetes_webhook" + assert any(command[:3] == ["kubectl", "--context", "test-cluster"] for command in commands) + assert any("apply" in command for command in commands) + assert any("run" in command for command in commands) + + +def test_kubernetes_backend_rejects_wrong_ack(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: + """An acknowledgement for another notification cannot produce PASS evidence.""" + monkeypatch.setattr( + provider, + "_run", + lambda *args, **kwargs: SimpleNamespace(stdout='{"delivery_id":"different","status":"delivered"}\n'), + ) + monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) + with pytest.raises(provider.DeliveryError, match="did not acknowledge"): + provider._deliver_kubernetes({"delivery_id": "delivery-6"}) + + +def test_kubernetes_backend_does_not_pass_when_cleanup_fails( + provider: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + """A delivered webhook cannot PASS while its ephemeral namespace remains.""" + monkeypatch.setattr( + provider, + "_run", + lambda *args, **kwargs: SimpleNamespace(stdout='{"delivery_id":"delivery-7","status":"delivered"}\n'), + ) + monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=1)) + with pytest.raises(provider.DeliveryError, match="cleanup failed"): + provider._deliver_kubernetes({"delivery_id": "delivery-7"}) diff --git a/isvtest/src/isvtest/validations/breakfix.py b/isvtest/src/isvtest/validations/breakfix.py index 99612620e..7c3003bcf 100644 --- a/isvtest/src/isvtest/validations/breakfix.py +++ b/isvtest/src/isvtest/validations/breakfix.py @@ -30,6 +30,7 @@ from __future__ import annotations +from datetime import datetime from typing import Any, ClassVar import pytest @@ -388,7 +389,8 @@ class PlannedMaintenanceNotificationCheck(_QueryableRecordsCheck): Step output: success, notification_channel_observable: bool - notifications: list[{machine_id, type, message, notified_at}] + notifications: list[{machine_id, type, message, notified_at, + scheduled_at, channel, delivery_status, delivery_id}] Requires a real notification record, not just the observable flag: the flag alone is the provider asserting its own capability. @@ -403,13 +405,18 @@ class PlannedMaintenanceNotificationCheck(_QueryableRecordsCheck): api_label: ClassVar[str] = "Planned maintenance notification" record_noun: ClassVar[str] = "notification" + def _is_evidence(self, record: Any) -> bool: + """Require successful delivery of a planned-maintenance notification.""" + return _is_delivered_notification(record, "planned_maintenance", require_schedule=True) + class FailureNotificationCheck(_QueryableRecordsCheck): """Validate tenants can be notified of immediate node failure (BFX06-01). Step output: success, notification_channel_observable: bool - notifications: list[{machine_id, type, message, notified_at}] + notifications: list[{machine_id, type, message, notified_at, failed_at, + channel, delivery_status, delivery_id}] Requires a real notification record, for the same reason as PlannedMaintenanceNotificationCheck. @@ -423,3 +430,53 @@ class FailureNotificationCheck(_QueryableRecordsCheck): absent_noun: ClassVar[str] = "immediate failure notifications" api_label: ClassVar[str] = "Immediate failure notification" record_noun: ClassVar[str] = "notification" + + def _is_evidence(self, record: Any) -> bool: + """Require successful delivery of an immediate node-failure notification.""" + if not _is_delivered_notification(record, "node_failure"): + return False + failed_at = _parse_timestamp(record.get("failed_at")) + notified_at = _parse_timestamp(record.get("notified_at")) + if failed_at is None or notified_at is None: + return False + delay_seconds = (notified_at - failed_at).total_seconds() + return 0 <= delay_seconds <= 300 + + +_NON_DELIVERY_CHANNELS = {"", "log", "none", "stdout", "unknown"} + + +def _parse_timestamp(value: Any) -> datetime | None: + """Parse a timezone-aware ISO 8601 timestamp.""" + if not isinstance(value, str) or not value.strip(): + return None + try: + parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00")) + except ValueError: + return None + return parsed if parsed.tzinfo is not None else None + + +def _is_delivered_notification(record: Any, expected_type: str, *, require_schedule: bool = False) -> bool: + """Return whether a normalized record proves tenant notification delivery.""" + if not isinstance(record, dict): + return False + required_strings = ("machine_id", "message", "delivery_id") + if any(not isinstance(record.get(key), str) or not record[key].strip() for key in required_strings): + return False + if record.get("type") != expected_type: + return False + channel = record.get("channel") + if ( + not isinstance(channel, str) + or channel.strip().lower() in _NON_DELIVERY_CHANNELS + or record.get("delivery_status") != "delivered" + ): + return False + notified_at = _parse_timestamp(record.get("notified_at")) + if notified_at is None: + return False + if not require_schedule: + return True + scheduled_at = _parse_timestamp(record.get("scheduled_at")) + return scheduled_at is not None and scheduled_at > notified_at diff --git a/isvtest/tests/test_breakfix.py b/isvtest/tests/test_breakfix.py index 485c8cbcb..a587ad98c 100644 --- a/isvtest/tests/test_breakfix.py +++ b/isvtest/tests/test_breakfix.py @@ -32,31 +32,43 @@ def _run(check_class: type[BaseValidation], step_output: dict[str, Any]) -> Base # (check class, observable flag key, record list key, one sample record) -# BFX05/BFX06 sit here too: a notification channel is held to the same evidence -# bar as the BFX02 query APIs, so the flag alone cannot pass the check. _QUERYABLE_CASES = [ (MaintenanceEventsCheck, "events_queryable", "events", {"machine_id": "m-1", "status": "maintenance"}), (RetirementNoticesCheck, "notices_queryable", "notices", {"machine_id": "m-1", "status": "scheduled"}), (RepairHistoryCheck, "history_queryable", "records", {"machine_id": "m-1", "entries": [{"status": "x"}]}), +] + +_NOTIFICATION_CASES = [ ( PlannedMaintenanceNotificationCheck, - "notification_channel_observable", - "notifications", - {"machine_id": "m-1", "type": "planned_maintenance"}, + "Planned maintenance", + { + "machine_id": "m-1", + "type": "planned_maintenance", + "message": "Scheduled firmware maintenance", + "notified_at": "2026-08-23T10:00:00Z", + "scheduled_at": "2026-08-24T10:00:00Z", + "channel": "webhook", + "delivery_status": "delivered", + "delivery_id": "delivery-1", + }, ), ( FailureNotificationCheck, - "notification_channel_observable", - "notifications", - {"machine_id": "m-1", "type": "node_failure"}, + "Immediate failure", + { + "machine_id": "m-1", + "type": "node_failure", + "message": "Node became unreachable", + "failed_at": "2026-08-23T10:00:00Z", + "notified_at": "2026-08-23T10:00:30Z", + "channel": "webhook", + "delivery_status": "delivered", + "delivery_id": "delivery-2", + }, ), ] -_NOTIFICATION_CASES = [ - (PlannedMaintenanceNotificationCheck, "Planned maintenance"), - (FailureNotificationCheck, "Immediate failure"), -] - class TestQueryableRecordChecks: """Cover the BFX02 query checks that share _QueryableRecordsCheck.""" @@ -170,25 +182,87 @@ def test_fails_when_existing_workloads_unreported(self) -> None: class TestNotificationChecks: """Cover the BFX05-01 planned and BFX06-01 immediate notification checks.""" - @pytest.mark.parametrize(("check_class", "label"), _NOTIFICATION_CASES) - def test_passes_when_a_notification_is_evidenced(self, check_class: type[BaseValidation], label: str) -> None: + @pytest.mark.parametrize(("check_class", "label", "record"), _NOTIFICATION_CASES) + def test_passes_when_a_notification_is_evidenced( + self, check_class: type[BaseValidation], label: str, record: dict[str, Any] + ) -> None: """An observable channel with a real notification passes and names the channel.""" step_output = { "success": True, "notification_channel_observable": True, - "notifications": [{"machine_id": "m-1", "message": "scheduled"}], + "notifications": [record], } check = _run(check_class, step_output) assert check.passed assert label in check.message - @pytest.mark.parametrize(("check_class", "label"), _NOTIFICATION_CASES) - def test_observable_flag_alone_is_not_evidence(self, check_class: type[BaseValidation], label: str) -> None: + @pytest.mark.parametrize(("check_class", "label", "record"), _NOTIFICATION_CASES) + def test_observable_flag_alone_is_not_evidence( + self, check_class: type[BaseValidation], label: str, record: dict[str, Any] + ) -> None: """The flag is the provider asserting its own capability; it is not evidence.""" with pytest.raises(pytest.skip.Exception): _run(check_class, {"success": True, "notification_channel_observable": True}) - @pytest.mark.parametrize(("check_class", "label"), _NOTIFICATION_CASES) - def test_fails_when_channel_unobservable(self, check_class: type[BaseValidation], label: str) -> None: + @pytest.mark.parametrize(("check_class", "label", "record"), _NOTIFICATION_CASES) + def test_fails_when_channel_unobservable( + self, check_class: type[BaseValidation], label: str, record: dict[str, Any] + ) -> None: """A channel the provider cannot observe fails.""" assert not _run(check_class, {"success": True, "notification_channel_observable": False}).passed + + @pytest.mark.parametrize(("check_class", "label", "record"), _NOTIFICATION_CASES) + @pytest.mark.parametrize( + ("field", "value"), + [ + ("delivery_status", "accepted"), + ("delivery_id", ""), + ("channel", "stdout"), + ("notified_at", "not-a-timestamp"), + ], + ) + def test_skips_when_delivery_proof_is_incomplete( + self, + check_class: type[BaseValidation], + label: str, + record: dict[str, Any], + field: str, + value: str, + ) -> None: + """A synthesized record without successful delivery proof cannot pass.""" + bad_record = {**record, field: value} + with pytest.raises(pytest.skip.Exception): + _run( + check_class, + { + "success": True, + "notification_channel_observable": True, + "notifications": [bad_record], + }, + ) + + def test_planned_notification_requires_a_future_schedule(self) -> None: + """Planned-maintenance evidence must identify maintenance after notification.""" + record = {**_NOTIFICATION_CASES[0][2], "scheduled_at": "2026-08-23T09:59:59Z"} + with pytest.raises(pytest.skip.Exception): + _run( + PlannedMaintenanceNotificationCheck, + {"success": True, "notification_channel_observable": True, "notifications": [record]}, + ) + + def test_failure_notification_must_be_immediate(self) -> None: + """Failure delivery more than five minutes after detection is not immediate.""" + record = {**_NOTIFICATION_CASES[1][2], "notified_at": "2026-08-23T10:05:01Z"} + with pytest.raises(pytest.skip.Exception): + _run( + FailureNotificationCheck, + {"success": True, "notification_channel_observable": True, "notifications": [record]}, + ) + + def test_provider_specific_communication_channel_is_supported(self) -> None: + """The provider-neutral contract accepts communication systems beyond built-ins.""" + record = {**_NOTIFICATION_CASES[0][2], "channel": "pagerduty"} + assert _run( + PlannedMaintenanceNotificationCheck, + {"success": True, "notification_channel_observable": True, "notifications": [record]}, + ).passed From 9bf7c8898ae8cbe9a2d4b13c140d2b27093e3e0b Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Sun, 23 Aug 2026 03:48:02 -0700 Subject: [PATCH 2/8] fix(breakfix): address notification review findings Signed-off-by: Hasan Khan --- .../providers/aws/config/bare_metal.yaml | 4 ++ .../breakfix/query_tenant_notification.py | 22 ++++++-- .../test_notification_delivery_provider.py | 52 ++++++++++++++++++- 3 files changed, 72 insertions(+), 6 deletions(-) diff --git a/isvctl/configs/providers/aws/config/bare_metal.yaml b/isvctl/configs/providers/aws/config/bare_metal.yaml index a07e6e990..ee80e4855 100644 --- a/isvctl/configs/providers/aws/config/bare_metal.yaml +++ b/isvctl/configs/providers/aws/config/bare_metal.yaml @@ -276,6 +276,8 @@ commands: - "aws" - "--event-type" - "planned_maintenance" + - "--machine-id" + - "{{steps.launch_instance.instance_id}}" - "--message" - "Planned node maintenance notification validation" - "--region" @@ -291,6 +293,8 @@ commands: - "aws" - "--event-type" - "node_failure" + - "--machine-id" + - "{{steps.launch_instance.instance_id}}" - "--message" - "Immediate node failure notification validation" - "--region" diff --git a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py index 7c84462fe..e4de949b4 100644 --- a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py +++ b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py @@ -94,6 +94,7 @@ def _payload(args: argparse.Namespace, delivery_id: str, started_at: datetime) - def _deliver_aws(payload: dict[str, str], region: str) -> str: """Publish through SNS and prove receipt through an ephemeral SQS subscriber.""" try: + # Lazy import: boto3 is needed only by the optional AWS transport. import boto3 except ImportError as exc: # pragma: no cover - dependency is present in the workspace raise DeliveryError("AWS notification backend requires boto3") from exc @@ -149,6 +150,7 @@ def _deliver_aws(payload: dict[str, str], region: str) -> str: except Exception as exc: raise DeliveryError(f"AWS notification delivery failed: {type(exc).__name__}") from exc finally: + pending_error = sys.exc_info()[1] cleanup_failed = False if topic_arn: try: @@ -161,7 +163,10 @@ def _deliver_aws(payload: dict[str, str], region: str) -> str: except Exception: cleanup_failed = True if cleanup_failed: - raise DeliveryError("AWS notification cleanup failed") + message = "AWS notification cleanup failed" + if isinstance(pending_error, DeliveryError): + message = f"{pending_error}; {message}" + raise DeliveryError(message) from pending_error def _kubectl_base() -> list[str]: @@ -289,6 +294,8 @@ def _deliver_kubernetes(payload: dict[str, str]) -> str: return "kubernetes_webhook" finally: if created_namespace: + pending_error = sys.exc_info()[1] + cleanup_failed = False try: cleanup = subprocess.run( [ @@ -305,10 +312,15 @@ def _deliver_kubernetes(payload: dict[str, str]) -> str: timeout=70, check=False, ) - except (OSError, subprocess.TimeoutExpired) as exc: - raise DeliveryError("Kubernetes notification cleanup failed") from exc - if cleanup.returncode != 0: - raise DeliveryError("Kubernetes notification cleanup failed") + except (OSError, subprocess.TimeoutExpired): + cleanup_failed = True + else: + cleanup_failed = cleanup.returncode != 0 + if cleanup_failed: + message = "Kubernetes notification cleanup failed" + if isinstance(pending_error, DeliveryError): + message = f"{pending_error}; {message}" + raise DeliveryError(message) from pending_error def _deliver_webhook(payload: dict[str, str], url: str, channel: str) -> str: diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index 52f52eb6d..04b9a0cc6 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -16,8 +16,10 @@ from typing import Any, ClassVar import pytest +import yaml SCRIPT = Path(__file__).parents[1] / "configs" / "providers" / "shared" / "breakfix" / "query_tenant_notification.py" +AWS_CONFIG = Path(__file__).parents[1] / "configs" / "providers" / "aws" / "config" / "bare_metal.yaml" def _load_module() -> ModuleType: @@ -66,6 +68,16 @@ def test_failure_payload_records_detection_time(provider: ModuleType) -> None: assert payload["type"] == "node_failure" +def test_aws_steps_bind_notification_to_launched_instance() -> None: + """Both AWS notification records identify the bare-metal instance under test.""" + config = yaml.safe_load(AWS_CONFIG.read_text()) + steps = {step["name"]: step for step in config["commands"]["bare_metal"]["steps"]} + for name in ("query_planned_notifications", "query_failure_notifications"): + args = steps[name]["args"] + index = args.index("--machine-id") + assert args[index + 1] == "{{steps.launch_instance.instance_id}}" + + class _CaptureHandler(BaseHTTPRequestHandler): """Capture one test webhook request and acknowledge it.""" @@ -86,7 +98,8 @@ def log_message(self, fmt: str, *args: object) -> None: def test_webhook_backends_accept_acknowledged_delivery(provider: ModuleType, channel: str) -> None: """Generic, Slack, and Teams webhook formats accept a 2xx acknowledgement.""" server = HTTPServer(("127.0.0.1", 0), _CaptureHandler) - thread = threading.Thread(target=server.handle_request) + server.timeout = 10 + thread = threading.Thread(target=server.handle_request, daemon=True) thread.start() try: payload = { @@ -199,6 +212,29 @@ def fail_delete(**_kwargs: Any) -> None: assert sqs.deleted +def test_aws_backend_preserves_delivery_and_cleanup_failures( + provider: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + """AWS diagnostics retain the delivery failure when cleanup also fails.""" + sqs = _FakeSqs() + sns = _FakeSns(sqs) + + def no_messages(**_kwargs: Any) -> dict[str, Any]: + return {} + + sqs.receive_message = no_messages + + def fail_delete(**_kwargs: Any) -> None: + raise RuntimeError("denied") + + sns.delete_topic = fail_delete + session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) + monkeypatch.setitem(provider.sys.modules, "boto3", SimpleNamespace(Session=lambda **kwargs: session)) + with pytest.raises(provider.DeliveryError, match=r"did not receive.*cleanup failed"): + provider._deliver_aws({"delivery_id": "delivery-4c"}, "us-west-2") + assert sqs.deleted + + def test_kubernetes_backend_requires_receiver_ack(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: """Kubernetes PASS requires the in-cluster receiver to echo the delivery ID.""" commands: list[list[str]] = [] @@ -243,3 +279,17 @@ def test_kubernetes_backend_does_not_pass_when_cleanup_fails( monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=1)) with pytest.raises(provider.DeliveryError, match="cleanup failed"): provider._deliver_kubernetes({"delivery_id": "delivery-7"}) + + +def test_kubernetes_backend_preserves_delivery_and_cleanup_failures( + provider: ModuleType, monkeypatch: pytest.MonkeyPatch +) -> None: + """Kubernetes diagnostics retain the acknowledgement failure when cleanup also fails.""" + monkeypatch.setattr( + provider, + "_run", + lambda *args, **kwargs: SimpleNamespace(stdout='{"delivery_id":"different","status":"delivered"}\n'), + ) + monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=1)) + with pytest.raises(provider.DeliveryError, match=r"did not acknowledge.*cleanup failed"): + provider._deliver_kubernetes({"delivery_id": "delivery-8"}) From 304b0be029a515d0e6e93a17899dd14dcecdcdd9 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Sun, 23 Aug 2026 03:54:59 -0700 Subject: [PATCH 3/8] test(breakfix): document notification helpers Signed-off-by: Hasan Khan --- isvctl/tests/test_notification_delivery_provider.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index 04b9a0cc6..b4964f42b 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -220,11 +220,13 @@ def test_aws_backend_preserves_delivery_and_cleanup_failures( sns = _FakeSns(sqs) def no_messages(**_kwargs: Any) -> dict[str, Any]: + """Simulate an empty receive-message poll.""" return {} sqs.receive_message = no_messages def fail_delete(**_kwargs: Any) -> None: + """Simulate a topic cleanup failure.""" raise RuntimeError("denied") sns.delete_topic = fail_delete From c8efbe73887b9a34f70d7df9cc8d985a271a1d63 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Mon, 24 Aug 2026 11:14:00 -0700 Subject: [PATCH 4/8] fix(breakfix): preserve notification failure JSON Signed-off-by: Hasan Khan --- .../breakfix/query_tenant_notification.py | 24 ++++++++++- .../test_notification_delivery_provider.py | 40 +++++++++++++++++++ 2 files changed, 62 insertions(+), 2 deletions(-) diff --git a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py index e4de949b4..bf88cb331 100644 --- a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py +++ b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py @@ -65,6 +65,14 @@ class DeliveryError(RuntimeError): """Raised when a notification cannot be proved delivered.""" +class ProviderArgumentParser(argparse.ArgumentParser): + """Raise provider errors instead of exiting without a JSON result.""" + + def error(self, message: str) -> None: + """Convert invalid arguments into the provider failure path.""" + raise DeliveryError(f"Invalid arguments: {message}") + + def _timestamp(value: datetime) -> str: """Render a UTC timestamp in the provider-neutral ISO 8601 format.""" return value.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z") @@ -365,7 +373,7 @@ def _deliver_webhook(payload: dict[str, str], url: str, channel: str) -> str: def _parse_args() -> argparse.Namespace: """Parse the delivery probe arguments.""" - parser = argparse.ArgumentParser(description="Deliver and verify one tenant notification") + parser = ProviderArgumentParser(description="Deliver and verify one tenant notification") parser.add_argument("--backend", choices=("aws", "kubernetes", "webhook"), required=True) parser.add_argument("--event-type", choices=("planned_maintenance", "node_failure"), required=True) parser.add_argument("--machine-id", default="notification-probe-node") @@ -390,7 +398,19 @@ def _parse_args() -> argparse.Namespace: def main() -> int: """Deliver a notification and emit normalized proof of receipt.""" - args = _parse_args() + try: + args = _parse_args() + except DeliveryError as exc: + return _emit( + { + "success": False, + "platform": "notification", + "test_name": "query_tenant_notification", + "notification_channel_observable": False, + "notifications": [], + "error": str(exc), + } + ) metadata = { "platform": args.backend, "test_name": ( diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index b4964f42b..5c914ce01 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -7,6 +7,7 @@ import importlib.util import json +import sys import threading from argparse import Namespace from datetime import UTC, datetime @@ -78,6 +79,45 @@ def test_aws_steps_bind_notification_to_launched_instance() -> None: assert args[index + 1] == "{{steps.launch_instance.instance_id}}" +@pytest.mark.parametrize( + "argv", + [ + [SCRIPT.name, "--backend", "invalid", "--event-type", "node_failure", "--message", "failed"], + [SCRIPT.name, "--backend", "aws", "--event-type", "node_failure"], + [ + SCRIPT.name, + "--backend", + "aws", + "--event-type", + "planned_maintenance", + "--message", + "planned", + "--schedule-hours=0", + ], + ], +) +def test_invalid_arguments_emit_failure_json( + argv: list[str], + monkeypatch: pytest.MonkeyPatch, + capsys: pytest.CaptureFixture[str], +) -> None: + """Parser failures must preserve the notification provider contract.""" + provider = _load_module() + monkeypatch.setattr(sys, "argv", argv) + + assert provider.main() == 1 + captured = capsys.readouterr() + payload = json.loads(captured.out) + + assert captured.err == "" + assert payload["success"] is False + assert payload["platform"] == "notification" + assert payload["test_name"] == "query_tenant_notification" + assert payload["notification_channel_observable"] is False + assert payload["notifications"] == [] + assert payload["error"].startswith("Invalid arguments:") + + class _CaptureHandler(BaseHTTPRequestHandler): """Capture one test webhook request and acknowledge it.""" From f0032b02b1f5b6ab2771138a2113b95fb3377761 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Mon, 24 Aug 2026 11:56:11 -0700 Subject: [PATCH 5/8] fix(breakfix): expose notifications in Kubernetes suites Signed-off-by: Hasan Khan --- docs/test-plan.yaml | 2 + isvctl/configs/providers/aws/config/eks.yaml | 36 +++++++++++++ isvctl/configs/providers/minikube.yaml | 32 ++++++++++++ .../configs/providers/my-isv/config/k8s.yaml | 34 ++++++++++++ isvctl/configs/suites/README.md | 4 ++ isvctl/configs/suites/bare_metal.yaml | 4 +- isvctl/configs/suites/k8s.yaml | 14 +++++ .../test_notification_delivery_provider.py | 52 +++++++++++++++++++ isvtest/src/isvtest/validations/breakfix.py | 8 +++ 9 files changed, 184 insertions(+), 2 deletions(-) diff --git a/docs/test-plan.yaml b/docs/test-plan.yaml index dd14fdbd3..74a81422f 100644 --- a/docs/test-plan.yaml +++ b/docs/test-plan.yaml @@ -2011,6 +2011,7 @@ domains: labels: - bare_metal - breakfix + - kubernetes dependencies: - LimitedEnv req_id: BFX05 @@ -2024,6 +2025,7 @@ domains: labels: - bare_metal - breakfix + - kubernetes dependencies: - LimitedEnv req_id: BFX06 diff --git a/isvctl/configs/providers/aws/config/eks.yaml b/isvctl/configs/providers/aws/config/eks.yaml index 62707c39c..c5650aac4 100644 --- a/isvctl/configs/providers/aws/config/eks.yaml +++ b/isvctl/configs/providers/aws/config/eks.yaml @@ -206,6 +206,42 @@ commands: TF_AUTO_APPROVE: "true" NODE_POOL_STATE_FILE: "terraform-delete.tfstate" + # BFX05/BFX06: prove acknowledged tenant notification delivery through + # an ephemeral in-cluster HTTP receiver, then remove all probe resources. + - name: query_planned_notifications + phase: test + continue_on_failure: true + command: "python3 ../../shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "kubernetes" + - "--event-type" + - "planned_maintenance" + - "--machine-id" + - "notification-probe-node" + - "--message" + - "Planned node maintenance notification validation" + requires_available_validations: + - K8sPlannedMaintenanceNotificationCheck + timeout: 300 + + - name: query_failure_notifications + phase: test + continue_on_failure: true + command: "python3 ../../shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "kubernetes" + - "--event-type" + - "node_failure" + - "--machine-id" + - "notification-probe-node" + - "--message" + - "Immediate node failure notification validation" + requires_available_validations: + - K8sFailureNotificationCheck + timeout: 300 + - name: teardown phase: teardown command: "../scripts/eks/teardown.sh" diff --git a/isvctl/configs/providers/minikube.yaml b/isvctl/configs/providers/minikube.yaml index 26b45ad42..0457d177d 100644 --- a/isvctl/configs/providers/minikube.yaml +++ b/isvctl/configs/providers/minikube.yaml @@ -34,6 +34,38 @@ commands: phase: setup command: "my-isv/scripts/k8s/setup_minikube.sh" timeout: 60 + - name: query_planned_notifications + phase: test + continue_on_failure: true + command: "python shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "kubernetes" + - "--event-type" + - "planned_maintenance" + - "--machine-id" + - "notification-probe-node" + - "--message" + - "Planned node maintenance notification validation" + requires_available_validations: + - K8sPlannedMaintenanceNotificationCheck + timeout: 300 + - name: query_failure_notifications + phase: test + continue_on_failure: true + command: "python shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "kubernetes" + - "--event-type" + - "node_failure" + - "--machine-id" + - "notification-probe-node" + - "--message" + - "Immediate node failure notification validation" + requires_available_validations: + - K8sFailureNotificationCheck + timeout: 300 - name: teardown phase: teardown command: "my-isv/scripts/k8s/teardown_minikube.sh" diff --git a/isvctl/configs/providers/my-isv/config/k8s.yaml b/isvctl/configs/providers/my-isv/config/k8s.yaml index dc21c126d..7f9f90399 100644 --- a/isvctl/configs/providers/my-isv/config/k8s.yaml +++ b/isvctl/configs/providers/my-isv/config/k8s.yaml @@ -73,6 +73,40 @@ commands: args: ["--region", "{{region}}"] timeout: 600 + - name: query_planned_notifications + phase: test + continue_on_failure: true + command: "python ../../shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "kubernetes" + - "--event-type" + - "planned_maintenance" + - "--machine-id" + - "notification-probe-node" + - "--message" + - "Planned node maintenance notification validation" + requires_available_validations: + - K8sPlannedMaintenanceNotificationCheck + timeout: 300 + + - name: query_failure_notifications + phase: test + continue_on_failure: true + command: "python ../../shared/breakfix/query_tenant_notification.py" + args: + - "--backend" + - "kubernetes" + - "--event-type" + - "node_failure" + - "--machine-id" + - "notification-probe-node" + - "--message" + - "Immediate node failure notification validation" + requires_available_validations: + - K8sFailureNotificationCheck + timeout: 300 + tests: description: "my-isv Kubernetes platform suite (real cluster required)" settings: diff --git a/isvctl/configs/suites/README.md b/isvctl/configs/suites/README.md index 214cb58ac..fab1a0caf 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -272,6 +272,10 @@ its plan item is not platform-scoped. | `query_planned_notifications` | test | provider implementation or `providers/shared/breakfix/query_tenant_notification.py` | `notification_channel_observable`, `notifications[].{machine_id,type,message,notified_at,scheduled_at,channel,delivery_status,delivery_id}`; PASS requires acknowledged delivery and a schedule after notification (BFX05-01) | | `query_failure_notifications` | test | provider implementation or `providers/shared/breakfix/query_tenant_notification.py` | `notification_channel_observable`, `notifications[].{machine_id,type,message,failed_at,notified_at,channel,delivery_status,delivery_id}`; PASS requires acknowledged delivery within five minutes of failure (BFX06-01) | +The canonical Kubernetes suite wires both notification checks through +provider-owned delivery steps. The shared Kubernetes backend proves receipt +with an ephemeral in-cluster HTTP receiver and removes the probe namespace. + ### Storage (`storage.yaml`) Umbrella suite for the storage capability area. Today it covers persistent block diff --git a/isvctl/configs/suites/bare_metal.yaml b/isvctl/configs/suites/bare_metal.yaml index 2c278277a..b3173c13f 100644 --- a/isvctl/configs/suites/bare_metal.yaml +++ b/isvctl/configs/suites/bare_metal.yaml @@ -869,14 +869,14 @@ tests: checks: PlannedMaintenanceNotificationCheck: test_id: "BFX05-01" - labels: ["bare_metal", "breakfix"] + labels: ["bare_metal", "breakfix", "kubernetes"] failure_notification: step: query_failure_notifications checks: FailureNotificationCheck: test_id: "BFX06-01" - labels: ["bare_metal", "breakfix"] + labels: ["bare_metal", "breakfix", "kubernetes"] exclude: labels: [] diff --git a/isvctl/configs/suites/k8s.yaml b/isvctl/configs/suites/k8s.yaml index 21000f172..53e05a540 100644 --- a/isvctl/configs/suites/k8s.yaml +++ b/isvctl/configs/suites/k8s.yaml @@ -403,5 +403,19 @@ tests: test_id: "BFX01-04" labels: ["kubernetes", "breakfix", "min_req"] + planned_maintenance_notification: + step: query_planned_notifications + checks: + K8sPlannedMaintenanceNotificationCheck: + test_id: "BFX05-01" + labels: ["bare_metal", "breakfix", "kubernetes"] + + failure_notification: + step: query_failure_notifications + checks: + K8sFailureNotificationCheck: + test_id: "BFX06-01" + labels: ["bare_metal", "breakfix", "kubernetes"] + exclude: labels: [] diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index 5c914ce01..9f1c1a858 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -21,6 +21,10 @@ SCRIPT = Path(__file__).parents[1] / "configs" / "providers" / "shared" / "breakfix" / "query_tenant_notification.py" AWS_CONFIG = Path(__file__).parents[1] / "configs" / "providers" / "aws" / "config" / "bare_metal.yaml" +K8S_SUITE = Path(__file__).parents[1] / "configs" / "suites" / "k8s.yaml" +MINIKUBE_CONFIG = Path(__file__).parents[1] / "configs" / "providers" / "minikube.yaml" +AWS_EKS_CONFIG = Path(__file__).parents[1] / "configs" / "providers" / "aws" / "config" / "eks.yaml" +MY_ISV_K8S_CONFIG = Path(__file__).parents[1] / "configs" / "providers" / "my-isv" / "config" / "k8s.yaml" def _load_module() -> ModuleType: @@ -79,6 +83,54 @@ def test_aws_steps_bind_notification_to_launched_instance() -> None: assert args[index + 1] == "{{steps.launch_instance.instance_id}}" +def test_k8s_suite_declares_notification_delivery_validations() -> None: + """The canonical Kubernetes suite must expose both notification checks.""" + config = yaml.safe_load(K8S_SUITE.read_text()) + validations = config["tests"]["validations"] + + assert validations["planned_maintenance_notification"] == { + "step": "query_planned_notifications", + "checks": { + "K8sPlannedMaintenanceNotificationCheck": { + "test_id": "BFX05-01", + "labels": ["bare_metal", "breakfix", "kubernetes"], + } + }, + } + assert validations["failure_notification"] == { + "step": "query_failure_notifications", + "checks": { + "K8sFailureNotificationCheck": { + "test_id": "BFX06-01", + "labels": ["bare_metal", "breakfix", "kubernetes"], + } + }, + } + + +@pytest.mark.parametrize( + ("provider_config", "command"), + [ + (MINIKUBE_CONFIG, "python shared/breakfix/query_tenant_notification.py"), + (AWS_EKS_CONFIG, "python3 ../../shared/breakfix/query_tenant_notification.py"), + (MY_ISV_K8S_CONFIG, "python ../../shared/breakfix/query_tenant_notification.py"), + ], +) +def test_kubernetes_providers_wire_notification_delivery(provider_config: Path, command: str) -> None: + """Kubernetes provider configs must execute both shared delivery probes.""" + config = yaml.safe_load(provider_config.read_text()) + steps = {item["name"]: item for item in config["commands"]["kubernetes"]["steps"]} + + planned = steps["query_planned_notifications"] + failure = steps["query_failure_notifications"] + assert planned["command"] == command + assert failure["command"] == command + assert planned["requires_available_validations"] == ["K8sPlannedMaintenanceNotificationCheck"] + assert failure["requires_available_validations"] == ["K8sFailureNotificationCheck"] + assert planned["args"][planned["args"].index("--backend") + 1] == "kubernetes" + assert failure["args"][failure["args"].index("--backend") + 1] == "kubernetes" + + @pytest.mark.parametrize( "argv", [ diff --git a/isvtest/src/isvtest/validations/breakfix.py b/isvtest/src/isvtest/validations/breakfix.py index 7c3003bcf..b18410bda 100644 --- a/isvtest/src/isvtest/validations/breakfix.py +++ b/isvtest/src/isvtest/validations/breakfix.py @@ -410,6 +410,10 @@ def _is_evidence(self, record: Any) -> bool: return _is_delivered_notification(record, "planned_maintenance", require_schedule=True) +class K8sPlannedMaintenanceNotificationCheck(PlannedMaintenanceNotificationCheck): + """Validate BFX05-01 through an acknowledged in-cluster delivery.""" + + class FailureNotificationCheck(_QueryableRecordsCheck): """Validate tenants can be notified of immediate node failure (BFX06-01). @@ -443,6 +447,10 @@ def _is_evidence(self, record: Any) -> bool: return 0 <= delay_seconds <= 300 +class K8sFailureNotificationCheck(FailureNotificationCheck): + """Validate BFX06-01 through an acknowledged in-cluster delivery.""" + + _NON_DELIVERY_CHANNELS = {"", "log", "none", "stdout", "unknown"} From 11a0d732a939a55abf8d833c7210d93817968d80 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Mon, 24 Aug 2026 15:05:51 -0700 Subject: [PATCH 6/8] fix(breakfix): timestamp failures at publish time Signed-off-by: Hasan Khan --- .../breakfix/query_tenant_notification.py | 9 +++++++ .../test_notification_delivery_provider.py | 27 ++++++++++++++++--- 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py index bf88cb331..0c1e5018e 100644 --- a/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py +++ b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py @@ -99,6 +99,12 @@ def _payload(args: argparse.Namespace, delivery_id: str, started_at: datetime) - return common +def _mark_failure_detected(payload: dict[str, str]) -> None: + """Timestamp a simulated node failure immediately before publication.""" + if payload.get("type") == "node_failure": + payload["failed_at"] = _timestamp(datetime.now(UTC)) + + def _deliver_aws(payload: dict[str, str], region: str) -> str: """Publish through SNS and prove receipt through an ephemeral SQS subscriber.""" try: @@ -140,6 +146,7 @@ def _deliver_aws(payload: dict[str, str], region: str) -> str: Attributes={"RawMessageDelivery": "true"}, ReturnSubscriptionArn=True, ) + _mark_failure_detected(payload) sns.publish(TopicArn=topic_arn, Message=json.dumps(payload, separators=(",", ":"))) for _ in range(3): @@ -269,6 +276,7 @@ def _deliver_kubernetes(payload: dict[str, str]) -> str: objects = json.dumps(_kubernetes_objects(namespace, receiver_name)) _run([*kubectl, "apply", "-f", "-"], stdin=objects) _run([*kubectl, "wait", "--for=condition=Ready", f"pod/{receiver_name}", "-n", namespace, "--timeout=120s"]) + _mark_failure_detected(payload) result = _run( [ *kubectl, @@ -335,6 +343,7 @@ def _deliver_webhook(payload: dict[str, str], url: str, channel: str) -> str: """Post to a configured Slack, Teams, or generic HTTP webhook.""" if not url.startswith("https://") and not url.startswith("http://127.0.0.1:"): raise DeliveryError("Webhook URL must use HTTPS (loopback HTTP is allowed for local testing)") + _mark_failure_detected(payload) outgoing: dict[str, Any] = payload text = ( f"{payload['message']}\nNode: {payload['machine_id']}\n" diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index 9f1c1a858..48d37e12b 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -187,7 +187,9 @@ def log_message(self, fmt: str, *args: object) -> None: @pytest.mark.parametrize("channel", ["webhook", "slack", "teams"]) -def test_webhook_backends_accept_acknowledged_delivery(provider: ModuleType, channel: str) -> None: +def test_webhook_backends_accept_acknowledged_delivery( + provider: ModuleType, channel: str, monkeypatch: pytest.MonkeyPatch +) -> None: """Generic, Slack, and Teams webhook formats accept a 2xx acknowledgement.""" server = HTTPServer(("127.0.0.1", 0), _CaptureHandler) server.timeout = 10 @@ -199,12 +201,15 @@ def test_webhook_backends_accept_acknowledged_delivery(provider: ModuleType, cha "machine_id": "node-1", "type": "node_failure", "message": "Node failed", + "failed_at": "before-webhook-setup", } + monkeypatch.setattr(provider, "_timestamp", lambda value: "webhook-publish-time") result = provider._deliver_webhook(payload, f"http://127.0.0.1:{server.server_port}/notify", channel) finally: thread.join(timeout=5) server.server_close() assert result == channel + assert payload["failed_at"] == "webhook-publish-time" if channel == "webhook": assert _CaptureHandler.payload["delivery_id"] == "delivery-3" elif channel == "slack": @@ -282,8 +287,15 @@ def test_aws_backend_proves_receipt_and_cleans_up(provider: ModuleType, monkeypa session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) fake_boto3 = SimpleNamespace(Session=lambda **kwargs: session) monkeypatch.setitem(provider.sys.modules, "boto3", fake_boto3) - payload = {"delivery_id": "delivery-4", "machine_id": "node-1", "type": "node_failure"} + monkeypatch.setattr(provider, "_timestamp", lambda value: "aws-publish-time") + payload = { + "delivery_id": "delivery-4", + "machine_id": "node-1", + "type": "node_failure", + "failed_at": "before-aws-setup", + } assert provider._deliver_aws(payload, "us-west-2") == "aws_sns" + assert json.loads(sqs.body)["failed_at"] == "aws-publish-time" assert sns.deleted assert sqs.deleted @@ -341,12 +353,21 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: monkeypatch.setattr(provider, "_run", fake_run) monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) + monkeypatch.setattr(provider, "_timestamp", lambda value: "kubernetes-publish-time") monkeypatch.setenv("KUBECTL", "kubectl --context test-cluster") - payload = {"delivery_id": "delivery-5", "machine_id": "node-1", "type": "planned_maintenance"} + payload = { + "delivery_id": "delivery-5", + "machine_id": "node-1", + "type": "node_failure", + "failed_at": "before-kubernetes-setup", + } assert provider._deliver_kubernetes(payload) == "kubernetes_webhook" assert any(command[:3] == ["kubectl", "--context", "test-cluster"] for command in commands) assert any("apply" in command for command in commands) assert any("run" in command for command in commands) + sender = next(command for command in commands if "run" in command) + sent_payload = json.loads(sender[sender.index("--data-binary") + 1]) + assert sent_payload["failed_at"] == "kubernetes-publish-time" def test_kubernetes_backend_rejects_wrong_ack(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: From 71753ec59e964863d3b0cfc2a9b317e760105953 Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Mon, 24 Aug 2026 15:13:34 -0700 Subject: [PATCH 7/8] test(breakfix): use typed timestamp stubs Signed-off-by: Hasan Khan --- .../test_notification_delivery_provider.py | 20 ++++++++++++++++--- 1 file changed, 17 insertions(+), 3 deletions(-) diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index 48d37e12b..c63297dd0 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -191,6 +191,11 @@ def test_webhook_backends_accept_acknowledged_delivery( provider: ModuleType, channel: str, monkeypatch: pytest.MonkeyPatch ) -> None: """Generic, Slack, and Teams webhook formats accept a 2xx acknowledgement.""" + + def webhook_publish_time(_value: datetime) -> str: + """Return the deterministic webhook publication time.""" + return "webhook-publish-time" + server = HTTPServer(("127.0.0.1", 0), _CaptureHandler) server.timeout = 10 thread = threading.Thread(target=server.handle_request, daemon=True) @@ -203,7 +208,7 @@ def test_webhook_backends_accept_acknowledged_delivery( "message": "Node failed", "failed_at": "before-webhook-setup", } - monkeypatch.setattr(provider, "_timestamp", lambda value: "webhook-publish-time") + monkeypatch.setattr(provider, "_timestamp", webhook_publish_time) result = provider._deliver_webhook(payload, f"http://127.0.0.1:{server.server_port}/notify", channel) finally: thread.join(timeout=5) @@ -282,12 +287,17 @@ def delete_queue(self, **kwargs: Any) -> None: def test_aws_backend_proves_receipt_and_cleans_up(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: """SNS PASS requires the exact delivery ID to arrive and cleans temporary resources.""" + + def aws_publish_time(_value: datetime) -> str: + """Return the deterministic AWS publication time.""" + return "aws-publish-time" + sqs = _FakeSqs() sns = _FakeSns(sqs) session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) fake_boto3 = SimpleNamespace(Session=lambda **kwargs: session) monkeypatch.setitem(provider.sys.modules, "boto3", fake_boto3) - monkeypatch.setattr(provider, "_timestamp", lambda value: "aws-publish-time") + monkeypatch.setattr(provider, "_timestamp", aws_publish_time) payload = { "delivery_id": "delivery-4", "machine_id": "node-1", @@ -345,6 +355,10 @@ def test_kubernetes_backend_requires_receiver_ack(provider: ModuleType, monkeypa """Kubernetes PASS requires the in-cluster receiver to echo the delivery ID.""" commands: list[list[str]] = [] + def kubernetes_publish_time(_value: datetime) -> str: + """Return the deterministic Kubernetes publication time.""" + return "kubernetes-publish-time" + def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: commands.append(command) if "run" in command: @@ -353,7 +367,7 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: monkeypatch.setattr(provider, "_run", fake_run) monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) - monkeypatch.setattr(provider, "_timestamp", lambda value: "kubernetes-publish-time") + monkeypatch.setattr(provider, "_timestamp", kubernetes_publish_time) monkeypatch.setenv("KUBECTL", "kubectl --context test-cluster") payload = { "delivery_id": "delivery-5", From 158ee88cee237606978a60d61872f3fc30756baa Mon Sep 17 00:00:00 2001 From: Hasan Khan Date: Mon, 24 Aug 2026 15:21:04 -0700 Subject: [PATCH 8/8] test(breakfix): type provider stubs Signed-off-by: Hasan Khan --- .../test_notification_delivery_provider.py | 72 ++++++++++++------- 1 file changed, 46 insertions(+), 26 deletions(-) diff --git a/isvctl/tests/test_notification_delivery_provider.py b/isvctl/tests/test_notification_delivery_provider.py index c63297dd0..ae852e762 100644 --- a/isvctl/tests/test_notification_delivery_provider.py +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -285,6 +285,42 @@ def delete_queue(self, **kwargs: Any) -> None: self.deleted = True +def _fake_boto3(sns: _FakeSns, sqs: _FakeSqs) -> SimpleNamespace: + """Return a minimal boto3 module backed by the supplied fake clients.""" + + def client(name: str) -> _FakeSns | _FakeSqs: + """Return the requested fake AWS client.""" + return sns if name == "sns" else sqs + + session = SimpleNamespace(client=client) + + def session_factory(**_kwargs: Any) -> SimpleNamespace: + """Return the shared fake AWS session.""" + return session + + return SimpleNamespace(Session=session_factory) + + +def _successful_subprocess(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + """Return a successful subprocess result.""" + return SimpleNamespace(returncode=0) + + +def _failed_subprocess(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + """Return a failed subprocess result.""" + return SimpleNamespace(returncode=1) + + +def _wrong_ack(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + """Return an acknowledgement for a different delivery.""" + return SimpleNamespace(stdout='{"delivery_id":"different","status":"delivered"}\n') + + +def _delivered_ack(*_args: Any, **_kwargs: Any) -> SimpleNamespace: + """Return an acknowledgement for the expected delivery.""" + return SimpleNamespace(stdout='{"delivery_id":"delivery-7","status":"delivered"}\n') + + def test_aws_backend_proves_receipt_and_cleans_up(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: """SNS PASS requires the exact delivery ID to arrive and cleans temporary resources.""" @@ -294,9 +330,7 @@ def aws_publish_time(_value: datetime) -> str: sqs = _FakeSqs() sns = _FakeSns(sqs) - session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) - fake_boto3 = SimpleNamespace(Session=lambda **kwargs: session) - monkeypatch.setitem(provider.sys.modules, "boto3", fake_boto3) + monkeypatch.setitem(provider.sys.modules, "boto3", _fake_boto3(sns, sqs)) monkeypatch.setattr(provider, "_timestamp", aws_publish_time) payload = { "delivery_id": "delivery-4", @@ -319,8 +353,7 @@ def fail_delete(**_kwargs: Any) -> None: raise RuntimeError("denied") sns.delete_topic = fail_delete - session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) - monkeypatch.setitem(provider.sys.modules, "boto3", SimpleNamespace(Session=lambda **kwargs: session)) + monkeypatch.setitem(provider.sys.modules, "boto3", _fake_boto3(sns, sqs)) with pytest.raises(provider.DeliveryError, match="cleanup failed"): provider._deliver_aws({"delivery_id": "delivery-4b"}, "us-west-2") assert sqs.deleted @@ -344,8 +377,7 @@ def fail_delete(**_kwargs: Any) -> None: raise RuntimeError("denied") sns.delete_topic = fail_delete - session = SimpleNamespace(client=lambda name: sns if name == "sns" else sqs) - monkeypatch.setitem(provider.sys.modules, "boto3", SimpleNamespace(Session=lambda **kwargs: session)) + monkeypatch.setitem(provider.sys.modules, "boto3", _fake_boto3(sns, sqs)) with pytest.raises(provider.DeliveryError, match=r"did not receive.*cleanup failed"): provider._deliver_aws({"delivery_id": "delivery-4c"}, "us-west-2") assert sqs.deleted @@ -366,7 +398,7 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: return SimpleNamespace(stdout="") monkeypatch.setattr(provider, "_run", fake_run) - monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) + monkeypatch.setattr(provider.subprocess, "run", _successful_subprocess) monkeypatch.setattr(provider, "_timestamp", kubernetes_publish_time) monkeypatch.setenv("KUBECTL", "kubectl --context test-cluster") payload = { @@ -386,12 +418,8 @@ def fake_run(command: list[str], **kwargs: Any) -> SimpleNamespace: def test_kubernetes_backend_rejects_wrong_ack(provider: ModuleType, monkeypatch: pytest.MonkeyPatch) -> None: """An acknowledgement for another notification cannot produce PASS evidence.""" - monkeypatch.setattr( - provider, - "_run", - lambda *args, **kwargs: SimpleNamespace(stdout='{"delivery_id":"different","status":"delivered"}\n'), - ) - monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=0)) + monkeypatch.setattr(provider, "_run", _wrong_ack) + monkeypatch.setattr(provider.subprocess, "run", _successful_subprocess) with pytest.raises(provider.DeliveryError, match="did not acknowledge"): provider._deliver_kubernetes({"delivery_id": "delivery-6"}) @@ -400,12 +428,8 @@ def test_kubernetes_backend_does_not_pass_when_cleanup_fails( provider: ModuleType, monkeypatch: pytest.MonkeyPatch ) -> None: """A delivered webhook cannot PASS while its ephemeral namespace remains.""" - monkeypatch.setattr( - provider, - "_run", - lambda *args, **kwargs: SimpleNamespace(stdout='{"delivery_id":"delivery-7","status":"delivered"}\n'), - ) - monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=1)) + monkeypatch.setattr(provider, "_run", _delivered_ack) + monkeypatch.setattr(provider.subprocess, "run", _failed_subprocess) with pytest.raises(provider.DeliveryError, match="cleanup failed"): provider._deliver_kubernetes({"delivery_id": "delivery-7"}) @@ -414,11 +438,7 @@ def test_kubernetes_backend_preserves_delivery_and_cleanup_failures( provider: ModuleType, monkeypatch: pytest.MonkeyPatch ) -> None: """Kubernetes diagnostics retain the acknowledgement failure when cleanup also fails.""" - monkeypatch.setattr( - provider, - "_run", - lambda *args, **kwargs: SimpleNamespace(stdout='{"delivery_id":"different","status":"delivered"}\n'), - ) - monkeypatch.setattr(provider.subprocess, "run", lambda *args, **kwargs: SimpleNamespace(returncode=1)) + monkeypatch.setattr(provider, "_run", _wrong_ack) + monkeypatch.setattr(provider.subprocess, "run", _failed_subprocess) with pytest.raises(provider.DeliveryError, match=r"did not acknowledge.*cleanup failed"): provider._deliver_kubernetes({"delivery_id": "delivery-8"})