From c0d5a5af3ddd10903abc4da8a6a28b33b4bb252a Mon Sep 17 00:00:00 2001 From: Petr Date: Tue, 18 Aug 2026 09:24:47 -0400 Subject: [PATCH] feat(notification): audit Flow Notification subscriptions fleet-wide (#600) 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 was auditable in minutes while the recipients that actually page someone when production breaks required opening each flow in the UI by hand. kbagent notification list [--project ALIAS ...] [--event NAME] [--component-id ID] [--config-id ID] [--branch ID] Reads GET /project-subscriptions on the derived notification.{stack} host with a plain Storage token, fans out across every registered project in parallel, and collects per-project failures in `errors` instead of aborting the run. Mirrored as GET /notifications on `kbagent serve`. 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 -- the same guarantee _billing_get gives against a real-money top-up. Three contract details follow the service's own OpenAPI rather than the shapes proposed in the issue: event names are kebab-case (job-failed, not jobFailed) and free-form rather than an enum; subscriptions bind to a flow through dotted filter fields (job.component.id, job.configuration.id, branch.id); and a recipient is a discriminated union where email carries `address` and webhook carries `url`. Two shapes that would otherwise read as broken rows are reported honestly: a subscription with no config filter is the catch-all (scope project-wide), 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, so --branch filters client-side and is never inferred from the project's active branch; inheriting it would silently hide the production recipients the audit exists to check. --- CLAUDE.md | 16 + plugins/kbagent/skills/kbagent/SKILL.md | 1 + .../kbagent/references/commands-reference.md | 3 + .../skills/kbagent/references/gotchas.md | 48 +++ .../references/notification-workflow.md | 81 +++++ src/keboola_agent_cli/changelog.py | 39 +++ src/keboola_agent_cli/cli.py | 5 + src/keboola_agent_cli/client/_client.py | 2 + src/keboola_agent_cli/client/_core.py | 30 ++ src/keboola_agent_cli/client/notifications.py | 52 ++++ src/keboola_agent_cli/commands/context.py | 25 ++ .../commands/notification.py | 167 ++++++++++ src/keboola_agent_cli/permissions.py | 1 + src/keboola_agent_cli/server/app.py | 13 + src/keboola_agent_cli/server/dependencies.py | 3 + .../server/routers/notifications.py | 43 +++ .../services/notification_service.py | 293 ++++++++++++++++++ tests/test_e2e.py | 80 +++++ tests/test_notification_cli.py | 249 +++++++++++++++ tests/test_notification_client.py | 125 ++++++++ tests/test_notification_service.py | 282 +++++++++++++++++ tests/test_server_router_calls.py | 89 ++++++ 22 files changed, 1647 insertions(+) create mode 100644 plugins/kbagent/skills/kbagent/references/notification-workflow.md create mode 100644 src/keboola_agent_cli/client/notifications.py create mode 100644 src/keboola_agent_cli/commands/notification.py create mode 100644 src/keboola_agent_cli/server/routers/notifications.py create mode 100644 src/keboola_agent_cli/services/notification_service.py create mode 100644 tests/test_notification_cli.py create mode 100644 tests/test_notification_client.py create mode 100644 tests/test_notification_service.py 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