From f62fd3ec72a8cdbfe50bfc901d1841b735699066 Mon Sep 17 00:00:00 2001 From: Petr Date: Mon, 24 Aug 2026 07:39:45 +0200 Subject: [PATCH] feat(serve): mirror describe-batch + unload-table, expose job idempotency (#657) Three MEDIUM gaps from the 0.89.0 serve audit. Each mirror deliberately differs from its CLI command where the CLI option names the CALLER's filesystem -- which, over serve, is the server's. - `POST /storage/describe-batch/{project}`: the whole bulk-documentation path was CLI-only. Sections travel inline in the body instead of a `--from-file` path. `parse_describe_batch_file` is split into a shared `parse_describe_batch_document`, and the write loop into `_apply_describe_batch`, so the two surfaces cannot drift into accepting different documents -- the half that drifted would be the one letting a malformed section reach the write loop. The sections are typed `Any` on purpose: declaring them `dict[str, str]` would have pydantic reject a wrong shape first with its generic 422, discarding #645's messages (offending key, actual type, copy-pasteable example). - `POST /storage/tables/{project}/{table_id}/unload`: no `--download` / `--output`; the response carries the Storage `file_id` and `GET /storage/files/{p}/{file_id}/download` fetches the bytes. That is also the only shape that works for parquet, whose export is sliced. - `POST /jobs/{p}/run` now accepts `idempotency_key` / `force_rerun`. `JobService.run_job` has taken both since #427, and retrying a POST is the canonical case that store was built for, so the mirror omitting them left the caller who needs them most without them. Omitting both produces a byte-identical call to the pre-change one. The issue's bonus ask, `GET /permissions/show`, is in the #655 PR instead -- it belongs with the enforcement it describes. Tests: 20 new in tests/test_server_missing_mirrors.py, including an exact-kwargs assertion that `unload` never passes `output_path`, a greedy-`{table_id:path}` case proving `/unload` is not eaten as part of a dotted table id, and a parametrised check that the shared validator rejects and accepts the same documents on both paths. Fixes #657 --- docs/web-server-endpoints.md | 6 +- docs/web-server.md | 25 +- .../skills/kbagent/references/gotchas.md | 13 + src/keboola_agent_cli/server/routers/jobs.py | 19 +- .../server/routers/storage.py | 104 +++++++ .../services/_describe_batch_input.py | 52 +++- .../services/storage_service.py | 71 ++++- tests/test_server_missing_mirrors.py | 282 ++++++++++++++++++ 8 files changed, 551 insertions(+), 21 deletions(-) create mode 100644 tests/test_server_missing_mirrors.py diff --git a/docs/web-server-endpoints.md b/docs/web-server-endpoints.md index cceeb3ecb..b329f5f28 100644 --- a/docs/web-server-endpoints.md +++ b/docs/web-server-endpoints.md @@ -9,7 +9,7 @@ auth, and the concepts behind these routes live in [`web-server.md`](web-server.md); a running server serves the same spec interactively at `/docs` (Swagger) and `/openapi.json`. -**231 operations** across **202 paths** and **30 routers**. +**233 operations** across **204 paths** and **30 routers**. Paths are shown as the server registers them. Reaching them through the Node BFF (or single-process `--ui` mode) prefixes every path with `/api`. @@ -166,7 +166,7 @@ Encrypt secret values for a specific project + component using the Keboola encry ## Data -### `storage` (32 operations) +### `storage` (34 operations) Buckets, tables, columns, files. Create, upload, download, describe, swap, delete. Mirrors `kbagent storage *`. @@ -196,7 +196,9 @@ Buckets, tables, columns, files. Create, upload, download, describe, swap, delet | `POST` | `/storage/table-from-snapshot/{project}` | Create a NEW table from a snapshot | | `POST` | `/storage/tables/{project}/{table_id}/describe` | Set table description | | `POST` | `/storage/columns/{project}/{table_id}/describe` | Set column descriptions | +| `POST` | `/storage/describe-batch/{project}` | Apply descriptions in bulk | | `POST` | `/storage/columns/{project}/describe-migrate` | Migrate legacy column descriptions | +| `POST` | `/storage/tables/{project}/{table_id}/unload` | Export a table to a file | | `GET` | `/storage/files` | List storage files | | `POST` | `/storage/files/upload` | Upload a file to Storage | | `GET` | `/storage/files/{project}/{file_id}` | Get file detail | diff --git a/docs/web-server.md b/docs/web-server.md index 9c7076f9d..fa7363010 100644 --- a/docs/web-server.md +++ b/docs/web-server.md @@ -106,8 +106,10 @@ read/audit endpoints, three deliberate gaps" below. Going the other way, several CLI surfaces are deliberately CLI-only: `sync` (filesystem-local by design), `permissions`, and `init`. The mirrors still -considered missing are tracked in #657, and the fact that `permissions` -constrains only `/auth/*` and not the other ~30 routers is #655. +considered missing are tracked in #657 — `describe-batch` (inline payload), +`unload-table` and `job run`'s idempotency fields came off that list in vNEXT; +the fact that `permissions` constrains only `/auth/*` and not the other ~30 +routers is #655. Auto-generated OpenAPI spec at `/openapi.json`, Swagger UI at `/docs`. @@ -572,6 +574,25 @@ for a human), then use `POST /auth/register-projects` — or `auth register-projects` on the CLI — to register the resulting session's projects for `serve` to use. +### CLI options that a REST mirror deliberately drops + +*(since vNEXT — issue #657)* + +Three mirrors landed with a shape that is not a field-for-field copy of the CLI +command, because the CLI's version of the option refers to the *caller's* +filesystem — which, over `serve`, is the server's: + +| CLI | REST | Difference | +|---|---|---| +| `storage describe-batch --from-file PATH` | `POST /storage/describe-batch/{project}` | Sections travel **inline in the body** (`buckets` / `tables` / `columns` / `branch_id`). Same validator (`services/_describe_batch_input.py`), so both surfaces accept and reject exactly the same documents, whole, before the first write. | +| `storage unload-table [--download] [--output]` | `POST /storage/tables/{project}/{table_id}/unload` | No `download` / `output`. The response carries the Storage `file_id`; fetch the bytes with `GET /storage/files/{project}/{file_id}/download`. That is also the only shape that works for `file_type: "parquet"`, whose export is sliced. | +| `job run --idempotency-key KEY [--force-rerun]` | `POST /jobs/{project}/run` | Same two fields, now on the body. Retrying a POST is the canonical case #427's store was built for, so the mirror omitting them left the caller who needs them most without them. Dedup is scoped to the **served** config dir, i.e. per machine. | + +One error-shape note: a malformed `describe-batch` body answers **422** +(`HTTP_ERROR`) where the CLI exits **2** (`INVALID_ARGUMENT`). The message is +the same one #645 writes — the offending key, its actual type, and an example — +but the code differs, so branch on the status for this one, not the code. + ### Manage tokens are per-request Operations that need a Keboola Manage API token (`org setup`, diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index 79993bab5..dd2f2df79 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -4506,6 +4506,19 @@ shapes. keep) must be in its allow list, or the server will not start either. Pass `--config-dir` to `serve` itself: the root-level `kbagent --config-dir ... serve` sets the dir for the CLI invocation, not for the server process. +- **Three `serve` mirrors landed in vNEXT (#657) with a deliberately different + shape from their CLI command** -- `POST /storage/describe-batch/{project}` + takes the `buckets`/`tables`/`columns` sections INLINE in the body (no + `--from-file` path: a REST caller's filesystem is not the server's), and it + answers **422** for a malformed document where the CLI exits 2 + `INVALID_ARGUMENT` -- same message, different code, so branch on the status. + `POST /storage/tables/{project}/{table_id}/unload` has no `--download` / + `--output`; it returns the Storage `file_id` and you fetch the bytes with + `GET /storage/files/{project}/{file_id}/download` (the only shape that works + for `file_type: "parquet"`, whose export is sliced). `POST /jobs/{p}/run` + finally accepts `idempotency_key` / `force_rerun` -- retrying a POST is + exactly what #427's store is for, and before vNEXT the REST caller could not + reach it; dedup is scoped to the SERVED config dir, i.e. per machine. - **A deny policy does NOT firewall the whole REST surface.** `permissions set --mode deny --deny cli:write` blocks `POST /auth/register-projects` (HTTP 403, `error_code: PERMISSION_DENIED`) but diff --git a/src/keboola_agent_cli/server/routers/jobs.py b/src/keboola_agent_cli/server/routers/jobs.py index d3c324c98..4d1bc5e05 100644 --- a/src/keboola_agent_cli/server/routers/jobs.py +++ b/src/keboola_agent_cli/server/routers/jobs.py @@ -31,6 +31,14 @@ class JobRun(BaseModel): variable_values_id: str | None = None no_variables: bool = False mode: str = DEFAULT_JOB_MODE + # Client-side de-duplication (#427, exposed over REST by #657). Retrying a + # POST is the canonical case the store was built for -- a timed-out or + # replayed request must not start a second job -- so the REST mirror + # dropping these two fields left the one caller that needs them most + # without them. The store is keyed inside the SERVED config dir, so dedup + # is per-machine, exactly as on the CLI. + idempotency_key: str | None = None + force_rerun: bool = False class JobTerminate(BaseModel): @@ -93,7 +101,14 @@ def run( log_tail_lines: int = DEFAULT_LOG_TAIL_LINES, registry: ServiceRegistry = Depends(get_registry), ) -> dict[str, Any]: - """Run a component configuration and optionally wait for completion. Mirrors `kbagent job run`.""" + """Run a component configuration and optionally wait for completion. + + Mirrors `kbagent job run`, including its client-side idempotency + (`idempotency_key` / `force_rerun`, #427): replaying this POST under the + same key returns the recorded job instead of starting a second one, unless + that job failed or `force_rerun` is set. Reusing a key for a *different* + component/config raises rather than return the wrong job. + """ return registry.job.run_job( alias=project, component_id=body.component_id, @@ -107,6 +122,8 @@ def run( poll_strategy=poll_strategy, log_tail_lines=log_tail_lines, mode=body.mode, + idempotency_key=body.idempotency_key, + force_rerun=body.force_rerun, ) diff --git a/src/keboola_agent_cli/server/routers/storage.py b/src/keboola_agent_cli/server/routers/storage.py index 832416fc6..8a5ecf828 100644 --- a/src/keboola_agent_cli/server/routers/storage.py +++ b/src/keboola_agent_cli/server/routers/storage.py @@ -15,6 +15,47 @@ router = APIRouter(prefix="/storage", tags=["storage"]) +class DescribeBatch(BaseModel): + """Inline payload for the bulk-documentation mirror (issue #657). + + The CLI takes a YAML *path*; a REST caller has a JSON *body*, and asking + it to first write a file onto the server's filesystem would be both + awkward and a way to smuggle a path in. So the three sections travel in + the body instead, and everything else -- the rules, the messages, the + reject-whole-before-first-write guarantee -- is literally the shared + validator (`services/_describe_batch_input.py`). + + The sections are typed ``Any`` on purpose. Declaring them + ``dict[str, str]`` would have pydantic reject a wrong shape first, with + its generic 422, discarding the whole point of #645's messages: the key + that is wrong, its actual type, and a copy-pasteable example. + """ + + buckets: Any = None + tables: Any = None + columns: Any = None + branch_id: int | None = None + + +class UnloadTable(BaseModel): + """Body for the `storage unload-table` mirror (issue #657). + + The CLI's `--download` / `--output` are deliberately absent: they write to + the *caller's* filesystem, which for a REST caller is the server's. + The route returns the Storage file id instead, and + `GET /storage/files/{project}/{file_id}/download` fetches the bytes -- + which is also the only shape that works for `parquet`, whose export is + sliced and has no single-file download even on the CLI. + """ + + columns: list[str] | None = None + limit: int | None = None + tags: list[str] | None = None + file_type: str = "csv" + keep_slices: bool = False + branch_id: int | None = None + + class CreateBucket(BaseModel): stage: str name: str @@ -580,6 +621,36 @@ def describe_columns( ) +@router.post("/describe-batch/{project}", summary="Apply descriptions in bulk") +def describe_batch( + project: str, + body: DescribeBatch, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Apply bucket/table/column descriptions in one call. + + Mirrors `kbagent storage describe-batch --from-file`, with the document + inline instead of on disk. Same validation, so the whole payload is + rejected before the first write if any section has the wrong shape; a + per-item API failure afterwards is accumulated into `errors` and does not + stop the remaining items. + + A shape error answers **422** (`HTTP_ERROR`) rather than the CLI's exit-2 + `INVALID_ARGUMENT` -- it is a malformed request body, the same class of + problem FastAPI already answers 422 for on this surface. The message is + the CLI's, naming the offending key and its actual type. + """ + document = {"buckets": body.buckets, "tables": body.tables, "columns": body.columns} + try: + return registry.storage.describe_batch_document( + alias=project, + document=document, + branch_id=body.branch_id, + ) + except ValueError as exc: + raise HTTPException(status_code=422, detail=str(exc)) from None + + @router.post("/columns/{project}/describe-migrate", summary="Migrate legacy column descriptions") def describe_migrate( project: str, @@ -597,6 +668,39 @@ def describe_migrate( ) +@router.post("/tables/{project}/{table_id:path}/unload", summary="Export a table to a file") +def unload_table( + project: str, + table_id: str, + body: UnloadTable | None = None, + registry: ServiceRegistry = Depends(get_registry), +) -> dict[str, Any]: + """Export a table into a Storage File and return its metadata. + + Mirrors `kbagent storage unload-table`. The sync `GET + /storage/table-preview/...` covers small reads; this covers the big ones, + where the async export + a presigned Storage file is the whole point. + + Nothing is written to the server's filesystem: the response carries the + `file_id`, and `GET /storage/files/{project}/{file_id}/download` fetches + the bytes. `file_type: "parquet"` produces a SLICED file with no + single-file download -- use `GET /storage/files/{project}/{file_id}` and + work with the slices. + """ + params = body or UnloadTable() + return registry.storage.unload_table_to_file( + alias=project, + table_id=table_id, + columns=params.columns, + limit=params.limit, + tags=params.tags, + download=False, + branch_id=params.branch_id, + file_type=params.file_type, + keep_slices=params.keep_slices, + ) + + # Registered AFTER the more specific /columns/.../describe route above: the # greedy {table_id:path} would otherwise shadow that POST and swallow a # ".../describe" suffix as part of the table id. diff --git a/src/keboola_agent_cli/services/_describe_batch_input.py b/src/keboola_agent_cli/services/_describe_batch_input.py index ee6575265..0817d2b3f 100644 --- a/src/keboola_agent_cli/services/_describe_batch_input.py +++ b/src/keboola_agent_cli/services/_describe_batch_input.py @@ -180,6 +180,41 @@ def _columns_section(raw: dict[str, Any]) -> dict[str, dict[str, str]]: return parsed +def parse_describe_batch_document(raw: Any, source_label: str) -> DescribeBatchInput: + """Validate an already-loaded describe-batch document. + + Split out from :func:`parse_describe_batch_file` so the REST mirror + (``POST /storage/describe-batch/{project}``, issue #657) validates the + JSON body through exactly the same rules as the CLI's YAML file. Two + copies of these rules would drift, and the half that drifted would be the + one that lets a malformed section reach the write loop. + + Args: + raw: The parsed document -- YAML from a file, or a JSON request body. + source_label: What to name in an error message ("describe.yml", + "request body"). Only ever used for the top-level shape error; + per-section errors already name their own key. + + Returns: + The validated sections, descriptions coerced to ``str``. Absent, + ``None`` and empty sections come back empty. + + Raises: + ValueError: The document is not a mapping at the top level, or any + section/entry has a wrong non-empty shape, a null description, or + a duplicate coerced key. + """ + if _is_empty(raw): + return DescribeBatchInput() + if not isinstance(raw, dict): + raise _shape_error(source_label, _TOP_LEVEL_SUBJECT, raw, _TOP_LEVEL_EXAMPLE) + return DescribeBatchInput( + buckets=_scalar_section(raw, "buckets"), + tables=_scalar_section(raw, "tables"), + columns=_columns_section(raw), + ) + + def parse_describe_batch_file(from_file: Path) -> DescribeBatchInput: """Read and validate a describe-batch YAML file. @@ -191,10 +226,9 @@ def parse_describe_batch_file(from_file: Path) -> DescribeBatchInput: ``None`` and empty sections come back empty. Raises: - ValueError: The file is missing, is not valid YAML, is not a mapping - at the top level, or any section/entry has a wrong non-empty - shape, a null description, or a duplicate coerced key. The command - layer maps this to ``INVALID_ARGUMENT`` and exit 2. + ValueError: The file is missing, is not valid YAML, or fails any check + in :func:`parse_describe_batch_document`. The command layer maps + this to ``INVALID_ARGUMENT`` and exit 2. """ import yaml @@ -204,12 +238,4 @@ def parse_describe_batch_file(from_file: Path) -> DescribeBatchInput: raw = yaml.safe_load(from_file.read_text(encoding="utf-8")) except yaml.YAMLError as exc: raise ValueError(f"Batch file is not valid YAML: {exc}") from None - if _is_empty(raw): - return DescribeBatchInput() - if not isinstance(raw, dict): - raise _shape_error(from_file.name, _TOP_LEVEL_SUBJECT, raw, _TOP_LEVEL_EXAMPLE) - return DescribeBatchInput( - buckets=_scalar_section(raw, "buckets"), - tables=_scalar_section(raw, "tables"), - columns=_columns_section(raw), - ) + return parse_describe_batch_document(raw, from_file.name) diff --git a/src/keboola_agent_cli/services/storage_service.py b/src/keboola_agent_cli/services/storage_service.py index 80d4c5bf9..5e23a7d28 100644 --- a/src/keboola_agent_cli/services/storage_service.py +++ b/src/keboola_agent_cli/services/storage_service.py @@ -9,7 +9,7 @@ import re from collections.abc import Callable from pathlib import Path -from typing import Any +from typing import TYPE_CHECKING, Any from ..constants import STORAGE_BRANCHES_FEATURE from ..errors import ConfigError, ErrorCode, KeboolaApiError @@ -19,6 +19,9 @@ from ._table_detail import build_table_detail from .table_usage import collect_table_usage, fetch_usage_components +if TYPE_CHECKING: + from ._describe_batch_input import DescribeBatchInput + logger = logging.getLogger(__name__) @@ -2509,10 +2512,72 @@ def describe_batch( Returns: Dict with project_alias, applied, errors, applied_count, error_count. """ - from ..errors import KeboolaApiError from ._describe_batch_input import parse_describe_batch_file - parsed = parse_describe_batch_file(from_file) + return self._apply_describe_batch( + alias, + parse_describe_batch_file(from_file), + branch_id=branch_id, + progress_callback=progress_callback, + ) + + def describe_batch_document( + self, + alias: str, + document: Any, + branch_id: int | None = None, + progress_callback: Callable[[str, str, int, int], None] | None = None, + ) -> dict[str, Any]: + """Apply descriptions from an already-loaded document (no file on disk). + + The inline-payload sibling of :meth:`describe_batch`, added for the + REST mirror in issue #657 -- a serve caller has a JSON body, not a + path, and asking it to write a temp file on the server's filesystem + would be both awkward and a way to smuggle a path in. + + Validation and application are literally the same code as the file + path (:func:`~keboola_agent_cli.services._describe_batch_input.parse_describe_batch_document` + plus :meth:`_apply_describe_batch`), so the two surfaces cannot drift + into accepting different documents. + + Args: + alias: Project alias. + document: The parsed document -- same three optional ``buckets`` / + ``tables`` / ``columns`` sections the YAML file carries. + branch_id: If set, target a specific dev branch. + progress_callback: See :meth:`describe_batch`. + + Returns: + Dict with project_alias, applied, errors, applied_count, error_count. + + Raises: + ValueError: The document has a wrong shape. Rejected whole, before + the first write, exactly as for a file. + """ + from ._describe_batch_input import parse_describe_batch_document + + return self._apply_describe_batch( + alias, + parse_describe_batch_document(document, "request body"), + branch_id=branch_id, + progress_callback=progress_callback, + ) + + def _apply_describe_batch( + self, + alias: str, + parsed: "DescribeBatchInput", + branch_id: int | None = None, + progress_callback: Callable[[str, str, int, int], None] | None = None, + ) -> dict[str, Any]: + """Write one validated describe-batch document, accumulating per-item errors. + + Shared by :meth:`describe_batch` (file) and + :meth:`describe_batch_document` (inline). A failure on one item does + not skip the rest -- all per-item API results, success and error, are + collected and returned. + """ + from ..errors import KeboolaApiError applied: list[dict[str, Any]] = [] errors: list[dict[str, Any]] = [] diff --git a/tests/test_server_missing_mirrors.py b/tests/test_server_missing_mirrors.py new file mode 100644 index 000000000..10cba52e8 --- /dev/null +++ b/tests/test_server_missing_mirrors.py @@ -0,0 +1,282 @@ +"""REST mirrors added for issue #657, plus the idempotency fields #427 built. + +Three gaps from the 0.89.0 serve audit: + +1. `storage describe-batch` had no route at all -- the whole bulk-documentation + path was CLI-only. +2. `storage unload-table` had no route despite `unload_table_to_file` existing; + the sync preview covers small reads, unload covers the big ones. +3. `POST /jobs/{p}/run` dropped `idempotency_key` / `force_rerun` even though + `JobService.run_job` takes both -- and retrying a POST is precisely the case + #427 built the store for. + +The service layer is mocked throughout: what is under test is the router -> +service contract (kwarg names, defaults, and which CLI-only options are +deliberately NOT exposed), not Keboola behaviour. +""" + +from __future__ import annotations + +import importlib.util +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +import pytest + +if importlib.util.find_spec("fastapi") is None: # pragma: no cover + pytest.skip( + "FastAPI not installed; run `uv pip install -e '.[server]'`", allow_module_level=True + ) + +from fastapi.testclient import TestClient + +from keboola_agent_cli.server import create_app +from keboola_agent_cli.server.dependencies import ServiceRegistry, get_registry + +AUTH = {"Authorization": "Bearer test-token"} +PROJECT = "my-proj" +TABLE_ID = "in.c-main.mytable" + + +def _client(tmp_path: Path, **services: Any) -> TestClient: + registry = ServiceRegistry.__new__(ServiceRegistry) + for name, mock in services.items(): + setattr(registry, name, mock) + app = create_app(config_dir=str(tmp_path), auth_token="test-token") + app.dependency_overrides[get_registry] = lambda: registry + return TestClient(app) + + +def _storage_mock() -> MagicMock: + svc = MagicMock() + svc.describe_batch_document.return_value = { + "project_alias": PROJECT, + "applied": [], + "errors": [], + "applied_count": 0, + "error_count": 0, + } + svc.unload_table_to_file.return_value = {"file_id": 123, "table_id": TABLE_ID} + return svc + + +class TestDescribeBatchRoute: + def test_sections_reach_the_service_as_one_document(self, tmp_path: Path) -> None: + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + res = client.post( + f"/storage/describe-batch/{PROJECT}", + headers=AUTH, + json={ + "buckets": {"in.c-sales": "All sales data"}, + "tables": {"in.c-sales.orders": "One row per order"}, + "columns": {"in.c-sales.orders": {"id": "Primary key"}}, + "branch_id": 42, + }, + ) + assert res.status_code == 200 + kwargs = storage.describe_batch_document.call_args.kwargs + assert kwargs["alias"] == PROJECT + assert kwargs["branch_id"] == 42 + assert kwargs["document"] == { + "buckets": {"in.c-sales": "All sales data"}, + "tables": {"in.c-sales.orders": "One row per order"}, + "columns": {"in.c-sales.orders": {"id": "Primary key"}}, + } + + def test_omitted_sections_travel_as_none(self, tmp_path: Path) -> None: + """An absent section must stay absent, not become an empty dict. + + `_describe_batch_input` treats absent and empty identically, but the + route inventing `{}` would hide a caller sending nothing at all. + """ + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + res = client.post( + f"/storage/describe-batch/{PROJECT}", + headers=AUTH, + json={"buckets": {"in.c-sales": "All sales data"}}, + ) + assert res.status_code == 200 + document = storage.describe_batch_document.call_args.kwargs["document"] + assert document["tables"] is None + assert document["columns"] is None + + def test_branch_id_defaults_to_none(self, tmp_path: Path) -> None: + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + client.post(f"/storage/describe-batch/{PROJECT}", headers=AUTH, json={}) + assert storage.describe_batch_document.call_args.kwargs["branch_id"] is None + + def test_shape_error_answers_422_with_the_cli_message(self, tmp_path: Path) -> None: + """The point of sharing the validator: the REST caller gets #645's message.""" + storage = MagicMock() + storage.describe_batch_document.side_effect = ValueError( + "'buckets' must be a mapping of bucket ID to description, got a list." + ) + with _client(tmp_path, storage=storage) as client: + res = client.post( + f"/storage/describe-batch/{PROJECT}", + headers=AUTH, + json={"buckets": ["in.c-sales"]}, + ) + assert res.status_code == 422 + assert "must be a mapping of bucket ID to description" in res.json()["error"]["message"] + + def test_per_item_api_errors_are_returned_not_raised(self, tmp_path: Path) -> None: + """Shape is a 422; an API refusal mid-batch is a 200 with `errors`.""" + storage = MagicMock() + storage.describe_batch_document.return_value = { + "project_alias": PROJECT, + "applied": [], + "errors": [{"type": "bucket", "id": "in.c-nope", "error": "Bucket not found"}], + "applied_count": 0, + "error_count": 1, + } + with _client(tmp_path, storage=storage) as client: + res = client.post( + f"/storage/describe-batch/{PROJECT}", + headers=AUTH, + json={"buckets": {"in.c-nope": "x"}}, + ) + assert res.status_code == 200 + assert res.json()["error_count"] == 1 + + +class TestUnloadTableRoute: + def test_defaults_never_touch_the_server_filesystem(self, tmp_path: Path) -> None: + """`download` is hard-wired False: the caller's disk is not the server's.""" + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + res = client.post(f"/storage/tables/{PROJECT}/{TABLE_ID}/unload", headers=AUTH, json={}) + assert res.status_code == 200 + kwargs = storage.unload_table_to_file.call_args.kwargs + assert kwargs["download"] is False + assert "output_path" not in kwargs + assert kwargs == { + "alias": PROJECT, + "table_id": TABLE_ID, + "columns": None, + "limit": None, + "tags": None, + "download": False, + "branch_id": None, + "file_type": "csv", + "keep_slices": False, + } + + def test_body_is_optional(self, tmp_path: Path) -> None: + """An unload with no options is the common case; requiring `{}` is noise.""" + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + res = client.post(f"/storage/tables/{PROJECT}/{TABLE_ID}/unload", headers=AUTH) + assert res.status_code == 200 + assert storage.unload_table_to_file.call_args.kwargs["file_type"] == "csv" + + def test_all_options_reach_the_service(self, tmp_path: Path) -> None: + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + client.post( + f"/storage/tables/{PROJECT}/{TABLE_ID}/unload", + headers=AUTH, + json={ + "columns": ["id", "amount"], + "limit": 100, + "tags": ["export"], + "file_type": "parquet", + "keep_slices": True, + "branch_id": 7, + }, + ) + kwargs = storage.unload_table_to_file.call_args.kwargs + assert kwargs["columns"] == ["id", "amount"] + assert kwargs["limit"] == 100 + assert kwargs["tags"] == ["export"] + assert kwargs["file_type"] == "parquet" + assert kwargs["keep_slices"] is True + assert kwargs["branch_id"] == 7 + + def test_dotted_table_id_survives_the_path_converter(self, tmp_path: Path) -> None: + """`{table_id:path}` is greedy -- prove `/unload` is not eaten as part of it.""" + storage = _storage_mock() + with _client(tmp_path, storage=storage) as client: + client.post( + f"/storage/tables/{PROJECT}/out.c-my.deeply.dotted/unload", headers=AUTH, json={} + ) + assert storage.unload_table_to_file.call_args.kwargs["table_id"] == "out.c-my.deeply.dotted" + + +class TestJobRunIdempotency: + def _job_mock(self) -> MagicMock: + job = MagicMock() + job.run_job.return_value = {"id": "job-1", "status": "created"} + return job + + def test_key_and_force_rerun_reach_run_job(self, tmp_path: Path) -> None: + job = self._job_mock() + with _client(tmp_path, job=job) as client: + res = client.post( + f"/jobs/{PROJECT}/run", + headers=AUTH, + json={ + "component_id": "keboola.ex-http", + "config_id": "42", + "idempotency_key": "nightly-2026-08-24", + "force_rerun": True, + }, + ) + assert res.status_code == 200 + kwargs = job.run_job.call_args.kwargs + assert kwargs["idempotency_key"] == "nightly-2026-08-24" + assert kwargs["force_rerun"] is True + + def test_defaults_preserve_the_pre_657_call(self, tmp_path: Path) -> None: + """Omitting both must be indistinguishable from the old body.""" + job = self._job_mock() + with _client(tmp_path, job=job) as client: + client.post( + f"/jobs/{PROJECT}/run", + headers=AUTH, + json={"component_id": "keboola.ex-http", "config_id": "42"}, + ) + kwargs = job.run_job.call_args.kwargs + assert kwargs["idempotency_key"] is None + assert kwargs["force_rerun"] is False + + +class TestDescribeBatchValidatorIsShared: + """The file and inline paths must accept and reject exactly the same documents.""" + + @pytest.mark.parametrize( + "document", + [ + {"buckets": ["in.c-sales"]}, + {"tables": {"in.c-sales.orders": None}}, + {"columns": {"in.c-sales.orders": "not-a-mapping"}}, + "not-a-mapping-at-all", + ], + ) + def test_rejected_documents_raise_valueerror(self, document: Any) -> None: + from keboola_agent_cli.services._describe_batch_input import ( + parse_describe_batch_document, + ) + + with pytest.raises(ValueError): + parse_describe_batch_document(document, "request body") + + @pytest.mark.parametrize("empty", [None, {}, {"buckets": None}, {"buckets": {}}]) + def test_empty_sections_stay_a_silent_no_op(self, empty: Any) -> None: + from keboola_agent_cli.services._describe_batch_input import ( + parse_describe_batch_document, + ) + + assert parse_describe_batch_document(empty, "request body").total == 0 + + def test_source_label_names_the_body_not_a_filename(self) -> None: + from keboola_agent_cli.services._describe_batch_input import ( + parse_describe_batch_document, + ) + + with pytest.raises(ValueError, match="'request body' must be"): + parse_describe_batch_document("nope", "request body")