diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 47d2dc2b..ed1ee38e 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -10,7 +10,7 @@ "plugins": [ { "name": "kbagent", - "version": "0.66.0", + "version": "0.66.1", "source": "./plugins/kbagent", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "category": "development" diff --git a/CLAUDE.md b/CLAUDE.md index 72b200fd..274ca0f7 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -593,6 +593,9 @@ kbagent flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes] # flow validate: with --project fetches the live schema (full validation; fetch failure -> # semantic-only + note); without --project runs semantic-only + a note. flow schema --full # requires --project (fetches live schema); plain flow schema is the offline YAML template. +# flow schedule (0.66.1+) also activates the config on the Scheduler Service so the cron fires; +# activation failure keeps the config written, sets activated=false + warning, exit stays 0. +# flow schedule-remove deregisters from the service before deleting each config. # Execute a flow with: kbagent job run --project NAME --component-id keboola.flow --config-id ID kbagent schedule list [--project NAME ...] [--enabled-only] [--branch ID] diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index 45549029..c63d6a31 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.66.0", + "version": "0.66.1", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "author": { "name": "Keboola", diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md index 817ee638..1223ef10 100644 --- a/plugins/kbagent/agents/keboola-expert.md +++ b/plugins/kbagent/agents/keboola-expert.md @@ -181,6 +181,13 @@ read it when a trigger fires. Each `(X.Y.Z+)` tag is the version floor. phases+tasks -- fetch `flow detail` first, merge locally, run `flow validate --file @merged.yaml --project ALIAS` (full schema) until clean, then push. +- **`flow schedule` activates on the Scheduler Service (0.66.1+)**: older + versions only wrote the `keboola.scheduler` config -- the schedule showed + `enabled` but the cron NEVER fired. On 0.66.1+ the command also registers it + with the service; check `activated: true` in the result. `activated: false` + + warning = config written but dormant (token lacks the privilege) -- re-run + with an admin token. **VERSION GATE**: schedules created by < 0.66.1 stay + dormant until `flow schedule` is re-run on 0.66.1+. - **Snowflake transformation scaffolding**: MCP `create_config` REFUSES `keboola.snowflake-transformation`. Use `config new --push --no-files` (0.33.0+) or `config new --output-dir` then `config update`, or MCP diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 35ade439..b1c1b908 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -225,8 +225,8 @@ Requires the project to be added with its **master ('owner') Storage API token** - `flow new --project NAME --name NAME [--description D] [--file @path.yaml|-|JSON] [--branch ID]` -- create a conditional flow; validated against the **live** CF schema fetched from the stack before the API call (`INVALID_FLOW_DEFINITION` on failure). A schema-fetch failure does NOT block the write: structural check skipped, semantic checks still run, a `structural schema validation skipped` warning is surfaced - `flow update --project NAME --flow-id ID [--name N] [--description D] [--file @path.yaml|-|JSON] [--branch ID]` -- update name, description, or phases/tasks; `--file` is a full-replace of phases+tasks; merge-aware validation against the live CF schema (same graceful semantic-only degradation on fetch failure); requires at least one of --name/--description/--file - `flow delete --project NAME --flow-id ID [--branch ID] [--yes]` -- delete a flow config (confirmation guard) -- `flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--timezone TZ] [--disabled] [--branch ID]` -- attach a cron schedule (stored as keboola.scheduler config, target.componentId=keboola.flow); replaces any existing schedule -- `flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes]` -- remove all cron schedules attached to a flow; idempotent +- `flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--timezone TZ] [--disabled] [--branch ID]` -- attach a cron schedule (stored as keboola.scheduler config, target.componentId=keboola.flow); replaces any existing schedule. Since v0.66.1 the config is also activated on the Scheduler Service so the trigger fires; an activation failure keeps the config written and surfaces a warning (`activated: false`, exit 0) +- `flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes]` -- remove all cron schedules attached to a flow (since v0.66.1 also deregisters them from the Scheduler Service); idempotent ## Schedule Discovery & Audit (Fleet-Wide) - `schedule list [--project NAME ...] [--enabled-only] [--branch ID]` -- fleet-wide list of every `keboola.scheduler` config across one, many, or all projects (parallel fan-out, no --project = all). Each row has `project_alias`, `schedule_id`, `schedule_name`, `parent_component_id`, `parent_config_id`, `parent_name`, `cron`, `timezone`, `enabled`. Answers "which configs are running on cron triggers across N projects?" without enumerating flows diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index b19d7a7f..76c2bf83 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -2284,6 +2284,21 @@ The trade-off is deliberate: one big call avoids the O(unique-parents) round-tri scheduler configs that target the flow. Pair it with `--dry-run` to see the affected configs (cron + timezone) without calling `delete_config`. +## Flow: `schedule` activates on the Scheduler Service (since v0.66.1) + +- `kbagent flow schedule` registers the `keboola.scheduler` config with the + **Scheduler Service** (`POST /schedules`) after writing it — writing the + Storage config alone leaves the cron trigger dormant. Older versions only + wrote the config: a schedule they created shows `state: enabled` but never + fires until `flow schedule` is re-run on v0.66.1+ (or the schedule is + re-saved in the UI). +- Activation failure is **non-fatal**: the config stays written, the result + carries `activated: false` + a warning, and the exit code stays 0. Typical + cause is a Storage token without the schedule-management privilege — re-run + with an admin token to activate. +- `flow schedule-remove` (v0.66.1+) deregisters each schedule from the + Scheduler Service before deleting its config, so removal stops the trigger. + ## `search` is a top-level command, not `config search` (since v0.30.0) `kbagent search QUERY` searches across **all item types** (tables, buckets, configs, flows, data apps, transformations) via the Storage API global-search endpoint. It is distinct from `kbagent config search --query Q` which scans only configuration JSON bodies. diff --git a/pyproject.toml b/pyproject.toml index 5117765e..0b0b0940 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-cli" -version = "0.66.0" +version = "0.66.1" description = "AI-friendly CLI for managing Keboola projects" readme = "README.md" requires-python = ">=3.12" diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index 9f14f21b..65757a36 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -24,6 +24,20 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.66.1": [ + "Fix (#479): `flow schedule` now activates the schedule on the Scheduler Service, so the cron " + "trigger actually fires. Previously the command only wrote the `keboola.scheduler` Storage " + "config; the schedule looked `enabled` but never ran until re-saved in the UI. The command " + "now calls `POST /schedules` on the Scheduler Service after the config upsert (also for " + "`--disabled`, which deregisters the trigger). An activation failure -- e.g. a token without " + "the schedule-management privilege -- keeps the config written, reports `activated: false` + " + "a warning, and exits 0. Schedules created by older kbagent versions stay dormant until " + "`flow schedule` is re-run on 0.66.1+.", + "Fix (#479): `flow schedule-remove` now deregisters each schedule from the Scheduler Service " + "(`DELETE /configurations/{id}`) before deleting its Storage config, so removed schedules " + "stop firing. Deregistration failures other than 404 are surfaced as warnings and do not " + "block the config deletion.", + ], "0.66.0": [ "New: device-enrollment primitives on the importable library -- a hosted Data App can now mint " "per-device credentials in-process (no CLI subprocess, no master token on the device). " diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index 93cd0238..95664ca7 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -625,11 +625,14 @@ kbagent flow schedule --project NAME --flow-id ID --cron "0 6 * * *" [--timezone TZ] [--enabled/--disabled] [--name NAME] [--branch ID] Upsert a cron schedule: updates the existing keboola.scheduler config if one exists, creates one otherwise. Calling twice with a new cron replaces the old schedule — no duplicates created. - Schedules are stored as Storage API configs, not a separate scheduler service. + The config is then activated on the Scheduler Service so the cron trigger fires; an activation + failure (e.g. token cannot manage schedules) keeps the config written, sets activated=false, and + surfaces a warning (exit stays 0). Re-run with a capable token to activate. kbagent flow schedule-remove --project NAME --flow-id ID [--branch ID] [--yes] - Remove all schedules bound to this flow (deletes all matching keboola.scheduler configs). - Idempotent: safe to run when no schedules exist. + Remove all schedules bound to this flow: each schedule is deregistered from the Scheduler + Service, then its keboola.scheduler config is deleted. Idempotent: safe to run when no + schedules exist. ### Schedule Discovery & Audit (Fleet-Wide) diff --git a/src/keboola_agent_cli/commands/flow.py b/src/keboola_agent_cli/commands/flow.py index 052d5a34..8cc53a77 100644 --- a/src/keboola_agent_cli/commands/flow.py +++ b/src/keboola_agent_cli/commands/flow.py @@ -791,6 +791,10 @@ def flow_schedule( created. If one already exists it is updated in-place — calling this command a second time will not create duplicates. + The schedule is also activated on the Scheduler Service so the cron + trigger actually fires. If activation fails (e.g. the token cannot + manage schedules), the config stays written and a warning is shown. + \b Examples: # Run daily at 6am UTC @@ -824,11 +828,18 @@ def flow_schedule( else: state_label = "[green]enabled[/green]" if enabled else "[yellow]disabled[/yellow]" action = result.get("status", "created") - formatter.success(f"Schedule {action}: {escape(cron)} ({escape(timezone)}) — {state_label}") + activation = "" + if result.get("activated"): + activation = " and activated" if enabled else " and deactivated" + formatter.success( + f"Schedule {action}{activation}: {escape(cron)} ({escape(timezone)}) — {state_label}" + ) formatter.console.print( f" Scheduler config: {escape(result.get('schedule_name', ''))} " f"[dim](ID: {escape(result.get('schedule_id', ''))})[/dim]" ) + for warning in result.get("warnings", []): + formatter.warning(warning) # --------------------------------------------------------------------------- @@ -860,6 +871,9 @@ def flow_schedule_remove( ) -> None: """Remove all schedules bound to a flow (deletes keboola.scheduler configs). + Each schedule is deregistered from the Scheduler Service before its + config is deleted, so the cron trigger stops firing. + Idempotent: safe to run even if no schedules exist. """ formatter = get_formatter(ctx) @@ -946,3 +960,5 @@ def flow_schedule_remove( formatter.console.print("[dim]No schedules found — nothing removed.[/dim]") else: formatter.success(f"Removed {count} schedule(s) from flow {escape(flow_id)}") + for warning in result.get("warnings", []): + formatter.warning(warning) diff --git a/src/keboola_agent_cli/scheduler_client.py b/src/keboola_agent_cli/scheduler_client.py new file mode 100644 index 00000000..6f87c56d --- /dev/null +++ b/src/keboola_agent_cli/scheduler_client.py @@ -0,0 +1,86 @@ +"""Keboola Scheduler Service API client with retry, timeouts, and token masking. + +This module communicates with the Keboola Scheduler Service, which registers +cron triggers for ``keboola.scheduler`` Storage configurations. Writing the +Storage config alone is not enough -- the service must be told to (re)load it +via ``POST /schedules`` before the cron trigger fires. Derives the Scheduler +Service URL from the Storage API stack URL by replacing 'connection.' with +'scheduler.' in the hostname. + +Inherits shared retry/error logic from BaseHttpClient. +""" + +import logging +from typing import Any +from urllib.parse import quote + +from .http_base import BaseHttpClient + +logger = logging.getLogger(__name__) + + +class SchedulerClient(BaseHttpClient): + """HTTP client for the Keboola Scheduler Service API. + + Provides schedule activation (register/refresh a ``keboola.scheduler`` + config with the service) and removal of the service-side registration, + with built-in retry logic (exponential backoff for 429/5xx), timeouts, + and automatic token masking in error messages. + + Inherits _do_request() and _raise_api_error() from BaseHttpClient. + """ + + def __init__(self, stack_url: str, token: str) -> None: + self._stack_url = stack_url.rstrip("/") + scheduler_base_url = self._derive_service_url(self._stack_url, "scheduler") + headers = { + "X-StorageApi-Token": token, + } + super().__init__( + base_url=scheduler_base_url, + token=token, + headers=headers, + ) + + def __enter__(self) -> "SchedulerClient": + return self + + def __exit__(self, *args: Any) -> None: + self.close() + + def activate_schedule(self, configuration_id: str) -> dict[str, Any]: + """Register (or refresh) a schedule with the Scheduler Service. + + The service loads the referenced config and applies its current + state; the operation is idempotent. + + Args: + configuration_id: ID of the ``keboola.scheduler`` Storage + configuration (not the target flow ID). + + Returns: + Dict with the registered schedule as returned by the service. + + Raises: + KeboolaApiError: On API errors (403 when the token lacks the + schedule-activation privilege, etc.). + """ + payload = {"configurationId": configuration_id} + response = self._do_request("POST", "/schedules", json=payload) + return response.json() + + def remove_schedule(self, configuration_id: str) -> None: + """Deregister a schedule from the Scheduler Service. + + The Storage configuration itself is untouched. + + Args: + configuration_id: ID of the ``keboola.scheduler`` Storage + configuration. + + Raises: + KeboolaApiError: On API errors (404 when the service has no + registration for this configuration, etc.). + """ + encoded_id = quote(configuration_id, safe="") + self._do_request("DELETE", f"/configurations/{encoded_id}") diff --git a/src/keboola_agent_cli/services/flow_service.py b/src/keboola_agent_cli/services/flow_service.py index d6354551..2ab5152b 100644 --- a/src/keboola_agent_cli/services/flow_service.py +++ b/src/keboola_agent_cli/services/flow_service.py @@ -9,7 +9,9 @@ Flows are semantic sugar over the Storage API config layer -- no separate HTTP client is needed. Schedules are stored as keboola.scheduler configs -whose ``target`` points at the flow. +whose ``target`` points at the flow, and are additionally registered with +the Scheduler Service (via ``SchedulerClient``) -- the Storage config alone +does not make the cron trigger fire. """ from __future__ import annotations @@ -24,6 +26,7 @@ from ..config_store import ConfigStore from ..errors import ErrorCode, KeboolaApiError from ..models import ComponentDetail, ProjectConfig +from ..scheduler_client import SchedulerClient from .base import BaseService, ClientFactory from .flow_validation import find_unreachable_phases, validate_conditional_flow @@ -34,6 +37,7 @@ SCHEDULER_COMPONENT_ID = "keboola.scheduler" AiClientFactory = Callable[[str, str], AiServiceClient] +SchedulerClientFactory = Callable[[str, str], SchedulerClient] @dataclass(frozen=True) @@ -55,6 +59,11 @@ def default_ai_client_factory(stack_url: str, token: str) -> AiServiceClient: return AiServiceClient(stack_url=stack_url, token=token) +def default_scheduler_client_factory(stack_url: str, token: str) -> SchedulerClient: + """Default factory: build a ``SchedulerClient`` for the given project.""" + return SchedulerClient(stack_url=stack_url, token=token) + + # --------------------------------------------------------------------------- # Internal helpers # --------------------------------------------------------------------------- @@ -70,6 +79,21 @@ def _parse_configuration(raw: Any) -> dict[str, Any]: return raw or {} +def _schedules_targeting_flow( + all_sched: list[dict[str, Any]], config_id: str +) -> list[dict[str, Any]]: + """Filter keboola.scheduler configs whose target is this keboola.flow config.""" + matching: list[dict[str, Any]] = [] + for sched in all_sched: + body = _parse_configuration(sched.get("configuration")) + target = body.get("target") or {} + if target.get("componentId") == FLOW_COMPONENT_ID and str( + target.get("configurationId", "") + ) == str(config_id): + matching.append(sched) + return matching + + def _collect_schedules_by_parent( client: Any, branch_id: int | None ) -> dict[tuple[str, str], list[dict[str, Any]]]: @@ -126,8 +150,9 @@ def _collect_schedules_by_parent( class FlowService(BaseService): """Business logic for conditional flow (keboola.flow) CRUD. - All schedule operations use keboola.scheduler component configs -- - no separate Scheduler Service HTTP client required. + Schedules are stored as keboola.scheduler component configs AND + registered with the Scheduler Service via ``scheduler_client_factory`` + -- writing the Storage config alone leaves the cron trigger dormant. The structural conditional-flow JSON Schema is fetched at runtime from the stack's component registry (AI Service ``configurationSchema`` for @@ -139,9 +164,13 @@ def __init__( config_store: ConfigStore, client_factory: ClientFactory | None = None, ai_client_factory: AiClientFactory | None = None, + scheduler_client_factory: SchedulerClientFactory | None = None, ) -> None: super().__init__(config_store, client_factory) self._ai_client_factory = ai_client_factory or default_ai_client_factory + self._scheduler_client_factory = ( + scheduler_client_factory or default_scheduler_client_factory + ) # ── schema fetch ───────────────────────────────────────────────── @@ -571,22 +600,17 @@ def list_flow_schedules( client.close() schedules: list[dict[str, Any]] = [] - for sched in all_sched: - body = _parse_configuration(sched.get("configuration")) - target = body.get("target") or {} - if target.get("componentId") == FLOW_COMPONENT_ID and str( - target.get("configurationId", "") - ) == str(config_id): - sched_info = body.get("schedule") or {} - schedules.append( - { - "schedule_id": str(sched.get("id", "")), - "name": sched.get("name", ""), - "cron_tab": sched_info.get("cronTab", ""), - "timezone": sched_info.get("timezone", "UTC"), - "state": sched_info.get("state", "disabled"), - } - ) + for sched in _schedules_targeting_flow(all_sched, config_id): + sched_info = _parse_configuration(sched.get("configuration")).get("schedule") or {} + schedules.append( + { + "schedule_id": str(sched.get("id", "")), + "name": sched.get("name", ""), + "cron_tab": sched_info.get("cronTab", ""), + "timezone": sched_info.get("timezone", "UTC"), + "state": sched_info.get("state", "disabled"), + } + ) return { "project_alias": alias, @@ -612,7 +636,11 @@ def set_flow_schedule( duplicate schedules when called repeatedly. The schedule is stored as a keboola.scheduler configuration whose - ``target`` points at the keboola.flow component + config. + ``target`` points at the keboola.flow component + config, then + registered with the Scheduler Service so the cron trigger actually + fires. An activation failure (e.g. token without the activation + privilege) is non-fatal: the config stays written and the failure is + reported via ``warnings`` + ``activated: False`` in the result. Args: alias: Project alias. @@ -662,15 +690,8 @@ def set_flow_schedule( else: raise - existing_id: str | None = None - for sched in existing: - body = _parse_configuration(sched.get("configuration")) - target = body.get("target") or {} - if target.get("componentId") == FLOW_COMPONENT_ID and str( - target.get("configurationId", "") - ) == str(config_id): - existing_id = str(sched.get("id", "")) - break + matching = _schedules_targeting_flow(existing, config_id) + existing_id = str(matching[0].get("id", "")) if matching else None if existing_id: result = client.update_config( @@ -692,17 +713,39 @@ def set_flow_schedule( finally: client.close() + schedule_id = str(result.get("id", existing_id or "")) + + # The service re-reads the config on activation, so this also + # deregisters a disabled schedule. + warnings: list[str] = [] + activated = False + with self._scheduler_client_factory(project.stack_url, project.token) as scheduler: + try: + scheduler.activate_schedule(schedule_id) + activated = True + except KeboolaApiError as exc: + logger.warning("Scheduler Service activation failed: %s", exc.message) + warnings.append( + f"Schedule config {schedule_id} was {status} but could not be " + f"activated on the Scheduler Service: {exc.message}. The service " + "may not reflect the updated configuration until activation " + "succeeds -- re-run this command with a token that can manage " + "schedules." + ) + return { "status": status, "project_alias": alias, - "schedule_id": str(result.get("id", existing_id or "")), + "schedule_id": schedule_id, "schedule_name": schedule_name, "component_id": FLOW_COMPONENT_ID, "config_id": config_id, "cron_tab": cron_tab, "timezone": timezone, "state": "enabled" if enabled else "disabled", + "activated": activated, "branch_id": effective_branch, + "warnings": warnings, } def remove_flow_schedule( @@ -713,6 +756,12 @@ def remove_flow_schedule( ) -> dict[str, Any]: """Delete all keboola.scheduler configs that target this flow. + Each schedule is first deregistered from the Scheduler Service (so + the cron trigger stops firing), then its Storage config is deleted. + A missing service-side registration is ignored; other deregistration + failures are reported via ``warnings`` and do not block the Storage + config deletion. + Idempotent: if no schedules exist, returns deleted_count=0. """ projects = self.resolve_projects([alias]) @@ -733,20 +782,30 @@ def remove_flow_schedule( deleted: list[str] = [] errors: list[str] = [] - for sched in all_sched: - body = _parse_configuration(sched.get("configuration")) - target = body.get("target") or {} - if target.get("componentId") == FLOW_COMPONENT_ID and str( - target.get("configurationId", "") - ) == str(config_id): - sched_id = str(sched.get("id", "")) - try: - client.delete_config( - SCHEDULER_COMPONENT_ID, sched_id, branch_id=effective_branch - ) - deleted.append(sched_id) - except KeboolaApiError as exc: - errors.append(f"{sched_id}: {exc.message}") + warnings: list[str] = [] + matching = _schedules_targeting_flow(all_sched, config_id) + if matching: + with self._scheduler_client_factory(project.stack_url, project.token) as scheduler: + for sched in matching: + sched_id = str(sched.get("id", "")) + try: + scheduler.remove_schedule(sched_id) + except KeboolaApiError as exc: + if exc.error_code != ErrorCode.NOT_FOUND: + logger.warning( + "Scheduler Service deregistration failed: %s", exc.message + ) + warnings.append( + f"Schedule {sched_id} could not be deregistered from " + f"the Scheduler Service: {exc.message}" + ) + try: + client.delete_config( + SCHEDULER_COMPONENT_ID, sched_id, branch_id=effective_branch + ) + deleted.append(sched_id) + except KeboolaApiError as exc: + errors.append(f"{sched_id}: {exc.message}") finally: client.close() @@ -766,4 +825,5 @@ def remove_flow_schedule( "deleted_schedule_ids": deleted, "deleted_count": len(deleted), "branch_id": effective_branch, + "warnings": warnings, } diff --git a/src/keboola_agent_cli/services/schedule_service.py b/src/keboola_agent_cli/services/schedule_service.py index 5c9689e4..5cb9028c 100644 --- a/src/keboola_agent_cli/services/schedule_service.py +++ b/src/keboola_agent_cli/services/schedule_service.py @@ -14,9 +14,11 @@ configuration.schedule.timezone = "Europe/Prague" configuration.schedule.state = "enabled" # or "disabled" -No separate Scheduler Service HTTP client is required -- everything reuses -``KeboolaClient.list_component_configs`` + ``get_config_detail`` + ``list_jobs`` -from the Storage and Queue APIs. +This read/audit path needs no Scheduler Service HTTP client -- everything +reuses ``KeboolaClient.list_component_configs`` + ``get_config_detail`` + +``list_jobs`` from the Storage and Queue APIs. (The WRITE path is different: +making a schedule actually fire requires registering it with the Scheduler +Service -- see ``SchedulerClient`` and ``FlowService.set_flow_schedule``.) """ from __future__ import annotations diff --git a/tests/test_flow_cli.py b/tests/test_flow_cli.py index e84e693e..407a852c 100644 --- a/tests/test_flow_cli.py +++ b/tests/test_flow_cli.py @@ -648,6 +648,129 @@ def test_schedule_with_timezone_and_disabled(self, tmp_path: Path) -> None: assert call_kwargs["timezone"] == "Europe/Prague" assert call_kwargs["enabled"] is False + def test_schedule_json_includes_activation_result(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.set_flow_schedule.return_value = { + "status": "created", + "project_alias": "prod", + "schedule_id": "sched-99", + "schedule_name": "Daily Run (Schedule)", + "component_id": "keboola.flow", + "config_id": "flow-1", + "cron_tab": "0 6 * * *", + "timezone": "UTC", + "state": "enabled", + "activated": False, + "branch_id": None, + "warnings": ["Schedule config sched-99 was created but could not be activated"], + } + result = _invoke( + store, + mock_flow, + [ + "--json", + "flow", + "schedule", + "--project", + "prod", + "--flow-id", + "flow-1", + "--cron", + "0 6 * * *", + ], + ) + assert result.exit_code == 0, result.output + data = json.loads(result.output) + assert data["data"]["activated"] is False + assert len(data["data"]["warnings"]) == 1 + + def test_schedule_human_mode_prints_activation_warning(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.set_flow_schedule.return_value = { + "status": "created", + "project_alias": "prod", + "schedule_id": "sched-99", + "schedule_name": "Daily Run (Schedule)", + "component_id": "keboola.flow", + "config_id": "flow-1", + "cron_tab": "0 6 * * *", + "timezone": "UTC", + "state": "enabled", + "activated": False, + "branch_id": None, + "warnings": ["could not be activated on the Scheduler Service"], + } + result = _invoke( + store, + mock_flow, + ["flow", "schedule", "--project", "prod", "--flow-id", "flow-1", "--cron", "0 6 * * *"], + ) + assert result.exit_code == 0, result.output + assert "could not be activated" in result.output + + def test_schedule_human_mode_reports_activated(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.set_flow_schedule.return_value = { + "status": "created", + "project_alias": "prod", + "schedule_id": "sched-99", + "schedule_name": "Daily Run (Schedule)", + "component_id": "keboola.flow", + "config_id": "flow-1", + "cron_tab": "0 6 * * *", + "timezone": "UTC", + "state": "enabled", + "activated": True, + "branch_id": None, + "warnings": [], + } + result = _invoke( + store, + mock_flow, + ["flow", "schedule", "--project", "prod", "--flow-id", "flow-1", "--cron", "0 6 * * *"], + ) + assert result.exit_code == 0, result.output + assert "created and activated" in result.output + + def test_schedule_human_mode_reports_deactivated_when_disabled(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.set_flow_schedule.return_value = { + "status": "created", + "project_alias": "prod", + "schedule_id": "sched-99", + "schedule_name": "Daily Run (Schedule)", + "component_id": "keboola.flow", + "config_id": "flow-1", + "cron_tab": "0 6 * * *", + "timezone": "UTC", + "state": "disabled", + "activated": True, + "branch_id": None, + "warnings": [], + } + result = _invoke( + store, + mock_flow, + [ + "flow", + "schedule", + "--project", + "prod", + "--flow-id", + "flow-1", + "--cron", + "0 6 * * *", + "--disabled", + ], + ) + assert result.exit_code == 0, result.output + assert "created and deactivated" in result.output + assert "and activated" not in result.output + # --------------------------------------------------------------------------- # flow schedule-remove @@ -686,6 +809,27 @@ def test_remove_with_yes(self, tmp_path: Path) -> None: assert data["data"]["deleted_count"] == 1 assert "component_id" not in mock_flow.remove_flow_schedule.call_args.kwargs + def test_remove_human_mode_prints_deregistration_warning(self, tmp_path: Path) -> None: + store = _setup_config(tmp_path / "cfg", {"prod": {}}) + mock_flow = MagicMock() + mock_flow.remove_flow_schedule.return_value = { + "status": "removed", + "project_alias": "prod", + "component_id": "keboola.flow", + "config_id": "flow-1", + "deleted_schedule_ids": ["sched-1"], + "deleted_count": 1, + "branch_id": None, + "warnings": ["Schedule sched-1 could not be deregistered from the Scheduler Service"], + } + result = _invoke( + store, + mock_flow, + ["flow", "schedule-remove", "--project", "prod", "--flow-id", "flow-1", "--yes"], + ) + assert result.exit_code == 0, result.output + assert "could not be deregistered" in result.output + def test_schedule_remove_dry_run_lists_schedules(self, tmp_path: Path) -> None: store = _setup_config(tmp_path / "cfg", {"prod": {}}) mock_flow = MagicMock() diff --git a/tests/test_flow_service.py b/tests/test_flow_service.py index 4a54a79c..b96ed913 100644 --- a/tests/test_flow_service.py +++ b/tests/test_flow_service.py @@ -72,19 +72,29 @@ def _make_ai_client(schema: dict | None = _FLOW_SCHEMA, raise_exc: Exception | N return ai +def _make_scheduler_client() -> MagicMock: + """Build a mock SchedulerClient usable as a context manager.""" + scheduler = MagicMock() + scheduler.__enter__.return_value = scheduler + return scheduler + + def _make_flow_service( mock_client: MagicMock, projects: dict | None = None, ai_client: MagicMock | None = None, + scheduler_client: MagicMock | None = None, ) -> FlowService: if projects is None: projects = {"prod": {"url": "https://connection.keboola.com", "token": "tok"}} cs = _mock_config_store(projects) ai = ai_client if ai_client is not None else _make_ai_client() + scheduler = scheduler_client if scheduler_client is not None else _make_scheduler_client() return FlowService( config_store=cs, client_factory=lambda url, token: mock_client, ai_client_factory=lambda url, token: ai, + scheduler_client_factory=lambda url, token: scheduler, ) @@ -405,6 +415,71 @@ def test_set_flow_schedule_targets_keboola_flow(): assert result["component_id"] == "keboola.flow" +def test_set_flow_schedule_activates_created_config(): + client = MagicMock() + client.get_config_detail.return_value = {"name": "CF"} + client.list_component_configs.return_value = [] + client.create_config.return_value = {"id": "77"} + scheduler = _make_scheduler_client() + svc = _make_flow_service(client, scheduler_client=scheduler) + result = svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *") + # activation targets the scheduler config id, not the flow id + scheduler.activate_schedule.assert_called_once_with("77") + assert result["activated"] is True + assert result["warnings"] == [] + + +def test_set_flow_schedule_activates_updated_config(): + client = MagicMock() + client.get_config_detail.return_value = {"name": "CF"} + client.list_component_configs.return_value = [ + { + "id": "88", + "configuration": { + "target": {"componentId": "keboola.flow", "configurationId": "5"}, + }, + } + ] + client.update_config.return_value = {"id": "88"} + scheduler = _make_scheduler_client() + svc = _make_flow_service(client, scheduler_client=scheduler) + result = svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *") + assert result["status"] == "updated" + scheduler.activate_schedule.assert_called_once_with("88") + assert result["activated"] is True + + +def test_set_flow_schedule_disabled_still_calls_scheduler_service(): + client = MagicMock() + client.get_config_detail.return_value = {"name": "CF"} + client.list_component_configs.return_value = [] + client.create_config.return_value = {"id": "77"} + scheduler = _make_scheduler_client() + svc = _make_flow_service(client, scheduler_client=scheduler) + svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *", enabled=False) + scheduler.activate_schedule.assert_called_once_with("77") + + +def test_set_flow_schedule_activation_failure_warns_not_raises(): + client = MagicMock() + client.get_config_detail.return_value = {"name": "CF"} + client.list_component_configs.return_value = [] + client.create_config.return_value = {"id": "77"} + scheduler = _make_scheduler_client() + scheduler.activate_schedule.side_effect = KeboolaApiError( + message="Access denied", + status_code=403, + error_code=ErrorCode.ACCESS_DENIED, + retryable=False, + ) + svc = _make_flow_service(client, scheduler_client=scheduler) + result = svc.set_flow_schedule(alias="prod", config_id="5", cron_tab="0 6 * * *") + assert result["status"] == "created" + assert result["activated"] is False + assert len(result["warnings"]) == 1 + assert "may not reflect" in result["warnings"][0] + + def test_remove_flow_schedule_filters_keboola_flow(): client = MagicMock() client.list_component_configs.return_value = [ @@ -419,3 +494,68 @@ def test_remove_flow_schedule_filters_keboola_flow(): result = svc.remove_flow_schedule(alias="prod", config_id="5") assert result["deleted_count"] == 1 assert result["component_id"] == "keboola.flow" + + +def test_remove_flow_schedule_deregisters_from_scheduler_service(): + client = MagicMock() + client.list_component_configs.return_value = [ + { + "id": "77", + "configuration": { + "target": {"componentId": "keboola.flow", "configurationId": "5"}, + }, + } + ] + scheduler = _make_scheduler_client() + svc = _make_flow_service(client, scheduler_client=scheduler) + result = svc.remove_flow_schedule(alias="prod", config_id="5") + scheduler.remove_schedule.assert_called_once_with("77") + assert result["deleted_count"] == 1 + assert result["warnings"] == [] + + +def test_remove_flow_schedule_tolerates_missing_service_registration(): + client = MagicMock() + client.list_component_configs.return_value = [ + { + "id": "77", + "configuration": { + "target": {"componentId": "keboola.flow", "configurationId": "5"}, + }, + } + ] + scheduler = _make_scheduler_client() + scheduler.remove_schedule.side_effect = KeboolaApiError( + message="Schedule not found", + status_code=404, + error_code=ErrorCode.NOT_FOUND, + retryable=False, + ) + svc = _make_flow_service(client, scheduler_client=scheduler) + result = svc.remove_flow_schedule(alias="prod", config_id="5") + assert result["deleted_count"] == 1 + assert result["warnings"] == [] + + +def test_remove_flow_schedule_service_failure_warns_but_deletes_config(): + client = MagicMock() + client.list_component_configs.return_value = [ + { + "id": "77", + "configuration": { + "target": {"componentId": "keboola.flow", "configurationId": "5"}, + }, + } + ] + scheduler = _make_scheduler_client() + scheduler.remove_schedule.side_effect = KeboolaApiError( + message="Access denied", + status_code=403, + error_code=ErrorCode.ACCESS_DENIED, + retryable=False, + ) + svc = _make_flow_service(client, scheduler_client=scheduler) + result = svc.remove_flow_schedule(alias="prod", config_id="5") + assert result["deleted_count"] == 1 + assert len(result["warnings"]) == 1 + client.delete_config.assert_called_once() diff --git a/tests/test_scheduler_client.py b/tests/test_scheduler_client.py new file mode 100644 index 00000000..8bcc8526 --- /dev/null +++ b/tests/test_scheduler_client.py @@ -0,0 +1,105 @@ +"""Tests for SchedulerClient -- URL derivation, request envelopes, error mapping. + +Mirrors the test_metastore_client.py pattern: drive the client through +pytest-httpx mocks and verify the verb-level contract (URL derivation, +POST /schedules body, DELETE /configurations/{id} path, auth header, +error normalization inherited from BaseHttpClient). +""" + +from __future__ import annotations + +import json + +import pytest + +from keboola_agent_cli.errors import ErrorCode, KeboolaApiError +from keboola_agent_cli.scheduler_client import SchedulerClient + +STACK_URL_US = "https://connection.keboola.com" +SCHEDULER_URL_US = "https://scheduler.keboola.com" +TOKEN = "901-55555-fakeTestTokenDoNotUseXXXXXXXX" + + +@pytest.fixture(autouse=True) +def _no_backoff_sleep(monkeypatch: pytest.MonkeyPatch) -> None: + """Disable retry-backoff sleeps so the suite stays fast.""" + import keboola_agent_cli.http_base as http_base_module + + monkeypatch.setattr(http_base_module.time, "sleep", lambda _x: None) + + +class TestUrlDerivation: + """Verify SchedulerClient maps ``connection.`` to ``scheduler.``.""" + + def test_us_stack(self) -> None: + result = SchedulerClient._derive_service_url("https://connection.keboola.com", "scheduler") + assert result == "https://scheduler.keboola.com" + + def test_eu_gcp_stack(self) -> None: + result = SchedulerClient._derive_service_url( + "https://connection.europe-west3.gcp.keboola.com", "scheduler" + ) + assert result == "https://scheduler.europe-west3.gcp.keboola.com" + + +class TestActivateSchedule: + """activate_schedule posts the scheduler config id to POST /schedules.""" + + def test_posts_configuration_id(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{SCHEDULER_URL_US}/schedules", + method="POST", + json={"id": "1", "configurationId": "77"}, + status_code=201, + ) + with SchedulerClient(stack_url=STACK_URL_US, token=TOKEN) as client: + result = client.activate_schedule("77") + request = httpx_mock.get_requests()[0] + assert json.loads(request.content) == {"configurationId": "77"} + assert request.headers["X-StorageApi-Token"] == TOKEN + assert result["configurationId"] == "77" + + def test_forbidden_maps_to_access_denied(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{SCHEDULER_URL_US}/schedules", + method="POST", + json={"error": "Insufficient permissions"}, + status_code=403, + ) + with ( + SchedulerClient(stack_url=STACK_URL_US, token=TOKEN) as client, + pytest.raises(KeboolaApiError) as excinfo, + ): + client.activate_schedule("77") + assert excinfo.value.error_code == ErrorCode.ACCESS_DENIED + assert excinfo.value.status_code == 403 + + +class TestRemoveSchedule: + """remove_schedule deregisters via DELETE /configurations/{id}.""" + + def test_deletes_configuration(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{SCHEDULER_URL_US}/configurations/77", + method="DELETE", + status_code=204, + ) + with SchedulerClient(stack_url=STACK_URL_US, token=TOKEN) as client: + client.remove_schedule("77") + request = httpx_mock.get_requests()[0] + assert request.method == "DELETE" + assert request.headers["X-StorageApi-Token"] == TOKEN + + def test_missing_registration_maps_to_not_found(self, httpx_mock) -> None: + httpx_mock.add_response( + url=f"{SCHEDULER_URL_US}/configurations/77", + method="DELETE", + json={"error": "Schedule not found"}, + status_code=404, + ) + with ( + SchedulerClient(stack_url=STACK_URL_US, token=TOKEN) as client, + pytest.raises(KeboolaApiError) as excinfo, + ): + client.remove_schedule("77") + assert excinfo.value.error_code == ErrorCode.NOT_FOUND diff --git a/uv.lock b/uv.lock index 9e5ad008..56711c64 100644 --- a/uv.lock +++ b/uv.lock @@ -590,7 +590,7 @@ wheels = [ [[package]] name = "keboola-cli" -version = "0.66.0" +version = "0.66.1" source = { editable = "." } dependencies = [ { name = "croniter" },