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/bare_metal.yaml b/isvctl/configs/providers/aws/config/bare_metal.yaml index 94c5bb374..ee80e4855 100644 --- a/isvctl/configs/providers/aws/config/bare_metal.yaml +++ b/isvctl/configs/providers/aws/config/bare_metal.yaml @@ -265,6 +265,42 @@ 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" + - "--machine-id" + - "{{steps.launch_instance.instance_id}}" + - "--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" + - "--machine-id" + - "{{steps.launch_instance.instance_id}}" + - "--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/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/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..0c1e5018e --- /dev/null +++ b/isvctl/configs/providers/shared/breakfix/query_tenant_notification.py @@ -0,0 +1,468 @@ +#!/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.""" + + +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") + + +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 _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: + # 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 + + 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, + ) + _mark_failure_detected(payload) + 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: + pending_error = sys.exc_info()[1] + 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: + 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]: + """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"]) + _mark_failure_detected(payload) + 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: + pending_error = sys.exc_info()[1] + cleanup_failed = False + 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): + 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: + """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" + 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 = 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") + 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.""" + 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": ( + "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..fab1a0caf 100644 --- a/isvctl/configs/suites/README.md +++ b/isvctl/configs/suites/README.md @@ -269,8 +269,12 @@ 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) | + +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`) 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 new file mode 100644 index 000000000..ae852e762 --- /dev/null +++ b/isvctl/tests/test_notification_delivery_provider.py @@ -0,0 +1,444 @@ +# 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 sys +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 +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" +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: + """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" + + +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}}" + + +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", + [ + [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.""" + + 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, 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) + thread.start() + try: + payload = { + "delivery_id": "delivery-3", + "machine_id": "node-1", + "type": "node_failure", + "message": "Node failed", + "failed_at": "before-webhook-setup", + } + 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) + 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": + 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 _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.""" + + def aws_publish_time(_value: datetime) -> str: + """Return the deterministic AWS publication time.""" + return "aws-publish-time" + + sqs = _FakeSqs() + sns = _FakeSns(sqs) + monkeypatch.setitem(provider.sys.modules, "boto3", _fake_boto3(sns, sqs)) + monkeypatch.setattr(provider, "_timestamp", 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 + + +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 + 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 + + +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]: + """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 + 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 + + +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 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: + return SimpleNamespace(stdout='{"delivery_id":"delivery-5","status":"delivered"}\n') + return SimpleNamespace(stdout="") + + monkeypatch.setattr(provider, "_run", fake_run) + monkeypatch.setattr(provider.subprocess, "run", _successful_subprocess) + monkeypatch.setattr(provider, "_timestamp", kubernetes_publish_time) + monkeypatch.setenv("KUBECTL", "kubectl --context test-cluster") + 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: + """An acknowledgement for another notification cannot produce PASS evidence.""" + 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"}) + + +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", _delivered_ack) + monkeypatch.setattr(provider.subprocess, "run", _failed_subprocess) + 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", _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"}) diff --git a/isvtest/src/isvtest/validations/breakfix.py b/isvtest/src/isvtest/validations/breakfix.py index 99612620e..b18410bda 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,22 @@ 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 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). 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 +434,57 @@ 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 + + +class K8sFailureNotificationCheck(FailureNotificationCheck): + """Validate BFX06-01 through an acknowledged in-cluster delivery.""" + + +_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