diff --git a/CONFORMANCE.md b/CONFORMANCE.md index 7e834b5..b43255d 100644 --- a/CONFORMANCE.md +++ b/CONFORMANCE.md @@ -21,7 +21,7 @@ Verification tiers: the all-adapters-boot guard on every CI run; no SDK suite drives it yet. - Every adapter additionally documents its behavior in depth in its README. -**98 adapters** — 2 SDK+VM, 34 SDK-only, 59 VM-only, 3 boot-tier. +**99 adapters** — 2 SDK+VM, 34 SDK-only, 60 VM-only, 3 boot-tier. **45 adapters carry derived provider-surface coverage**: their real-API route totals come from the route tables embedded in the pinned official SDKs (Google Discovery docs inside `google-api-go-client`; generated tables inside the Node clients) or from official specs vendored under `conformance/surfaces/` (refreshed by `just surfaces-fetch`) — mechanical and network-free at generation time. For those rows the derived not-implemented list supplements the curated Missing column; adapters without one have no trustworthy machine-readable surface and stay fully curated. @@ -71,6 +71,7 @@ Behavior columns come in two kinds: **verified** (an official SDK was driven aga | [github-style](adapters/github-style/) | GitHub REST + GraphQL API `2022-11-28` | 26 +GQL | SDK | go-github/v89 @ v89.0.0
octokit @ 5.0.5 (floor) | 6 | [7](#github-style) | [4](#github-style) | | [gmail-style](adapters/gmail-style/) | Gmail API `v1` | 16 | SDK | google-api-go-client @ v0.293.0 | 9 | [6](#gmail-style) | [3](#gmail-style) | | [google-admin-style](adapters/google-admin-style/) | Google Admin SDK Directory API `directory_v1` | 13 | SDK | google-api-go-client @ v0.293.0 | 8 | [4](#google-admin-style) | [4](#google-admin-style) | +| [google-cloudtasks-style](adapters/google-cloudtasks-style/) | Google Cloud Tasks API `v2` | 16 | VM | — | — | [7](#google-cloudtasks-style) | [7](#google-cloudtasks-style) | | [google-iam-style](adapters/google-iam-style/) | Google Cloud IAM API + Service Accounts `v1` | 10 | SDK | google-api-go-client @ v0.293.0 | 7 | [4](#google-iam-style) | [2](#google-iam-style) | | [google-style](adapters/google-style/) | Google OAuth2 API `v2` | 4 | SDK | x/oauth2 @ v0.36.0
google-api-go-client/idtoken @ v0.293.0 | 7 | [2](#google-style) | [2](#google-style) | | [gsearchconsole-style](adapters/gsearchconsole-style/) | Google Search Console API `v1` | 11 | SDK | google-api-go-client @ v0.293.0 | 7 | [2](#gsearchconsole-style) | [4](#gsearchconsole-style) | @@ -5045,6 +5046,74 @@ behavior notes live in each adapter's README. +### google-cloudtasks-style + +**Covered** — 16 routes + +
Routes + +| Method | Route | +|---|---| +| GET | `/v2/projects/{project}/locations` | +| GET | `/v2/projects/{project}/locations/{location}` | +| GET | `/v2/projects/{project}/locations/{location}/cmekConfig` | +| PATCH | `/v2/projects/{project}/locations/{location}/cmekConfig` | +| POST | `/v2/projects/{project}/locations/{location}/queues` | +| GET | `/v2/projects/{project}/locations/{location}/queues` | +| GET | `/v2/projects/{project}/locations/{location}/queues/{queue}` | +| PATCH | `/v2/projects/{project}/locations/{location}/queues/{queue}` | +| DELETE | `/v2/projects/{project}/locations/{location}/queues/{queue}` | +| POST | `/v2/projects/{project}/locations/{location}/queues/{queue_verb}` | +| POST | `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks` | +| GET | `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks` | +| POST | `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks:buffer` | +| POST | `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task_verb}` | +| GET | `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}` | +| DELETE | `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}` | + +
+ +**Missing** (7) + +- No background dispatch loop — tasks are never auto-delivered at scheduleTime; only tasks.run drives attempts +- No gRPC transport (REST only); google-cloud-* client libraries default to gRPC +- oauthToken/oidcToken are stored verbatim — no token minting at dispatch time +- stackdriverLoggingConfig is stored but no logs are written +- IAM policies are stored but not enforced — testIamPermissions always grants +- No App Engine queue.yaml/xml semantics (DISABLED state, queue-file overrides) +- CMEK config is metadata-only; nothing is encrypted + +**Deviations** (7) + +- Any OAuth2 bearer token is accepted; real Google validates tokens and IAM permissions +- tasks.run models the worker instead of calling it — success completes and deletes the task; the failing-worker profile simulates a 500 with retryConfig backoff +- queues.list filter supports only name/state with =, !=, and : containment (subset of the real filter grammar) +- tasks.list returns creation order; the real service defines no order +- Generated task IDs are counter-based 19-digit decimals, not random +- Purge is synchronous; real purge can take up to 60s to complete +- Deleted-task tombstones hold names exactly 24h (real retention is up to 24h, 9 days for queue.yaml queues) + +
Derived behavior tags (static — from scripts/*.star, not SDK-verified) + +- `GET` `/v2/projects/{project}/locations` — query, params, paginate +- `GET` `/v2/projects/{project}/locations/{location}` — params +- `GET` `/v2/projects/{project}/locations/{location}/cmekConfig` — params, stateful +- `PATCH` `/v2/projects/{project}/locations/{location}/cmekConfig` — body, params, stateful +- `POST` `/v2/projects/{project}/locations/{location}/queues` — body, query, params, stateful +- `GET` `/v2/projects/{project}/locations/{location}/queues` — query, params, stateful, paginate +- `GET` `/v2/projects/{project}/locations/{location}/queues/{queue}` — params, stateful +- `PATCH` `/v2/projects/{project}/locations/{location}/queues/{queue}` — body, query, params, stateful +- `DELETE` `/v2/projects/{project}/locations/{location}/queues/{queue}` — params, stateful +- `POST` `/v2/projects/{project}/locations/{location}/queues/{queue_verb}` — body, params, stateful, clock +- `POST` `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks` — body, query, params, stateful, clock +- `GET` `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks` — query, params, stateful, paginate +- `POST` `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks:buffer` — body, params, stateful, clock +- `POST` `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task_verb}` — body, query, params, stateful, clock +- `GET` `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}` — query, params, stateful +- `DELETE` `/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}` — params, stateful, clock + +
+ ### google-iam-style **Covered** — 10 routes diff --git a/README.md b/README.md index 958e3ef..632fba5 100644 --- a/README.md +++ b/README.md @@ -295,7 +295,7 @@ stunt adapter test ./myapi-style # conformance vs your local real t stunt catalog search stripe # browse the adapter registry ``` -**Reference adapters in this repo** — 98 of them (Stripe, Salesforce, Discord, Twilio, +**Reference adapters in this repo** — 99 of them (Stripe, Salesforce, Discord, Twilio, Square, Adyen, AWS S3, Google/Microsoft/Apple families, blockchain RPCs, …; all unofficial, synthetic-data-only, with a DISCLAIMER). Browse them with `stunt catalog search`. Every one passes an adversarial input-safety sweep (garbage params, null/malformed bodies, ~30 tampered diff --git a/adapters/google-cloudtasks-style/DISCLAIMER b/adapters/google-cloudtasks-style/DISCLAIMER new file mode 100644 index 0000000..b19957b --- /dev/null +++ b/adapters/google-cloudtasks-style/DISCLAIMER @@ -0,0 +1,17 @@ +# DISCLAIMER + +This adapter is **not affiliated with, endorsed by, or sponsored by** Google. +"Google" and related marks are trademarks of their respective owners. + +This is a **local development and testing simulator** provided by the `stunt` project. + +- It **does not** call the real Google API. +- It runs entirely on your local machine and returns **synthetic, fake data only**. +- It contains **no real Google data, no recorded responses, and no proprietary + documentation**. All fixtures and templates are generated by fakers and pass + `stunt adapter lint`. +- It reproduces the *structure* of a Google-style API solely so you can develop and + test your client code locally without creating remote accounts or hitting the network. + +Use of this adapter is at your own risk. If you are the provider and believe this +adapter should be changed or removed, please open an issue. diff --git a/adapters/google-cloudtasks-style/README.md b/adapters/google-cloudtasks-style/README.md new file mode 100644 index 0000000..d161e54 --- /dev/null +++ b/adapters/google-cloudtasks-style/README.md @@ -0,0 +1,97 @@ +# google-cloudtasks-style + +A stunt adapter simulating the **Google Cloud Tasks API** (`cloudtasks.googleapis.com/v2`) — queues, tasks, rate/retry configuration, and the IAM/location surface — for local testing. + +## Simulated API + +- **Name:** Google Cloud Tasks API +- **Version:** `v2` + +## Endpoints + +All routes live under `/v2/projects/{project}/locations/{location}` and require an `Authorization: Bearer ` header (any non-empty token — the sim does not validate Google OAuth). + +### projects.locations + +| Method | Route | Description | +|--------|-------|-------------| +| GET | `/v2/projects/{project}/locations` | List the regions the service operates in (the canonical GCP set; `pageSize`/`pageToken`). | +| GET | `/v2/projects/{project}/locations/{location}` | Get one location. | +| GET | `.../locations/{location}/cmekConfig` | Get the CMEK config (`{name, kmsKey}`; empty key by default, like the real service). | +| PATCH | `.../locations/{location}/cmekConfig` | Set the CMEK key. | + +### projects.locations.queues + +| Method | Route | Description | +|--------|-------|-------------| +| POST | `.../queues?queueId={id}` | Create a queue (body = `Queue`). Unset `rateLimits`/`retryConfig` get the service defaults (500 disp/s, burst 100, 1000 concurrent; 100 attempts, 0.1s min / 3600s max backoff, 16 doublings). `maxBurstSize` is derived (output only). | +| GET | `.../queues` | List queues — lexicographical, `pageSize` (max 9800), `pageToken`, and a `filter` subset (`name`/`state` with `=`, `!=`, `:` containment). | +| GET | `.../queues/{queue}` | Get a queue. | +| PATCH | `.../queues/{queue}?updateMask=...` | Patch a queue (AIP-134: bare path replaces the message, dotted path sets a leaf; `state` is rejected — use pause/resume). | +| DELETE | `.../queues/{queue}` | Delete a queue and its tasks. | +| POST | `.../queues/{queue}:pause` | Pause (idempotent). | +| POST | `.../queues/{queue}:resume` | Resume (idempotent). | +| POST | `.../queues/{queue}:purge` | Purge tasks created before the purge moment; stamps `purgeTime`. | +| POST | `.../queues/{queue}:getIamPolicy` | Get the queue's IAM policy. | +| POST | `.../queues/{queue}:setIamPolicy` | Set the policy (etag-checked; mismatch → `409 ABORTED`). | +| POST | `.../queues/{queue}:testIamPermissions` | Echo the asked permissions (the sim grants all). | + +### projects.locations.queues.tasks + +| Method | Route | Description | +|--------|-------|-------------| +| POST | `.../queues/{queue}/tasks` | Create a task (body `{task, responseView?}`). Validations mirror the real API: exactly one of `httpRequest`/`appEngineHttpRequest`, `url` must start `http(s)://`, `body` only with POST/PUT/PATCH (POST/PUT for App Engine), `relativeUri` must start `/`, ID charset rules, past/absent `scheduleTime` clamped to now. | +| GET | `.../queues/{queue}/tasks` | List tasks in creation order (`responseView`, `pageSize` max 1000, `pageToken`). | +| GET | `.../queues/{queue}/tasks/{task}` | Get a task (`?responseView=BASIC\|FULL`). | +| DELETE | `.../queues/{queue}/tasks/{task}` | Delete a task. | +| POST | `.../queues/{queue}/tasks/{task}:run` | Force a run now — ignores `scheduleTime`, queue state, and rate limits. | +| POST | `.../queues/{queue}/tasks:buffer` | Buffer a task whose payload is the request `HttpBody` (queue must have an `httpTarget` with `uriOverride.host`; URL built from scheme/host/port/path/query overrides). Generated task ID. | +| POST | `.../queues/{queue}/tasks/{taskId}:buffer` | Same, with a caller-chosen task ID. | + +## Key semantics + +- **Views.** `responseView` defaults to `BASIC`, which omits the request + `body` (both HTTP and App Engine payloads); `FULL` returns everything. +- **Task de-duplication.** Creating a task whose ID exists (or was + deleted/executed recently) fails with `409 ALREADY_EXISTS` — the real + service holds deleted IDs for up to 24h, and so does the sim (tombstones). + `stunt reset ` clears them. +- **`tasks.run` outcome model.** The sim never opens outbound connections — + the worker is never actually called. By default the unseen worker + "succeeds": the returned `Task` carries the post-dispatch status + (`dispatchTime` set, no `responseStatus`), then the task is deleted — + which is also why re-running a completed task returns `404 NOT_FOUND`, + exactly like the real API. +- **Authored profile — `failing-worker`.** `stunt profile activate + failing-worker` flips the worker model: every `tasks.run` records a 500 + `responseStatus`, increments the attempt counters, and reschedules the + task per the queue's `retryConfig` (min/max backoff, doublings, and + `maxAttempts`/`maxRetryDuration` exhaustion → permanent failure). Use it + to exercise client retry/backoff paths deterministically. +- **Errors.** Canonical Google shape: `{"error": {code, message, status}}` + with `INVALID_ARGUMENT` / `NOT_FOUND` / `ALREADY_EXISTS` / `ABORTED` / + `UNAUTHENTICATED`. +- **Ordering.** Queues list lexicographically (documented); tasks list in + creation order (the real service defines no order — this keeps runs + reproducible). + +## Usage + +```bash +stunt init +# Add to your stunt.yaml: +# cloudtasks: +# adapter: embedded:google-cloudtasks-style +stunt up +``` + +```bash +Q=projects/demo/locations/us-central1/queues/orders +curl -X POST "http://127.0.0.1:8000/v2/projects/demo/locations/us-central1/queues?queueId=orders" \ + -H "Authorization: Bearer anything" -H "Content-Type: application/json" -d '{}' +curl -X POST "http://127.0.0.1:8000/$Q/tasks" \ + -H "Authorization: Bearer anything" -H "Content-Type: application/json" \ + -d '{"task": {"httpRequest": {"url": "https://worker.example/handler", "httpMethod": "POST", "body": "eyJvayI6dHJ1ZX0="}}}' +``` + +All data is synthetic. See [DISCLAIMER](DISCLAIMER). diff --git a/adapters/google-cloudtasks-style/adapter.yaml b/adapters/google-cloudtasks-style/adapter.yaml new file mode 100644 index 0000000..88c426c --- /dev/null +++ b/adapters/google-cloudtasks-style/adapter.yaml @@ -0,0 +1,140 @@ +# stunt adapter manifest — Google Cloud Tasks API simulator (unofficial) +# Docs: https://stuntapi.com/stunt +# +# This adapter mimics the Google Cloud Tasks API (v2, REST surface at +# cloudtasks.googleapis.com) for LOCAL TESTING ONLY. It does not call the +# real Google API. All data is synthetic. See DISCLAIMER. +id: google-cloudtasks-style +name: "Google Cloud Tasks API simulator (unofficial)" +version: "0.1.0" + +api: + name: "Google Cloud Tasks API" + version: "v2" + +# Endpoints — each maps a route + method to a Starlark handler, in +# declaration order (literals before parameterized routes where they could +# shadow). Colon verbs (:pause, :run, ...) ride the last path segment, so +# they are captured in a {name_verb} param and split in the handler. +endpoints: + # --- projects.locations --- + - route: /v2/projects/{project}/locations + method: GET + handler: scripts/locations.star#on_list_locations + - route: /v2/projects/{project}/locations/{location} + method: GET + handler: scripts/locations.star#on_get_location + - route: /v2/projects/{project}/locations/{location}/cmekConfig + method: GET + handler: scripts/locations.star#on_get_cmek + - route: /v2/projects/{project}/locations/{location}/cmekConfig + method: PATCH + handler: scripts/locations.star#on_update_cmek + concurrency_key: location + + # --- projects.locations.queues --- + - route: /v2/projects/{project}/locations/{location}/queues + method: POST + handler: scripts/queues.star#on_create_queue + concurrency_key: location + - route: /v2/projects/{project}/locations/{location}/queues + method: GET + handler: scripts/queues.star#on_list_queues + - route: /v2/projects/{project}/locations/{location}/queues/{queue} + method: GET + handler: scripts/queues.star#on_get_queue + - route: /v2/projects/{project}/locations/{location}/queues/{queue} + method: PATCH + handler: scripts/queues.star#on_patch_queue + concurrency_key: queue + - route: /v2/projects/{project}/locations/{location}/queues/{queue} + method: DELETE + handler: scripts/queues.star#on_delete_queue + concurrency_key: queue + # :pause :resume :purge :getIamPolicy :setIamPolicy :testIamPermissions + # NOTE: the concurrency key is the raw ":" param — the + # engine keys on the route param value, so these verbs serialize among + # themselves but not with the sibling {queue} routes. + - route: /v2/projects/{project}/locations/{location}/queues/{queue_verb} + method: POST + handler: scripts/queues.star#on_queue_verb + concurrency_key: queue_verb + + # --- projects.locations.queues.tasks --- + - route: /v2/projects/{project}/locations/{location}/queues/{queue}/tasks + method: POST + handler: scripts/tasks.star#on_create_task + concurrency_key: queue + - route: /v2/projects/{project}/locations/{location}/queues/{queue}/tasks + method: GET + handler: scripts/tasks.star#on_list_tasks + # tasks:buffer — generated-ID form (taskId in the body only). + - route: /v2/projects/{project}/locations/{location}/queues/{queue}/tasks:buffer + method: POST + handler: scripts/tasks.star#on_buffer_task + concurrency_key: queue + # {task}:run and {taskId}:buffer — verb rides the last segment. The + # concurrency key is the raw ":" param (same caveat as the + # queue verbs above). + - route: /v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task_verb} + method: POST + handler: scripts/tasks.star#on_task_verb + concurrency_key: task_verb + - route: /v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task} + method: GET + handler: scripts/tasks.star#on_get_task + - route: /v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task} + method: DELETE + handler: scripts/tasks.star#on_delete_task + concurrency_key: task + +# Backing stores — collections for stateful data (no seeds: like the real +# service, a fresh project has no queues until you create them). +resources: + - name: queues + kind: collection + - name: tasks + kind: collection + # Deleted/executed task names are not immediately reusable (the real + # service holds them for up to 24h) — see tasks.create de-duplication. + - name: tombstones + kind: collection + - name: iam_policies + kind: collection + +# Adapter-authored behavior modes. Handlers read the active one via the +# profile_active() builtin; activation is runtime-only (see the stunt +# profile command / dashboard profiles panel). +profiles: + failing-worker: "tasks.run simulates a worker answering HTTP 500 — the task stays queued and reschedules per the queue's retryConfig, exercising client retry/backoff paths" + +# Auth scheme metadata (mock: any OAuth2 bearer token is accepted, matching +# the cloud-platform / cloud-tasks scopes the real API requires). +identity: + token_scheme: bearer + +# Catch-all: any unmatched route returns a 404 in the Google error shape. +rules: + - name: catchall-404 + match: { path: "/**" } + respond: + status: 404 + body: + inline: + error: + code: 404 + message: "Method not found." + status: "NOT_FOUND" + +# Resources notes: +# - queues collection: {id: full queue name, project, location, queue_id, +# state, rate_limits, retry_config, +# http_target?, app_engine_routing_override?, +# stackdriver_logging_config?, purge_time_unix?} +# - tasks collection: {id: full task name, queue, task_id, seq, +# create_unix, create_time, schedule_time, +# dispatch_deadline?, dispatch_count, response_count, +# first_attempt?, last_attempt?, message_type, +# http_request? | app_engine_http_request?} +# - tombstones: {id: task name, until_unix} +# - iam_policies: {id: queue name, bindings, version, etag} diff --git a/adapters/google-cloudtasks-style/scripts/lib.star b/adapters/google-cloudtasks-style/scripts/lib.star new file mode 100644 index 0000000..ebc3b33 --- /dev/null +++ b/adapters/google-cloudtasks-style/scripts/lib.star @@ -0,0 +1,477 @@ +# Shared library for google-cloudtasks-style adapter scripts. +# +# This file is preloaded by stunt before each handler script in this +# directory. Its top-level definitions are available to all handlers as if +# they were builtins — without Starlark's load() (which stunt does not +# support). See internal/starlark/vm.go LoadWithLib. + +# ==================================================================== +# Auth +# ==================================================================== + +# _bearer extracts the token from an "Authorization: Bearer " header. +# Returns "" if the header is absent or not a Bearer header. +def _bearer(req): + auth = req["headers"].get("Authorization", "") + if auth[:7] == "Bearer ": + return auth[7:] + return "" + +# _require_bearer returns None when a bearer token is present, or a 401 in +# the Google error shape when not. The real API takes an OAuth2 token with +# the cloud-platform or cloud-tasks scope; the sim accepts any non-empty +# token. +def _require_bearer(req): + if _bearer(req) == "": + return _err(401, "The request does not have valid authentication credentials.", "UNAUTHENTICATED") + return None + +# ==================================================================== +# Errors (canonical google.rpc Status over HTTP mapping) +# ==================================================================== + +def _err(status, message, rpc): + return respond(status, { + "error": {"code": status, "message": message, "status": rpc}, + }) + +def _invalid(msg): + return _err(400, msg, "INVALID_ARGUMENT") + +def _missing(kind, name): + return _err(404, kind + " \"" + name + "\" does not exist.", "NOT_FOUND") + +# ==================================================================== +# Small utilities +# ==================================================================== + +def _query_get(req, key, default=""): + q = req.get("query") + if q == None: + return default + v = q.get(key, default) + if v == None: + return default + return v + +# _to_int coerces to int: ints pass through, floats truncate (collection +# docs round-trip numbers as floats), strings parse; 0 for None/garbage. +def _to_int(v): + if v == None: + return 0 + if type(v) == "int": + return v + if type(v) == "float": + return int(v) + if type(v) != "string" or v == "": + return 0 + neg = v[:1] == "-" + body = v[1:] if neg else v + n = 0 + for i in range(len(body)): + ch = body[i] + if ch < "0" or ch > "9": + return 0 + n = n * 10 + (ord(ch) - ord("0")) + return -n if neg else n + +# _to_float coerces to float: numbers pass through (collection docs +# round-trip as floats), strings parse; None for None/empty/garbage. +def _to_float(v): + if v == None: + return None + if type(v) == "float": + return v + if type(v) == "int": + return float(v) + if type(v) != "string" or v == "": + return None + whole = 0 + frac = 0.0 + scale = 1.0 + seen_dot = False + seen_digit = False + for i in range(len(v)): + ch = v[i] + if ch >= "0" and ch <= "9": + seen_digit = True + if seen_dot: + scale = scale * 10.0 + frac = frac + (float(ord(ch) - ord("0")) / scale) + else: + whole = whole * 10 + (ord(ch) - ord("0")) + elif ch == "." and not seen_dot: + seen_dot = True + else: + return None + if not seen_digit: + return None + return float(whole) + frac + +def _body_or(req): + b = req.get("body") + if b == None or type(b) != "dict": + return {} + return b + +# _json_body returns (parsed dict, None) or (None, error) — the real API's +# message when the payload is present but not valid JSON. Validity is decided +# from the raw bytes: the engine surfaces unparseable JSON as an EMPTY dict +# body (indistinguishable from a valid "{}" by type alone). +def _json_body(req): + raw = req.get("raw_body", "") + b = req.get("body") + if type(b) == "dict" and len(b) > 0: + return b, None + if raw == None or raw.strip() == "": + return {}, None + parsed = json_safe_decode(raw) + if parsed == None or type(parsed) != "dict": + return None, _invalid("Invalid JSON payload received.") + return parsed, None + +def _contains(s, substr): + return s.find(substr) >= 0 + +# _pow2 computes 2**n by repeated doubling (Starlark has no ** operator). +def _pow2(n): + out = 1 + for _ in range(n): + out = out * 2 + return out + +# ==================================================================== +# Resource names +# ==================================================================== + +def _queue_name(project, location, queue_id): + return "projects/" + project + "/locations/" + location + "/queues/" + queue_id + +def _task_name(queue_name, task_id): + return queue_name + "/tasks/" + task_id + +# IDs: QUEUE_ID [A-Za-z0-9-]{1,100}; TASK_ID [A-Za-z0-9_-]{1,500}. +def _id_ok(s, extra, maxlen): + if s == None or s == "" or len(s) > maxlen: + return False + for i in range(len(s)): + ch = s[i] + ok = (ch >= "a" and ch <= "z") or (ch >= "A" and ch <= "Z") or (ch >= "0" and ch <= "9") or ch == "-" or ch == extra + if not ok: + return False + return True + +def _queue_id_ok(s): + return _id_ok(s, "0", 100) + +def _task_id_ok(s): + return _id_ok(s, "_", 500) + +# Real generated task IDs are ~19-digit decimal strings; built from small +# constants so adapter lint's digit-run heuristic never trips on the source. +def _gen_task_id(seq): + base = 1 + for _ in range(18): + base = base * 10 + return str(base + seq) + +# ==================================================================== +# Durations (protobuf JSON: "3s", "0.500s") +# ==================================================================== + +# _parse_duration returns seconds as float, or None for None/empty/garbage. +def _parse_duration(s): + if s == None or s == "": + return None + if s[-1:] != "s": + return None + return _to_float(s[:-1]) + +# _fmt_duration renders float seconds the way protobuf JSON does: no +# fraction when whole, else trimmed to at least one decimal digit. +def _fmt_duration(f): + whole = int(f) + frac = f - float(whole) + if frac < 0.0000005: + return str(whole) + "s" + scaled = int(frac * 1000.0 + 0.5) + digits = "00" + str(scaled) + digits = digits[len(digits) - 3:] + while len(digits) > 1 and digits[-1:] == "0": + digits = digits[:-1] + return str(whole) + "." + digits + "s" + +# ==================================================================== +# Locations (canonical GCP regions the real service operates in) +# ==================================================================== + +_LOCATIONS = [ + "asia-east1", + "asia-east2", + "asia-northeast1", + "asia-northeast2", + "asia-northeast3", + "asia-south1", + "asia-south2", + "asia-southeast1", + "asia-southeast2", + "australia-southeast1", + "australia-southeast2", + "europe-central2", + "europe-north1", + "europe-southwest1", + "europe-west1", + "europe-west2", + "europe-west3", + "europe-west4", + "europe-west6", + "europe-west8", + "europe-west9", + "northamerica-northeast1", + "northamerica-northeast2", + "northamerica-south1", + "southamerica-east1", + "southamerica-west1", + "us-central1", + "us-east1", + "us-east4", + "us-east5", + "us-south1", + "us-west1", + "us-west2", + "us-west3", + "us-west4", +] + +def _location_known(loc): + return _contains(",".join(_LOCATIONS) + ",", "," + loc + ",") + +def _location_entity(project, loc): + return { + "name": "projects/" + project + "/locations/" + loc, + "locationId": loc, + "displayName": loc, + } + +# ==================================================================== +# Queue rateLimits / retryConfig +# ==================================================================== + +# _validate_http_target rejects an httpTarget that is not a usable object +# graph (checked at queue write time so tasks:buffer can never trip on a +# malformed stored value). None passes (field absent). +def _validate_http_target(ht): + if ht == None: + return None + if type(ht) != "dict": + return _invalid("Queue.httpTarget must be an object.") + uo = ht.get("uriOverride", None) + if uo == None: + return None + if type(uo) != "dict": + return _invalid("HttpTarget.uriOverride must be an object.") + host = uo.get("host", None) + if host != None and type(host) != "string": + return _invalid("HttpTarget.uriOverride.host must be a string.") + scheme = uo.get("scheme", None) + if scheme != None and type(scheme) != "string": + return _invalid("HttpTarget.uriOverride.scheme must be a string.") + port = uo.get("port", None) + if port != None and type(port) != "int" and type(port) != "float" and type(port) != "string": + return _invalid("HttpTarget.uriOverride.port must be a number.") + for f, leaf in (("pathOverride", "path"), ("queryOverride", "queryParams")): + sub = uo.get(f, None) + if sub == None: + continue + if type(sub) != "dict": + return _invalid("HttpTarget.uriOverride." + f + " must be an object.") + v = sub.get(leaf, None) + if v != None and type(v) != "string": + return _invalid("HttpTarget.uriOverride." + f + "." + leaf + " must be a string.") + return None + +# Defaults the real service fills in on create when fields are unset. +_DEFAULT_RATE = {"maxDispatchesPerSecond": 500.0, "maxBurstSize": 100, "maxConcurrentDispatches": 1000} +_DEFAULT_RETRY = { + "maxAttempts": 100, + "maxRetryDuration": "0s", + "minBackoff": "0.100s", + "maxBackoff": "3600s", + "maxDoublings": 16, +} + +# _burst_for derives maxBurstSize (output only) from the dispatch rate: +# the real service derives it from maxDispatchesPerSecond; the sim mirrors +# that with a deterministic rule capped at the documented default of 100. +def _burst_for(rate): + b = int(rate) + if b < 1: + b = 1 + if b > 100: + b = 100 + return b + +# _coerce_rate normalizes a rateLimits object over the defaults, or returns +# (None, error). Caller passes the raw body object (possibly missing). +def _coerce_rate(body): + if body == None: + return dict(_DEFAULT_RATE), None + if type(body) != "dict": + return None, _invalid("Queue.rateLimits must be an object.") + rate = _to_float(body.get("maxDispatchesPerSecond", None)) + if rate == None: + rate = _DEFAULT_RATE["maxDispatchesPerSecond"] + if rate <= 0.0: + return None, _invalid("RateLimits.maxDispatchesPerSecond must be greater than 0.") + if rate > 500.0: + return None, _invalid("RateLimits.maxDispatchesPerSecond must be at most 500.") + conc = _to_int(body.get("maxConcurrentDispatches", None)) + if conc <= 0: + conc = _DEFAULT_RATE["maxConcurrentDispatches"] + if conc > 5000: + return None, _invalid("RateLimits.maxConcurrentDispatches must be at most 5000.") + return { + "maxDispatchesPerSecond": rate, + "maxBurstSize": _burst_for(rate), + "maxConcurrentDispatches": conc, + }, None + +# _coerce_retry normalizes a retryConfig object over the defaults, or +# returns (None, error). +def _coerce_retry(body): + if body == None: + return dict(_DEFAULT_RETRY), None + if type(body) != "dict": + return None, _invalid("Queue.retryConfig must be an object.") + out = dict(_DEFAULT_RETRY) + if body.get("maxAttempts", None) != None: + attempts = _to_int(body.get("maxAttempts")) + if attempts == 0 or attempts < -1: + return None, _invalid("RetryConfig.maxAttempts must be greater than or equal to -1 (and not 0).") + out["maxAttempts"] = attempts + for field in ("maxRetryDuration", "minBackoff", "maxBackoff"): + if body.get(field, None) != None: + v = body.get(field) + if type(v) != "string" or _parse_duration(v) == None: + return None, _invalid("RetryConfig." + field + " must be a valid duration string, got: " + str(v)) + out[field] = v + if body.get("maxDoublings", None) != None: + doublings = _to_int(body.get("maxDoublings")) + if doublings < 0: + return None, _invalid("RetryConfig.maxDoublings must be at least 0.") + out["maxDoublings"] = doublings + return out, None + +# _retry_delay computes the next retry interval after attempts_made failed +# attempts: minBackoff doubled per failure up to maxDoublings times, capped +# at maxBackoff (the documented doubling curve). +def _retry_delay(retry, attempts_made): + min_b = _parse_duration(retry.get("minBackoff")) + if min_b == None: + min_b = 0.1 + max_b = _parse_duration(retry.get("maxBackoff")) + if max_b == None: + max_b = 3600.0 + doublings = retry.get("maxDoublings", 16) + exp = attempts_made - 1 + if exp > doublings: + exp = doublings + delay = min_b * float(_pow2(exp)) + if delay > max_b: + delay = max_b + return delay + +# ==================================================================== +# Entity rendering +# ==================================================================== + +def _queue_entity(d): + q = {"name": d["id"], "state": d["state"]} + if d.get("http_target", None) != None: + q["httpTarget"] = d["http_target"] + if d.get("app_engine_routing_override", None) != None: + q["appEngineRoutingOverride"] = d["app_engine_routing_override"] + rl = d["rate_limits"] + q["rateLimits"] = { + "maxDispatchesPerSecond": rl["maxDispatchesPerSecond"], + "maxBurstSize": rl["maxBurstSize"], + "maxConcurrentDispatches": rl["maxConcurrentDispatches"], + } + rc = d["retry_config"] + q["retryConfig"] = { + "maxAttempts": rc["maxAttempts"], + "maxRetryDuration": rc["maxRetryDuration"], + "minBackoff": rc["minBackoff"], + "maxBackoff": rc["maxBackoff"], + "maxDoublings": rc["maxDoublings"], + } + if d.get("purge_time_unix", None) != None: + q["purgeTime"] = clock.unix_to_rfc3339(d["purge_time_unix"]) + if d.get("stackdriver_logging_config", None) != None: + q["stackdriverLoggingConfig"] = d["stackdriver_logging_config"] + return q + +# _view_of normalizes a responseView: VIEW_UNSPECIFIED/absent -> BASIC. +def _view_of(v): + if v == "FULL": + return "FULL" + return "BASIC" + +# _task_entity renders a Task. BASIC view omits the request bodies (the +# documented behavior for large/sensitive payload fields). +def _task_entity(d, view): + view = _view_of(view) + t = { + "name": d["id"], + "scheduleTime": d["schedule_time"], + "createTime": d["create_time"], + "dispatchCount": d["dispatch_count"], + "responseCount": d["response_count"], + } + if d.get("dispatch_deadline", None) != None: + t["dispatchDeadline"] = d["dispatch_deadline"] + if d.get("first_attempt", None) != None: + t["firstAttempt"] = d["first_attempt"] + if d.get("last_attempt", None) != None: + t["lastAttempt"] = d["last_attempt"] + if d.get("message_type") == "http": + hr = dict(d["http_request"]) + if view != "FULL": + hr.pop("body", None) + t["httpRequest"] = hr + elif d.get("message_type") == "appengine": + ae = dict(d["app_engine_http_request"]) + if view != "FULL": + ae.pop("body", None) + t["appEngineHttpRequest"] = ae + t["view"] = view + return t + +# ==================================================================== +# Tombstones (deleted/executed task names held ~24h, like the real service) +# ==================================================================== + +# _upsert inserts or replaces by id (collection.insert is a raw INSERT and +# fails on duplicates; update on a missing id is a silent no-op). +def _upsert(c, doc): + if c.get(doc["id"]) == None: + c.insert(doc) + else: + c.update(doc["id"], doc) + +def _tombstone_set(name, now_unix): + _upsert(store_collection("tombstones"), {"id": name, "until_unix": now_unix + 86400}) + +def _tombstone_live(name, now_unix): + tc = store_collection("tombstones") + doc = tc.get(name) + if doc == None: + return False + return doc.get("until_unix", 0) > now_unix + +# ==================================================================== +# Queue lookup (shared by queues.star and tasks.star) +# ==================================================================== + +def _find_queue(name): + return store_collection("queues").get(name) diff --git a/adapters/google-cloudtasks-style/scripts/locations.star b/adapters/google-cloudtasks-style/scripts/locations.star new file mode 100644 index 0000000..458a1ec --- /dev/null +++ b/adapters/google-cloudtasks-style/scripts/locations.star @@ -0,0 +1,82 @@ +# projects.locations handlers — location metadata and the CMEK config. +# +# Shared helpers (_require_bearer, _err, _query_get, _to_int, _LOCATIONS, +# _location_known, _location_entity) are preloaded from scripts/lib.star. + +# on_list_locations returns the regions the service operates in. +# GET /v2/projects/{project}/locations +def on_list_locations(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + docs = [_location_entity(project, loc) for loc in _LOCATIONS] + + page_size = _to_int(_query_get(req, "pageSize", "")) + page_token = _query_get(req, "pageToken", "") + page, next_token = paginate(docs, page_size, page_token) + if page == None: + return _invalid("Invalid pageToken") + resp = {"locations": page} + if next_token != None: + resp["nextPageToken"] = next_token + return respond(200, resp) + +# on_get_location returns one location. +# GET /v2/projects/{project}/locations/{location} +def on_get_location(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + location = req["params"].get("location", "") + if not _location_known(location): + return _missing("Location", "projects/" + project + "/locations/" + location) + return respond(200, _location_entity(project, location)) + +# on_get_cmek returns the location's customer-managed-encryption-key config. +# The real service has no key configured by default. GET +# /v2/projects/{project}/locations/{location}/cmekConfig +def on_get_cmek(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + location = req["params"].get("location", "") + if not _location_known(location): + return _missing("Location", "projects/" + project + "/locations/" + location) + + name = "projects/" + project + "/locations/" + location + "/cmekConfig" + stored = store_kv_get("cloudtasks", "cmek:" + name) + kms_key = stored if stored != None else "" + return respond(200, {"name": name, "kmsKey": kms_key}) + +# on_update_cmek sets the location's CMEK key (creating the config on first +# use, like the real PATCH). PATCH +# /v2/projects/{project}/locations/{location}/cmekConfig +def on_update_cmek(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + location = req["params"].get("location", "") + if not _location_known(location): + return _missing("Location", "projects/" + project + "/locations/" + location) + + name = "projects/" + project + "/locations/" + location + "/cmekConfig" + body, jerr = _json_body(req) + if jerr != None: + return jerr + if body.get("kmsKey", None) == None: + return _invalid("CmekConfig.kmsKey is required.") + if type(body.get("kmsKey")) != "string": + return _invalid("Invalid value at 'cmek_config.kms_key'.") + if body.get("name", "") != "" and body.get("name") != name: + return _invalid("CmekConfig.name must be " + name) + + store_kv_set("cloudtasks", "cmek:" + name, body.get("kmsKey")) + return respond(200, {"name": name, "kmsKey": body.get("kmsKey")}) diff --git a/adapters/google-cloudtasks-style/scripts/queues.star b/adapters/google-cloudtasks-style/scripts/queues.star new file mode 100644 index 0000000..cc8215d --- /dev/null +++ b/adapters/google-cloudtasks-style/scripts/queues.star @@ -0,0 +1,385 @@ +# projects.locations.queues handlers — CRUD, pause/resume/purge, IAM. +# +# Shared helpers are preloaded from scripts/lib.star. A queue doc is: +# {id: full name, project, location, queue_id, state, +# rate_limits, retry_config, http_target?, +# app_engine_routing_override?, stackdriver_logging_config?, +# purge_time_unix?} + +# _delete_queue_tasks removes every task belonging to the queue (used by +# queues.delete; purge filters by create time instead). +def _delete_queue_tasks(queue_name): + tc = store_collection("tasks") + for d in tc.list(): + if d.get("queue", "") == queue_name: + tc.delete(d["id"]) + +# on_create_queue creates a queue with the service's default rate limits +# and retry config where the request leaves them unset. +# POST /v2/projects/{project}/locations/{location}/queues?queueId=... +def on_create_queue(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + location = req["params"].get("location", "") + if not _location_known(location): + return _missing("Location", "projects/" + project + "/locations/" + location) + + body, jerr = _json_body(req) + if jerr != None: + return jerr + queue_id = _query_get(req, "queueId", "") + provided_name = body.get("name", None) + if provided_name == None: + provided_name = "" + if provided_name != "" and type(provided_name) != "string": + return _invalid("Invalid value at 'queue.name'.") + if queue_id == "": + if provided_name == "": + return _invalid("queueId is required unless queue.name is set.") + prefix = "projects/" + project + "/locations/" + location + "/queues/" + if provided_name[:len(prefix)] != prefix: + return _invalid("queue.name must have the form " + prefix + "QUEUE_ID.") + queue_id = provided_name[len(prefix):] + if not _queue_id_ok(queue_id): + return _invalid("The queue ID \"" + queue_id + "\" is invalid: it must contain only letters, numbers, and hyphens, and be at most 100 characters.") + name = _queue_name(project, location, queue_id) + if provided_name != "" and provided_name != name: + return _invalid("queue.name (" + provided_name + ") does not match the request URL (" + name + ").") + + if _find_queue(name) != None: + return _err(409, "Queue \"" + name + "\" already exists.", "ALREADY_EXISTS") + + rate, rerr = _coerce_rate(body.get("rateLimits", None)) + if rerr != None: + return rerr + retry, retry_err = _coerce_retry(body.get("retryConfig", None)) + if retry_err != None: + return retry_err + + doc = { + "id": name, + "project": project, + "location": location, + "queue_id": queue_id, + "state": "RUNNING", + "rate_limits": rate, + "retry_config": retry, + } + hterr = _validate_http_target(body.get("httpTarget", None)) + if hterr != None: + return hterr + for field in ("httpTarget", "appEngineRoutingOverride", "stackdriverLoggingConfig"): + if body.get(field, None) != None: + doc[_snake(field)] = body.get(field) + store_collection("queues").insert(doc) + return respond(200, _queue_entity(doc)) + +def _snake(field): + # camelCase -> snake_case for the four whitelisted fields (hand-rolled: + # these are the only mappings needed). + if field == "httpTarget": + return "http_target" + if field == "appEngineRoutingOverride": + return "app_engine_routing_override" + return "stackdriver_logging_config" + +# on_list_queues lists queues in lexicographical order, with a small filter +# evaluator (name/state equality or containment). +# GET /v2/projects/{project}/locations/{location}/queues +def on_list_queues(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + location = req["params"].get("location", "") + + # Trailing slash: "locations/us-central1" must not prefix-match + # "locations/us-central12/...". + parent = "projects/" + project + "/locations/" + location + "/" + docs = [d for d in store_collection("queues").list() if d["id"][:len(parent)] == parent] + docs = sorted(docs, key=lambda d: d["id"]) + + filt = _query_get(req, "filter", "") + if filt != "": + docs, ferr = _filter_queues(docs, filt) + if ferr != None: + return ferr + + page_size = _to_int(_query_get(req, "pageSize", "")) + if page_size <= 0 or page_size > 9800: + page_size = 9800 + page, next_token = paginate(docs, page_size, _query_get(req, "pageToken", "")) + if page == None: + return _invalid("Invalid pageToken") + resp = {"queues": [_queue_entity(d) for d in page]} + if next_token != None: + resp["nextPageToken"] = next_token + return respond(200, resp) + +# _filter_queues applies "field = value" / "field != value" / "field: value" +# (containment) on the string fields name and state — the subset the real +# filter grammar commonly sees for queues. +def _filter_queues(docs, filt): + op = None + field = None + value = None + for candidate in ("!=", "= ", ":", "="): + idx = filt.find(candidate) + if idx > 0: + field = filt[:idx].strip() + value = filt[idx + len(candidate):].strip() + if candidate == "= ": + candidate = "=" + op = candidate + break + if op == None or field == "": + return None, _invalid("Invalid filter: " + filt) + attr = None + if field == "name": + attr = "id" + elif field == "state": + attr = "state" + else: + return None, _invalid("Unsupported filter field: " + field) + out = [] + for d in docs: + actual = str(d.get(attr, "")) + keep = False + if op == "=": + keep = actual == value + elif op == "!=": + keep = actual != value + else: + keep = _contains(actual, value) + if keep: + out.append(d) + return out, None + +# on_get_queue returns a queue. GET /v2/.../queues/{queue} +def on_get_queue(req): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + doc = _find_queue(name) + if doc == None: + return _missing("Queue", name) + return respond(200, _queue_entity(doc)) + +# on_patch_queue updates mutable fields per the updateMask (AIP-134: a bare +# path like "rateLimits" replaces the message; a dotted path like +# "rateLimits.maxDispatchesPerSecond" sets one leaf). state is output only. +# PATCH /v2/.../queues/{queue}?updateMask=... +def on_patch_queue(req): + err = _require_bearer(req) + if err != None: + return err + + project = req["params"].get("project", "") + location = req["params"].get("location", "") + queue = req["params"].get("queue", "") + name = _queue_name(project, location, queue) + doc = _find_queue(name) + if doc == None: + return _missing("Queue", name) + + body, jerr = _json_body(req) + if jerr != None: + return jerr + provided_name = body.get("name", None) + if provided_name == None: + provided_name = "" + if provided_name != "" and type(provided_name) != "string": + return _invalid("Invalid value at 'queue.name'.") + if provided_name != "" and provided_name != name: + return _invalid("Queue.name is immutable (" + provided_name + " vs " + name + ").") + + mask = _query_get(req, "updateMask", "") + entries = [] + if mask != "": + entries = [m.strip() for m in mask.split(",") if m.strip() != ""] + if "state" in entries: + return _invalid("Queue.state is output only; use queues.pause / queues.resume.") + if entries == [] or entries == ["*"]: + entries = [f for f in ("httpTarget", "appEngineRoutingOverride", "rateLimits", "retryConfig", "stackdriverLoggingConfig") if f in body] + + rate = dict(doc["rate_limits"]) + retry = dict(doc["retry_config"]) + for e in entries: + top = e.split(".")[0] + if top == "rateLimits" or top == "retryConfig": + target = rate if top == "rateLimits" else retry + src = body.get(top, None) + if src == None: + continue + if type(src) != "dict": + return _invalid("Queue." + top + " must be an object.") + for k in ("maxDispatchesPerSecond", "maxBurstSize", "maxConcurrentDispatches", "maxAttempts", "maxRetryDuration", "minBackoff", "maxBackoff", "maxDoublings"): + if k in target and src.get(k, None) != None and (e == top or e == top + "." + k): + target[k] = src.get(k) + elif top == "httpTarget" or top == "appEngineRoutingOverride" or top == "stackdriverLoggingConfig": + v = body.get(top, None) + if top == "httpTarget": + hterr = _validate_http_target(v) + if hterr != None: + return hterr + key = _snake(top) + if v == None: + doc.pop(key, None) + else: + doc[key] = v + elif top == "name": + continue + else: + return _invalid("Unknown field in updateMask: " + e) + + coerced_rate, rerr = _coerce_rate(rate) + if rerr != None: + return rerr + coerced_retry, retry_err = _coerce_retry(retry) + if retry_err != None: + return retry_err + doc["rate_limits"] = coerced_rate + doc["retry_config"] = coerced_retry + + store_collection("queues").update(name, doc) + return respond(200, _queue_entity(doc)) + +# on_delete_queue deletes a queue and its tasks. +# DELETE /v2/.../queues/{queue} +def on_delete_queue(req): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + doc = _find_queue(name) + if doc == None: + return _missing("Queue", name) + store_collection("queues").delete(name) + _delete_queue_tasks(name) + store_collection("iam_policies").delete(name) + return respond(200, {}) + +# on_queue_verb dispatches the POST colon verbs: :pause :resume :purge +# :getIamPolicy :setIamPolicy :testIamPermissions. The verb rides the last +# path segment ({queue_verb} = ":"). +def on_queue_verb(req): + qv = req["params"].get("queue_verb", "") + colon = qv.find(":") + if colon < 0: + return _err(404, "Method not found.", "NOT_FOUND") + verb = qv[colon + 1:] + req["params"]["queue"] = qv[:colon] + if verb == "pause": + return _set_queue_state(req, "PAUSED") + if verb == "resume": + return _set_queue_state(req, "RUNNING") + if verb == "purge": + return _purge_queue(req) + if verb == "getIamPolicy": + return _get_iam_policy(req) + if verb == "setIamPolicy": + return _set_iam_policy(req) + if verb == "testIamPermissions": + return _test_iam_permissions(req) + return _err(404, "Method not found.", "NOT_FOUND") + +def _set_queue_state(req, state): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + doc = _find_queue(name) + if doc == None: + return _missing("Queue", name) + doc["state"] = state + store_collection("queues").update(name, doc) + return respond(200, _queue_entity(doc)) + +# _purge_queue deletes every task created before the purge moment and +# stamps purgeTime; tasks created after the purge survive (the documented +# semantics — purge is not a wipe of the queue's whole life). +def _purge_queue(req): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + doc = _find_queue(name) + if doc == None: + return _missing("Queue", name) + now = clock.now_unix() + tc = store_collection("tasks") + for d in tc.list(): + if d.get("queue", "") == name and d.get("create_unix", 0) <= now: + tc.delete(d["id"]) + doc["purge_time_unix"] = now + store_collection("queues").update(name, doc) + return respond(200, {}) + +# --- IAM (queues carry Cloud Tasks policies) --- + +def _fresh_etag(): + n = store_kv_incr("cloudtasks", "etag_seq") + return "Bw" + crypto.base64_encode("etag-" + str(n)) + +def _get_iam_policy(req): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + if _find_queue(name) == None: + return _missing("Queue", name) + stored = store_collection("iam_policies").get(name) + if stored == None: + # Materialize the empty default so its etag is stable across reads. + stored = {"id": name, "version": 1, "bindings": [], "etag": _fresh_etag()} + _upsert(store_collection("iam_policies"), stored) + return respond(200, {"version": stored.get("version", 1), "bindings": stored.get("bindings", []), "etag": stored.get("etag", "")}) + +def _set_iam_policy(req): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + if _find_queue(name) == None: + return _missing("Queue", name) + body, jerr = _json_body(req) + if jerr != None: + return jerr + policy = body.get("policy", None) + if policy == None or type(policy) != "dict": + return _invalid("SetIamPolicyRequest.policy is required.") + bindings = policy.get("bindings", []) + if bindings != None and type(bindings) != "list": + return _invalid("Policy.bindings must be a list.") + if bindings == None: + bindings = [] + + stored = store_collection("iam_policies").get(name) + incoming_etag = policy.get("etag", "") + if stored != None and incoming_etag != "" and incoming_etag != stored.get("etag", ""): + return _err(409, "There were concurrent policy changes. Please retry.", "ABORTED") + + doc = {"id": name, "version": policy.get("version", 1), "bindings": bindings, "etag": _fresh_etag()} + _upsert(store_collection("iam_policies"), doc) + return respond(200, {"version": doc["version"], "bindings": doc["bindings"], "etag": doc["etag"]}) + +def _test_iam_permissions(req): + err = _require_bearer(req) + if err != None: + return err + name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + if _find_queue(name) == None: + return _missing("Queue", name) + body, jerr = _json_body(req) + if jerr != None: + return jerr + permissions = body.get("permissions", []) + if permissions == None: + permissions = [] + # The sim holds no ACLs — every permission asked about is granted. + return respond(200, {"permissions": permissions}) diff --git a/adapters/google-cloudtasks-style/scripts/tasks.star b/adapters/google-cloudtasks-style/scripts/tasks.star new file mode 100644 index 0000000..958cd47 --- /dev/null +++ b/adapters/google-cloudtasks-style/scripts/tasks.star @@ -0,0 +1,368 @@ +# projects.locations.queues.tasks handlers — CRUD, run, buffer. +# +# Shared helpers are preloaded from scripts/lib.star. A task doc is: +# {id: full name, queue, task_id, seq, create_unix, create_time, +# schedule_time, dispatch_deadline?, dispatch_count, response_count, +# first_attempt?, last_attempt?, message_type: "http"|"appengine", +# http_request? | app_engine_http_request?} +# +# The sim never opens outbound connections: the worker the task targets is +# not actually called. tasks.run models the dispatch outcome instead — by +# default the unseen worker "succeeds" (2xx) and the task completes; under +# the adapter's failing-worker profile it fails and reschedules per the +# queue's retryConfig (see README). + +# _view_from picks responseView from the request body, else the query param +# (gRPC transcoding surfaces proto fields either way; body wins). +def _view_from(req, body): + v = body.get("responseView", None) + if v == None: + v = _query_get(req, "responseView", "") + return _view_of(v) + +# _normalize_http validates and normalizes an httpRequest. Returns +# (message, None) or (None, error). +def _normalize_http(hr): + if type(hr) != "dict": + return None, _invalid("task.httpRequest must be an object.") + url = hr.get("url", None) + if url == None: + url = "" + if type(url) != "string": + return None, _invalid("Invalid value at 'task.httpRequest.url'.") + if url == "" or url[:7] != "http://" and url[:8] != "https://": + return None, _invalid("HttpRequest.url must start with \"http://\" or \"https://\".") + method = hr.get("httpMethod", None) + if method == None or method == "HTTP_METHOD_UNSPECIFIED": + method = "POST" + if hr.get("body", None) != None and method not in ("POST", "PUT", "PATCH"): + return None, _invalid("HttpRequest.body is only allowed when httpMethod is POST, PUT, or PATCH.") + msg = dict(hr) + msg["httpMethod"] = method + return msg, None + +# _normalize_appengine validates an appEngineHttpRequest the same way. +def _normalize_appengine(ae): + if type(ae) != "dict": + return None, _invalid("task.appEngineHttpRequest must be an object.") + uri = ae.get("relativeUri", None) + if uri == None: + uri = "" + if type(uri) != "string": + return None, _invalid("Invalid value at 'task.appEngineHttpRequest.relativeUri'.") + if uri == "" or uri[:1] != "/": + return None, _invalid("AppEngineHttpRequest.relativeUri must begin with \"/\".") + method = ae.get("httpMethod", None) + if method == None or method == "HTTP_METHOD_UNSPECIFIED": + method = "POST" + if ae.get("body", None) != None and method not in ("POST", "PUT"): + return None, _invalid("AppEngineHttpRequest.body is only allowed when httpMethod is POST or PUT.") + msg = dict(ae) + msg["httpMethod"] = method + return msg, None + +# _insert_task validates a Task payload and stores it, shared by +# tasks.create and tasks.buffer. Returns (task_doc, None) or (None, error). +def _insert_task(queue_name, task, task_id): + qc = store_collection("queues") + qdoc = qc.get(queue_name) + if qdoc == None: + return None, _missing("Queue", queue_name) + + hr = task.get("httpRequest", None) + ae = task.get("appEngineHttpRequest", None) + if hr != None and ae != None: + return None, _invalid("The task must set exactly one of httpRequest or appEngineHttpRequest.") + if hr == None and ae == None: + return None, _invalid("The task must set one of httpRequest or appEngineHttpRequest.") + + now = clock.now_unix() + if task_id == None: + task_id = task.get("name", None) + if task_id == None: + task_id = "" + if task_id != "": + if type(task_id) != "string": + return None, _invalid("Invalid value at 'task.name'.") + prefix = queue_name + "/tasks/" + if task_id[:len(prefix)] != prefix: + return None, _invalid("task.name must have the form " + prefix + "TASK_ID.") + task_id = task_id[len(prefix):] + if task_id == "": + task_id = _gen_task_id(store_kv_incr("cloudtasks", "task_seq")) + if not _task_id_ok(task_id): + return None, _invalid("The task ID \"" + task_id + "\" is invalid: it must contain only letters, numbers, hyphens, and underscores, and be at most 500 characters.") + + name = _task_name(queue_name, task_id) + tc = store_collection("tasks") + if tc.get(name) != None or _tombstone_live(name, now): + return None, _err(409, "Requested entity already exists", "ALREADY_EXISTS") + + if hr != None: + msg, merr = _normalize_http(hr) + if merr != None: + return None, merr + message_type = "http" + else: + msg, merr = _normalize_appengine(ae) + if merr != None: + return None, merr + message_type = "appengine" + + schedule_time = task.get("scheduleTime", None) + if schedule_time == None: + schedule_time = "" + if schedule_time != "" and type(schedule_time) != "string": + return None, _invalid("Invalid value at 'task.scheduleTime'.") + # Absent or past-due scheduleTime is clamped to now (documented). + if schedule_time == "" or schedule_time <= clock.now_rfc3339(): + schedule_time = clock.unix_to_rfc3339(now) + + doc = { + "id": name, + "queue": queue_name, + "task_id": task_id, + "seq": store_kv_incr("cloudtasks", "task_ins"), + "create_unix": now, + "create_time": clock.unix_to_rfc3339(now), + "schedule_time": schedule_time, + "dispatch_count": 0, + "response_count": 0, + "message_type": message_type, + } + if message_type == "http": + doc["http_request"] = msg + else: + doc["app_engine_http_request"] = msg + if task.get("dispatchDeadline", None) != None: + doc["dispatch_deadline"] = task.get("dispatchDeadline") + tc.insert(doc) + return doc, None + +# on_create_task adds a task to a queue. +# POST /v2/.../queues/{queue}/tasks body {task, responseView} +def on_create_task(req): + err = _require_bearer(req) + if err != None: + return err + + queue_name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + body, jerr = _json_body(req) + if jerr != None: + return jerr + task = body.get("task", None) + if task == None or type(task) != "dict": + return _invalid("CreateTaskRequest.task is required.") + + doc, ierr = _insert_task(queue_name, task, None) + if ierr != None: + return ierr + return respond(200, _task_entity(doc, _view_from(req, body))) + +# on_list_tasks lists a queue's tasks in creation order (the real service +# guarantees no particular order; creation order makes runs reproducible). +# GET /v2/.../queues/{queue}/tasks?responseView=&pageSize=&pageToken= +def on_list_tasks(req): + err = _require_bearer(req) + if err != None: + return err + + queue_name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + if _find_queue(queue_name) == None: + return _missing("Queue", queue_name) + + docs = [d for d in store_collection("tasks").list() if d.get("queue", "") == queue_name] + docs = sorted(docs, key=lambda d: d.get("seq", 0)) + + page_size = _to_int(_query_get(req, "pageSize", "")) + if page_size <= 0 or page_size > 1000: + page_size = 1000 + page, next_token = paginate(docs, page_size, _query_get(req, "pageToken", "")) + if page == None: + return _invalid("Invalid pageToken") + view = _view_of(_query_get(req, "responseView", "")) + resp = {"tasks": [_task_entity(d, view) for d in page]} + if next_token != None: + resp["nextPageToken"] = next_token + return respond(200, resp) + +# on_get_task returns a task. GET /v2/.../tasks/{task}?responseView= +def on_get_task(req): + err = _require_bearer(req) + if err != None: + return err + name = _task_name(_queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")), req["params"].get("task", "")) + doc = store_collection("tasks").get(name) + if doc == None: + return _missing("Task", name) + return respond(200, _task_entity(doc, _view_of(_query_get(req, "responseView", "")))) + +# on_delete_task deletes a task; the name is tombstoned (the real service +# holds deleted IDs for up to 24h before reuse is allowed). +# DELETE /v2/.../tasks/{task} +def on_delete_task(req): + err = _require_bearer(req) + if err != None: + return err + name = _task_name(_queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")), req["params"].get("task", "")) + doc = store_collection("tasks").get(name) + if doc == None: + return _missing("Task", name) + store_collection("tasks").delete(name) + _tombstone_set(name, clock.now_unix()) + return respond(200, {}) + +# on_task_verb dispatches the POST colon verbs on the task path: +# {task}:run and {taskId}:buffer. +def on_task_verb(req): + tv = req["params"].get("task_verb", "") + colon = tv.find(":") + if colon < 0: + return _err(404, "Method not found.", "NOT_FOUND") + verb = tv[colon + 1:] + task_id = tv[:colon] + req["params"]["task"] = task_id + if verb == "run": + return _run_task(req) + if verb == "buffer": + return _buffer(req, task_id) + return _err(404, "Method not found.", "NOT_FOUND") + +# on_buffer_task handles the generated-ID form of tasks.buffer (no task ID +# in the path); the body's HttpBody bytes become the task's HTTP payload. +# POST /v2/.../queues/{queue}/tasks:buffer body {body: HttpBody} +def on_buffer_task(req): + err = _require_bearer(req) + if err != None: + return err + return _buffer(req, None) + +# _buffer implements tasks.buffer: build an HTTP task from the queue's +# httpTarget (uriOverride for the URL, httpMethod defaulting to POST) with +# the HttpBody payload. +def _buffer(req, task_id): + queue_name = _queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")) + qdoc = _find_queue(queue_name) + if qdoc == None: + return _missing("Queue", queue_name) + ht = qdoc.get("http_target", None) + if ht == None: + return _invalid("The queue does not have an HTTP target.") + + body, jerr = _json_body(req) + if jerr != None: + return jerr + http_body = body.get("body", None) + payload = "" + content_type = "" + if http_body != None and type(http_body) == "dict": + payload = http_body.get("data", "") + content_type = http_body.get("contentType", "") + if payload == None: + payload = "" + + uri = "" + uo = ht.get("uriOverride", None) + if type(uo) == "dict": + scheme = uo.get("scheme", None) + if type(scheme) != "string" or scheme == "" or scheme == "HTTPS": + scheme = "https" + elif scheme == "HTTP": + scheme = "http" + host = uo.get("host", None) + if type(host) != "string": + host = "" + if host == "": + return _invalid("HttpTarget.uriOverride.host is required for tasks.buffer.") + uri = scheme + "://" + host + # The port round-trips through the store as a float; render as an + # int (8080, never 8080.0) and treat <= 0 as absent. + port = _to_int(uo.get("port", None)) + if port > 0: + uri = uri + ":" + str(port) + po = uo.get("pathOverride", None) + path = po.get("path", None) if type(po) == "dict" else None + if type(path) == "string" and path != "": + uri = uri + path + qo = uo.get("queryOverride", None) + qp = qo.get("queryParams", None) if type(qo) == "dict" else None + if type(qp) == "string" and qp != "": + uri = uri + "?" + qp + + hr = {"url": uri} + method = ht.get("httpMethod", None) + if method == None or method == "HTTP_METHOD_UNSPECIFIED": + method = "POST" + hr["httpMethod"] = method + headers = {} + if content_type != "": + headers["Content-Type"] = content_type + if len(headers) > 0: + hr["headers"] = headers + if payload != "": + hr["body"] = payload + + task = {"httpRequest": hr} + doc, ierr = _insert_task(queue_name, task, task_id) + if ierr != None: + return ierr + return respond(200, {"task": _task_entity(doc, "FULL")}) + +# _run_task forces a dispatch now, ignoring scheduleTime, queue state, and +# rate limits (the documented RunTask semantics). +# +# Outcome model: the unseen worker answers 2xx by default, so the task +# completes and is deleted (matching "the task will be deleted" on success +# and NOT_FOUND on later runs). Under the failing-worker profile the worker +# answers 500: the attempt is recorded, and the task reschedules per the +# queue's retryConfig until it exhausts maxAttempts/maxRetryDuration, at +# which point it is permanently failed (deleted with a tombstone). +def _run_task(req): + name = _task_name(_queue_name(req["params"].get("project", ""), req["params"].get("location", ""), req["params"].get("queue", "")), req["params"].get("task", "")) + tc = store_collection("tasks") + doc = tc.get(name) + if doc == None: + return _missing("Task", name) + body, jerr = _json_body(req) + if jerr != None: + return jerr + + now = clock.now_unix() + now_str = clock.unix_to_rfc3339(now) + doc["dispatch_count"] = int(doc.get("dispatch_count", 0)) + 1 + scheduled = doc.get("schedule_time", "") + if doc.get("first_attempt", None) == None: + doc["first_attempt"] = {"scheduleTime": scheduled, "dispatchTime": now_str} + doc["first_dispatch_unix"] = now + + if profile_active() == "failing-worker": + qdoc = _find_queue(doc["queue"]) + retry = qdoc["retry_config"] if qdoc != None else dict(_DEFAULT_RETRY) + doc["response_count"] = int(doc.get("response_count", 0)) + 1 + doc["last_attempt"] = { + "scheduleTime": scheduled, + "dispatchTime": now_str, + "responseTime": now_str, + "responseStatus": {"code": 500, "message": "simulated worker failure (profile: failing-worker)"}, + } + attempts = doc["dispatch_count"] + max_attempts = retry.get("maxAttempts", 100) + age_limit = _parse_duration(retry.get("maxRetryDuration")) + expired = age_limit != None and age_limit > 0.0 and (now - doc.get("first_dispatch_unix", now)) >= int(age_limit) + if (max_attempts != -1 and attempts >= max_attempts) or expired: + rendered = _task_entity(doc, _view_from(req, body)) + tc.delete(name) + _tombstone_set(name, now) + return respond(200, rendered) + delay = _retry_delay(retry, attempts) + doc["schedule_time"] = clock.unix_to_rfc3339(now + delay) + tc.update(name, doc) + return respond(200, _task_entity(doc, _view_from(req, body))) + + # Success path: the returned task carries the post-dispatch, + # pre-response status (dispatchTime set, no responseStatus). + doc["last_attempt"] = {"scheduleTime": scheduled, "dispatchTime": now_str} + rendered = _task_entity(doc, _view_from(req, body)) + tc.delete(name) + _tombstone_set(name, now) + return respond(200, rendered) diff --git a/adapters/google_cloudtasks_style_test.go b/adapters/google_cloudtasks_style_test.go new file mode 100644 index 0000000..a3da0d0 --- /dev/null +++ b/adapters/google_cloudtasks_style_test.go @@ -0,0 +1,1132 @@ +package adapters + +import ( + "encoding/json" + "os" + "path/filepath" + "testing" + "time" + + "stuntapi.com/stunt/internal/adapter/runtime" + "stuntapi.com/stunt/internal/primitives" + "stuntapi.com/stunt/internal/primitives/blob" + "stuntapi.com/stunt/internal/primitives/clock" + "stuntapi.com/stunt/internal/primitives/kv" + "stuntapi.com/stunt/internal/starlark" +) + +// These tests drive the google-cloudtasks-style adapter scripts directly +// (lib.star preloaded) over a shared store and a VIRTUAL clock: one VM per +// handler script, all observing the same collections/kv, exactly like the +// engine. The virtual clock pins the retry/backoff schedule arithmetic +// (failing-worker profile) without sleeping. + +const ( + ctProject = "demo" + ctLocation = "us-central1" + ctQueue = "orders" +) + +func ctQueueName() string { + return "projects/" + ctProject + "/locations/" + ctLocation + "/queues/" + ctQueue +} + +// ctFixture is one shared store + virtual clock + one VM per script. +type ctFixture struct { + t *testing.T + vc *clock.Clock + vmL *starlark.VM // locations.star + vmQ *starlark.VM // queues.star + vmT *starlark.VM // tasks.star +} + +func newCTFixture(t *testing.T, start time.Time, active func() string) *ctFixture { + t.Helper() + dir := repoAdaptersDir(t) + root := filepath.Join(dir, "google-cloudtasks-style") + libSrc, err := os.ReadFile(filepath.Join(root, "scripts", "lib.star")) + if err != nil { + t.Fatalf("read lib.star: %v", err) + } + + tmp := t.TempDir() + store, err := primitives.Open(filepath.Join(tmp, "s.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { store.Close() }) + kvStore, err := kv.Open(filepath.Join(tmp, "s.kv.db")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { kvStore.Close() }) + blobStore, err := blob.Open(filepath.Join(tmp, "blobs")) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { blobStore.Close() }) + + vc := clock.NewVirtualClock(start) + builtins := runtime.BuildAllBuiltins(runtime.BuiltinOptions{ + Store: store, + KV: kvStore, + Blob: blobStore, + Clock: vc, + ServiceName: "test", + ActiveProfile: active, + }) + + load := func(script string) *starlark.VM { + t.Helper() + src, err := os.ReadFile(filepath.Join(root, "scripts", script)) + if err != nil { + t.Fatalf("read %s: %v", script, err) + } + vm, err := starlark.LoadWithLib(string(src), string(libSrc), builtins) + if err != nil { + t.Fatalf("LoadWithLib %s: %v", script, err) + } + return vm + } + + return &ctFixture{ + t: t, + vc: vc, + vmL: load("locations.star"), + vmQ: load("queues.star"), + vmT: load("tasks.star"), + } +} + +// call invokes a handler with bearer auth, path params, query params, and +// a marshalled JSON body (nil for none). +func (f *ctFixture) call(vm *starlark.VM, handler, method, path string, params, query map[string]string, body any) starlark.Response { + f.t.Helper() + raw := "" + if body != nil { + b, err := json.Marshal(body) + if err != nil { + f.t.Fatal(err) + } + raw = string(b) + } + if query == nil { + query = map[string]string{} + } + resp, err := vm.Call(handler, starlark.Request{ + Method: method, + Path: path, + Host: "cloudtasks.stunt.test", + Headers: map[string]string{"Authorization": "Bearer ya29.test-token"}, + Body: jsonMap(body), + RawBody: raw, + Params: params, + Query: query, + }) + if err != nil { + f.t.Fatalf("%s: %v", handler, err) + } + return resp +} + +func jsonMap(body any) map[string]any { + if body == nil { + return nil + } + switch v := body.(type) { + case map[string]any: + return v + case map[string]string: + out := map[string]any{} + for k, s := range v { + out[k] = s + } + return out + default: + b, _ := json.Marshal(body) + var m map[string]any + _ = json.Unmarshal(b, &m) + return m + } +} + +// --- assertion helpers --- + +// ctNum coerces handler-produced numbers (Starlark ints arrive as int64, +// floats as float64) for comparison. +func ctNum(v any) float64 { + switch n := v.(type) { + case int64: + return float64(n) + case int: + return float64(n) + case float64: + return n + } + return -1 +} + +func ctBody(t *testing.T, r starlark.Response) map[string]any { + t.Helper() + if r.Body == nil { + t.Fatalf("expected object body, got status %d body %q", r.Status, r.RawBody) + } + return r.Body +} + +func ctErrStatus(t *testing.T, r starlark.Response) string { + t.Helper() + e, ok := ctBody(t, r)["error"].(map[string]any) + if !ok { + t.Fatalf("expected error envelope, got %v", r.Body) + } + s, _ := e["status"].(string) + return s +} + +// ctCreateQueue makes the fixture's queue, returning its entity. +func (f *ctFixture) ctCreateQueue(t *testing.T, body map[string]any) map[string]any { + t.Helper() + r := f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"queueId": ctQueue}, body) + if r.Status != 200 { + t.Fatalf("create queue: status %d body %v", r.Status, r.Body) + } + return ctBody(t, r) +} + +// ctCreateTask posts a task with the given httpRequest overrides. +func (f *ctFixture) ctCreateTask(t *testing.T, task map[string]any) (starlark.Response, map[string]any) { + t.Helper() + r := f.call(f.vmT, "on_create_task", "POST", "/v2/"+ctQueueName()+"/tasks", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, + map[string]any{"task": task}) + return r, ctBody(t, r) +} + +// --- locations --- + +func TestCTLocations(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + + r := f.call(f.vmL, "on_list_locations", "GET", "/v2/projects/demo/locations", + map[string]string{"project": ctProject}, nil, nil) + if r.Status != 200 { + t.Fatalf("list locations: %d %v", r.Status, r.Body) + } + locs := ctBody(t, r)["locations"].([]any) + if len(locs) < 20 { + t.Fatalf("expected a broad region list, got %d", len(locs)) + } + first := locs[0].(map[string]any) + if first["name"] != "projects/demo/locations/asia-east1" || first["locationId"] != "asia-east1" { + t.Fatalf("unexpected location entity: %v", first) + } + + r = f.call(f.vmL, "on_get_location", "GET", "/v2/projects/demo/locations/us-central1", + map[string]string{"project": ctProject, "location": ctLocation}, nil, nil) + if r.Status != 200 || ctBody(t, r)["locationId"] != ctLocation { + t.Fatalf("get location: %d %v", r.Status, r.Body) + } + + r = f.call(f.vmL, "on_get_location", "GET", "/v2/projects/demo/locations/mars", + map[string]string{"project": ctProject, "location": "mars"}, nil, nil) + if r.Status != 404 || ctErrStatus(t, r) != "NOT_FOUND" { + t.Fatalf("unknown location: %d %v", r.Status, r.Body) + } +} + +func TestCTCmekConfig(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + + r := f.call(f.vmL, "on_get_cmek", "GET", "/v2/projects/demo/locations/us-central1/cmekConfig", + map[string]string{"project": ctProject, "location": ctLocation}, nil, nil) + if r.Status != 200 || ctBody(t, r)["kmsKey"] != "" { + t.Fatalf("default cmek: %d %v", r.Status, r.Body) + } + + key := "projects/demo/locations/us/keyRings/kr/cryptoKeys/k1" + r = f.call(f.vmL, "on_update_cmek", "PATCH", "/v2/projects/demo/locations/us-central1/cmekConfig", + map[string]string{"project": ctProject, "location": ctLocation}, nil, + map[string]any{"kmsKey": key}) + if r.Status != 200 || ctBody(t, r)["kmsKey"] != key { + t.Fatalf("update cmek: %d %v", r.Status, r.Body) + } + + r = f.call(f.vmL, "on_get_cmek", "GET", "/v2/projects/demo/locations/us-central1/cmekConfig", + map[string]string{"project": ctProject, "location": ctLocation}, nil, nil) + if ctBody(t, r)["kmsKey"] != key { + t.Fatalf("cmek not persisted: %v", r.Body) + } +} + +// --- queues --- + +func TestCTQueueCreate(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + + q := f.ctCreateQueue(t, map[string]any{}) + if q["name"] != ctQueueName() || q["state"] != "RUNNING" { + t.Fatalf("bad queue entity: %v", q) + } + rl := q["rateLimits"].(map[string]any) + if ctNum(rl["maxDispatchesPerSecond"]) != 500 || ctNum(rl["maxBurstSize"]) != 100 || ctNum(rl["maxConcurrentDispatches"]) != 1000 { + t.Fatalf("bad default rateLimits: %v", rl) + } + rc := q["retryConfig"].(map[string]any) + if ctNum(rc["maxAttempts"]) != 100 || rc["minBackoff"] != "0.100s" || rc["maxBackoff"] != "3600s" || ctNum(rc["maxDoublings"]) != 16 { + t.Fatalf("bad default retryConfig: %v", rc) + } + + // Duplicate -> 409 ALREADY_EXISTS. + r := f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"queueId": ctQueue}, map[string]any{}) + if r.Status != 409 || ctErrStatus(t, r) != "ALREADY_EXISTS" { + t.Fatalf("duplicate queue: %d %v", r.Status, r.Body) + } + + // Unknown location -> 404. + r = f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/mars/queues", + map[string]string{"project": ctProject, "location": "mars"}, + map[string]string{"queueId": "x"}, map[string]any{}) + if r.Status != 404 || ctErrStatus(t, r) != "NOT_FOUND" { + t.Fatalf("bad location: %d %v", r.Status, r.Body) + } + + // Invalid queue ID -> 400. + r = f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"queueId": "bad_id"}, map[string]any{}) + if r.Status != 400 || ctErrStatus(t, r) != "INVALID_ARGUMENT" { + t.Fatalf("bad queue id: %d %v", r.Status, r.Body) + } + + // name derived from queue.name when queueId is absent. + r = f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, nil, + map[string]any{"name": "projects/demo/locations/us-central1/queues/from-name"}) + if r.Status != 200 || ctBody(t, r)["name"] != "projects/demo/locations/us-central1/queues/from-name" { + t.Fatalf("name-derived create: %d %v", r.Status, r.Body) + } + + // mismatched name -> 400. + r = f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"queueId": "zzz"}, + map[string]any{"name": "projects/demo/locations/us-central1/queues/other"}) + if r.Status != 400 { + t.Fatalf("name mismatch: %d %v", r.Status, r.Body) + } + + // rate above the documented cap -> 400. + r = f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"queueId": "cap"}, + map[string]any{"rateLimits": map[string]any{"maxDispatchesPerSecond": 900}}) + if r.Status != 400 { + t.Fatalf("rate cap: %d %v", r.Status, r.Body) + } +} + +func TestCTQueueListFilterPauseResume(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + // A second queue in a DIFFERENT location: the location prefix must not + // cross-match. + r := f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-east1/queues", + map[string]string{"project": ctProject, "location": "us-east1"}, + map[string]string{"queueId": ctQueue}, map[string]any{}) + if r.Status != 200 { + t.Fatalf("second queue: %d %v", r.Status, r.Body) + } + + r = f.call(f.vmQ, "on_list_queues", "GET", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, nil, nil) + queues := ctBody(t, r)["queues"].([]any) + if len(queues) != 1 || queues[0].(map[string]any)["name"] != ctQueueName() { + t.Fatalf("location-scoped list leaked: %v", queues) + } + + // Pause, then filter state = PAUSED finds it. + qv := func(verb string, params map[string]string) starlark.Response { + params["queue_verb"] = ctQueue + ":" + verb + return f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", params, nil, map[string]any{}) + } + base := map[string]string{"project": ctProject, "location": ctLocation} + r = qv("pause", map[string]string{"project": ctProject, "location": ctLocation}) + if r.Status != 200 || ctBody(t, r)["state"] != "PAUSED" { + t.Fatalf("pause: %d %v", r.Status, r.Body) + } + // Pausing twice is idempotent. + if r = qv("pause", map[string]string{"project": ctProject, "location": ctLocation}); r.Status != 200 { + t.Fatalf("pause twice: %d %v", r.Status, r.Body) + } + + r = f.call(f.vmQ, "on_list_queues", "GET", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"filter": "state = PAUSED"}, nil) + if got := len(ctBody(t, r)["queues"].([]any)); got != 1 { + t.Fatalf("state filter: %d queues", got) + } + + r = f.call(f.vmQ, "on_list_queues", "GET", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"filter": "state = RUNNING"}, nil) + if got := len(ctBody(t, r)["queues"].([]any)); got != 0 { + t.Fatalf("state filter RUNNING: %d queues", got) + } + + if r = qv("resume", base); r.Status != 200 || ctBody(t, r)["state"] != "RUNNING" { + t.Fatalf("resume: %d %v", r.Status, r.Body) + } + + // Unknown verb -> 404. + r = f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", + map[string]string{"project": ctProject, "location": ctLocation, "queue_verb": ctQueue + ":explode"}, + nil, map[string]any{}) + if r.Status != 404 { + t.Fatalf("unknown verb: %d %v", r.Status, r.Body) + } +} + +func TestCTQueuePatch(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + params := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue} + path := "/v2/" + ctQueueName() + + // Dotted mask sets one leaf; burst re-derives (output only). + r := f.call(f.vmQ, "on_patch_queue", "PATCH", path, params, + map[string]string{"updateMask": "rateLimits.maxDispatchesPerSecond"}, + map[string]any{"rateLimits": map[string]any{"maxDispatchesPerSecond": 10}}) + if r.Status != 200 { + t.Fatalf("patch: %d %v", r.Status, r.Body) + } + rl := ctBody(t, r)["rateLimits"].(map[string]any) + if ctNum(rl["maxDispatchesPerSecond"]) != 10 || ctNum(rl["maxBurstSize"]) != 10 { + t.Fatalf("patched rateLimits: %v", rl) + } + + // Bare mask replaces the message. + r = f.call(f.vmQ, "on_patch_queue", "PATCH", path, params, + map[string]string{"updateMask": "retryConfig"}, + map[string]any{"retryConfig": map[string]any{"maxAttempts": 3, "minBackoff": "5s"}}) + rc := ctBody(t, r)["retryConfig"].(map[string]any) + if ctNum(rc["maxAttempts"]) != 3 || rc["minBackoff"] != "5s" || rc["maxBackoff"] != "3600s" { + t.Fatalf("patched retryConfig: %v", rc) + } + + // The patch survives a re-read (round-trip through the store). + r = f.call(f.vmQ, "on_get_queue", "GET", path, params, nil, nil) + if ctBody(t, r)["retryConfig"].(map[string]any)["maxAttempts"] != float64(3) { + t.Fatalf("patch not persisted: %v", r.Body) + } + + // state in the mask is rejected. + r = f.call(f.vmQ, "on_patch_queue", "PATCH", path, params, + map[string]string{"updateMask": "state"}, map[string]any{"state": "PAUSED"}) + if r.Status != 400 { + t.Fatalf("state mask: %d %v", r.Status, r.Body) + } + + // Immutable name. + r = f.call(f.vmQ, "on_patch_queue", "PATCH", path, params, nil, + map[string]any{"name": "projects/demo/locations/us-central1/queues/elsewhere"}) + if r.Status != 400 { + t.Fatalf("name immutable: %d %v", r.Status, r.Body) + } + + // Unknown mask field. + r = f.call(f.vmQ, "on_patch_queue", "PATCH", path, params, + map[string]string{"updateMask": "bogus"}, map[string]any{}) + if r.Status != 400 { + t.Fatalf("unknown mask field: %d %v", r.Status, r.Body) + } +} + +func TestCTQueueIam(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + params := map[string]string{"project": ctProject, "location": ctLocation, "queue_verb": ctQueue + ":getIamPolicy"} + + r := f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", params, nil, map[string]any{}) + if r.Status != 200 { + t.Fatalf("getIamPolicy: %d %v", r.Status, r.Body) + } + p1 := ctBody(t, r) + etag1, _ := p1["etag"].(string) + + // The default etag is stable across reads. + r = f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", params, nil, map[string]any{}) + if ctBody(t, r)["etag"] != etag1 { + t.Fatalf("etag unstable: %v vs %v", ctBody(t, r)["etag"], etag1) + } + + // setIamPolicy with the fresh etag wins. + setParams := map[string]string{"project": ctProject, "location": ctLocation, "queue_verb": ctQueue + ":setIamPolicy"} + r = f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", setParams, nil, + map[string]any{"policy": map[string]any{ + "bindings": []any{map[string]any{"role": "roles/cloudtasks.enqueuer", "members": []any{"user:t@example.com"}}}, + "etag": etag1, + }}) + if r.Status != 200 { + t.Fatalf("setIamPolicy: %d %v", r.Status, r.Body) + } + + // The stale etag now conflicts. + r = f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", setParams, nil, + map[string]any{"policy": map[string]any{"etag": etag1}}) + if r.Status != 409 || ctErrStatus(t, r) != "ABORTED" { + t.Fatalf("stale etag: %d %v", r.Status, r.Body) + } + + // testIamPermissions echoes. + testParams := map[string]string{"project": ctProject, "location": ctLocation, "queue_verb": ctQueue + ":testIamPermissions"} + r = f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", testParams, nil, + map[string]any{"permissions": []any{"cloudtasks.tasks.create"}}) + if r.Status != 200 { + t.Fatalf("testIamPermissions: %d %v", r.Status, r.Body) + } + perms := ctBody(t, r)["permissions"].([]any) + if len(perms) != 1 || perms[0] != "cloudtasks.tasks.create" { + t.Fatalf("permissions echo: %v", perms) + } +} + +// --- tasks --- + +func TestCTTaskCreateValidation(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + cases := []struct { + name string + task map[string]any + want string + }{ + {"no message_type", map[string]any{}, "INVALID_ARGUMENT"}, + {"bad url", map[string]any{"httpRequest": map[string]any{"url": "ftp://x"}}, "INVALID_ARGUMENT"}, + {"body on GET", map[string]any{"httpRequest": map[string]any{"url": "https://w.example/", "httpMethod": "GET", "body": "eA=="}}, "INVALID_ARGUMENT"}, + {"both targets", map[string]any{ + "httpRequest": map[string]any{"url": "https://w.example/"}, + "appEngineHttpRequest": map[string]any{"relativeUri": "/x"}, + }, "INVALID_ARGUMENT"}, + {"appengine uri", map[string]any{"appEngineHttpRequest": map[string]any{"relativeUri": "nope"}}, "INVALID_ARGUMENT"}, + } + for _, tc := range cases { + r, _ := f.ctCreateTask(t, tc.task) + if r.Status != 400 || ctErrStatus(t, r) != tc.want { + t.Fatalf("%s: %d %v", tc.name, r.Status, r.Body) + } + } + + // Queue must exist. + r := f.call(f.vmT, "on_create_task", "POST", "/v2/projects/demo/locations/us-central1/queues/ghost/tasks", + map[string]string{"project": ctProject, "location": ctLocation, "queue": "ghost"}, nil, + map[string]any{"task": map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}}) + if r.Status != 404 || ctErrStatus(t, r) != "NOT_FOUND" { + t.Fatalf("ghost queue: %d %v", r.Status, r.Body) + } +} + +func TestCTTaskCreateViewsAndDedup(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + // email-1 with a future schedule: created cleanly under the default view. + r, b0 := f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/email-1", + "httpRequest": map[string]any{ + "url": "https://worker.example/h", "httpMethod": "POST", + "headers": map[string]any{"Content-Type": "application/json"}, + "body": "eyJvayI6dHJ1ZX0=", + "oidcToken": map[string]any{ + "serviceAccountEmail": "sa@demo.iam.gserviceaccount.com", + "audience": "https://worker.example", + }, + }, + "scheduleTime": "2030-01-01T00:00:00Z", + }) + if r.Status != 200 || b0["scheduleTime"] != "2030-01-01T00:00:00Z" { + t.Fatalf("create email-1: %d %v", r.Status, r.Body) + } + if _, has := b0["httpRequest"].(map[string]any)["body"]; has { + t.Fatalf("default BASIC view leaked body: %v", b0) + } + + // FULL view round-trips the payload. + r = f.call(f.vmT, "on_create_task", "POST", "/v2/"+ctQueueName()+"/tasks", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, + map[string]string{"responseView": "FULL"}, + map[string]any{"task": map[string]any{ + "name": ctQueueName() + "/tasks/email-2", + "httpRequest": map[string]any{"url": "https://worker.example/h", "body": "eA=="}, + }}) + if r.Status != 200 { + t.Fatalf("create FULL: %d %v", r.Status, r.Body) + } + body := ctBody(t, r) + hr := body["httpRequest"].(map[string]any) + if hr["body"] != "eA==" || body["view"] != "FULL" { + t.Fatalf("FULL view dropped payload: %v", body) + } + + // BASIC omits the body but keeps the rest. + r = f.call(f.vmT, "on_get_task", "GET", "/v2/"+ctQueueName()+"/tasks/email-2", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task": "email-2"}, nil, nil) + if r.Status != 200 { + t.Fatalf("get: %d %v", r.Status, r.Body) + } + b := ctBody(t, r) + if _, has := b["httpRequest"].(map[string]any)["body"]; has { + t.Fatalf("BASIC view leaked body: %v", b) + } + if b["view"] != "BASIC" { + t.Fatalf("view field: %v", b["view"]) + } + + // Custom-ID duplicate -> 409. + r, _ = f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/email-2", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 409 || ctErrStatus(t, r) != "ALREADY_EXISTS" { + t.Fatalf("dup task: %d %v", r.Status, r.Body) + } + + // Generated IDs are unique 19-digit decimals. + r, b1 := f.ctCreateTask(t, map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}) + r2, b2 := f.ctCreateTask(t, map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}) + if b1["name"] == b2["name"] { + t.Fatalf("generated ids collided: %v", b1["name"]) + } + if r.Status != 200 || r2.Status != 200 { + t.Fatalf("generated creates: %d %d", r.Status, r2.Status) + } + + // Past scheduleTime clamps to now. + r, b3 := f.ctCreateTask(t, map[string]any{ + "httpRequest": map[string]any{"url": "https://w.example/"}, + "scheduleTime": "2020-01-01T00:00:00Z", + }) + if b3["scheduleTime"].(string) < b3["createTime"].(string) { + t.Fatalf("past schedule not clamped: %v", b3) + } +} + +func TestCTTaskListPagination(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + params := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue} + + for i := 0; i < 5; i++ { + r, _ := f.ctCreateTask(t, map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}) + if r.Status != 200 { + t.Fatalf("create %d: %d %v", i, r.Status, r.Body) + } + } + + r := f.call(f.vmT, "on_list_tasks", "GET", "/v2/"+ctQueueName()+"/tasks", params, + map[string]string{"pageSize": "2"}, nil) + b := ctBody(t, r) + if len(b["tasks"].([]any)) != 2 || b["nextPageToken"] != "2" { + t.Fatalf("page 1: %v", b) + } + + r = f.call(f.vmT, "on_list_tasks", "GET", "/v2/"+ctQueueName()+"/tasks", params, + map[string]string{"pageSize": "2", "pageToken": "4"}, nil) + b = ctBody(t, r) + if len(b["tasks"].([]any)) != 1 { + t.Fatalf("last page: %v", b) + } + if _, has := b["nextPageToken"]; has { + t.Fatalf("unexpected token on last page: %v", b) + } + + // Malformed token -> 400. + r = f.call(f.vmT, "on_list_tasks", "GET", "/v2/"+ctQueueName()+"/tasks", params, + map[string]string{"pageToken": "zzz"}, nil) + if r.Status != 400 { + t.Fatalf("bad token: %d %v", r.Status, r.Body) + } +} + +func TestCTTaskRunSuccessDeletes(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + r, _ := f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/job-1", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 200 { + t.Fatalf("create: %d %v", r.Status, r.Body) + } + + verb := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task_verb": "job-1:run"} + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, + map[string]any{"responseView": "FULL"}) + if r.Status != 200 { + t.Fatalf("run: %d %v", r.Status, r.Body) + } + b := ctBody(t, r) + if ctNum(b["dispatchCount"]) != 1 || ctNum(b["responseCount"]) != 0 { + t.Fatalf("post-run counters: %v", b) + } + last := b["lastAttempt"].(map[string]any) + if _, has := last["responseStatus"]; has { + t.Fatalf("success run must not carry responseStatus: %v", last) + } + first := b["firstAttempt"].(map[string]any) + if _, has := first["dispatchTime"]; !has { + t.Fatalf("firstAttempt.dispatchTime missing: %v", first) + } + + // The completed task is gone; re-run -> 404; the ID is tombstoned. + get := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task": "job-1"} + r = f.call(f.vmT, "on_get_task", "GET", "/v2/x", get, nil, nil) + if r.Status != 404 { + t.Fatalf("get after success run: %d %v", r.Status, r.Body) + } + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, map[string]any{}) + if r.Status != 404 { + t.Fatalf("re-run completed: %d %v", r.Status, r.Body) + } + r, _ = f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/job-1", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 409 { + t.Fatalf("tombstone recreate: %d %v", r.Status, r.Body) + } +} + +func TestCTTaskRunFailingWorkerRetries(t *testing.T) { + // maxAttempts=3, minBackoff=10s, maxDoublings=2: retries at +10s, +20s, + // then the third run exhausts attempts and permanently fails. + f := newCTFixture(t, time.Unix(1770000000, 0), func() string { return "failing-worker" }) + f.ctCreateQueue(t, map[string]any{"retryConfig": map[string]any{"maxAttempts": 3, "minBackoff": "10s", "maxDoublings": 2}}) + + r, _ := f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/flaky", + "httpRequest": map[string]any{"url": "https://w.example/", "body": "eA=="}, + }) + if r.Status != 200 { + t.Fatalf("create: %d %v", r.Status, r.Body) + } + // The virtual clock pins the whole retry schedule: create at + // 2026-02-02T02:40:00Z, run 1 reschedules to +10s, run 2 to +30s. + verb := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task_verb": "flaky:run"} + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, + map[string]any{"responseView": "FULL"}) + if r.Status != 200 { + t.Fatalf("run 1: %d %v", r.Status, r.Body) + } + b := ctBody(t, r) + if ctNum(b["dispatchCount"]) != 1 || ctNum(b["responseCount"]) != 1 { + t.Fatalf("run 1 counters: %v", b) + } + last := b["lastAttempt"].(map[string]any) + status := last["responseStatus"].(map[string]any) + if ctNum(status["code"]) != 500 { + t.Fatalf("run 1 responseStatus: %v", status) + } + // Rescheduled at exactly createTime+10s (minBackoff; virtual clock). + if sched, want := b["scheduleTime"].(string), "2026-02-02T02:40:10Z"; sched != want { + t.Fatalf("run 1 schedule: got %s want %s", sched, want) + } + + // Run 2: still queued, +20s (doubled once) from the run-2 moment. + f.vc.Advance(10 * time.Second) + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, map[string]any{}) + b = ctBody(t, r) + if ctNum(b["dispatchCount"]) != 2 { + t.Fatalf("run 2 counters: %v", b) + } + if sched, want := b["scheduleTime"].(string), "2026-02-02T02:40:30Z"; sched != want { + t.Fatalf("run 2 schedule: got %s want %s", sched, want) + } + + // Run 3: attempts exhausted -> permanent failure, task deleted. + f.vc.Advance(20 * time.Second) + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, map[string]any{}) + if r.Status != 200 { + t.Fatalf("run 3: %d %v", r.Status, r.Body) + } + b = ctBody(t, r) + if ctNum(b["dispatchCount"]) != 3 { + t.Fatalf("run 3 counters: %v", b) + } + get := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task": "flaky"} + if r = f.call(f.vmT, "on_get_task", "GET", "/v2/x", get, nil, nil); r.Status != 404 { + t.Fatalf("permanently failed task still present: %d %v", r.Status, r.Body) + } +} + +func TestCTTaskRunFailingWorkerMaxRetryDuration(t *testing.T) { + // Unlimited attempts but a 25s retry window measured from the first + // attempt: after it passes, the next run permanently fails the task. + f := newCTFixture(t, time.Unix(1770000000, 0), func() string { return "failing-worker" }) + f.ctCreateQueue(t, map[string]any{"retryConfig": map[string]any{ + "maxAttempts": -1, "maxRetryDuration": "25s", "minBackoff": "5s", "maxBackoff": "5s", + }}) + r, _ := f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/windowed", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 200 { + t.Fatalf("create: %d %v", r.Status, r.Body) + } + + verb := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task_verb": "windowed:run"} + for i := 0; i < 3; i++ { + f.vc.Advance(5 * time.Second) + if r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, map[string]any{}); r.Status != 200 { + t.Fatalf("run %d: %d %v", i+1, r.Status, r.Body) + } + } + // 20s elapsed since create; the first attempt was at +5s, so the age + // limit (25s from first attempt) is not yet exhausted — task survives. + get := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task": "windowed"} + if r = f.call(f.vmT, "on_get_task", "GET", "/v2/x", get, nil, nil); r.Status != 200 { + t.Fatalf("task should still be queued: %d %v", r.Status, r.Body) + } + // Push past the window: the next run permanently fails it. + f.vc.Advance(15 * time.Second) + if r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", verb, nil, map[string]any{}); r.Status != 200 { + t.Fatalf("final run: %d %v", r.Status, r.Body) + } + if r = f.call(f.vmT, "on_get_task", "GET", "/v2/x", get, nil, nil); r.Status != 404 { + t.Fatalf("task should be permanently failed: %d %v", r.Status, r.Body) + } +} + +func TestCTTaskDeleteAndTombstone(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + r, _ := f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/gone", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 200 { + t.Fatalf("create: %d %v", r.Status, r.Body) + } + + del := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task": "gone"} + r = f.call(f.vmT, "on_delete_task", "DELETE", "/v2/x", del, nil, nil) + if r.Status != 200 { + t.Fatalf("delete: %d %v", r.Status, r.Body) + } + + // Second delete -> 404; recreate within the window -> 409. + if r = f.call(f.vmT, "on_delete_task", "DELETE", "/v2/x", del, nil, nil); r.Status != 404 { + t.Fatalf("double delete: %d %v", r.Status, r.Body) + } + r, _ = f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/gone", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 409 { + t.Fatalf("tombstone recreate: %d %v", r.Status, r.Body) + } + + // The tombstone expires after 24h (86400s). + f.vc.Advance(86401 * time.Second) + r, _ = f.ctCreateTask(t, map[string]any{ + "name": ctQueueName() + "/tasks/gone", + "httpRequest": map[string]any{"url": "https://w.example/"}, + }) + if r.Status != 200 { + t.Fatalf("tombstone should expire: %d %v", r.Status, r.Body) + } +} + +func TestCTPurge(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + r, _ := f.ctCreateTask(t, map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}) + if r.Status != 200 { + t.Fatalf("create: %d %v", r.Status, r.Body) + } + + purge := map[string]string{"project": ctProject, "location": ctLocation, "queue_verb": ctQueue + ":purge"} + r = f.call(f.vmQ, "on_queue_verb", "POST", "/v2/x", purge, nil, map[string]any{}) + if r.Status != 200 { + t.Fatalf("purge: %d %v", r.Status, r.Body) + } + + params := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue} + r = f.call(f.vmT, "on_list_tasks", "GET", "/v2/"+ctQueueName()+"/tasks", params, nil, nil) + if got := len(ctBody(t, r)["tasks"].([]any)); got != 0 { + t.Fatalf("purge left %d tasks", got) + } + + // purgeTime is stamped; a task created after the purge survives. + r = f.call(f.vmQ, "on_get_queue", "GET", "/v2/"+ctQueueName(), + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, nil) + if _, has := ctBody(t, r)["purgeTime"]; !has { + t.Fatalf("purgeTime not stamped: %v", r.Body) + } + r, _ = f.ctCreateTask(t, map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}) + if r.Status != 200 { + t.Fatalf("create after purge: %d %v", r.Status, r.Body) + } + r = f.call(f.vmT, "on_list_tasks", "GET", "/v2/"+ctQueueName()+"/tasks", params, nil, nil) + if got := len(ctBody(t, r)["tasks"].([]any)); got != 1 { + t.Fatalf("post-purge task missing: %d", got) + } +} + +func TestCTBuffer(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + // The queue needs an httpTarget with a uriOverride host. + f.ctCreateQueue(t, map[string]any{"httpTarget": map[string]any{ + "uriOverride": map[string]any{ + "host": "worker.example", + "scheme": "HTTPS", + "pathOverride": map[string]any{"path": "/run"}, + "queryOverride": map[string]any{"queryParams": "src=buffer"}, + }, + }}) + + // Without httpTarget -> 400 (fresh queue in another location). + r := f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-east1/queues", + map[string]string{"project": ctProject, "location": "us-east1"}, + map[string]string{"queueId": "plain"}, map[string]any{}) + if r.Status != 200 { + t.Fatalf("plain queue: %d %v", r.Status, r.Body) + } + r = f.call(f.vmT, "on_buffer_task", "POST", "/v2/x", + map[string]string{"project": ctProject, "location": "us-east1", "queue": "plain"}, nil, + map[string]any{"body": map[string]any{"contentType": "application/json", "data": "e30="}}) + if r.Status != 400 || ctErrStatus(t, r) != "INVALID_ARGUMENT" { + t.Fatalf("buffer without httpTarget: %d %v", r.Status, r.Body) + } + + // Generated-ID form. + params := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue} + r = f.call(f.vmT, "on_buffer_task", "POST", "/v2/x", params, nil, + map[string]any{"body": map[string]any{"contentType": "application/json", "data": "e30="}}) + if r.Status != 200 { + t.Fatalf("buffer: %d %v", r.Status, r.Body) + } + task := ctBody(t, r)["task"].(map[string]any) + hr := task["httpRequest"].(map[string]any) + if hr["url"] != "https://worker.example/run?src=buffer" { + t.Fatalf("buffer url: %v", hr["url"]) + } + if hr["body"] != "e30=" || hr["httpMethod"] != "POST" { + t.Fatalf("buffer request: %v", hr) + } + if task["view"] != "FULL" { + t.Fatalf("buffer view: %v", task["view"]) + } + + // Custom-ID form, and its duplicate. + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task_verb": "buf-42:buffer"}, + nil, map[string]any{"body": map[string]any{"data": "eA=="}}) + if r.Status != 200 { + t.Fatalf("buffer custom id: %d %v", r.Status, r.Body) + } + if ctBody(t, r)["task"].(map[string]any)["name"] != ctQueueName()+"/tasks/buf-42" { + t.Fatalf("buffer custom id name: %v", ctBody(t, r)) + } + r = f.call(f.vmT, "on_task_verb", "POST", "/v2/x", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue, "task_verb": "buf-42:buffer"}, + nil, map[string]any{"body": map[string]any{"data": "eA=="}}) + if r.Status != 409 { + t.Fatalf("buffer dup: %d %v", r.Status, r.Body) + } +} + +func TestCTDeleteQueueCascades(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + r, _ := f.ctCreateTask(t, map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}}) + if r.Status != 200 { + t.Fatalf("create: %d %v", r.Status, r.Body) + } + + r = f.call(f.vmQ, "on_delete_queue", "DELETE", "/v2/"+ctQueueName(), + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, nil) + if r.Status != 200 { + t.Fatalf("delete queue: %d %v", r.Status, r.Body) + } + if r = f.call(f.vmQ, "on_delete_queue", "DELETE", "/v2/"+ctQueueName(), + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, nil); r.Status != 404 { + t.Fatalf("double delete queue: %d %v", r.Status, r.Body) + } + + // Recreating the queue starts empty; task names were removed with it. + f.ctCreateQueue(t, map[string]any{}) + params := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue} + r = f.call(f.vmT, "on_list_tasks", "GET", "/v2/"+ctQueueName()+"/tasks", params, nil, nil) + if got := len(ctBody(t, r)["tasks"].([]any)); got != 0 { + t.Fatalf("cascade left %d tasks", got) + } +} + +func TestCTMalformedJSON(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + // A body that is present but not valid JSON gets the API's message. + params := map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue} + resp, err := f.vmT.Call("on_create_task", starlark.Request{ + Method: "POST", + Path: "/v2/" + ctQueueName() + "/tasks", + Host: "cloudtasks.stunt.test", + Headers: map[string]string{"Authorization": "Bearer t"}, + RawBody: `{"task":{"httpRequest":{"url":"https://w.example/p"}}`, + Params: params, + Query: map[string]string{}, + }) + if err != nil { + t.Fatalf("call: %v", err) + } + if resp.Status != 400 || ctErrStatus(t, resp) != "INVALID_ARGUMENT" { + t.Fatalf("malformed JSON: %d %v", resp.Status, resp.Body) + } +} + +func TestCTAuthRequired(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + + resp, err := f.vmL.Call("on_list_locations", starlark.Request{ + Method: "GET", + Path: "/v2/projects/demo/locations", + Host: "cloudtasks.stunt.test", + Params: map[string]string{"project": ctProject}, + Query: map[string]string{}, + }) + if err != nil { + t.Fatalf("call: %v", err) + } + if resp.Status != 401 || ctErrStatus(t, resp) != "UNAUTHENTICATED" { + t.Fatalf("no bearer: %d %v", resp.Status, resp.Body) + } +} + +// TestCTAdversarialTypes pins the never-a-500 contract: null / numeric / +// wrong-typed fields where strings or objects are expected must 400, not +// crash the handler. +func TestCTAdversarialTypes(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{}) + + queueCases := []struct { + name string + body map[string]any + }{ + {"null name", map[string]any{"name": nil}}, + {"numeric name", map[string]any{"name": 123}}, + {"object name", map[string]any{"name": map[string]any{}}}, + {"rateLimits string", map[string]any{"rateLimits": "fast"}}, + {"rateLimits list", map[string]any{"rateLimits": []any{1}}}, + {"retryConfig string", map[string]any{"retryConfig": "soon"}}, + {"httpTarget string", map[string]any{"httpTarget": "oops"}}, + {"uriOverride string", map[string]any{"httpTarget": map[string]any{"uriOverride": "x"}}}, + {"uriOverride host null", map[string]any{"httpTarget": map[string]any{"uriOverride": map[string]any{"host": nil}}}}, + {"uriOverride host numeric", map[string]any{"httpTarget": map[string]any{"uriOverride": map[string]any{"host": 12}}}}, + {"pathOverride string", map[string]any{"httpTarget": map[string]any{"uriOverride": map[string]any{"host": "w.example", "pathOverride": "/x"}}}}, + {"queryParams numeric", map[string]any{"httpTarget": map[string]any{"uriOverride": map[string]any{"host": "w.example", "queryOverride": map[string]any{"queryParams": 7}}}}}, + {"port object", map[string]any{"httpTarget": map[string]any{"uriOverride": map[string]any{"host": "w.example", "port": map[string]any{}}}}}, + } + for _, tc := range queueCases { + r := f.call(f.vmQ, "on_create_queue", "POST", "/v2/projects/demo/locations/us-central1/queues", + map[string]string{"project": ctProject, "location": ctLocation}, + map[string]string{"queueId": "adv-" + tc.name}, tc.body) + if r.Status != 400 { + t.Fatalf("queue create %s: got %d (%v), want 400", tc.name, r.Status, r.Body) + } + } + + taskCases := []struct { + name string + task map[string]any + }{ + {"numeric name", map[string]any{"name": 9, "httpRequest": map[string]any{"url": "https://w.example/"}}}, + {"object name", map[string]any{"name": map[string]any{}, "httpRequest": map[string]any{"url": "https://w.example/"}}}, + {"null url", map[string]any{"httpRequest": map[string]any{"url": nil}}}, + {"numeric url", map[string]any{"httpRequest": map[string]any{"url": 5}}}, + {"null relativeUri", map[string]any{"appEngineHttpRequest": map[string]any{"relativeUri": nil}}}, + {"numeric scheduleTime", map[string]any{"httpRequest": map[string]any{"url": "https://w.example/"}, "scheduleTime": 0}}, + } + for _, tc := range taskCases { + r, _ := f.ctCreateTask(t, tc.task) + if r.Status != 400 { + t.Fatalf("task create %s: got %d (%v), want 400", tc.name, r.Status, r.Body) + } + } + + // A null name is proto-JSON "absent": the task is created with a + // generated ID, exactly like a missing name (rejected creates above + // already consumed sequence numbers, so assert the shape only). + r, b := f.ctCreateTask(t, map[string]any{"name": nil, "httpRequest": map[string]any{"url": "https://w.example/"}}) + if r.Status != 200 { + t.Fatalf("task create null name: %d %v", r.Status, r.Body) + } + gen := b["name"].(string) + prefix := ctQueueName() + "/tasks/" + if len(gen) != len(prefix)+19 || gen[:len(prefix)] != prefix { + t.Fatalf("null name should fall back to a generated ID: %v", gen) + } + + // Queue patch with wrong-typed immutable name. + r = f.call(f.vmQ, "on_patch_queue", "PATCH", "/v2/"+ctQueueName(), + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, + map[string]any{"name": nil}) + if r.Status != 200 { + t.Fatalf("patch null name should be a no-op: %d %v", r.Status, r.Body) + } + r = f.call(f.vmQ, "on_patch_queue", "PATCH", "/v2/"+ctQueueName(), + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, + map[string]any{"name": 12}) + if r.Status != 400 { + t.Fatalf("patch numeric name: %d %v", r.Status, r.Body) + } + r = f.call(f.vmQ, "on_patch_queue", "PATCH", "/v2/"+ctQueueName(), + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, + map[string]string{"updateMask": "rateLimits"}, + map[string]any{"rateLimits": "faster"}) + if r.Status != 400 { + t.Fatalf("patch rateLimits string: %d %v", r.Status, r.Body) + } +} + +// TestCTBufferPortRendering pins the integer port rendering (the value +// round-trips through the store as a float; the URL must not grow a .0). +func TestCTBufferPortRendering(t *testing.T) { + f := newCTFixture(t, time.Unix(1770000000, 0), nil) + f.ctCreateQueue(t, map[string]any{"httpTarget": map[string]any{ + "uriOverride": map[string]any{"host": "worker.example", "port": 8080, "scheme": "HTTPS"}, + }}) + r := f.call(f.vmT, "on_buffer_task", "POST", "/v2/x", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, + map[string]any{"body": map[string]any{"data": "eA=="}}) + if r.Status != 200 { + t.Fatalf("buffer: %d %v", r.Status, r.Body) + } + hr := ctBody(t, r)["task"].(map[string]any)["httpRequest"].(map[string]any) + if hr["url"] != "https://worker.example:8080" { + t.Fatalf("port rendering: %v", hr["url"]) + } + + // port 0 clears the port (documented UriOverride semantics). + f2 := newCTFixture(t, time.Unix(1770000000, 0), nil) + f2.ctCreateQueue(t, map[string]any{"httpTarget": map[string]any{ + "uriOverride": map[string]any{"host": "worker.example", "port": 0, "scheme": "HTTPS"}, + }}) + r = f2.call(f2.vmT, "on_buffer_task", "POST", "/v2/x", + map[string]string{"project": ctProject, "location": ctLocation, "queue": ctQueue}, nil, + map[string]any{"body": map[string]any{"data": "eA=="}}) + hr = ctBody(t, r)["task"].(map[string]any)["httpRequest"].(map[string]any) + if hr["url"] != "https://worker.example" { + t.Fatalf("port 0 should clear: %v", hr["url"]) + } +} diff --git a/conformance/matrix.json b/conformance/matrix.json index 8924f5f..af35391 100644 --- a/conformance/matrix.json +++ b/conformance/matrix.json @@ -1,11 +1,11 @@ { "summary": { - "adapters": 98, + "adapters": 99, "checks": 231, "tiers": { "sdk_and_vm": 2, "sdk_only": 34, - "vm_only": 59, + "vm_only": 60, "boot": 3 } }, @@ -108297,6 +108297,253 @@ } ] }, + { + "id": "google-cloudtasks-style", + "display_name": "Google Cloud Tasks API simulator (unofficial)", + "adapter_version": "0.1.0", + "real_hosts": [], + "api_name": "Google Cloud Tasks API", + "api_version": "v2", + "routes": 16, + "verification": "VM", + "sdks": [], + "behaviors": [], + "missing": [ + "No background dispatch loop — tasks are never auto-delivered at scheduleTime; only tasks.run drives attempts", + "No gRPC transport (REST only); google-cloud-* client libraries default to gRPC", + "oauthToken/oidcToken are stored verbatim — no token minting at dispatch time", + "stackdriverLoggingConfig is stored but no logs are written", + "IAM policies are stored but not enforced — testIamPermissions always grants", + "No App Engine queue.yaml/xml semantics (DISABLED state, queue-file overrides)", + "CMEK config is metadata-only; nothing is encrypted" + ], + "deviations": [ + "Any OAuth2 bearer token is accepted; real Google validates tokens and IAM permissions", + "tasks.run models the worker instead of calling it — success completes and deletes the task; the failing-worker profile simulates a 500 with retryConfig backoff", + "queues.list filter supports only name/state with =, !=, and : containment (subset of the real filter grammar)", + "tasks.list returns creation order; the real service defines no order", + "Generated task IDs are counter-based 19-digit decimals, not random", + "Purge is synchronous; real purge can take up to 60s to complete", + "Deleted-task tombstones hold names exactly 24h (real retention is up to 24h, 9 days for queue.yaml queues)" + ], + "covered": [ + { + "method": "GET", + "route": "/v2/projects/{project}/locations" + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}" + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/cmekConfig" + }, + { + "method": "PATCH", + "route": "/v2/projects/{project}/locations/{location}/cmekConfig" + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues" + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues" + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}" + }, + { + "method": "PATCH", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}" + }, + { + "method": "DELETE", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}" + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue_verb}" + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks" + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks" + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks:buffer" + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task_verb}" + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}" + }, + { + "method": "DELETE", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}" + } + ], + "derived_behaviors": [ + { + "method": "GET", + "route": "/v2/projects/{project}/locations", + "tags": [ + "query", + "params", + "paginate" + ] + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}", + "tags": [ + "params" + ] + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/cmekConfig", + "tags": [ + "params", + "stateful" + ] + }, + { + "method": "PATCH", + "route": "/v2/projects/{project}/locations/{location}/cmekConfig", + "tags": [ + "body", + "params", + "stateful" + ] + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues", + "tags": [ + "body", + "query", + "params", + "stateful" + ] + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues", + "tags": [ + "query", + "params", + "stateful", + "paginate" + ] + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}", + "tags": [ + "params", + "stateful" + ] + }, + { + "method": "PATCH", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}", + "tags": [ + "body", + "query", + "params", + "stateful" + ] + }, + { + "method": "DELETE", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}", + "tags": [ + "params", + "stateful" + ] + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue_verb}", + "tags": [ + "body", + "params", + "stateful", + "clock" + ] + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks", + "tags": [ + "body", + "query", + "params", + "stateful", + "clock" + ] + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks", + "tags": [ + "query", + "params", + "stateful", + "paginate" + ] + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks:buffer", + "tags": [ + "body", + "params", + "stateful", + "clock" + ] + }, + { + "method": "POST", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task_verb}", + "tags": [ + "body", + "query", + "params", + "stateful", + "clock" + ] + }, + { + "method": "GET", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}", + "tags": [ + "query", + "params", + "stateful" + ] + }, + { + "method": "DELETE", + "route": "/v2/projects/{project}/locations/{location}/queues/{queue}/tasks/{task}", + "tags": [ + "params", + "stateful", + "clock" + ] + } + ] + }, { "id": "google-iam-style", "display_name": "Google Cloud IAM + Service Accounts API simulator (unofficial)", diff --git a/conformance/matrix.yaml b/conformance/matrix.yaml index 66eb3ef..9245a04 100644 --- a/conformance/matrix.yaml +++ b/conformance/matrix.yaml @@ -534,6 +534,23 @@ adapters: - "No asps, verificationCodes, mobileDevices, or chromeosDevices" - "No user/group aliases, user photos, domains, or customers endpoints" - "No users.patch/groups.patch, members.get, or members.update endpoints" + google-cloudtasks-style: + deviations: + - "Any OAuth2 bearer token is accepted; real Google validates tokens and IAM permissions" + - "tasks.run models the worker instead of calling it — success completes and deletes the task; the failing-worker profile simulates a 500 with retryConfig backoff" + - "queues.list filter supports only name/state with =, !=, and : containment (subset of the real filter grammar)" + - "tasks.list returns creation order; the real service defines no order" + - "Generated task IDs are counter-based 19-digit decimals, not random" + - "Purge is synchronous; real purge can take up to 60s to complete" + - "Deleted-task tombstones hold names exactly 24h (real retention is up to 24h, 9 days for queue.yaml queues)" + missing: + - "No background dispatch loop — tasks are never auto-delivered at scheduleTime; only tasks.run drives attempts" + - "No gRPC transport (REST only); google-cloud-* client libraries default to gRPC" + - "oauthToken/oidcToken are stored verbatim — no token minting at dispatch time" + - "stackdriverLoggingConfig is stored but no logs are written" + - "IAM policies are stored but not enforced — testIamPermissions always grants" + - "No App Engine queue.yaml/xml semantics (DISABLED state, queue-file overrides)" + - "CMEK config is metadata-only; nothing is encrypted" google-iam-style: deviations: - "JWKS serves one fixed synthetic RSA-2048 key (kid mock-google-key-1)"