diff --git a/CLAUDE.md b/CLAUDE.md index d8cc8ab6..0f204ec2 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -761,6 +761,22 @@ kbagent schedule list [--project NAME ...] [--enabled-only] [--branch ID] kbagent schedule detail --project NAME --schedule-id ID [--branch ID] kbagent schedule find [--cron-window START-END] [--not-run-since DAYS] [--project NAME ...] [--branch ID] +kbagent notification list [--project NAME ...] [--event NAME] [--component-id ID] [--config-id ID] [--branch ID] +# notification (0.84.2+, #600): read-only fleet audit of Flow Notification subscriptions -- the +# Flow Builder Notifications tab (bell icon: Success / Error / Processing-delay cards). Backed by +# the Notification Service (`GET /project-subscriptions` on `notification.{stack}`, plain Storage +# token, no elevated scope). NOT in the flow's `configuration`, so `flow detail` / `config detail` +# cannot show these -- the in-flow `type: "notification"` TASK is a different mechanism and IS +# visible there. Event names are KEBAB-case (`job-failed`, `job-succeeded`, +# `job-succeeded-with-warning`, `job-processing-long`, + `phase-job-*`); `--event` goes to the API, +# `--component-id`/`--config-id`/`--branch` match client-side on the subscription's own filter +# fields (`job.component.id` / `job.configuration.id` / `branch.id`). The endpoint is NOT +# branch-scoped: without `--branch` dev-branch subscriptions come back alongside production, and +# `--branch` is NEVER inferred from the project's active branch (that would hide the production +# recipients an audit exists to check). A subscription with no config filter is the catch-all +# (scope `project-wide`); one pointing at a deleted config keeps its id with an empty +# `config_name` -- a finding, not an error. Create/delete are deliberately NOT exposed. + kbagent context kbagent init [--from-global] [--project ALIAS ...] # `--project ALIAS` (repeatable) copies only the named project(s) from the global config and implies --from-global. diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index a259e152..9d4a2404 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -220,6 +220,7 @@ When working inside a git repository or project directory, run `kbagent init` (o | List cron schedules (keboola.scheduler configs) across projects | `kbagent schedule list` | | Show full detail for a single cron schedule | `kbagent schedule detail --project PROJECT --schedule-id SCHEDULE-ID` | | Audit schedules by cron window or job-freshness | `kbagent schedule find` | +| List Flow Notification subscriptions (the Notifications tab) across projects | `kbagent notification list` | | List development branches from connected projects | `kbagent branch list` | | Create a new development branch and auto-activate it | `kbagent branch create --project PROJECT --name NAME` | | Set an existing development branch as active | `kbagent branch use --project PROJECT --branch BRANCH` | diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 3015c390..4f1262fe 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -105,6 +105,9 @@ The `permissions` subcommands persist a write/destructive policy to config.json ## Billing (PAYG Credits) (since v0.84.2) - `billing credits [--project ALIAS ...]` -- read-only PAYG credit balance (`GET /credits` on `billing.{stack}`, plain Storage token). Fans out across all registered projects in parallel by default; `--project` (repeatable) narrows. Per-project failures degrade individually and are collected in `errors`, never abort the run. A project without the `pay-as-you-go` `owner.features` flag never calls the billing host (NXDOMAIN on some non-PAYG stacks) -- it gets an `error_code: PAYG_NOT_AVAILABLE` entry instead. `--json` emits `{"credits": [...], "errors": [...]}`. Rows carry the API's native unit (`consumed`/`remaining` credits) plus derived `*_minutes` fields (1 credit = 60 minutes, matching the Keboola UI). Gives the current balance only -- purchase history / Stripe invoice IDs are not reachable with a project token (issue #594 primary ask, still open; that data lives on `connection.{stack}` `/pay-as-you-go/billing/*`). See [billing-workflow.md](billing-workflow.md) for the full shape of the invoice-history gap and why it must not be worked around. +## Flow Notifications (since v0.84.2) +- `notification list [--project ALIAS ...] [--event NAME] [--component-id ID] [--config-id ID] [--branch ID]` -- fleet-wide audit of Flow Notification subscriptions, i.e. the Flow Builder **Notifications tab** (bell icon: Success / Error / Processing-delay cards). Backed by the Notification Service (`GET /project-subscriptions` on `notification.{stack}`) with a plain Storage token -- no `canManageTokens` or other elevated scope. **These are NOT in the flow's `configuration`**, so `flow detail` / `config detail` cannot show them; the in-flow `type: "notification"` task is a different mechanism and IS visible there. Multi-project fan-out in parallel by default; per-project failures land in `errors` and never abort the run. `--json` emits `{"subscriptions": [...], "errors": [...]}`; each row has `project_alias`, `subscription_id`, `event`, `scope` (`config` | `project-wide`), `component_id`, `config_id`, `config_name`, `branch_id`, `channel` (`email` | `webhook`), `address` (the email address OR the webhook URL), `expires_at`, and the raw `filters` list. Event names are **kebab-case**: `job-failed`, `job-succeeded`, `job-succeeded-with-warning`, `job-processing-long`, plus `phase-job-*` variants. `--event` is forwarded to the API; `--component-id`, `--config-id` and `--branch` match client-side against the subscription's own filter fields (`job.component.id`, `job.configuration.id`, `branch.id`). Read-only: create/delete subscriptions are deliberately not exposed. See [notification-workflow.md](notification-workflow.md). + ## Feature Flags (since v0.48.0) Requires a **super-admin** Manage API token (same kind as `org setup`). Same default-deny token policy: interactive hidden prompt by default, or `--allow-env-manage-token` + `KBC_MANAGE_API_TOKEN` for CI. `--project ALIAS` resolves the stack URL (and, for project ops, the numeric `project_id`) from config -- the alias is the only handle you pass. - `feature list --project ALIAS` -- the stack-wide feature catalogue (`GET /manage/features`). Returns `{alias, stack_url, features: [{name, title, description, type, ...}]}`. Only `name` is a stable identifier; extra fields pass through unmodified. diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index 84f75c4c..e9aeee36 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -3714,3 +3714,51 @@ mapping in favor of `changed_since: adaptive`, which tracks a assumption: here the empty state is the expensive, surprising path, and a seeded checkpoint is the conservative one. Do not assume "no state = safe default" when adaptive is involved. + +## `notification list`: the Notifications tab is a different mechanism from an in-flow notification task (since v0.84.2) + +`kbagent notification list` closes the one flow-notification surface no CLI +could reach (#600). Everything below is behavior an agent gets wrong by +default. + +- **Two unrelated mechanisms share the word "notification".** The Flow + Builder **Notifications tab** (bell icon: Success / Error / + Processing-delay cards) lives in the Notification Service, NOT in the + flow's `configuration` JSON -- `flow detail` and `config detail` cannot + see it, and never could. The in-flow **notification task** (a phase task + with `type: "notification"` and `recipients: [{channel, address}]`) lives + inside the configuration and IS visible through `flow detail`. Auditing + "who gets paged when this flow breaks" requires BOTH; reporting either one + alone silently under-reports. +- **Event names are kebab-case.** `job-failed`, `job-succeeded`, + `job-succeeded-with-warning`, `job-processing-long`, plus the + `phase-job-*` variants. Not `jobFailed`. The service types `event` as a + free-form string rather than an enum, so kbagent does not restrict it -- + an unknown value produces the API's own 400 rather than a client-side + rejection, and a newly shipped event type works immediately. +- **The filter fields are dotted, not camelCase.** A subscription is bound + to a flow through `filters: [{field: "job.component.id", ...}, + {field: "job.configuration.id", ...}]` -- not `component` / + `configurationId`. `--component-id` and `--config-id` match against those + client-side (only `--event` is served API-side). +- **A subscription with no config filter is the catch-all, not a broken + row.** It fires for every job in the project and is reported as + `scope: "project-wide"` with an empty `config_id`. This is usually the + most important row in an audit -- do not filter it out as noise. +- **A row whose `config_name` is empty but `config_id` is set points at a + deleted configuration.** That is a finding (a subscription paging someone + about a flow that no longer exists), never an error to retry. +- **The endpoint is NOT branch-scoped.** It answers with every branch's + subscriptions at once; a dev-branch one carries a `branch.id` filter, + a production one carries none. `--branch` filters client-side and, unlike + every other branch-aware command, is **never** inferred from the + project's active branch -- inheriting it would hide exactly the + production recipients the audit is checking. A `--branch` value is + meaningful in one project only, so it requires exactly one `--project` + (exit 2 otherwise). +- **Read-only by construction.** The service also exposes create/delete for + subscriptions -- changing who gets paged when production breaks -- and + kbagent deliberately exposes neither. Do not suggest kbagent can add or + remove a recipient; that is still a UI (or direct API) operation. +- **Plain Storage token, no elevated scope.** The read path needs no + `canManageTokens` and no manage token. diff --git a/plugins/kbagent/skills/kbagent/references/notification-workflow.md b/plugins/kbagent/skills/kbagent/references/notification-workflow.md new file mode 100644 index 00000000..a04faea1 --- /dev/null +++ b/plugins/kbagent/skills/kbagent/references/notification-workflow.md @@ -0,0 +1,81 @@ +# Flow Notification audit workflow (`kbagent notification list`) + +Answers the fleet-wide question **"who gets paged when a production flow +breaks, and are those recipients still valid?"** -- across every registered +project, in one command. + +Available since **v0.84.2** (issue #600). Read-only. + +## Why this command exists + +Auditing flow notifications used to be only half-possible from a CLI. Three +of the four surfaces were already reachable: + +| Surface | Where it lives | Reachable before v0.84.2 | +|---|---|---| +| Owner / contact emails in flow descriptions | flow `description` | yes (`flow list`, `config detail`) | +| In-flow notification **task** (`type: "notification"`) | the flow's `configuration` JSON | yes (`flow detail`) | +| Email-sending component configs (e.g. `kds-team.app-email-smtp-sender`) | component configs | yes (`config search`) | +| **Notifications tab** (bell icon: Success / Error / Processing-delay) | **Notification Service** | **no -- UI only** | + +The last row is the one that actually pages a human when a production flow +fails, and it was the only one that required opening each flow in the web UI +by hand. On the 20-project / 276-flow fleet that prompted the issue, that was +the entire cost of the audit. + +## The command + +```bash +# Every subscription, every registered project +kbagent notification list + +# One project, only failures +kbagent notification list --project prod --event job-failed + +# Everything pointed at one specific flow +kbagent notification list --project prod --component-id keboola.flow --config-id 9001 + +# Machine-readable, for joining against your own inventory +kbagent --json notification list > subscriptions.json +``` + +Rows carry `project_alias`, `subscription_id`, `event`, `scope`, +`component_id`, `config_id`, `config_name`, `branch_id`, `channel`, +`address`, `expires_at`, and the raw `filters` list. + +## Reading the output + +- **`scope: "project-wide"`** -- no config filter: the subscription fires for + every job in the project. The catch-all "tell me about any failure". Often + the most important row; never noise. +- **`config_name` empty while `config_id` is set** -- the subscription points + at a configuration that no longer exists. A dangling recipient, i.e. a + finding. +- **`branch_id` set** -- the subscription is filtered to a dev branch. + Production subscriptions carry no branch filter. The endpoint is not + branch-scoped, so both come back together unless you pass `--branch`. +- **`channel: "webhook"`** -- `address` holds the webhook URL rather than an + email address; the two share a column because both answer "where does this + go". + +## A complete audit + +1. `kbagent --json notification list > tab.json` -- the Notifications tab. +2. `kbagent --json flow list` + `flow detail` -- the in-flow notification + **tasks** (a different mechanism; see gotchas.md). +3. `kbagent config search --query "@"` -- addresses hiding in descriptions + and in email-sender component configs. +4. Join all three against your directory of valid addresses. Typical + findings: placeholder addresses that were never replaced, recipients who + have left, flows with no owner at all, and subscriptions surviving the + flow they watched. + +## What this command will not do + +- **It cannot add or remove a recipient.** The Notification Service has + create/delete endpoints; kbagent deliberately wraps neither, at every layer + (the HTTP dispatcher takes no method argument). Changing who gets paged + stays a deliberate UI/API action. +- **It does not read the in-flow notification task.** That is `flow detail`. +- **It needs no elevated token.** The read path works with the same plain + Storage token every other project command uses. diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 427f1e79..f7d9baf9 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -25,6 +25,45 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { "0.84.2": [ + "New: `kbagent notification list` audits Flow Notification subscriptions across " + "the whole fleet (closes #600). `[--project ALIAS ...] [--event NAME] " + "[--component-id ID] [--config-id ID] [--branch ID]`. The Flow Builder " + "**Notifications tab** -- the bell icon's Success / Error / Processing-delay cards -- " + "is backed by the Notification Service, not by the flow's `configuration`, so " + "`flow detail` and `config detail` never could show it. It was the one notification " + "surface with no CLI path at all: on the 20-project / 276-flow fleet that prompted " + "the issue, every other surface (owner emails in descriptions, in-flow " + '`type: "notification"` tasks, email-sender component configs) was auditable in ' + "minutes, while the recipients that actually page someone when production breaks " + "required opening each flow in the UI by hand. Reads `GET /project-subscriptions` on " + "the derived `notification.{stack}` host with a plain Storage token (no elevated " + "scope), fans out across every registered project in parallel, and collects " + "per-project failures in `errors` instead of aborting the run. `--json` emits " + "`{subscriptions: [...], errors: [...]}`; each row carries project_alias, " + "subscription_id, event, scope, component_id, config_id, config_name, branch_id, " + "channel, address, expires_at and the raw filters. Read-only by construction: the " + "service's create/delete endpoints change who gets paged when production breaks, and " + "the HTTP dispatcher takes no method argument, so no future caller can reach them " + "through it. New surfaces: `client/notifications.py`, " + "`services/notification_service.py`, `commands/notification.py`, " + "`GET /notifications` on `kbagent serve`, permission `notification.list = read`.", + "Note: three details of the notification contract are easy to guess wrong. " + "The implementation follows the service's own OpenAPI, not the guess. " + "Event names are KEBAB-case (`job-failed`, `job-succeeded-with-warning`, " + "`job-processing-long`, `phase-job-*`), and `event` is typed as a free-form string " + "rather than an enum -- so no client-side allow-list is imposed and a newly shipped " + "event type works the day Keboola ships it. Subscriptions bind to a flow through " + "dotted filter fields (`job.component.id`, `job.configuration.id`, `branch.id`), not " + "`component`/`configurationId`. A recipient is a discriminated union: `email` carries " + "`address`, `webhook` carries `url` -- both render in one `address` column with " + "`channel` alongside. Two shapes that would otherwise read as broken rows are " + "reported honestly instead: a subscription with no config filter is the catch-all " + "(`scope: project-wide`, empty config_id), and one pointing at a deleted " + "configuration keeps its id with an empty `config_name` -- a finding, not an error. " + "The endpoint is NOT branch-scoped: it answers with every branch's subscriptions, and " + "`--branch` filters client-side and is NEVER inferred from the project's active " + "branch, because inheriting it would silently hide the production recipients the " + "audit exists to check.", "New: `kbagent config clone` duplicates a configuration WHOLE (closes #587). " "`--project P --component-id C --config-id ID --name N [--target-project P2] " "[--set PATH=VALUE ...] [--secret PATH=VALUE ...] [--dry-run]`. Until now there was " diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index c72f43c0..bbc6647c 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -26,6 +26,7 @@ from .commands.job import job_app from .commands.kai import kai_app from .commands.lineage import lineage_app +from .commands.notification import notification_app from .commands.org import org_app from .commands.permissions import permissions_app from .commands.project import project_app @@ -69,6 +70,7 @@ from .services.lineage_service import LineageService from .services.mcp_service import McpService from .services.member_service import MemberService +from .services.notification_service import NotificationService from .services.org_service import OrgService from .services.project_service import ProjectService from .services.repo_validate_service import RepoValidateService @@ -140,6 +142,7 @@ _FLOWS = "Flows" app.add_typer(flow_app, name="flow", rich_help_panel=_FLOWS) app.add_typer(schedule_app, name="schedule", rich_help_panel=_FLOWS) +app.add_typer(notification_app, name="notification", rich_help_panel=_FLOWS) # -- Development -- _DEV = "Development" @@ -346,6 +349,7 @@ def main( encrypt_service = EncryptService(config_store=config_store) flow_service = FlowService(config_store=config_store) schedule_service = ScheduleService(config_store=config_store) + notification_service = NotificationService(config_store=config_store) workspace_service = WorkspaceService(config_store=config_store) data_app_service = DataAppService(config_store=config_store) data_app_git_service = DataAppGitService(config_store=config_store) @@ -406,6 +410,7 @@ def main( ctx.obj["encrypt_service"] = encrypt_service ctx.obj["flow_service"] = flow_service ctx.obj["schedule_service"] = schedule_service + ctx.obj["notification_service"] = notification_service ctx.obj["workspace_service"] = workspace_service ctx.obj["data_app_service"] = data_app_service ctx.obj["data_app_git_service"] = data_app_git_service diff --git a/src/keboola_agent_cli/client/_client.py b/src/keboola_agent_cli/client/_client.py index ce753950..a2d7b87e 100644 --- a/src/keboola_agent_cli/client/_client.py +++ b/src/keboola_agent_cli/client/_client.py @@ -17,6 +17,7 @@ from .branches import _BranchesMixin from .configs import _ConfigsMixin from .misc import _MiscMixin +from .notifications import _NotificationMixin from .query import _QueryMixin from .queue import _QueueMixin from .storage_files import _StorageFilesMixin @@ -37,6 +38,7 @@ class KeboolaClient( _QueryMixin, _WorkspacesMixin, _BillingMixin, + _NotificationMixin, _MiscMixin, _CoreClient, ): diff --git a/src/keboola_agent_cli/client/_core.py b/src/keboola_agent_cli/client/_core.py index 8c38774c..cef38c50 100644 --- a/src/keboola_agent_cli/client/_core.py +++ b/src/keboola_agent_cli/client/_core.py @@ -46,6 +46,7 @@ def __init__(self, stack_url: str, token: str, *, http_auth: httpx.Auth | None = self._encrypt_client: httpx.Client | None = None self._sync_actions_client: httpx.Client | None = None self._billing_client: httpx.Client | None = None + self._notification_client: httpx.Client | None = None # Lazily built on first Data Streams call (per-device OTLP sources); the # Stream control plane is a sibling host reachable from this stack+token. self._stream_client: StreamClient | None = None @@ -76,6 +77,10 @@ def _sync_actions_base_url(self) -> str: def _billing_base_url(self) -> str: return self._derive_service_url(self._stack_url, "billing") + @property + def _notification_base_url(self) -> str: + return self._derive_service_url(self._stack_url, "notification") + def close(self) -> None: """Close the underlying HTTP clients.""" super().close() @@ -89,6 +94,8 @@ def close(self) -> None: self._sync_actions_client.close() if self._billing_client is not None: self._billing_client.close() + if self._notification_client is not None: + self._notification_client.close() if self._stream_client is not None: self._stream_client.close() @@ -188,6 +195,29 @@ def _billing_get(self, path: str, **kwargs: Any) -> httpx.Response: "GET", path, client=client, base_url=self._billing_base_url, **kwargs ) + def _notification_get(self, path: str, **kwargs: Any) -> httpx.Response: + """Execute a read-only Notification API request with retry. + + The notification service is a sibling host derived from the stack URL + (``notification.{stack-suffix}``); the sub-client inherits the main + client's headers, so the ``X-StorageApi-Token`` auth carries over, and + the bearer hook rides along for session-mode projects (the service + accepts a ``kbc_at_*`` bearer paired with ``X-KBC-ProjectId``, which + ``BearerAuth`` already stamps). + + GET-only for the same reason as ``_billing_get``: this service also + exposes ``POST`` / ``DELETE /project-subscriptions``, which add and + remove who gets paged when production breaks. kbagent's notification + surface is read-only by design (issue #600 scopes the write path out), + and hardcoding the verb means a future caller cannot reach the write + path through this dispatcher at all -- a guarantee in the signature + rather than a convention a reviewer has to catch. + """ + client = self._get_or_create_sub_client("_notification_client", self._notification_base_url) + return self._do_request( + "GET", path, client=client, base_url=self._notification_base_url, **kwargs + ) + def _wait_for_storage_job( self, job: dict[str, Any], diff --git a/src/keboola_agent_cli/client/notifications.py b/src/keboola_agent_cli/client/notifications.py new file mode 100644 index 00000000..4a505eef --- /dev/null +++ b/src/keboola_agent_cli/client/notifications.py @@ -0,0 +1,52 @@ +"""Flow Notification subscriptions -- GET /project-subscriptions. + +New for issue #600. These are the per-flow **Notifications tab** recipients +(the bell icon in Flow Builder: Success / Error / Processing-delay cards), a +different mechanism from the in-flow ``type: "notification"`` task -- the +latter lives inside the flow's own ``configuration`` JSON and is already +visible through ``flow detail``, while these live in a separate service and +were previously unreachable from any CLI. + +The service also exposes ``POST`` and ``DELETE /project-subscriptions``. They +are deliberately NOT wrapped: kbagent's notification surface is read-only, +and the GET-only dispatcher in ``_core.py`` makes that structural. +""" + +from typing import Any + +from ._core import _CoreClient + + +class _NotificationMixin(_CoreClient): + """Read-only access to the project's notification subscriptions.""" + + def list_project_subscriptions(self, event: str | None = None) -> list[dict[str, Any]]: + """List every notification subscription for the token's project. + + ``GET /project-subscriptions`` on the ``notification.{stack-suffix}`` + host. The project is resolved server-side from the token, so there is + no project parameter. + + ``event`` is passed through as the ``?event=`` filter. The service + types it as a free-form string (not an enum), so no client-side + allow-list is imposed here -- an unknown value is the server's 400 to + raise, and a new event type works the day Keboola ships it. Known + values are kebab-case: ``job-failed``, ``job-succeeded``, + ``job-succeeded-with-warning``, ``job-processing-long`` and their + ``phase-job-*`` counterparts. + + Returns the raw list verbatim; each item carries ``id``, ``event``, + optional ``expiresAt``, optional ``filters`` (a list of + ``{field, value, operator?}``) and a ``recipient`` whose shape depends + on its ``channel``: ``email`` carries ``address``, ``webhook`` carries + ``url``. Shaping happens in the service layer. + """ + params = {"event": event} if event else None + response = self._notification_get("/project-subscriptions", params=params) + payload = response.json() + # The endpoint is documented to answer with a bare array; tolerate a + # wrapped shape rather than raising a TypeError deep in the service. + if isinstance(payload, dict): + wrapped = payload.get("subscriptions") + return wrapped if isinstance(wrapped, list) else [] + return payload if isinstance(payload, list) else [] diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index a229b3fd..be41d880 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -991,6 +991,31 @@ cells as positive match signals. Queue API is not branch-aware: --branch + --not-run-since still compares against production jobs. +### Flow Notifications (Notifications tab -- since v0.84.2) + + kbagent notification list [--project NAME ...] [--event NAME] [--component-id ID] [--config-id ID] [--branch ID] + Fleet-wide audit of Flow Notification subscriptions -- the Flow Builder + Notifications tab (bell icon: Success / Error / Processing-delay cards). + These live in the notification service (GET /project-subscriptions on + notification.{{stack}}, plain Storage token), NOT in the flow's configuration, + so flow detail / config detail CANNOT show them. The in-flow + type: "notification" TASK is a different mechanism and IS visible via + flow detail -- do not conflate the two. + Rows: project_alias, subscription_id, event, scope, component_id, config_id, + config_name, branch_id, channel (email|webhook), address, expires_at, filters. + Event names are KEBAB-case: job-failed, job-succeeded, + job-succeeded-with-warning, job-processing-long (+ phase-job-* variants). + --event is sent to the API; --component-id/--config-id/--branch match + client-side against the subscription's own filter fields + (job.component.id / job.configuration.id / branch.id). + NOT branch-scoped: without --branch you get dev-branch subscriptions + alongside production, and --branch is NEVER inferred from the project's + active branch (that would hide production recipients from an audit). + A subscription with no config filter is the catch-all -- scope + "project-wide", empty config_id. One pointing at a deleted config keeps + its config_id with an empty config_name (a finding, not an error). + Read-only: create/delete subscriptions are deliberately not exposed. + ### Development Branches kbagent branch list [--project NAME] diff --git a/src/keboola_agent_cli/commands/notification.py b/src/keboola_agent_cli/commands/notification.py new file mode 100644 index 00000000..510a035e --- /dev/null +++ b/src/keboola_agent_cli/commands/notification.py @@ -0,0 +1,167 @@ +"""Flow Notification subscription commands (issue #600). + +Thin CLI layer over :class:`NotificationService`. One subcommand: + +- ``notification list`` -- who gets notified when a job fails, succeeds, or + runs long, across one or more projects, sourced from ``GET + /project-subscriptions`` on the ``notification.`` host. + +This is the Flow Builder **Notifications tab** (the bell icon: Success / +Error / Processing-delay cards). It is a different mechanism from the in-flow +``type: "notification"`` task, which lives inside the flow's own +configuration and is already visible through ``kbagent flow detail``. + +Read-only: safe under ``--deny-writes``. The service's create/delete +endpoints -- which change who gets paged when production breaks -- are +deliberately not exposed. +""" + +from __future__ import annotations + +from typing import Any + +import typer +from rich.markup import escape +from rich.table import Table + +from ..errors import ConfigError, ErrorCode +from ._helpers import check_cli_permission, get_formatter, get_service + +notification_app = typer.Typer( + help="Audit Flow Notification subscriptions across projects (issue #600). " + "Read-only -- the Notifications tab recipients, which flow detail cannot show." +) + + +@notification_app.callback(invoke_without_command=True) +def _notification_permission_check(ctx: typer.Context) -> None: + check_cli_permission(ctx, "notification") + + +def _format_subscription_table(formatter: Any, subscriptions: list[dict[str, Any]]) -> None: + tbl = Table( + "Project", + "Event", + "Flow / scope", + "Component", + "Branch", + "Channel", + "Recipient", + "Expires", + show_header=True, + header_style="bold cyan", + ) + for row in subscriptions: + config_id = row.get("config_id", "") + if config_id: + # A subscription pointing at a deleted config resolves to no name; + # show the bare id rather than an empty cell, since a dangling + # subscription is exactly what an audit is hunting for. + target = escape(row.get("config_name") or config_id) + else: + target = "[dim]project-wide[/dim]" + tbl.add_row( + escape(row.get("project_alias", "")), + escape(row.get("event", "")), + target, + escape(row.get("component_id", "") or "-"), + escape(row.get("branch_id", "") or "production"), + escape(row.get("channel", "")), + escape(row.get("address", "")), + escape(row.get("expires_at", "") or "-"), + ) + formatter.console.print(tbl) + + +def _emit_errors(formatter: Any, errors: list[dict[str, Any]]) -> None: + for err in errors: + formatter.warning( + f"Project '{escape(str(err.get('project_alias', '?')))}': " + f"{escape(str(err.get('message', 'error')))}" + ) + + +@notification_app.command("list") +def notification_list( + ctx: typer.Context, + project: list[str] | None = typer.Option( + None, + "--project", + help="Project alias (repeatable; omit for all registered projects)", + ), + event: str | None = typer.Option( + None, + "--event", + help="Event name, e.g. job-failed, job-succeeded, " + "job-succeeded-with-warning, job-processing-long (also phase-job-*)", + ), + component_id: str | None = typer.Option( + None, "--component-id", help="Only subscriptions filtered to this component" + ), + config_id: str | None = typer.Option( + None, "--config-id", help="Only subscriptions filtered to this configuration" + ), + branch: int | None = typer.Option( + None, "--branch", help="Only subscriptions carrying this branch.id filter" + ), +) -> None: + """List Flow Notification subscriptions (the Notifications tab) across projects. + + Answers "who gets paged when this flow breaks" for a whole fleet at once. + These recipients live in the notification service, not in the flow's + configuration, so `flow detail` / `config detail` cannot show them -- + they are the notification surface that used to require opening each flow + in the UI by hand. + + Each row shows: project alias, event, the flow the subscription is + filtered to (or `project-wide` when it carries no config filter -- the + catch-all "notify me on any job failure"), component, branch, channel + (email or webhook), the recipient address or URL, and expiry. + + `--event` is passed to the API; `--component-id`, `--config-id` and + `--branch` match client-side against the subscription's own filter + fields. NOTE the endpoint is not branch-scoped: it answers with every + branch's subscriptions, so without `--branch` the output includes + dev-branch ones alongside production. Unlike branch-scoped commands, + `--branch` here is NEVER inferred from the project's active branch -- + doing so would silently hide the production recipients this audit + exists to check. + + Read-only. The in-flow `type: "notification"` task is a different + mechanism -- see `kbagent flow detail` for those. + """ + formatter = get_formatter(ctx) + service = get_service(ctx, "notification_service") + + # A branch id is only meaningful inside one project, and it is a filter + # here rather than a scope -- so it is never inferred from the project's + # active branch the way branch-scoped commands do it (see the docstring). + if branch is not None and (not project or len(project) != 1): + formatter.error( + message="--branch requires exactly one --project", + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) from None + + try: + result = service.list_subscriptions( + aliases=project, + event=event, + component_id=component_id, + config_id=config_id, + branch_id=branch, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + + if formatter.json_mode: + formatter.output(result) + return + + subscriptions = result.get("subscriptions", []) + if not subscriptions: + formatter.console.print("[dim]No notification subscriptions found.[/dim]") + else: + _format_subscription_table(formatter, subscriptions) + _emit_errors(formatter, result.get("errors", [])) diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index e98261b8..4dadfff8 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -346,6 +346,7 @@ "schedule.list": "read", "schedule.detail": "read", "schedule.find": "read", + "notification.list": "read", # PAYG credit balance (issue #594) -- read-only, GET /credits only. "billing.credits": "read", # Top-level commands diff --git a/src/keboola_agent_cli/server/app.py b/src/keboola_agent_cli/server/app.py index 4d7e1080..486d0eb0 100644 --- a/src/keboola_agent_cli/server/app.py +++ b/src/keboola_agent_cli/server/app.py @@ -50,6 +50,7 @@ lineage, mcp, members, + notifications, org, projects, schedules, @@ -239,6 +240,17 @@ "Mirrors `kbagent schedule list|detail|find`." ), }, + { + "name": "notifications", + "description": ( + "**Execution.** " + "Flow Notification subscriptions -- the Notifications-tab " + "recipients paged when a job fails, succeeds, or runs long " + "(read-only; a different mechanism from an in-flow notification " + "task, which `flow detail` already shows). " + "Mirrors `kbagent notification list`." + ), + }, { "name": "data-apps", "description": ( @@ -693,6 +705,7 @@ async def _generic_handler(_request, exc: Exception): app.include_router(workspaces.router) app.include_router(flows.router) app.include_router(schedules.router) + app.include_router(notifications.router) app.include_router(lineage.router) app.include_router(sharing.router) app.include_router(data_apps.router) diff --git a/src/keboola_agent_cli/server/dependencies.py b/src/keboola_agent_cli/server/dependencies.py index 4d52a060..4eb4df40 100644 --- a/src/keboola_agent_cli/server/dependencies.py +++ b/src/keboola_agent_cli/server/dependencies.py @@ -33,6 +33,7 @@ from ..services.lineage_service import LineageService from ..services.mcp_service import McpService from ..services.member_service import MemberService +from ..services.notification_service import NotificationService from ..services.org_service import OrgService from ..services.project_service import ProjectService from ..services.repo_validate_service import RepoValidateService @@ -107,6 +108,7 @@ class ServiceRegistry: workspace: WorkspaceService = field(init=False) flow: FlowService = field(init=False) schedule: ScheduleService = field(init=False) + notification: NotificationService = field(init=False) lineage: LineageService = field(init=False) deep_lineage: DeepLineageService = field(init=False) sharing: SharingService = field(init=False) @@ -144,6 +146,7 @@ def __post_init__(self) -> None: self.workspace = WorkspaceService(config_store=cs) self.flow = FlowService(config_store=cs) self.schedule = ScheduleService(config_store=cs) + self.notification = NotificationService(config_store=cs) self.lineage = LineageService(config_store=cs) self.deep_lineage = DeepLineageService(config_store=cs) self.sharing = SharingService(config_store=cs) diff --git a/src/keboola_agent_cli/server/routers/notifications.py b/src/keboola_agent_cli/server/routers/notifications.py new file mode 100644 index 00000000..1618f3cc --- /dev/null +++ b/src/keboola_agent_cli/server/routers/notifications.py @@ -0,0 +1,43 @@ +"""Flow Notification subscription endpoints (issue #600). + +Read-only by design: the upstream notification service on +``notification.{stack}`` exposes ``POST`` / ``DELETE +/project-subscriptions``, which change who gets paged when production +breaks. This router -- and the client/service layers it delegates to -- +only ever issue GET requests. Do not add a write endpoint here. +""" + +from __future__ import annotations + +from typing import Any + +from fastapi import APIRouter, Depends, Query + +from ..dependencies import ServiceRegistry, get_registry + +router = APIRouter(prefix="/notifications", tags=["notifications"]) + + +@router.get("", summary="Flow Notification subscriptions across projects") +def list_subscriptions( + project: list[str] | None = Query(None), + event: str | None = Query(None), + component_id: str | None = Query(None), + config_id: str | None = Query(None), + branch: int | None = Query(None), + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Notification-tab recipients per project. Mirrors `kbagent notification list`. + + `event` is forwarded to the upstream service; `component_id`, + `config_id` and `branch` match client-side against each subscription's + own filter fields. The upstream endpoint is not branch-scoped, so + omitting `branch` returns dev-branch subscriptions alongside production. + """ + return registry.notification.list_subscriptions( + aliases=project, + event=event, + component_id=component_id, + config_id=config_id, + branch_id=branch, + ) diff --git a/src/keboola_agent_cli/services/notification_service.py b/src/keboola_agent_cli/services/notification_service.py new file mode 100644 index 00000000..170296fb --- /dev/null +++ b/src/keboola_agent_cli/services/notification_service.py @@ -0,0 +1,293 @@ +"""Fleet-wide audit of Flow Notification subscriptions (issue #600). + +Wraps ``KeboolaClient.list_project_subscriptions`` (``GET +/project-subscriptions`` on the derived ``notification.{stack}`` host) with +the fan-out / per-project-error shape every other multi-project service uses +(``ScheduleService`` is the closest sibling -- same question shape: "audit a +per-project sibling-service concept across every registered project, joined +against the config it points at"). + +These subscriptions are the Flow Builder **Notifications tab** (the bell +icon: Success / Error / Processing-delay cards), which lives in the +notification service and NOT in the flow's ``configuration`` JSON. The +in-flow ``type: "notification"`` task is a different mechanism entirely and +is already visible through ``flow detail``. + +READ-ONLY BY DESIGN: the service also exposes ``POST`` / ``DELETE +/project-subscriptions``, which change who gets paged when production +breaks. Nothing here writes; the GET-only dispatcher in ``client/_core.py`` +enforces the same restriction one layer down. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from ..errors import KeboolaApiError +from ..models import ProjectConfig +from .base import BaseService, project_error_entry + +logger = logging.getLogger(__name__) + +# Filter fields the notification service matches subscriptions on. Verified +# against the service's own OpenAPI examples -- NOT the camelCase +# `configurationId` / `component` an outside reader would guess. +FILTER_FIELD_COMPONENT_ID = "job.component.id" +FILTER_FIELD_CONFIG_ID = "job.configuration.id" +FILTER_FIELD_BRANCH_ID = "branch.id" + +# A subscription carrying no config filter fires for EVERY job in the +# project. That is a legitimate, and operationally important, state -- it is +# the "page me on any failure" catch-all -- so it must render as its own +# scope rather than as a flow with a missing name. +SCOPE_CONFIG = "config" +SCOPE_PROJECT_WIDE = "project-wide" + + +def _index_filters(filters: Any) -> dict[str, Any]: + """Index a subscription's ``filters`` list by field name. + + The service models filters as ``[{field, value, operator?}]``. Only the + identity fields are indexed by name here; a threshold filter such as + ``durationOvertimePercentage >= 0.75`` is preserved verbatim in the row's + ``filters`` key instead, since collapsing it to a scalar would drop the + operator that gives it meaning. + """ + indexed: dict[str, Any] = {} + if not isinstance(filters, list): + return indexed + for entry in filters: + if not isinstance(entry, dict): + continue + field = entry.get("field") + if isinstance(field, str) and field: + indexed[field] = entry.get("value") + return indexed + + +def _recipient_address(recipient: Any) -> tuple[str, str]: + """Return ``(channel, address)`` for either recipient shape. + + The service discriminates on ``channel``: an ``email`` recipient carries + ``address``, a ``webhook`` recipient carries ``url``. Both are "where the + notification goes", so they share one column -- keeping ``channel`` + alongside means the caller can still tell them apart. + """ + if not isinstance(recipient, dict): + return "", "" + channel = str(recipient.get("channel", "") or "") + address = recipient.get("address") or recipient.get("url") or "" + return channel, str(address) + + +def _shape_subscription(raw: dict[str, Any]) -> dict[str, Any]: + """Project one raw subscription into the CLI-facing row (name unresolved).""" + filters = raw.get("filters") + indexed = _index_filters(filters) + channel, address = _recipient_address(raw.get("recipient")) + + component_id = str(indexed.get(FILTER_FIELD_COMPONENT_ID) or "") + config_id = str(indexed.get(FILTER_FIELD_CONFIG_ID) or "") + branch_filter = indexed.get(FILTER_FIELD_BRANCH_ID) + + return { + "subscription_id": str(raw.get("id", "")), + "event": str(raw.get("event", "")), + "scope": SCOPE_CONFIG if config_id else SCOPE_PROJECT_WIDE, + "component_id": component_id, + "config_id": config_id, + "config_name": "", + "branch_id": str(branch_filter) if branch_filter is not None else "", + "channel": channel, + "address": address, + "expires_at": str(raw.get("expiresAt") or ""), + # Kept verbatim so a threshold filter (durationOvertimePercentage with + # its `>=` operator) or a field this version does not know about is + # still auditable from `--json` output. + "filters": filters if isinstance(filters, list) else [], + } + + +class NotificationService(BaseService): + """Fleet-wide discovery for notification subscriptions. + + Every method is read-only and accumulates per-project errors in the + ``errors`` field of the returned dict rather than aborting the fan-out. + """ + + def list_subscriptions( + self, + aliases: list[str] | None = None, + event: str | None = None, + component_id: str | None = None, + config_id: str | None = None, + branch_id: int | None = None, + ) -> dict[str, Any]: + """List notification subscriptions across one, many, or all projects. + + Args: + aliases: Project aliases to query. ``None`` / empty means every + registered project. + event: Passed to the service as the ``?event=`` filter. Free-form + (the service types it as a string, not an enum); known values + are kebab-case, e.g. ``job-failed``. + component_id: Keep only subscriptions filtered to this component. + config_id: Keep only subscriptions filtered to this configuration. + branch_id: Keep only subscriptions carrying this ``branch.id`` + filter. NOTE this is a client-side filter over the returned + rows, not a scoped request: the list endpoint is not + branch-scoped and answers with every branch's subscriptions. + + Returns: + ``{"subscriptions": [...], "errors": [...]}``. Each row carries + ``project_alias``, ``subscription_id``, ``event``, ``scope``, + ``component_id``, ``config_id``, ``config_name``, ``branch_id``, + ``channel``, ``address``, ``expires_at`` and raw ``filters``. + """ + projects = self.resolve_projects(aliases) + + def worker(alias: str, project: ProjectConfig) -> tuple[Any, ...]: + return self._fetch_project_subscriptions( + alias, + project, + event=event, + component_id=component_id, + config_id=config_id, + branch_id=branch_id, + ) + + successes, errors = self._run_parallel(projects, worker) + + subscriptions: list[dict[str, Any]] = [] + for result in successes: + subscriptions.extend(result[1]) + subscriptions.sort( + key=lambda s: ( + s.get("project_alias", ""), + s.get("component_id", ""), + (s.get("config_name") or "").lower(), + s.get("event", ""), + s.get("address", ""), + ) + ) + errors.sort(key=lambda e: e.get("project_alias", "")) + + return {"subscriptions": subscriptions, "errors": errors} + + # ------------------------------------------------------------------ + + def _fetch_project_subscriptions( + self, + alias: str, + project: ProjectConfig, + event: str | None, + component_id: str | None, + config_id: str | None, + branch_id: int | None, + ) -> tuple[Any, ...]: + """Fetch, filter and name-resolve subscriptions for a single project. + + ``project.active_branch_id`` is deliberately NOT consulted. Every other + branch-aware command treats a branch as a scope and inherits the + project's active one; here it is one filter field on the subscription, + so inheriting it would silently drop every production recipient from an + audit run inside a project that happens to have a dev branch selected. + """ + client = self._client_factory(project.stack_url, project.token) + try: + raw_subscriptions = client.list_project_subscriptions(event=event) + rows = [_shape_subscription(raw) for raw in raw_subscriptions if isinstance(raw, dict)] + rows = _apply_row_filters( + rows, + component_id=component_id, + config_id=config_id, + branch_id=branch_id, + ) + self._resolve_config_names(client, rows) + for row in rows: + row["project_alias"] = alias + return (alias, rows, True) + except KeboolaApiError as exc: + return (alias, project_error_entry(alias, exc)) + except Exception as exc: + # One project's failure must never abort the fan-out. + return (alias, project_error_entry(alias, exc)) + finally: + client.close() + + def _resolve_config_names(self, client: Any, rows: list[dict[str, Any]]) -> None: + """Fill each row's ``config_name`` in place. + + One ``list_component_configs`` call per distinct (branch, component) + pair actually referenced -- in practice one (production + ``keboola.flow``), and none at all for a project whose subscriptions + are all project-wide. The heavier ``list_components_with_configs`` + used by ``ScheduleService`` is deliberately avoided: it downloads + every configuration BODY in the project (megabytes on the 276-flow + fleet this issue came from) to recover a handful of names. + + Grouping by branch matters because a subscription may be filtered to a + dev branch, whose configs are invisible from production -- looking + that name up in the wrong branch would silently report it as deleted. + + Best-effort throughout: a lookup that fails leaves ``config_name`` + empty rather than failing the project. A subscription pointing at a + deleted flow is a real state worth surfacing, not an error -- and it + is exactly the kind of stale recipient an audit is looking for. + """ + wanted = { + (row["branch_id"], row["component_id"]) + for row in rows + if row["component_id"] and row["config_id"] + } + names: dict[tuple[str, str, str], str] = {} + for branch_key, component in sorted(wanted): + try: + configs = client.list_component_configs( + component, branch_id=int(branch_key) if branch_key else None + ) + except (KeboolaApiError, ValueError) as exc: + logger.debug( + "Config name lookup failed for %s (branch %s): %s", + component, + branch_key or "production", + exc, + ) + continue + for cfg in configs: + if isinstance(cfg, dict): + names[(branch_key, component, str(cfg.get("id", "")))] = str( + cfg.get("name", "") or "" + ) + + for row in rows: + row["config_name"] = names.get( + (row["branch_id"], row["component_id"], row["config_id"]), "" + ) + + +def _apply_row_filters( + rows: list[dict[str, Any]], + component_id: str | None, + config_id: str | None, + branch_id: int | None, +) -> list[dict[str, Any]]: + """Apply the client-side row filters. + + The list endpoint takes only ``?event=``; component, config and branch + are matched here against the subscription's own filter fields. + + A ``branch_id`` request keeps ONLY subscriptions carrying that + ``branch.id`` filter. Production subscriptions carry no branch filter at + all, so they are not silently folded into a dev-branch view. + """ + result = rows + if component_id: + result = [r for r in result if r["component_id"] == component_id] + if config_id: + result = [r for r in result if r["config_id"] == config_id] + if branch_id is not None: + wanted = str(branch_id) + result = [r for r in result if r["branch_id"] == wanted] + return result diff --git a/tests/test_e2e.py b/tests/test_e2e.py index ee413f64..17bc089a 100644 --- a/tests/test_e2e.py +++ b/tests/test_e2e.py @@ -13401,3 +13401,83 @@ def test_set_guard_rejects_state_prefix_exit_2(self) -> None: "parameters.foo=1", )["data"] assert data["configuration"]["parameters"]["foo"] == 1 + + +# --------------------------------------------------------------------------- +# Flow Notification subscriptions (issue #600) +# --------------------------------------------------------------------------- + + +@skip_without_credentials +@pytest.mark.e2e +class TestE2ENotificationList: + """End-to-end test for `kbagent notification list`. + + The E2E project carries no notification subscriptions of its own, so the + honest contract to assert here is the envelope and the wiring, not a + populated list: `--json` must return a well-formed + `{"subscriptions": [...], "errors": [...]}`, exit 0, and reach the + derived `notification.{stack}` host with the project's plain Storage + token (no elevated scope needed for the read path). + + Row-shape assertions run only when the project actually has + subscriptions, so the test starts covering the populated path the day a + fixture flow gains a Notifications-tab recipient -- without a rewrite. + """ + + @pytest.fixture(autouse=True) + def setup(self, tmp_path: Path) -> None: + self.token = os.environ[ENV_TOKEN] + raw_url = os.environ.get(ENV_URL, "connection.keboola.com") + self.url = raw_url if raw_url.startswith("https://") else f"https://{raw_url}" + self.alias = f"{RUN_ID}-notification" + self.config_dir = tmp_path / "config" + self.config_dir.mkdir() + + result = _invoke( + self.config_dir, + [ + "--json", + "project", + "add", + "--project", + self.alias, + "--url", + self.url, + "--token", + self.token, + ], + ) + assert result.exit_code == 0, f"project add failed: {result.output}" + + def _run(self, *args: str) -> Any: + return _invoke(self.config_dir, ["--json", *args]) + + def test_notification_list_returns_well_formed_envelope(self) -> None: + result = self._run("notification", "list", "--project", self.alias) + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert isinstance(data["subscriptions"], list) + assert isinstance(data["errors"], list) + + for row in data["subscriptions"]: + assert row["project_alias"] == self.alias + assert row["subscription_id"] + assert row["event"] + # A subscription is either filtered to one config or fires + # project-wide; there is no third state. + assert row["scope"] in ("config", "project-wide") + assert (row["scope"] == "config") == bool(row["config_id"]) + assert row["channel"] in ("email", "webhook") + + def test_event_filter_narrows_without_error(self) -> None: + """`--event` is forwarded to the API; an empty result is still exit 0.""" + result = self._run("notification", "list", "--project", self.alias, "--event", "job-failed") + assert result.exit_code == 0, result.output + data = json.loads(result.output)["data"] + assert all(row["event"] == "job-failed" for row in data["subscriptions"]) + + def test_branch_requires_a_single_project(self) -> None: + """A branch id is meaningless across projects -- usage error, exit 2.""" + result = self._run("notification", "list", "--branch", "1234") + assert result.exit_code == 2, result.output diff --git a/tests/test_notification_cli.py b/tests/test_notification_cli.py new file mode 100644 index 00000000..bb897368 --- /dev/null +++ b/tests/test_notification_cli.py @@ -0,0 +1,249 @@ +"""Tests for `kbagent notification list` via CliRunner (issue #600). + +Mirrors tests/test_billing_cli.py: patch ConfigStore + the service class used +inside `keboola_agent_cli.cli`, invoke through the real Typer app, and assert +on the JSON envelope, human-mode rendering, argument forwarding and exit codes. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock, patch + +from typer.testing import CliRunner + +from keboola_agent_cli.cli import app +from keboola_agent_cli.config_store import ConfigStore +from keboola_agent_cli.errors import ConfigError +from keboola_agent_cli.models import ProjectConfig + +runner = CliRunner() +TEST_TOKEN = "999-token-abc" + + +def _setup_config(config_dir: Path, projects: dict[str, dict] | None = None) -> ConfigStore: + store = ConfigStore(config_dir=config_dir) + if projects: + for alias, info in projects.items(): + store.add_project( + alias, + ProjectConfig( + stack_url=info.get("stack_url", "https://connection.keboola.com"), + token=info.get("token", TEST_TOKEN), + project_name=info.get("project_name", alias), + project_id=info.get("project_id", 1234), + ), + ) + return store + + +def _run(args: list[str], store: ConfigStore, mock_service: MagicMock) -> Any: + with ( + patch("keboola_agent_cli.cli.ConfigStore") as MockStore, + patch("keboola_agent_cli.cli.NotificationService") as MockNS, + ): + MockStore.return_value = store + MockNS.return_value = mock_service + return runner.invoke(app, args) + + +def _row(**overrides: Any) -> dict[str, Any]: + row = { + "project_alias": "prod", + "subscription_id": "101", + "event": "job-failed", + "scope": "config", + "component_id": "keboola.flow", + "config_id": "9001", + "config_name": "Daily ingest", + "branch_id": "", + "channel": "email", + "address": "ops@example.com", + "expires_at": "", + "filters": [], + } + row.update(overrides) + return row + + +class TestNotificationListCli: + def test_json_output_emits_envelope_verbatim(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + envelope = {"subscriptions": [_row()], "errors": []} + service.list_subscriptions.return_value = envelope + + result = _run(["--json", "notification", "list", "--project", "prod"], store, service) + + assert result.exit_code == 0, result.output + assert json.loads(result.output)["data"] == envelope + + def test_human_mode_shows_flow_name_and_recipient(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [_row()], "errors": []} + + result = _run(["notification", "list"], store, service) + + assert result.exit_code == 0, result.output + assert "Daily ingest" in result.output + assert "ops@example.com" in result.output + assert "job-failed" in result.output + + def test_project_wide_subscription_is_labelled_not_blank(self, tmp_path: Path) -> None: + """The catch-all must read as a scope, not as a flow with no name.""" + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = { + "subscriptions": [ + _row(scope="project-wide", component_id="", config_id="", config_name="") + ], + "errors": [], + } + + result = _run(["notification", "list"], store, service) + + assert "project-wide" in result.output + + def test_dangling_subscription_falls_back_to_config_id(self, tmp_path: Path) -> None: + """A subscription pointing at a deleted flow is the audit's whole point.""" + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = { + "subscriptions": [_row(config_name="")], + "errors": [], + } + + result = _run(["notification", "list"], store, service) + + assert "9001" in result.output + + def test_production_subscription_renders_as_production(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [_row()], "errors": []} + + result = _run(["notification", "list"], store, service) + + assert "production" in result.output + + def test_filters_are_forwarded(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"a": {}, "b": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + + result = _run( + [ + "--json", + "notification", + "list", + "--project", + "a", + "--project", + "b", + "--event", + "job-failed", + "--component-id", + "keboola.flow", + "--config-id", + "9001", + ], + store, + service, + ) + + assert result.exit_code == 0, result.output + service.list_subscriptions.assert_called_once_with( + aliases=["a", "b"], + event="job-failed", + component_id="keboola.flow", + config_id="9001", + branch_id=None, + ) + + def test_no_project_flag_forwards_none(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"a": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + + result = _run(["--json", "notification", "list"], store, service) + + assert result.exit_code == 0, result.output + assert service.list_subscriptions.call_args.kwargs["aliases"] is None + + def test_branch_is_passed_through_verbatim(self, tmp_path: Path) -> None: + """Never inferred from the active branch -- only what the caller typed.""" + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + + result = _run( + ["--json", "notification", "list", "--project", "prod", "--branch", "1234"], + store, + service, + ) + + assert result.exit_code == 0, result.output + assert service.list_subscriptions.call_args.kwargs["branch_id"] == 1234 + + def test_branch_without_single_project_is_usage_error(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"a": {}, "b": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + + result = _run(["notification", "list", "--branch", "1234"], store, service) + + assert result.exit_code == 2 + service.list_subscriptions.assert_not_called() + + def test_per_project_errors_surface_as_warnings_exit_0(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"a": {}, "b": {}}) + service = MagicMock() + service.list_subscriptions.return_value = { + "subscriptions": [_row(project_alias="a")], + "errors": [ + { + "project_alias": "b", + "error_code": "INVALID_TOKEN", + "message": "Invalid or expired token", + } + ], + } + + result = _run(["notification", "list"], store, service) + + assert result.exit_code == 0, result.output + assert "b" in result.output + + def test_empty_result_is_stated_not_silent(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + + result = _run(["notification", "list"], store, service) + + assert result.exit_code == 0, result.output + assert "No notification subscriptions found." in result.output + + def test_unknown_alias_exits_5(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.side_effect = ConfigError("Project 'nope' not found") + + result = _run(["notification", "list", "--project", "nope"], store, service) + + assert result.exit_code == 5 + + +class TestNotificationPermissions: + def test_read_only_command_survives_deny_writes(self, tmp_path: Path) -> None: + """An audit must stay available under the write firewall.""" + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + service = MagicMock() + service.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + + result = _run(["--deny-writes", "--json", "notification", "list"], store, service) + + assert result.exit_code == 0, result.output diff --git a/tests/test_notification_client.py b/tests/test_notification_client.py new file mode 100644 index 00000000..41c0d3e7 --- /dev/null +++ b/tests/test_notification_client.py @@ -0,0 +1,125 @@ +"""Client-layer tests for the notification service mixin (issue #600). + +Pins the wire contract: the derived host, the GET-only dispatcher, the +`?event=` passthrough, and tolerance of the payload shapes the endpoint can +answer with. +""" + +from __future__ import annotations + +from keboola_agent_cli.client import KeboolaClient + +STACK_URL = "https://connection.north-europe.azure.keboola.com" +NOTIFICATION_URL = "https://notification.north-europe.azure.keboola.com" +TOKEN = "901-55555-fakeTestTokenDoNotUseXXXXXXXX" + +SUBSCRIPTION = { + "id": "101", + "event": "job-failed", + "filters": [ + {"field": "job.component.id", "value": "keboola.flow"}, + {"field": "job.configuration.id", "value": "9001"}, + ], + "recipient": {"channel": "email", "address": "ops@example.com"}, +} + + +def _client() -> KeboolaClient: + return KeboolaClient(stack_url=STACK_URL, token=TOKEN) + + +class TestNotificationHost: + def test_base_url_is_derived_from_the_stack(self) -> None: + """No hardcoded hostname: connection. -> notification..""" + client = _client() + try: + assert client._notification_base_url == NOTIFICATION_URL + finally: + client.close() + + def test_list_hits_project_subscriptions_with_the_storage_token(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{NOTIFICATION_URL}/project-subscriptions", + json=[SUBSCRIPTION], + ) + + client = _client() + try: + assert client.list_project_subscriptions() == [SUBSCRIPTION] + finally: + client.close() + + request = httpx_mock.get_requests()[0] + assert request.method == "GET" + assert request.headers["X-StorageApi-Token"] == TOKEN + + def test_event_is_sent_as_a_query_param(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{NOTIFICATION_URL}/project-subscriptions?event=job-failed", + json=[SUBSCRIPTION], + ) + + client = _client() + try: + client.list_project_subscriptions(event="job-failed") + finally: + client.close() + + assert httpx_mock.get_requests()[0].url.params["event"] == "job-failed" + + def test_no_event_sends_a_clean_url(self, httpx_mock) -> None: + """An empty `?event=` would be a different request; omit the param.""" + httpx_mock.add_response(url=f"{NOTIFICATION_URL}/project-subscriptions", json=[]) + + client = _client() + try: + client.list_project_subscriptions() + finally: + client.close() + + assert str(httpx_mock.get_requests()[0].url) == ( + f"{NOTIFICATION_URL}/project-subscriptions" + ) + + +class TestPayloadTolerance: + def test_wrapped_payload_is_unwrapped(self, httpx_mock) -> None: + """Documented as a bare array; a wrapped shape must not raise.""" + httpx_mock.add_response( + url=f"{NOTIFICATION_URL}/project-subscriptions", + json={"subscriptions": [SUBSCRIPTION]}, + ) + + client = _client() + try: + assert client.list_project_subscriptions() == [SUBSCRIPTION] + finally: + client.close() + + def test_unexpected_payload_degrades_to_empty(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{NOTIFICATION_URL}/project-subscriptions", json={"unexpected": True} + ) + + client = _client() + try: + assert client.list_project_subscriptions() == [] + finally: + client.close() + + +class TestReadOnlyByConstruction: + def test_dispatcher_exposes_no_verb_parameter(self) -> None: + """The write path must be unreachable through this dispatcher. + + `POST` / `DELETE /project-subscriptions` change who gets paged when + production breaks. `_notification_get` hardcodes the verb, so no + future caller can construct such a request through it -- the same + guarantee `_billing_get` gives against a real-money top-up. + """ + import inspect + + from keboola_agent_cli.client._core import _CoreClient + + params = inspect.signature(_CoreClient._notification_get).parameters + assert "method" not in params diff --git a/tests/test_notification_service.py b/tests/test_notification_service.py new file mode 100644 index 00000000..6918792c --- /dev/null +++ b/tests/test_notification_service.py @@ -0,0 +1,282 @@ +"""Unit tests for NotificationService (issue #600). + +Tests the business logic in isolation using mocked KeboolaClient instances. +Payload shapes below are taken verbatim from the notification service's own +OpenAPI examples -- kebab-case event names, `job.component.id` / +`job.configuration.id` / `branch.id` filter fields, and the two recipient +shapes (email carries `address`, webhook carries `url`). +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +import pytest + +from keboola_agent_cli.errors import ConfigError, ErrorCode, KeboolaApiError +from keboola_agent_cli.services.notification_service import NotificationService + +_TOKEN_A = "901-storage-aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +_TOKEN_B = "901-storage-bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + +FLOW_COMPONENT = "keboola.flow" + +# One subscription per shape the service can answer with. +SUB_FLOW_FAILED: dict[str, Any] = { + "id": "101", + "event": "job-failed", + "filters": [ + {"field": "job.component.id", "value": FLOW_COMPONENT}, + {"field": "job.configuration.id", "value": "9001"}, + ], + "recipient": {"channel": "email", "address": "ops@example.com"}, +} +SUB_PROJECT_WIDE: dict[str, Any] = { + "id": "102", + "event": "job-failed", + "recipient": {"channel": "email", "address": "catchall@example.com"}, +} +SUB_WEBHOOK_LONG: dict[str, Any] = { + "id": "103", + "event": "job-processing-long", + "filters": [ + {"field": "job.component.id", "value": FLOW_COMPONENT}, + {"field": "job.configuration.id", "value": "9002"}, + {"field": "durationOvertimePercentage", "operator": ">=", "value": 0.75}, + ], + "recipient": {"channel": "webhook", "url": "https://hooks.example.com/kbc"}, + "expiresAt": "2026-01-07T14:00:00+01:00", +} +SUB_BRANCHED: dict[str, Any] = { + "id": "104", + "event": "job-failed", + "filters": [ + {"field": "branch.id", "value": "1234"}, + {"field": "job.component.id", "value": FLOW_COMPONENT}, + {"field": "job.configuration.id", "value": "9001"}, + ], + "recipient": {"channel": "email", "address": "dev@example.com"}, +} + +FLOW_CONFIGS = [ + {"id": "9001", "name": "Daily ingest"}, + {"id": "9002", "name": "Nightly rebuild"}, +] + + +def _mock_config_store(projects: dict) -> MagicMock: + cs = MagicMock() + config = MagicMock() + config.projects = { + alias: MagicMock( + stack_url=v["url"], + token=v["token"], + active_branch_id=v.get("active_branch_id"), + project_id=v.get("project_id"), + ) + for alias, v in projects.items() + } + config.max_parallel_workers = 10 + cs.load.return_value = config + cs.get_project.side_effect = lambda alias: config.projects.get(alias) + return cs + + +def _make_service(client_or_map: Any, projects: dict | None = None) -> NotificationService: + if projects is None: + projects = { + "prod": {"url": "https://connection.keboola.com", "token": _TOKEN_A}, + } + cs = _mock_config_store(projects) + if isinstance(client_or_map, dict): + + def factory(url: str, token: str) -> MagicMock: + return client_or_map[token] + else: + + def factory(url: str, token: str) -> MagicMock: + return client_or_map + + return NotificationService(config_store=cs, client_factory=factory) + + +def _client(subscriptions: list[dict[str, Any]], configs: Any = None) -> MagicMock: + client = MagicMock() + client.list_project_subscriptions.return_value = subscriptions + client.list_component_configs.return_value = FLOW_CONFIGS if configs is None else configs + return client + + +class TestShaping: + def test_config_scoped_row_is_fully_resolved(self) -> None: + service = _make_service(_client([SUB_FLOW_FAILED])) + result = service.list_subscriptions(aliases=["prod"]) + + assert result["errors"] == [] + (row,) = result["subscriptions"] + assert row["project_alias"] == "prod" + assert row["subscription_id"] == "101" + assert row["event"] == "job-failed" + assert row["scope"] == "config" + assert row["component_id"] == FLOW_COMPONENT + assert row["config_id"] == "9001" + assert row["config_name"] == "Daily ingest" + assert row["channel"] == "email" + assert row["address"] == "ops@example.com" + assert row["branch_id"] == "" + assert row["expires_at"] == "" + + def test_subscription_without_filters_is_project_wide(self) -> None: + """The catch-all 'page me on any failure' must not read as a broken flow row.""" + client = _client([SUB_PROJECT_WIDE]) + service = _make_service(client) + result = service.list_subscriptions(aliases=["prod"]) + + (row,) = result["subscriptions"] + assert row["scope"] == "project-wide" + assert row["config_id"] == "" + assert row["config_name"] == "" + assert row["address"] == "catchall@example.com" + # No config to name -> no reason to pay for a config listing at all. + client.list_component_configs.assert_not_called() + + def test_webhook_recipient_uses_url_field(self) -> None: + """email carries `address`, webhook carries `url` -- both are the recipient.""" + service = _make_service(_client([SUB_WEBHOOK_LONG])) + (row,) = service.list_subscriptions(aliases=["prod"])["subscriptions"] + + assert row["channel"] == "webhook" + assert row["address"] == "https://hooks.example.com/kbc" + assert row["expires_at"] == "2026-01-07T14:00:00+01:00" + + def test_threshold_filter_survives_verbatim(self) -> None: + """A `>=` filter loses its meaning if collapsed to a scalar -- keep it raw.""" + service = _make_service(_client([SUB_WEBHOOK_LONG])) + (row,) = service.list_subscriptions(aliases=["prod"])["subscriptions"] + + assert { + "field": "durationOvertimePercentage", + "operator": ">=", + "value": 0.75, + } in row["filters"] + + def test_branch_filter_is_surfaced(self) -> None: + service = _make_service(_client([SUB_BRANCHED])) + (row,) = service.list_subscriptions(aliases=["prod"])["subscriptions"] + + assert row["branch_id"] == "1234" + + def test_dangling_subscription_keeps_empty_name(self) -> None: + """A subscription pointing at a deleted flow is a finding, not an error.""" + service = _make_service(_client([SUB_FLOW_FAILED], configs=[])) + (row,) = service.list_subscriptions(aliases=["prod"])["subscriptions"] + + assert row["config_id"] == "9001" + assert row["config_name"] == "" + + +class TestFilters: + def test_event_is_forwarded_to_the_api(self) -> None: + client = _client([SUB_FLOW_FAILED]) + service = _make_service(client) + service.list_subscriptions(aliases=["prod"], event="job-failed") + + client.list_project_subscriptions.assert_called_once_with(event="job-failed") + + def test_config_id_filters_client_side(self) -> None: + service = _make_service(_client([SUB_FLOW_FAILED, SUB_WEBHOOK_LONG])) + result = service.list_subscriptions(aliases=["prod"], config_id="9002") + + assert [r["subscription_id"] for r in result["subscriptions"]] == ["103"] + + def test_component_id_filters_client_side(self) -> None: + service = _make_service(_client([SUB_FLOW_FAILED, SUB_PROJECT_WIDE])) + result = service.list_subscriptions(aliases=["prod"], component_id="keboola.orchestrator") + + assert result["subscriptions"] == [] + + def test_branch_filter_keeps_only_that_branch(self) -> None: + service = _make_service(_client([SUB_FLOW_FAILED, SUB_BRANCHED])) + result = service.list_subscriptions(aliases=["prod"], branch_id=1234) + + assert [r["subscription_id"] for r in result["subscriptions"]] == ["104"] + + def test_active_branch_never_narrows_the_audit(self) -> None: + """Production recipients must stay visible on a project with an active branch. + + Branch is a filter field here, not a scope -- inheriting the project's + active branch would silently hide exactly what the audit looks for. + """ + projects = { + "prod": { + "url": "https://connection.keboola.com", + "token": _TOKEN_A, + "active_branch_id": 1234, + } + } + service = _make_service(_client([SUB_FLOW_FAILED, SUB_BRANCHED]), projects) + result = service.list_subscriptions(aliases=["prod"]) + + assert {r["subscription_id"] for r in result["subscriptions"]} == {"101", "104"} + + +class TestFanOut: + def test_per_project_error_does_not_abort_the_run(self) -> None: + good = _client([SUB_FLOW_FAILED]) + bad = MagicMock() + bad.list_project_subscriptions.side_effect = KeboolaApiError( + message="Invalid or expired token", + status_code=401, + error_code=ErrorCode.INVALID_TOKEN, + ) + projects = { + "prod": {"url": "https://connection.keboola.com", "token": _TOKEN_A}, + "dev": {"url": "https://connection.keboola.com", "token": _TOKEN_B}, + } + service = _make_service({_TOKEN_A: good, _TOKEN_B: bad}, projects) + result = service.list_subscriptions() + + assert [r["project_alias"] for r in result["subscriptions"]] == ["prod"] + assert len(result["errors"]) == 1 + assert result["errors"][0]["project_alias"] == "dev" + assert result["errors"][0]["error_code"] == ErrorCode.INVALID_TOKEN + + def test_clients_are_always_closed(self) -> None: + client = _client([SUB_FLOW_FAILED]) + service = _make_service(client) + service.list_subscriptions(aliases=["prod"]) + + client.close.assert_called_once() + + def test_unknown_alias_raises_config_error(self) -> None: + service = _make_service(_client([])) + with pytest.raises(ConfigError): + service.list_subscriptions(aliases=["nope"]) + + def test_config_names_cost_one_call_per_branch_and_component(self) -> None: + """Name resolution is O(branch x component), never N+1 over subscriptions.""" + client = _client([SUB_FLOW_FAILED, SUB_WEBHOOK_LONG, SUB_BRANCHED]) + service = _make_service(client) + service.list_subscriptions(aliases=["prod"]) + + # Two production flows share one call; the dev-branch one needs its own, + # because a branch config is invisible from production. + assert client.list_component_configs.call_count == 2 + assert { + call.kwargs["branch_id"] for call in client.list_component_configs.call_args_list + } == {None, 1234} + # The heavy whole-project fetch is what this deliberately avoids. + client.list_components_with_configs.assert_not_called() + + def test_branch_config_name_is_looked_up_in_that_branch(self) -> None: + """A dev-branch subscription must not be reported as pointing at a deleted flow.""" + client = MagicMock() + client.list_project_subscriptions.return_value = [SUB_BRANCHED] + client.list_component_configs.side_effect = lambda component, branch_id=None: ( + [{"id": "9001", "name": "Daily ingest (branch)"}] if branch_id == 1234 else [] + ) + service = _make_service(client) + (row,) = service.list_subscriptions(aliases=["prod"])["subscriptions"] + + assert row["config_name"] == "Daily ingest (branch)" diff --git a/tests/test_server_router_calls.py b/tests/test_server_router_calls.py index a0db874f..8934e97a 100644 --- a/tests/test_server_router_calls.py +++ b/tests/test_server_router_calls.py @@ -1949,3 +1949,92 @@ def test_config_state_set_forwards_row_id_branch_id_and_dry_run(tmp_path: Path) branch_id=456, dry_run=True, ) + + +# --------------------------------------------------------------------------- +# notifications.py GET /notifications +# Service: notification.list_subscriptions(...) (mirrors `kbagent notification list`) +# --------------------------------------------------------------------------- + + +def test_notifications_list_defaults_to_every_project(tmp_path: Path) -> None: + """GET /notifications with no query params fans out over all projects.""" + notification_svc = MagicMock() + notification_svc.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + registry = _mock_registry(notification=notification_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.get("/notifications", headers=AUTH) + + assert res.status_code == 200, res.text + notification_svc.list_subscriptions.assert_called_once_with( + aliases=None, + event=None, + component_id=None, + config_id=None, + branch_id=None, + ) + + +def test_notifications_list_forwards_every_filter(tmp_path: Path) -> None: + """Each query param maps onto the service kwarg of the same meaning.""" + notification_svc = MagicMock() + notification_svc.list_subscriptions.return_value = {"subscriptions": [], "errors": []} + registry = _mock_registry(notification=notification_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.get( + "/notifications", + headers=AUTH, + params={ + "project": ["a", "b"], + "event": "job-failed", + "component_id": "keboola.flow", + "config_id": "9001", + "branch": 1234, + }, + ) + + assert res.status_code == 200, res.text + notification_svc.list_subscriptions.assert_called_once_with( + aliases=["a", "b"], + event="job-failed", + component_id="keboola.flow", + config_id="9001", + branch_id=1234, + ) + + +def test_notifications_list_returns_service_envelope_unchanged(tmp_path: Path) -> None: + """The router must return the service's envelope verbatim.""" + notification_svc = MagicMock() + envelope = { + "subscriptions": [ + { + "project_alias": "prod", + "subscription_id": "101", + "event": "job-failed", + "scope": "config", + "component_id": "keboola.flow", + "config_id": "9001", + "config_name": "Daily ingest", + "branch_id": "", + "channel": "email", + "address": "ops@example.com", + "expires_at": "", + "filters": [], + } + ], + "errors": [], + } + notification_svc.list_subscriptions.return_value = envelope + registry = _mock_registry(notification=notification_svc) + app = _make_app_with_registry(tmp_path, registry) + + with TestClient(app) as client: + res = client.get("/notifications", headers=AUTH) + + assert res.status_code == 200, res.text + assert res.json() == envelope