From e11bcef64975ddfc30a7e6d671f895ed9cbcb64e Mon Sep 17 00:00:00 2001 From: Harish Seshadri Date: Fri, 25 Sep 2026 09:59:43 -0700 Subject: [PATCH 1/2] feat(cloudflare-pages): add verified production rollback Add a module-owned rollback so consumers can recover when a production deploy fails its live smoke check. - previous_production_deployment: read-only; returns the deployment production serves now, to record as the rollback target before deploy. - rollback: POSTs the documented Pages rollback endpoint over the same pinned curl transport, to an explicit id or the previous successful production deployment. Refuses preview, unsuccessful, foreign, absent, or already-live targets before any write, then re-reads the project until canonical_deployment equals the target (1/2/4/8s, 60s deadline) and fails closed otherwise. - The real Dagger mock-provider contract now runs the rollback through real curl and the pinned jq projection. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_015oBArfm762nN1r4F4Fst5a --- docs/dagger-modules.md | 66 +++ .../.dagger/src/cloudflare_pages/api.py | 166 +++++- .../.dagger/src/cloudflare_pages/main.py | 183 +++++- .../.dagger/src/cloudflare_pages/models.py | 21 + .../.dagger/tests/test_deploy_contract.py | 57 +- .../.dagger/tests/test_rollback.py | 548 ++++++++++++++++++ 6 files changed, 1010 insertions(+), 31 deletions(-) create mode 100644 modules/cloudflare-pages/.dagger/tests/test_rollback.py diff --git a/docs/dagger-modules.md b/docs/dagger-modules.md index 653d8bb..9482ee4 100644 --- a/docs/dagger-modules.md +++ b/docs/dagger-modules.md @@ -310,6 +310,72 @@ request and Wrangler boundaries without access to Cloudflare and cannot mutate C cross-language fixtures also tamper with an envelope and prove rejection occurs before GitHub or Cloudflare transport. +## Roll back after a failed live smoke check + +**TL;DR:** record what production serves before you deploy. If the post-deploy smoke check fails, +call `rollback` with that ID, then fail the job anyway. Production recovers, and the red run still +tells you the release was bad. + +Two module functions cover this. Both take the same typed credentials as `deploy`: + +| Function | Writes? | Returns | +| --- | --- | --- | +| `previous-production-deployment(cloudflare-api-token, cloudflare-account-id, project)` | No | `project`, `deployment-id`, `deployment-url` of the deployment production serves **now** | +| `rollback(cloudflare-api-token, cloudflare-account-id, project, deployment-id="")` | One `POST .../deployments/{id}/rollback` | `project`, `from-deployment-id`, `to-deployment-id`, `live-deployment-id`, `live-deployment-url` | + +`previous-production-deployment` is "previous" from the point of view of the deploy you are about +to run: call it before `deploy`, and the ID it returns is what `rollback` should restore. With no +`deployment-id`, `rollback` picks the newest successful production deployment older than the one +live now. + +`rollback` fails closed. It refuses, before any write, when: + +- the project has no live production deployment, or the target is not in the 10 most recent + production deployments; +- the target is a preview deployment, is not a successful `deploy` stage, or belongs to another + project; +- the target is already live (a no-op must be explicit, so this raises instead of passing); +- no `deployment-id` is given and there is no older successful production deployment. + +After the POST it requires the response to name the target, then re-reads the project until its +live (`canonical_deployment`) ID equals the target, with 1, 2, 4, 8 second delays under a 60-second +deadline. If production never serves the target, it raises. A Cloudflare 4xx or 5xx raises a +sanitized `CloudflareApiError`, and the POST is never retried. + +Wire it in the consumer's Dagger release function: + +```python +pages = dag.cloudflare_pages() +before = pages.previous_production_deployment( + cloudflare_api_token=cloudflare_api_token, + cloudflare_account_id=cloudflare_account_id, + project=PROJECT, +) +target = await before.deployment_id() # 1. record the rollback target + +evidence = pages.deploy(...) # 2. deploy exactly as above +deployed = await evidence.deployment_id() + +if not await live_smoke_passes(LIVE_URL): # 3. the consumer's own smoke check + rolled = pages.rollback( # 4. restore the recorded target + cloudflare_api_token=cloudflare_api_token, + cloudflare_account_id=cloudflare_account_id, + project=PROJECT, + deployment_id=target, + ) + live = await rolled.live_deployment_id() + # 5. fail the job: the release was bad even though production recovered + raise RuntimeError(f"live smoke failed for {deployed}; production rolled back to {live}") +``` + +Record the target before `deploy`, not after: once the new deployment is live, "the one before it" +is only a guess. If `rollback` itself raises, the job must still fail. Production then needs a +human, and the error names the reason. + +What the module does not prove: it checks the rollback through the Cloudflare API (the project's +live deployment ID), the same way `deploy` checks convergence. It does not fetch the custom +domain. The consumer's smoke check should run again after a rollback if it needs that proof. + ## Composition proofs Warm proof from a repository checkout: diff --git a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/api.py b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/api.py index 8e9764f..e0215bb 100644 --- a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/api.py +++ b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/api.py @@ -5,15 +5,18 @@ import asyncio import json import re +from dataclasses import dataclass from typing import Final, Protocol from urllib.parse import urlsplit from pydantic import BaseModel, JsonValue, TypeAdapter, ValidationError from .models import ( + DEPLOYMENT_ID_PATTERN, ApiProblem, AttemptIdentity, CreatedDeployment, + DeploymentResponse, DeploymentsResponse, GitHubEvidence, ListedPagesDeployment, @@ -22,6 +25,7 @@ PagesTarget, ProjectResponse, ProviderDeploymentEvidence, + RollbackEvidence, ) API_ORIGIN: Final = "https://api.cloudflare.com/client/v4" @@ -31,6 +35,8 @@ CONTROL_PATTERN: Final = re.compile(r"[\x00-\x1f\x7f]+") BEARER_PATTERN: Final = re.compile(r"(?i)bearer\s+\S+") JSON_ADAPTER: Final[TypeAdapter[JsonValue]] = TypeAdapter(JsonValue) +ROLLBACK_DELAYS: Final = (1, 2, 4, 8) +ROLLBACK_DEADLINE_SECONDS: Final = 60 class CloudflareError(RuntimeError): @@ -61,6 +67,27 @@ async def upload(self, artifact: ArtifactT, source_sha: str) -> CreatedDeploymen async def sleep(self, seconds: int) -> None: ... +class RollbackOperations(Protocol): + """Read-and-rollback provider operations for one Pages project.""" + + async def get_project(self) -> str: ... + + async def get_deployments(self) -> str: ... + + async def rollback(self, deployment_id: str) -> str: ... + + async def sleep(self, seconds: int) -> None: ... + + +@dataclass(frozen=True) +class ProductionState: + """The bound project, its live deployment id, and recent production history.""" + + project: PagesProject + live_id: str + history: tuple[ListedPagesDeployment, ...] + + def project_path(account_ref: str, target: PagesTarget) -> str: """Return the only supported project API path.""" _require_account_ref(account_ref) @@ -381,13 +408,13 @@ def _deployment_identity(deployment: PagesDeployment, target: PagesTarget) -> tu deployment.deployment_trigger.type, deployment.project_id, metadata.commit_dirty, - _valid_deployment_url(deployment.url, target, deployment.short_id), + _valid_deployment_url(deployment.url, target.project, deployment.short_id), ) -def _valid_deployment_url(value: str, target: PagesTarget, short_id: str) -> bool: +def _valid_deployment_url(value: str, project: str, short_id: str) -> bool: parsed = urlsplit(value) - hostname = f"{short_id}.{target.project}.pages.dev" + hostname = f"{short_id}.{project}.pages.dev" identity = ( parsed.scheme, parsed.hostname, @@ -570,3 +597,136 @@ def _model[ModelT: BaseModel](model: type[ModelT], payload: dict[str, JsonValue] return model.model_validate_json(json.dumps(payload)) except (ValidationError, ValueError, TypeError): raise CloudflarePolicyError("Cloudflare response schema mismatch") from None + + +def parse_live_deployment_id(raw: str) -> str | None: + """Return the project's canonical (currently served) production deployment id.""" + parse_project_response(raw) + result = _object(_required(_json_object(raw), "result")) + live = _required(result, "canonical_deployment") + if live is None: + return None + value = _required(_object(live), "id") + if not isinstance(value, str) or DEPLOYMENT_ID_PATTERN.fullmatch(value) is None: + raise CloudflarePolicyError("Cloudflare response schema mismatch") + return value + + +def parse_deployment_response(raw: str) -> ListedPagesDeployment: + """Parse a strict projection of one documented deployment response.""" + payload = _json_object(raw) + fields = _response_fields(payload) + if fields["success"] is not True: + raise CloudflareApiError(provider_error_message(raw)) + projected = fields | {"result": _deployment_result(_required(payload, "result"))} + response = _model(DeploymentResponse, projected) + _require_success(response.success, response.errors) + return response.result + + +async def live_production_deployment( + operations: RollbackOperations, project: str +) -> ListedPagesDeployment: + """Read-only: the deployment production serves now, to record as a rollback target.""" + state = await read_production_state(operations, project) + return _history_row(state.history, state.live_id) + + +async def rollback_production( + operations: RollbackOperations, project: str, deployment_id: str | None +) -> RollbackEvidence: + """Roll production back to one eligible deployment and verify it is served.""" + state = await read_production_state(operations, project) + target = _rollback_target(state, deployment_id) + response = parse_deployment_response(await operations.rollback(target.id)) + if response.id != target.id: + raise CloudflarePolicyError("Cloudflare rollback response identity differs") + live_id = await _await_live(operations, target.id) + return RollbackEvidence(project, state.live_id, target.id, live_id, target.url) + + +async def read_production_state(operations: RollbackOperations, project: str) -> ProductionState: + """Read the bound project, its live deployment, and the recent production page.""" + raw = await operations.get_project() + pages = parse_project_response(raw) + if pages.name != project: + raise CloudflarePolicyError("Cloudflare project binding differs") + live_id = parse_live_deployment_id(raw) + if live_id is None: + raise CloudflarePolicyError("Cloudflare project has no live production deployment") + deployments = parse_deployments_response(await operations.get_deployments()) + _require_pagination(deployments) + return ProductionState(pages, live_id, deployments.result) + + +def _rollback_target(state: ProductionState, deployment_id: str | None) -> ListedPagesDeployment: + if deployment_id is None: + return _previous_successful(state) + target = _history_row(state.history, deployment_id) + if target.id == state.live_id: + raise CloudflarePolicyError("Rollback target is already live; refusing a no-op rollback") + _require_rollback_target(target, state.project) + return target + + +def _previous_successful(state: ProductionState) -> ListedPagesDeployment: + live = _history_row(state.history, state.live_id) + older = state.history[state.history.index(live) + 1 :] + for row in older: + if _rollback_eligible(row, state.project): + return row + raise CloudflarePolicyError("Cloudflare has no previous successful production deployment") + + +def _history_row( + history: tuple[ListedPagesDeployment, ...], deployment_id: str +) -> ListedPagesDeployment: + for row in history: + if row.id == deployment_id: + return row + raise CloudflarePolicyError("Cloudflare deployment is not in recent production history") + + +def _rollback_eligible(row: ListedPagesDeployment, project: PagesProject) -> bool: + try: + _require_rollback_target(row, project) + except CloudflarePolicyError: + return False + return True + + +def _require_rollback_target(row: ListedPagesDeployment, project: PagesProject) -> None: + if row.environment != "production": + raise CloudflarePolicyError("Rollback target is not a production deployment") + if (row.latest_stage.name, row.latest_stage.status) != ("deploy", "success"): + raise CloudflarePolicyError("Rollback target is not a successful deployment") + identity = (row.project_name, row.project_id) + url_valid = _valid_deployment_url(row.url, project.name, row.short_id) + if identity != (project.name, project.id) or not url_valid: + raise CloudflarePolicyError("Cloudflare deployment identity differs") + + +async def _await_live(operations: RollbackOperations, expected: str) -> str: + try: + async with asyncio.timeout(ROLLBACK_DEADLINE_SECONDS): + return await _bounded_live(operations, expected) + except TimeoutError: + raise _not_serving() from None + + +async def _bounded_live(operations: RollbackOperations, expected: str) -> str: + for delay in ROLLBACK_DELAYS: + if await _live_id(operations) == expected: + return expected + await operations.sleep(delay) + if await _live_id(operations) == expected: + return expected + raise _not_serving() + + +async def _live_id(operations: RollbackOperations) -> str | None: + return parse_live_deployment_id(await operations.get_project()) + + +def _not_serving() -> CloudflarePolicyError: + return CloudflarePolicyError("Cloudflare production does not serve the rollback target") diff --git a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py index 3685b06..5bdc863 100644 --- a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py +++ b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py @@ -20,17 +20,23 @@ CloudflarePolicyError, deploy_verified_artifact, disable_git_payload, + live_production_deployment, preflight_provider, provider_error_message, require_evidence_binding, + rollback_production, verify_current_deployment, ) from .models import ( + DEPLOYMENT_ID_PATTERN, + PROJECT_PATTERN, AttemptIdentity, CreatedDeployment, GitHubEvidence, + ListedPagesDeployment, PagesTarget, ProviderDeploymentEvidence, + RollbackEvidence, WranglerBuildMetadata, WranglerOutput, ) @@ -101,6 +107,7 @@ PurePosixPath("/usr/local/lib/node_modules/wrangler/templates/pages-template-worker.ts"), } ) +type CurlMethod = Literal["GET", "PATCH", "POST"] FUNCTIONS_SOURCE_EXCLUDES: Final = [ "**/node_modules", "**/package.json", @@ -151,6 +158,26 @@ class DeploymentEvidence: run_attempt: int = field() +@object_type +class ProductionDeployment: + """The production deployment served now: the rollback target to record.""" + + project: str = field() + deployment_id: str = field() + deployment_url: str = field() + + +@object_type +class ProductionRollbackEvidence: + """Non-secret proof that production now serves the rollback target.""" + + project: str = field() + from_deployment_id: str = field() + to_deployment_id: str = field() + live_deployment_id: str = field() + live_deployment_url: str = field() + + @dataclass(frozen=True) class CurlPagesOperations: """Pinned, secret-safe Dagger adapters for one Pages target.""" @@ -190,29 +217,14 @@ async def upload(self, artifact: dagger.Directory, source_sha: str) -> CreatedDe async def sleep(self, seconds: int) -> None: await asyncio.sleep(seconds) - async def _request(self, method: Literal["GET", "PATCH"], suffix: str, body: str = "") -> str: - request = self._request_container(method, suffix, body) - try: - result = await asyncio.wait_for(_request_result(request), CURL_DEADLINE_SECONDS) - except (TimeoutError, dagger.QueryError): - raise CloudflareApiError("Cloudflare network request failed") from None - return _require_http_success(*result) - - def _request_container( - self, method: Literal["GET", "PATCH"], suffix: str, body: str - ) -> dagger.Container: - base = dag.container(platform=dagger.Platform("linux/amd64")).from_(CURL_IMAGE) - base = base.with_entrypoint([]).with_user("0").with_mounted_temp("/work") - base = base.with_workdir("/work") - base = base.with_mounted_secret("/run/secrets/token", self.api_token) - base = base.with_mounted_secret("/run/secrets/account", self.account_id) - base = base.with_mounted_file("/run/jq", _jq_binary()) - if self.api_service is not None: - base = base.with_service_binding("api.cloudflare.com", self.api_service) - if self.ca_certificate is not None: - base = base.with_mounted_file("/run/mock-ca.pem", self.ca_certificate) - command = ["/bin/sh", "-euc", _curl_script(), "--", method, suffix] - return _uncached(base).with_exec(command, stdin=body) + async def _request(self, method: CurlMethod, suffix: str, body: str = "") -> str: + return await _send_request(self._request_container(method, suffix, body)) + + def _request_container(self, method: CurlMethod, suffix: str, body: str) -> dagger.Container: + transport = CurlTransport( + self.api_token, self.account_id, self.api_service, self.ca_certificate + ) + return _curl_request_container(transport, method, suffix, body) def _upload_container(self, artifact: dagger.Directory, source_sha: str) -> dagger.Container: base = _wrangler_base().with_mounted_directory("/artifact", artifact, read_only=True) @@ -232,6 +244,56 @@ def _project_suffix(self) -> str: return f"/pages/projects/{self.target.project}" +@dataclass(frozen=True) +class CurlTransport: + """Secret-safe credentials and optional test-only mock bindings for curl.""" + + api_token: dagger.Secret + account_id: dagger.Secret + api_service: dagger.Service | None = None + ca_certificate: dagger.File | None = None + + +@dataclass(frozen=True) +class CurlRollbackOperations: + """Pinned curl adapter for read-only history reads and one rollback POST.""" + + api_token: dagger.Secret + account_id: dagger.Secret + project: str + api_service: dagger.Service | None = None + ca_certificate: dagger.File | None = None + + def __post_init__(self) -> None: + if PROJECT_PATTERN.fullmatch(self.project) is None: + raise CloudflarePolicyError("Cloudflare Pages project name is malformed") + + async def get_project(self) -> str: + return await self._request("GET", self._project_suffix()) + + async def get_deployments(self) -> str: + suffix = f"{self._project_suffix()}/deployments?env=production&per_page=10" + return await self._request("GET", suffix) + + async def rollback(self, deployment_id: str) -> str: + if DEPLOYMENT_ID_PATTERN.fullmatch(deployment_id) is None: + raise CloudflarePolicyError("Cloudflare Pages deployment id is malformed") + suffix = f"{self._project_suffix()}/deployments/{deployment_id}/rollback" + return await self._request("POST", suffix) + + async def sleep(self, seconds: int) -> None: + await asyncio.sleep(seconds) + + async def _request(self, method: CurlMethod, suffix: str, body: str = "") -> str: + transport = CurlTransport( + self.api_token, self.account_id, self.api_service, self.ca_certificate + ) + return await _send_request(_curl_request_container(transport, method, suffix, body)) + + def _project_suffix(self) -> str: + return f"/pages/projects/{self.project}" + + @object_type class CloudflarePages: """Deploy only foundation-verified artifacts to one bound Pages target.""" @@ -272,6 +334,30 @@ async def deploy( inputs, consumer_identity, producing_identity, allowed_roots) # fmt: on + @function(cache="never") # type: ignore[call-overload,untyped-decorator] # SDK stub gap + async def previous_production_deployment( + self, + cloudflare_api_token: dagger.Secret, + cloudflare_account_id: dagger.Secret, + project: str, + ) -> ProductionDeployment: + """Read-only: record the deployment production serves now, before a new deploy.""" + operations = CurlRollbackOperations(cloudflare_api_token, cloudflare_account_id, project) + return _public_production(project, await live_production_deployment(operations, project)) + + @function(cache="never") # type: ignore[call-overload,untyped-decorator] # SDK stub gap + async def rollback( + self, + cloudflare_api_token: dagger.Secret, + cloudflare_account_id: dagger.Secret, + project: str, + deployment_id: str = "", + ) -> ProductionRollbackEvidence: + """Roll production back to deployment_id, or the previous successful one.""" + operations = CurlRollbackOperations(cloudflare_api_token, cloudflare_account_id, project) + evidence = await rollback_production(operations, project, deployment_id or None) + return _public_rollback(evidence) + # fmt: off @function(cache="never") # type: ignore[call-overload,untyped-decorator] # SDK stub gap async def verify( @@ -716,6 +802,31 @@ def _wrangler_base() -> dagger.Container: return base.with_env_variable("WRANGLER_SEND_METRICS", "false").with_exec(install) +def _curl_request_container( + transport: CurlTransport, method: CurlMethod, suffix: str, body: str +) -> dagger.Container: + base = dag.container(platform=dagger.Platform("linux/amd64")).from_(CURL_IMAGE) + base = base.with_entrypoint([]).with_user("0").with_mounted_temp("/work") + base = base.with_workdir("/work") + base = base.with_mounted_secret("/run/secrets/token", transport.api_token) + base = base.with_mounted_secret("/run/secrets/account", transport.account_id) + base = base.with_mounted_file("/run/jq", _jq_binary()) + if transport.api_service is not None: + base = base.with_service_binding("api.cloudflare.com", transport.api_service) + if transport.ca_certificate is not None: + base = base.with_mounted_file("/run/mock-ca.pem", transport.ca_certificate) + command = ["/bin/sh", "-euc", _curl_script(), "--", method, suffix] + return _uncached(base).with_exec(command, stdin=body) + + +async def _send_request(request: dagger.Container) -> str: + try: + result = await asyncio.wait_for(_request_result(request), CURL_DEADLINE_SECONDS) + except (TimeoutError, dagger.QueryError): + raise CloudflareApiError("Cloudflare network request failed") from None + return _require_http_success(*result) + + def _jq_binary() -> dagger.File: image = dag.container(platform=dagger.Platform("linux/amd64")).from_(JQ_IMAGE) return image.file("/jq") @@ -794,6 +905,24 @@ def _public_evidence(source: ProviderDeploymentEvidence) -> DeploymentEvidence: return evidence +def _public_production(project: str, source: ListedPagesDeployment) -> ProductionDeployment: + deployment = ProductionDeployment.__new__(ProductionDeployment) + deployment.project = project + deployment.deployment_id = source.id + deployment.deployment_url = source.url + return deployment + + +def _public_rollback(source: RollbackEvidence) -> ProductionRollbackEvidence: + evidence = ProductionRollbackEvidence.__new__(ProductionRollbackEvidence) + evidence.project = source.project + evidence.from_deployment_id = source.from_deployment_id + evidence.to_deployment_id = source.to_deployment_id + evidence.live_deployment_id = source.live_deployment_id + evidence.live_deployment_url = source.live_deployment_url + return evidence + + def _curl_script() -> str: projection = _jq_projection() return f""" @@ -819,7 +948,7 @@ def _curl_script() -> str: printf 'retry = 0\\n' fi if [ -f /run/mock-ca.pem ]; then printf 'cacert = "/run/mock-ca.pem"\\n'; fi - if [ "$method" = PATCH ]; then + if [ "$method" = PATCH ] || [ "$method" = POST ]; then printf 'header = "Content-Type: application/json"\\n' printf 'data-binary = "@{REQUEST_PATH}"\\n' fi @@ -842,14 +971,16 @@ def _jq_projection() -> str: r"production_deployments_enabled,preview_deployment_setting})}" ) project = r"{id,name,production_branch,domains,source:(.source|if .==null then null else " - project += source + r" end)}" + project += source + r" end),canonical_deployment:(.canonical_deployment|" + project += r"if .==null then null else {id} end)}" metadata = r"{branch,commit_hash,commit_dirty}" trigger = r"{type,metadata:(.metadata|" + metadata + r")}" deployment = r"{id,short_id,url,project_id,project_name,environment,latest_stage:" deployment += r"(.latest_stage|{name,status}),deployment_trigger:(.deployment_trigger|" deployment += trigger + r")}" result = r'if (.result|type)=="array" then [.result[]|' + deployment + r"] " - result += r"else (.result|" + project + r") end" + result += r'elif (.result|type)=="object" and (.result|has("latest_stage")) then (.result|' + result += deployment + r") else (.result|" + project + r") end" info = r"result_info:(.result_info|{count,page,per_page,total_count,total_pages})" prefix = r"{errors:[.errors[]|" + problem + r"],messages:[.messages[]|" + problem return prefix + r"],success,result:(" + result + r")," + info + r"}" diff --git a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/models.py b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/models.py index 3ae0dd9..dce059e 100644 --- a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/models.py +++ b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/models.py @@ -23,6 +23,7 @@ NUMERIC_ID_TEXT: Final = r"\A[1-9][0-9]*\z" TIMESTAMP_TEXT: Final = r"\A\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z\z" DEPLOY_ROOT_PATTERN: Final = re.compile(r"[A-Za-z0-9][A-Za-z0-9._-]{0,63}") +DEPLOYMENT_ID_PATTERN: Final = re.compile(r"[0-9a-f][0-9a-f-]{0,63}") class ClosedModel(BaseModel): # type: ignore[explicit-any] # Pydantic v2 base stub @@ -217,6 +218,15 @@ class DeploymentsResponse(ClosedModel): # type: ignore[explicit-any] # Pydanti result_info: ResultInfo +class DeploymentResponse(ClosedModel): # type: ignore[explicit-any] # Pydantic v2 base stub + """Strict projected Cloudflare response for one Pages deployment.""" + + errors: tuple[ApiProblem, ...] + messages: tuple[ApiProblem, ...] + result: ListedPagesDeployment + success: bool + + class WranglerOutput(ClosedModel): # type: ignore[explicit-any] # Pydantic v2 base stub """Pinned Wrangler 4.103.0 pages-deploy JSONL record.""" @@ -310,6 +320,17 @@ class ProviderDeploymentEvidence: attempt_identity: AttemptIdentity +@dataclass(frozen=True) +class RollbackEvidence: + """Non-secret proof that production moved from one deployment to another.""" + + project: str + from_deployment_id: str + to_deployment_id: str + live_deployment_id: str + live_deployment_url: str + + @dataclass(frozen=True) class CreatedDeployment: """The immutable deployment identity returned by this Wrangler upload.""" diff --git a/modules/cloudflare-pages/.dagger/tests/test_deploy_contract.py b/modules/cloudflare-pages/.dagger/tests/test_deploy_contract.py index 69db062..25a026c 100644 --- a/modules/cloudflare-pages/.dagger/tests/test_deploy_contract.py +++ b/modules/cloudflare-pages/.dagger/tests/test_deploy_contract.py @@ -81,6 +81,30 @@ def deployments() -> dict[str, object]: "result_info": {"count": len(result), "page": 1, "per_page": 10, "total_count": len(result), "total_pages": int(bool(result))}} +ROLL = "/client/v4/accounts/" + ACCOUNT + "/pages/projects/edge-roll" +ROLL_LIVE = "11111111" + +def rid(short: str) -> str: + return short + "-0000-4000-8000-000000000000" + +def roll_row(short: str, status: str = "success") -> dict[str, object]: + return {"id": rid(short), "short_id": short, "url": "https://" + short + ".edge-roll.pages.dev", + "project_id": "roll-project-id", "project_name": "edge-roll", "environment": "production", + "latest_stage": {"name": "deploy", "status": status, "ended_on": None}, + "deployment_trigger": {"type": "ad_hoc", "metadata": {"branch": "main", + "commit_hash": "", "commit_dirty": False, "commit_message": "ignored"}}, "aliases": None} + +def roll_project() -> dict[str, object]: + return {"errors": [], "messages": [], "success": True, "result": {"id": "roll-project-id", + "name": "edge-roll", "production_branch": "main", "domains": ["edge-roll.pages.dev"], + "source": None, "canonical_deployment": roll_row(ROLL_LIVE), + "latest_deployment": roll_row("11111111")}} + +def roll_history() -> dict[str, object]: + result = [roll_row("11111111"), roll_row("33333333", "failure"), roll_row("22222222")] + return {"errors": [], "messages": [], "success": True, "result": result, + "result_info": {"count": 3, "page": 1, "per_page": 10, "total_count": 3, "total_pages": 1}} + class Handler(BaseHTTPRequestHandler): def log_message(self, format: str, *args: object) -> None: return def send(self, value: object) -> None: @@ -93,6 +117,10 @@ def do_GET(self) -> None: EVENTS.append("get-project"); self.send(project(not GIT_DISABLED, "none" if GIT_DISABLED else "all")); return if self.path == PROJECT + "/deployments?env=production&per_page=10": EVENTS.append("get-deployments"); self.send(deployments()); return + if self.path == ROLL: + EVENTS.append("get-roll-project"); self.send(roll_project()); return + if self.path == ROLL + "/deployments?env=production&per_page=10": + EVENTS.append("get-roll-deployments"); self.send(roll_history()); return if self.path.endswith("/__mock/events"): payload = project(False, "none"); payload["result"]["domains"] = EVENTS self.send(payload); return @@ -110,6 +138,14 @@ def do_PATCH(self) -> None: GIT_DISABLED = True; EVENTS.append("disable-git"); self.send(project(False, "none")); return self.send_error(404) + def do_POST(self) -> None: + global ROLL_LIVE + assert self.headers.get("Content-Length") == "0" + if self.path == ROLL + "/deployments/" + rid("22222222") + "/rollback": + ROLL_LIVE = "22222222"; EVENTS.append("rollback") + self.send({"errors": [], "messages": [], "success": True, "result": roll_row("22222222")}); return + self.send_error(404) + health = HTTPServer(("0.0.0.0", 8080), Handler) threading.Thread(target=health.serve_forever, daemon=True).start() server = HTTPServer(("0.0.0.0", 443), Handler) @@ -121,8 +157,9 @@ def do_PATCH(self) -> None: import json import dagger from dagger import dag, function, object_type -from cloudflare_pages.api import CloudflarePolicyError, deploy_verified_artifact -from cloudflare_pages.main import (CurlPagesOperations, NODE_IMAGE, WRANGLER_OUTPUT_PATH, _jq_binary, +from cloudflare_pages.api import (CloudflarePolicyError, deploy_verified_artifact, + live_production_deployment, rollback_production) +from cloudflare_pages.main import (CurlPagesOperations, CurlRollbackOperations, NODE_IMAGE, WRANGLER_OUTPUT_PATH, _jq_binary, _prepare_deploy_artifact, _uncached, _verify_envelope, _wrangler_script, wrangler_deploy_args) from cloudflare_pages.models import AttemptIdentity, CreatedDeployment, GitHubEvidence, PagesTarget @@ -262,6 +299,21 @@ async def functions_contract(token: dagger.Secret, account: dagger.Secret, events = json.loads(await operations._request("GET", "/__mock/events"))["result"]["domains"] assert events.count("upload") == 2 +async def rollback_contract(token: dagger.Secret, account: dagger.Secret, + mock: dagger.Service, cert: dagger.File) -> None: + roll = CurlRollbackOperations(token, account, "edge-roll", mock, cert) + recorded = await live_production_deployment(roll, "edge-roll") + assert recorded.id == "11111111-0000-4000-8000-000000000000" + rolled = await rollback_production(roll, "edge-roll", None) + assert rolled.from_deployment_id == recorded.id + assert rolled.to_deployment_id == rolled.live_deployment_id == "22222222-0000-4000-8000-000000000000" + assert rolled.live_deployment_url == "https://22222222.edge-roll.pages.dev" + try: await rollback_production(roll, "edge-roll", rolled.live_deployment_id) + except CloudflarePolicyError: pass + else: raise ValueError("no-op rollback reached provider transport") + events = json.loads(await roll._request("GET", "/__mock/events"))["result"]["domains"] + assert events.count("rollback") == 1 + @object_type class ProviderContract: @function @@ -277,6 +329,7 @@ async def contract(self) -> str: assert events == ["wrangler-preflight", "get-project", "get-deployments", "disable-git", "get-project", "upload", "get-deployments"] assert result.source_sha == SHA await functions_contract(token, account, mock, fixture_files().file("ca.pem")) + await rollback_contract(token, account, mock, fixture_files().file("ca.pem")) tampered = envelope.with_new_file("artifact/dist/index.html", "tampered") try: await _verify_envelope(tampered, "hseshadr/edge-reco@" + SHA, "b" * 40 + ":44", ["dist"]) except dagger.QueryError: return "provider order, runnable module tree, multipart, escape, conflict, and tamper rejection passed" diff --git a/modules/cloudflare-pages/.dagger/tests/test_rollback.py b/modules/cloudflare-pages/.dagger/tests/test_rollback.py new file mode 100644 index 0000000..537235a --- /dev/null +++ b/modules/cloudflare-pages/.dagger/tests/test_rollback.py @@ -0,0 +1,548 @@ +"""Production rollback policy: explicit targets, refusals, and verified convergence.""" + +from __future__ import annotations + +import ast +import asyncio +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import cast + +import dagger +import pytest + +import cloudflare_pages.main as main_module +from cloudflare_pages.api import ( + CloudflareApiError, + CloudflarePolicyError, + live_production_deployment, + rollback_production, +) +from cloudflare_pages.main import CurlRollbackOperations +from cloudflare_pages.models import RollbackEvidence + +PROJECT = "edge-reco" +PROJECT_ID = "7b162ea7-7367-4d4a-a28a-cb84f88f6" +LIVE = "11111111" +PREVIOUS = "22222222" +FAILED = "33333333" +OLDER = "44444444" +SECRET_TYPE = "dagger." + "Secret" + + +def _id(short_id: str) -> str: + return f"{short_id}-0000-4000-8000-000000000000" + + +def _row( + short_id: str, + *, + status: str = "success", + environment: str = "production", + stage: str = "deploy", +) -> dict[str, object]: + return { + "id": _id(short_id), + "short_id": short_id, + "url": f"https://{short_id}.{PROJECT}.pages.dev", + "project_id": PROJECT_ID, + "project_name": PROJECT, + "environment": environment, + "latest_stage": {"name": stage, "status": status}, + "deployment_trigger": { + "type": "ad_hoc", + "metadata": {"branch": "main", "commit_hash": "", "commit_dirty": False}, + }, + } + + +def _project(canonical: str | None, name: str = PROJECT) -> str: + live = None if canonical is None else {"id": _id(canonical)} + result = { + "id": PROJECT_ID, + "name": name, + "production_branch": "main", + "domains": [f"{PROJECT}.pages.dev"], + "source": None, + "canonical_deployment": live, + } + return json.dumps({"errors": [], "messages": [], "success": True, "result": result}) + + +def _deployments(*rows: dict[str, object]) -> str: + info = {"count": len(rows), "page": 1, "per_page": 10, "total_count": 30, "total_pages": 3} + payload = {"errors": [], "messages": [], "success": True, "result": list(rows)} + return json.dumps(payload | {"result_info": info}) + + +def _rollback_response(short_id: str) -> str: + return json.dumps({"errors": [], "messages": [], "success": True, "result": _row(short_id)}) + + +def _history() -> tuple[dict[str, object], ...]: + return (_row(LIVE), _row(FAILED, status="failure"), _row(PREVIOUS), _row(OLDER)) + + +CloudflareError = CloudflareApiError | CloudflarePolicyError + + +@dataclass +class FakeRollbackOperations: + """Stubbed Pages API that switches production only when told to converge.""" + + rows: tuple[dict[str, object], ...] = field(default_factory=_history) + live: str | None = LIVE + converge: bool = True + project_name: str = PROJECT + rollback_error: CloudflareError | None = None + response_id: str | None = None + events: list[str] = field(default_factory=list) + + async def get_project(self) -> str: + self.events.append("get-project") + return _project(self.live, self.project_name) + + async def get_deployments(self) -> str: + self.events.append("get-deployments") + return _deployments(*self.rows) + + async def rollback(self, deployment_id: str) -> str: + self.events.append(f"rollback:{deployment_id}") + if self.rollback_error is not None: + raise self.rollback_error + short_id = deployment_id.split("-", maxsplit=1)[0] + if self.converge: + self.live = short_id + return _rollback_response(self.response_id or short_id) + + async def sleep(self, seconds: int) -> None: + self.events.append(f"sleep:{seconds}") + + +def _rollbacks(operations: FakeRollbackOperations) -> list[str]: + return [event for event in operations.events if event.startswith("rollback:")] + + +@pytest.mark.asyncio +async def test_should_roll_back_to_explicit_production_deployment_and_verify_live() -> None: + operations = FakeRollbackOperations() + + evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert evidence == RollbackEvidence( + PROJECT, + _id(LIVE), + _id(PREVIOUS), + _id(PREVIOUS), + f"https://{PREVIOUS}.{PROJECT}.pages.dev", + ) + assert operations.events == [ + "get-project", + "get-deployments", + f"rollback:{_id(PREVIOUS)}", + "get-project", + ] + + +@pytest.mark.asyncio +async def test_should_default_to_previous_successful_production_deployment() -> None: + operations = FakeRollbackOperations() + + evidence = await rollback_production(operations, PROJECT, None) + + assert evidence.to_deployment_id == _id(PREVIOUS) + assert _rollbacks(operations) == [f"rollback:{_id(PREVIOUS)}"] + + +@pytest.mark.asyncio +async def test_should_refuse_when_no_previous_successful_deployment_exists() -> None: + operations = FakeRollbackOperations(rows=(_row(LIVE), _row(FAILED, status="failure"))) + + with pytest.raises(CloudflarePolicyError, match="no previous successful production"): + await rollback_production(operations, PROJECT, None) + + assert _rollbacks(operations) == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("row", "reason"), + ( + (_row(PREVIOUS, environment="preview"), "not a production deployment"), + (_row(PREVIOUS, status="failure"), "not a successful deployment"), + (_row(PREVIOUS, status="active"), "not a successful deployment"), + (_row(PREVIOUS, stage="build", status="success"), "not a successful deployment"), + ), +) +async def test_should_refuse_preview_or_unsuccessful_target( + row: dict[str, object], reason: str +) -> None: + operations = FakeRollbackOperations(rows=(_row(LIVE), row)) + + with pytest.raises(CloudflarePolicyError, match=reason): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert _rollbacks(operations) == [] + + +@pytest.mark.asyncio +async def test_should_refuse_target_absent_from_recent_production_history() -> None: + operations = FakeRollbackOperations(rows=(_row(LIVE),)) + + with pytest.raises(CloudflarePolicyError, match="not in recent production history"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert _rollbacks(operations) == [] + + +@pytest.mark.asyncio +async def test_should_refuse_foreign_project_target() -> None: + foreign = _row(PREVIOUS) | {"project_id": "other-project"} + operations = FakeRollbackOperations(rows=(_row(LIVE), foreign)) + + with pytest.raises(CloudflarePolicyError, match="identity differs"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + +@pytest.mark.asyncio +async def test_should_refuse_explicit_noop_when_target_is_already_live() -> None: + operations = FakeRollbackOperations() + + with pytest.raises(CloudflarePolicyError, match="already live"): + await rollback_production(operations, PROJECT, _id(LIVE)) + + assert _rollbacks(operations) == [] + + +@pytest.mark.asyncio +async def test_should_refuse_rollback_without_a_live_production_deployment() -> None: + operations = FakeRollbackOperations(live=None) + + with pytest.raises(CloudflarePolicyError, match="no live production deployment"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert _rollbacks(operations) == [] + + +@pytest.mark.asyncio +async def test_should_refuse_mismatched_project_binding() -> None: + operations = FakeRollbackOperations(project_name="other") + + with pytest.raises(CloudflarePolicyError, match="project binding differs"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "error", + ( + CloudflareApiError("Cloudflare 8000009: rollback rejected"), + CloudflareApiError("Cloudflare API request failed"), + ), +) +async def test_should_fail_closed_when_rollback_api_rejects(error: CloudflareApiError) -> None: + operations = FakeRollbackOperations(rollback_error=error) + + with pytest.raises(CloudflareApiError): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert operations.events[-1] == f"rollback:{_id(PREVIOUS)}" + + +@pytest.mark.asyncio +async def test_should_refuse_rollback_response_for_different_deployment() -> None: + operations = FakeRollbackOperations(response_id=OLDER) + + with pytest.raises(CloudflarePolicyError, match="rollback response identity differs"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + +@pytest.mark.asyncio +async def test_should_fail_closed_when_production_does_not_serve_rollback_target() -> None: + operations = FakeRollbackOperations(converge=False) + + with pytest.raises(CloudflarePolicyError, match="does not serve the rollback target"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + sleeps = [event for event in operations.events if event.startswith("sleep:")] + assert sleeps == ["sleep:1", "sleep:2", "sleep:4", "sleep:8"] + + +@pytest.mark.asyncio +async def test_should_accept_rollback_after_delayed_production_convergence() -> None: + operations = DelayedOperations() + + evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert evidence.live_deployment_id == _id(PREVIOUS) + assert "sleep:1" in operations.events + + +@pytest.mark.asyncio +async def test_should_bound_rollback_verification_with_hard_deadline( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(asyncio, "timeout", lambda _: ImmediateTimeout()) + operations = FakeRollbackOperations() + + with pytest.raises(CloudflarePolicyError, match="does not serve the rollback target"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + +@dataclass +class DelayedOperations(FakeRollbackOperations): + """Production switches only after the first verification read.""" + + pending: str | None = None + + async def rollback(self, deployment_id: str) -> str: + self.events.append(f"rollback:{deployment_id}") + self.pending = deployment_id.split("-", maxsplit=1)[0] + return _rollback_response(self.pending) + + async def sleep(self, seconds: int) -> None: + self.events.append(f"sleep:{seconds}") + self.live = self.pending + + +class ImmediateTimeout: + async def __aenter__(self) -> None: + raise TimeoutError + + async def __aexit__(self, error_type: object, error: object, traceback: object) -> None: + return None + + +@pytest.mark.asyncio +async def test_should_report_live_deployment_as_recorded_rollback_target() -> None: + operations = FakeRollbackOperations() + + deployment = await live_production_deployment(operations, PROJECT) + + assert (deployment.id, deployment.url) == (_id(LIVE), f"https://{LIVE}.{PROJECT}.pages.dev") + assert operations.events == ["get-project", "get-deployments"] + + +@pytest.mark.asyncio +async def test_should_refuse_to_record_target_without_live_deployment() -> None: + with pytest.raises(CloudflarePolicyError, match="no live production deployment"): + await live_production_deployment(FakeRollbackOperations(live=None), PROJECT) + + +@pytest.mark.asyncio +async def test_should_refuse_to_record_live_target_missing_from_history() -> None: + operations = FakeRollbackOperations(rows=(_row(PREVIOUS),)) + + with pytest.raises(CloudflarePolicyError, match="not in recent production history"): + await live_production_deployment(operations, PROJECT) + + +@pytest.mark.asyncio +async def test_should_refuse_malformed_live_deployment_reference() -> None: + operations = MalformedLiveOperations() + + with pytest.raises(CloudflarePolicyError, match="schema mismatch"): + await live_production_deployment(operations, PROJECT) + + +@dataclass +class MalformedLiveOperations(FakeRollbackOperations): + async def get_project(self) -> str: + payload = json.loads(_project(LIVE)) + payload["result"]["canonical_deployment"] = {"id": 7} + return json.dumps(payload) + + +@dataclass(frozen=True) +class RecordingRollbackTransport(CurlRollbackOperations): + """Curl adapter double that records the exact method, suffix, and body.""" + + requests: list[tuple[str, str, str]] = field(default_factory=list) + + async def _request(self, method: str, suffix: str, body: str = "") -> str: + self.requests.append((method, suffix, body)) + return "response" + + +def _transport() -> RecordingRollbackTransport: + secret = cast(dagger.Secret, object()) + return RecordingRollbackTransport(secret, secret, PROJECT) + + +@pytest.mark.asyncio +async def test_should_build_only_documented_rollback_requests() -> None: + transport = _transport() + + await transport.get_project() + await transport.get_deployments() + await transport.rollback(_id(PREVIOUS)) + + assert transport.requests == [ + ("GET", "/pages/projects/edge-reco", ""), + ("GET", "/pages/projects/edge-reco/deployments?env=production&per_page=10", ""), + ("POST", f"/pages/projects/edge-reco/deployments/{_id(PREVIOUS)}/rollback", ""), + ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", ("", "../../projects/x", "ABC", "a" * 65, "a/b")) +async def test_should_refuse_malformed_rollback_id_before_transport(value: str) -> None: + transport = _transport() + + with pytest.raises(CloudflarePolicyError, match="deployment id is malformed"): + await transport.rollback(value) + + assert transport.requests == [] + + +def test_should_refuse_malformed_project_before_transport() -> None: + secret = cast(dagger.Secret, object()) + + with pytest.raises(CloudflarePolicyError, match="project name is malformed"): + CurlRollbackOperations(secret, secret, "../edge") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", ("400", "403", "409", "500", "503")) +async def test_should_raise_api_error_for_rollback_http_failure( + monkeypatch: pytest.MonkeyPatch, status: str +) -> None: + container = FailingContainer(f'{status}\n{{"errors":[],"messages":[],"success":false}}') + monkeypatch.setattr(main_module, "_curl_request_container", lambda *_: container) + secret = cast(dagger.Secret, object()) + + with pytest.raises(CloudflareApiError, match="API request failed"): + await CurlRollbackOperations(secret, secret, PROJECT).rollback(_id(PREVIOUS)) + + +@dataclass +class FailingContainer: + output: str + + async def stdout(self) -> str: + return self.output + + +def test_should_send_post_with_empty_json_body_and_no_retry() -> None: + script = main_module._curl_script() + + assert 'if [ "$method" = PATCH ] || [ "$method" = POST ]; then' in script + assert "retry = 0" in script + + +def _public_methods(*names: str) -> list[ast.AsyncFunctionDef]: + tree = ast.parse(Path(main_module.__file__).read_text()) + public = next( + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == "CloudflarePages" + ) + return [ + node + for node in public.body + if isinstance(node, ast.AsyncFunctionDef) and node.name in names + ] + + +def test_should_expose_rollback_functions_uncached_with_secret_credentials() -> None: + methods = _public_methods("rollback", "previous_production_deployment") + + assert {method.name for method in methods} == {"rollback", "previous_production_deployment"} + for method in methods: + decorator = next(item for item in method.decorator_list if isinstance(item, ast.Call)) + cache = next(item.value for item in decorator.keywords if item.arg == "cache") + assert isinstance(cache, ast.Constant) + assert cache.value == "never" + arguments = { + argument.arg: ast.unparse(argument.annotation or ast.Constant(None)) + for argument in method.args.args + } + credentials = (arguments["cloudflare_api_token"], arguments["cloudflare_account_id"]) + assert credentials == (SECRET_TYPE, SECRET_TYPE) + assert not set(arguments).intersection({"url", "origin", "command", "cmd", "script"}) + + +@pytest.mark.asyncio +async def test_should_raise_api_error_for_unsuccessful_rollback_body() -> None: + operations = UnsuccessfulRollbackOperations() + + with pytest.raises(CloudflareApiError, match="API request failed"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + +@dataclass +class UnsuccessfulRollbackOperations(FakeRollbackOperations): + async def rollback(self, deployment_id: str) -> str: + self.events.append(f"rollback:{deployment_id}") + return json.dumps({"errors": [], "messages": [], "success": False, "result": None}) + + +@pytest.mark.asyncio +async def test_should_accept_rollback_served_after_final_bounded_delay() -> None: + operations = LateOperations() + + evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + assert evidence.live_deployment_id == _id(PREVIOUS) + assert [event for event in operations.events if event.startswith("sleep:")][-1] == "sleep:8" + + +@dataclass +class LateOperations(DelayedOperations): + """Production switches only during the final bounded delay.""" + + async def sleep(self, seconds: int) -> None: + self.events.append(f"sleep:{seconds}") + if seconds == 8: + self.live = self.pending + + +def _patched_pages( + monkeypatch: pytest.MonkeyPatch, operations: FakeRollbackOperations +) -> main_module.CloudflarePages: + monkeypatch.setattr(main_module, "CurlRollbackOperations", lambda *_: operations) + return main_module.CloudflarePages() + + +@pytest.mark.asyncio +async def test_should_return_public_rollback_evidence(monkeypatch: pytest.MonkeyPatch) -> None: + secret = cast(dagger.Secret, object()) + pages = _patched_pages(monkeypatch, FakeRollbackOperations()) + + evidence = await pages.rollback(secret, secret, PROJECT) + + public = (evidence.project, evidence.from_deployment_id, evidence.to_deployment_id) + assert public == (PROJECT, _id(LIVE), _id(PREVIOUS)) + live = (evidence.live_deployment_id, evidence.live_deployment_url) + assert live == (_id(PREVIOUS), f"https://{PREVIOUS}.{PROJECT}.pages.dev") + + +@pytest.mark.asyncio +async def test_should_return_public_recorded_target(monkeypatch: pytest.MonkeyPatch) -> None: + secret = cast(dagger.Secret, object()) + operations = FakeRollbackOperations() + pages = _patched_pages(monkeypatch, operations) + + recorded = await pages.previous_production_deployment(secret, secret, PROJECT) + + assert (recorded.project, recorded.deployment_id) == (PROJECT, _id(LIVE)) + assert recorded.deployment_url == f"https://{LIVE}.{PROJECT}.pages.dev" + assert _rollbacks(operations) == [] + + +@pytest.mark.asyncio +async def test_should_sleep_between_rollback_verification_reads( + monkeypatch: pytest.MonkeyPatch, +) -> None: + delays: list[float] = [] + + async def record(seconds: float) -> None: + delays.append(seconds) + + monkeypatch.setattr(asyncio, "sleep", record) + secret = cast(dagger.Secret, object()) + + await CurlRollbackOperations(secret, secret, PROJECT).sleep(2) + + assert delays == [2] From 468c14e5d44f27eae93ce9c388a28288ea424be8 Mon Sep 17 00:00:00 2001 From: Harish Seshadri Date: Fri, 25 Sep 2026 10:08:51 -0700 Subject: [PATCH 2/2] test(cloudflare-pages): bring rollback tests to the python-quality contract - Every rollback test now reads Given / When / Then and is named test_should__when_. - Deployment fixture rows are a frozen Row dataclass instead of dicts passed through parametrize; magic numbers are named Final constants. - The public-signature AST check is split into two single-behavior tests. - _jq_projection is split into _jq_project and _jq_deployment helpers so every function is 15 lines or fewer; the emitted jq filter is byte-identical. Co-Authored-By: Claude Opus 5.5 (1M context) Claude-Session: https://claude.ai/code/session_015oBArfm762nN1r4F4Fst5a --- .../.dagger/src/cloudflare_pages/main.py | 26 +- .../.dagger/tests/test_rollback.py | 560 +++++++++++------- 2 files changed, 346 insertions(+), 240 deletions(-) diff --git a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py index 5bdc863..e5cce3b 100644 --- a/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py +++ b/modules/cloudflare-pages/.dagger/src/cloudflare_pages/main.py @@ -966,24 +966,30 @@ def _curl_script() -> str: def _jq_projection() -> str: problem = r'{code:.code,message:"Provider error",documentation_url:null,source:null}' + deployment = _jq_deployment() + result = r'if (.result|type)=="array" then [.result[]|' + deployment + r"] " + result += r'elif (.result|type)=="object" and (.result|has("latest_stage")) then (.result|' + result += deployment + r") else (.result|" + _jq_project() + r") end" + info = r"result_info:(.result_info|{count,page,per_page,total_count,total_pages})" + prefix = r"{errors:[.errors[]|" + problem + r"],messages:[.messages[]|" + problem + return prefix + r"],success,result:(" + result + r")," + info + r"}" + + +def _jq_project() -> str: source = ( r"{type:.type,config:(.config|{owner,repo_name,production_branch," r"production_deployments_enabled,preview_deployment_setting})}" ) project = r"{id,name,production_branch,domains,source:(.source|if .==null then null else " project += source + r" end),canonical_deployment:(.canonical_deployment|" - project += r"if .==null then null else {id} end)}" - metadata = r"{branch,commit_hash,commit_dirty}" - trigger = r"{type,metadata:(.metadata|" + metadata + r")}" + return project + r"if .==null then null else {id} end)}" + + +def _jq_deployment() -> str: + trigger = r"{type,metadata:(.metadata|{branch,commit_hash,commit_dirty})}" deployment = r"{id,short_id,url,project_id,project_name,environment,latest_stage:" deployment += r"(.latest_stage|{name,status}),deployment_trigger:(.deployment_trigger|" - deployment += trigger + r")}" - result = r'if (.result|type)=="array" then [.result[]|' + deployment + r"] " - result += r'elif (.result|type)=="object" and (.result|has("latest_stage")) then (.result|' - result += deployment + r") else (.result|" + project + r") end" - info = r"result_info:(.result_info|{count,page,per_page,total_count,total_pages})" - prefix = r"{errors:[.errors[]|" + problem + r"],messages:[.messages[]|" + problem - return prefix + r"],success,result:(" + result + r")," + info + r"}" + return deployment + trigger + r")}" def _wrangler_script() -> str: diff --git a/modules/cloudflare-pages/.dagger/tests/test_rollback.py b/modules/cloudflare-pages/.dagger/tests/test_rollback.py index 537235a..da44031 100644 --- a/modules/cloudflare-pages/.dagger/tests/test_rollback.py +++ b/modules/cloudflare-pages/.dagger/tests/test_rollback.py @@ -7,7 +7,7 @@ import json from dataclasses import dataclass, field from pathlib import Path -from typing import cast +from typing import Final, cast import dagger import pytest @@ -22,39 +22,59 @@ from cloudflare_pages.main import CurlRollbackOperations from cloudflare_pages.models import RollbackEvidence -PROJECT = "edge-reco" -PROJECT_ID = "7b162ea7-7367-4d4a-a28a-cb84f88f6" -LIVE = "11111111" -PREVIOUS = "22222222" -FAILED = "33333333" -OLDER = "44444444" -SECRET_TYPE = "dagger." + "Secret" +PROJECT: Final = "edge-reco" +PROJECT_ID: Final = "7b162ea7-7367-4d4a-a28a-cb84f88f6" +LIVE: Final = "11111111" +PREVIOUS: Final = "22222222" +FAILED: Final = "33333333" +OLDER: Final = "44444444" +SECRET_TYPE: Final = "dagger." + "Secret" +BOUNDED_DELAYS: Final = ["sleep:1", "sleep:2", "sleep:4", "sleep:8"] +FINAL_DELAY_SECONDS: Final = 8 +ESCAPE_HATCHES: Final = frozenset({"url", "origin", "command", "cmd", "script"}) +ROLLBACK_FUNCTIONS: Final = frozenset({"rollback", "previous_production_deployment"}) + +type CloudflareError = CloudflareApiError | CloudflarePolicyError def _id(short_id: str) -> str: return f"{short_id}-0000-4000-8000-000000000000" -def _row( - short_id: str, - *, - status: str = "success", - environment: str = "production", - stage: str = "deploy", -) -> dict[str, object]: - return { - "id": _id(short_id), - "short_id": short_id, - "url": f"https://{short_id}.{PROJECT}.pages.dev", - "project_id": PROJECT_ID, - "project_name": PROJECT, - "environment": environment, - "latest_stage": {"name": stage, "status": status}, - "deployment_trigger": { - "type": "ad_hoc", - "metadata": {"branch": "main", "commit_hash": "", "commit_dirty": False}, - }, - } +def _url(short_id: str) -> str: + return f"https://{short_id}.{PROJECT}.pages.dev" + + +def _secret() -> dagger.Secret: + return cast(dagger.Secret, object()) + + +@dataclass(frozen=True) +class Row: + """One provider deployment-list row, serialized only at the JSON boundary.""" + + short_id: str + status: str = "success" + environment: str = "production" + stage: str = "deploy" + project_id: str = PROJECT_ID + + def payload(self) -> dict[str, object]: + metadata = {"branch": "main", "commit_hash": "", "commit_dirty": False} + return { + "id": _id(self.short_id), + "short_id": self.short_id, + "url": _url(self.short_id), + "project_id": self.project_id, + "project_name": PROJECT, + "environment": self.environment, + "latest_stage": {"name": self.stage, "status": self.status}, + "deployment_trigger": {"type": "ad_hoc", "metadata": metadata}, + } + + +def _envelope(result: object) -> dict[str, object]: + return {"errors": [], "messages": [], "success": True, "result": result} def _project(canonical: str | None, name: str = PROJECT) -> str: @@ -67,31 +87,28 @@ def _project(canonical: str | None, name: str = PROJECT) -> str: "source": None, "canonical_deployment": live, } - return json.dumps({"errors": [], "messages": [], "success": True, "result": result}) + return json.dumps(_envelope(result)) -def _deployments(*rows: dict[str, object]) -> str: +def _deployments(rows: tuple[Row, ...]) -> str: info = {"count": len(rows), "page": 1, "per_page": 10, "total_count": 30, "total_pages": 3} - payload = {"errors": [], "messages": [], "success": True, "result": list(rows)} + payload = _envelope([row.payload() for row in rows]) return json.dumps(payload | {"result_info": info}) def _rollback_response(short_id: str) -> str: - return json.dumps({"errors": [], "messages": [], "success": True, "result": _row(short_id)}) - + return json.dumps(_envelope(Row(short_id).payload())) -def _history() -> tuple[dict[str, object], ...]: - return (_row(LIVE), _row(FAILED, status="failure"), _row(PREVIOUS), _row(OLDER)) - -CloudflareError = CloudflareApiError | CloudflarePolicyError +def _history() -> tuple[Row, ...]: + return (Row(LIVE), Row(FAILED, status="failure"), Row(PREVIOUS), Row(OLDER)) @dataclass class FakeRollbackOperations: """Stubbed Pages API that switches production only when told to converge.""" - rows: tuple[dict[str, object], ...] = field(default_factory=_history) + rows: tuple[Row, ...] = field(default_factory=_history) live: str | None = LIVE converge: bool = True project_name: str = PROJECT @@ -105,7 +122,7 @@ async def get_project(self) -> str: async def get_deployments(self) -> str: self.events.append("get-deployments") - return _deployments(*self.rows) + return _deployments(self.rows) async def rollback(self, deployment_id: str) -> str: self.events.append(f"rollback:{deployment_id}") @@ -120,23 +137,136 @@ async def sleep(self, seconds: int) -> None: self.events.append(f"sleep:{seconds}") +@dataclass +class DelayedOperations(FakeRollbackOperations): + """Production switches only after the first verification read.""" + + pending: str | None = None + + async def rollback(self, deployment_id: str) -> str: + self.events.append(f"rollback:{deployment_id}") + self.pending = deployment_id.split("-", maxsplit=1)[0] + return _rollback_response(self.pending) + + async def sleep(self, seconds: int) -> None: + self.events.append(f"sleep:{seconds}") + self.live = self.pending + + +@dataclass +class LateOperations(DelayedOperations): + """Production switches only during the final bounded delay.""" + + async def sleep(self, seconds: int) -> None: + self.events.append(f"sleep:{seconds}") + if seconds == FINAL_DELAY_SECONDS: + self.live = self.pending + + +@dataclass +class UnsuccessfulRollbackOperations(FakeRollbackOperations): + """The rollback POST returns 2xx with an unsuccessful Cloudflare envelope.""" + + async def rollback(self, deployment_id: str) -> str: + self.events.append(f"rollback:{deployment_id}") + return json.dumps(_envelope(None) | {"success": False}) + + +@dataclass +class MalformedLiveOperations(FakeRollbackOperations): + """The project names its live deployment with a non-string id.""" + + async def get_project(self) -> str: + payload = json.loads(_project(LIVE)) + payload["result"]["canonical_deployment"] = {"id": 7} + return json.dumps(payload) + + +class ImmediateTimeout: + """Async deadline that expires before the first verification read.""" + + async def __aenter__(self) -> None: + raise TimeoutError + + async def __aexit__(self, error_type: object, error: object, traceback: object) -> None: + return None + + +@dataclass(frozen=True) +class RecordingRollbackTransport(CurlRollbackOperations): + """Curl adapter double that records the exact method, suffix, and body.""" + + requests: list[tuple[str, str, str]] = field(default_factory=list) + + async def _request(self, method: str, suffix: str, body: str = "") -> str: + self.requests.append((method, suffix, body)) + return "response" + + +@dataclass +class FailingContainer: + """Transport container whose curl run reports one HTTP status.""" + + output: str + + async def stdout(self) -> str: + return self.output + + def _rollbacks(operations: FakeRollbackOperations) -> list[str]: return [event for event in operations.events if event.startswith("rollback:")] +def _sleeps(operations: FakeRollbackOperations) -> list[str]: + return [event for event in operations.events if event.startswith("sleep:")] + + +def _transport() -> RecordingRollbackTransport: + return RecordingRollbackTransport(_secret(), _secret(), PROJECT) + + +def _patched_pages( + monkeypatch: pytest.MonkeyPatch, operations: FakeRollbackOperations +) -> main_module.CloudflarePages: + monkeypatch.setattr(main_module, "CurlRollbackOperations", lambda *_: operations) + return main_module.CloudflarePages() + + +def _public_methods() -> list[ast.AsyncFunctionDef]: + tree = ast.parse(Path(main_module.__file__).read_text()) + classes = (node for node in tree.body if isinstance(node, ast.ClassDef)) + public = next(node for node in classes if node.name == "CloudflarePages") + return [ + node + for node in public.body + if isinstance(node, ast.AsyncFunctionDef) and node.name in ROLLBACK_FUNCTIONS + ] + + +def _cache_value(method: ast.AsyncFunctionDef) -> object: + decorator = next(item for item in method.decorator_list if isinstance(item, ast.Call)) + cache = next(item.value for item in decorator.keywords if item.arg == "cache") + return cache.value if isinstance(cache, ast.Constant) else None + + +def _annotations(method: ast.AsyncFunctionDef) -> dict[str, str]: + return { + argument.arg: ast.unparse(argument.annotation or ast.Constant(None)) + for argument in method.args.args + } + + @pytest.mark.asyncio -async def test_should_roll_back_to_explicit_production_deployment_and_verify_live() -> None: +async def test_should_serve_explicit_target_when_rollback_converges() -> None: + # Given operations = FakeRollbackOperations() + # When evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) - assert evidence == RollbackEvidence( - PROJECT, - _id(LIVE), - _id(PREVIOUS), - _id(PREVIOUS), - f"https://{PREVIOUS}.{PROJECT}.pages.dev", - ) + # Then + expected = RollbackEvidence(PROJECT, _id(LIVE), _id(PREVIOUS), _id(PREVIOUS), _url(PREVIOUS)) + assert evidence == expected assert operations.events == [ "get-project", "get-deployments", @@ -146,22 +276,26 @@ async def test_should_roll_back_to_explicit_production_deployment_and_verify_liv @pytest.mark.asyncio -async def test_should_default_to_previous_successful_production_deployment() -> None: +async def test_should_pick_previous_successful_deployment_when_no_target_is_given() -> None: + # Given operations = FakeRollbackOperations() + # When evidence = await rollback_production(operations, PROJECT, None) + # Then assert evidence.to_deployment_id == _id(PREVIOUS) assert _rollbacks(operations) == [f"rollback:{_id(PREVIOUS)}"] @pytest.mark.asyncio async def test_should_refuse_when_no_previous_successful_deployment_exists() -> None: - operations = FakeRollbackOperations(rows=(_row(LIVE), _row(FAILED, status="failure"))) + # Given + operations = FakeRollbackOperations(rows=(Row(LIVE), Row(FAILED, status="failure"))) + # When / Then with pytest.raises(CloudflarePolicyError, match="no previous successful production"): await rollback_production(operations, PROJECT, None) - assert _rollbacks(operations) == [] @@ -169,68 +303,76 @@ async def test_should_refuse_when_no_previous_successful_deployment_exists() -> @pytest.mark.parametrize( ("row", "reason"), ( - (_row(PREVIOUS, environment="preview"), "not a production deployment"), - (_row(PREVIOUS, status="failure"), "not a successful deployment"), - (_row(PREVIOUS, status="active"), "not a successful deployment"), - (_row(PREVIOUS, stage="build", status="success"), "not a successful deployment"), + (Row(PREVIOUS, environment="preview"), "not a production deployment"), + (Row(PREVIOUS, status="failure"), "not a successful deployment"), + (Row(PREVIOUS, status="active"), "not a successful deployment"), + (Row(PREVIOUS, stage="build"), "not a successful deployment"), ), ) -async def test_should_refuse_preview_or_unsuccessful_target( - row: dict[str, object], reason: str -) -> None: - operations = FakeRollbackOperations(rows=(_row(LIVE), row)) +async def test_should_refuse_when_target_is_preview_or_unsuccessful(row: Row, reason: str) -> None: + # Given + operations = FakeRollbackOperations(rows=(Row(LIVE), row)) + # When / Then with pytest.raises(CloudflarePolicyError, match=reason): await rollback_production(operations, PROJECT, _id(PREVIOUS)) - assert _rollbacks(operations) == [] @pytest.mark.asyncio -async def test_should_refuse_target_absent_from_recent_production_history() -> None: - operations = FakeRollbackOperations(rows=(_row(LIVE),)) +async def test_should_refuse_when_target_is_absent_from_recent_history() -> None: + # Given + operations = FakeRollbackOperations(rows=(Row(LIVE),)) + # When / Then with pytest.raises(CloudflarePolicyError, match="not in recent production history"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) - assert _rollbacks(operations) == [] @pytest.mark.asyncio -async def test_should_refuse_foreign_project_target() -> None: - foreign = _row(PREVIOUS) | {"project_id": "other-project"} - operations = FakeRollbackOperations(rows=(_row(LIVE), foreign)) +async def test_should_refuse_when_target_belongs_to_another_project() -> None: + # Given + foreign = Row(PREVIOUS, project_id="other-project") + operations = FakeRollbackOperations(rows=(Row(LIVE), foreign)) + # When / Then with pytest.raises(CloudflarePolicyError, match="identity differs"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) + assert _rollbacks(operations) == [] @pytest.mark.asyncio async def test_should_refuse_explicit_noop_when_target_is_already_live() -> None: + # Given operations = FakeRollbackOperations() + # When / Then with pytest.raises(CloudflarePolicyError, match="already live"): await rollback_production(operations, PROJECT, _id(LIVE)) - assert _rollbacks(operations) == [] @pytest.mark.asyncio -async def test_should_refuse_rollback_without_a_live_production_deployment() -> None: +async def test_should_refuse_rollback_when_project_has_no_live_deployment() -> None: + # Given operations = FakeRollbackOperations(live=None) + # When / Then with pytest.raises(CloudflarePolicyError, match="no live production deployment"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) - assert _rollbacks(operations) == [] @pytest.mark.asyncio -async def test_should_refuse_mismatched_project_binding() -> None: +async def test_should_refuse_rollback_when_project_name_differs() -> None: + # Given operations = FakeRollbackOperations(project_name="other") + # When / Then with pytest.raises(CloudflarePolicyError, match="project binding differs"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) + assert _rollbacks(operations) == [] @pytest.mark.asyncio @@ -242,142 +384,139 @@ async def test_should_refuse_mismatched_project_binding() -> None: ), ) async def test_should_fail_closed_when_rollback_api_rejects(error: CloudflareApiError) -> None: + # Given operations = FakeRollbackOperations(rollback_error=error) + # When / Then with pytest.raises(CloudflareApiError): await rollback_production(operations, PROJECT, _id(PREVIOUS)) - assert operations.events[-1] == f"rollback:{_id(PREVIOUS)}" @pytest.mark.asyncio -async def test_should_refuse_rollback_response_for_different_deployment() -> None: +async def test_should_raise_api_error_when_rollback_body_is_unsuccessful() -> None: + # Given + operations = UnsuccessfulRollbackOperations() + + # When / Then + with pytest.raises(CloudflareApiError, match="API request failed"): + await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + +@pytest.mark.asyncio +async def test_should_refuse_when_rollback_response_names_another_deployment() -> None: + # Given operations = FakeRollbackOperations(response_id=OLDER) + # When / Then with pytest.raises(CloudflarePolicyError, match="rollback response identity differs"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) @pytest.mark.asyncio -async def test_should_fail_closed_when_production_does_not_serve_rollback_target() -> None: +async def test_should_fail_closed_when_production_never_serves_rollback_target() -> None: + # Given operations = FakeRollbackOperations(converge=False) + # When / Then with pytest.raises(CloudflarePolicyError, match="does not serve the rollback target"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) - - sleeps = [event for event in operations.events if event.startswith("sleep:")] - assert sleeps == ["sleep:1", "sleep:2", "sleep:4", "sleep:8"] + assert _sleeps(operations) == BOUNDED_DELAYS @pytest.mark.asyncio -async def test_should_accept_rollback_after_delayed_production_convergence() -> None: +async def test_should_accept_rollback_when_production_converges_after_first_delay() -> None: + # Given operations = DelayedOperations() + # When evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) + # Then assert evidence.live_deployment_id == _id(PREVIOUS) - assert "sleep:1" in operations.events + assert _sleeps(operations) == ["sleep:1"] @pytest.mark.asyncio -async def test_should_bound_rollback_verification_with_hard_deadline( +async def test_should_accept_rollback_when_production_converges_after_final_delay() -> None: + # Given + operations = LateOperations() + + # When + evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) + + # Then + assert evidence.live_deployment_id == _id(PREVIOUS) + assert _sleeps(operations) == BOUNDED_DELAYS + + +@pytest.mark.asyncio +async def test_should_fail_closed_when_verification_deadline_expires( monkeypatch: pytest.MonkeyPatch, ) -> None: + # Given monkeypatch.setattr(asyncio, "timeout", lambda _: ImmediateTimeout()) operations = FakeRollbackOperations() + # When / Then with pytest.raises(CloudflarePolicyError, match="does not serve the rollback target"): await rollback_production(operations, PROJECT, _id(PREVIOUS)) -@dataclass -class DelayedOperations(FakeRollbackOperations): - """Production switches only after the first verification read.""" - - pending: str | None = None - - async def rollback(self, deployment_id: str) -> str: - self.events.append(f"rollback:{deployment_id}") - self.pending = deployment_id.split("-", maxsplit=1)[0] - return _rollback_response(self.pending) - - async def sleep(self, seconds: int) -> None: - self.events.append(f"sleep:{seconds}") - self.live = self.pending - - -class ImmediateTimeout: - async def __aenter__(self) -> None: - raise TimeoutError - - async def __aexit__(self, error_type: object, error: object, traceback: object) -> None: - return None - - @pytest.mark.asyncio -async def test_should_report_live_deployment_as_recorded_rollback_target() -> None: +async def test_should_record_live_deployment_when_reading_rollback_target() -> None: + # Given operations = FakeRollbackOperations() + # When deployment = await live_production_deployment(operations, PROJECT) - assert (deployment.id, deployment.url) == (_id(LIVE), f"https://{LIVE}.{PROJECT}.pages.dev") + # Then + assert (deployment.id, deployment.url) == (_id(LIVE), _url(LIVE)) assert operations.events == ["get-project", "get-deployments"] @pytest.mark.asyncio -async def test_should_refuse_to_record_target_without_live_deployment() -> None: +async def test_should_refuse_to_record_target_when_nothing_is_live() -> None: + # Given + operations = FakeRollbackOperations(live=None) + + # When / Then with pytest.raises(CloudflarePolicyError, match="no live production deployment"): - await live_production_deployment(FakeRollbackOperations(live=None), PROJECT) + await live_production_deployment(operations, PROJECT) @pytest.mark.asyncio -async def test_should_refuse_to_record_live_target_missing_from_history() -> None: - operations = FakeRollbackOperations(rows=(_row(PREVIOUS),)) +async def test_should_refuse_to_record_target_when_live_is_missing_from_history() -> None: + # Given + operations = FakeRollbackOperations(rows=(Row(PREVIOUS),)) + # When / Then with pytest.raises(CloudflarePolicyError, match="not in recent production history"): await live_production_deployment(operations, PROJECT) @pytest.mark.asyncio -async def test_should_refuse_malformed_live_deployment_reference() -> None: +async def test_should_refuse_to_record_target_when_live_reference_is_malformed() -> None: + # Given operations = MalformedLiveOperations() + # When / Then with pytest.raises(CloudflarePolicyError, match="schema mismatch"): await live_production_deployment(operations, PROJECT) -@dataclass -class MalformedLiveOperations(FakeRollbackOperations): - async def get_project(self) -> str: - payload = json.loads(_project(LIVE)) - payload["result"]["canonical_deployment"] = {"id": 7} - return json.dumps(payload) - - -@dataclass(frozen=True) -class RecordingRollbackTransport(CurlRollbackOperations): - """Curl adapter double that records the exact method, suffix, and body.""" - - requests: list[tuple[str, str, str]] = field(default_factory=list) - - async def _request(self, method: str, suffix: str, body: str = "") -> str: - self.requests.append((method, suffix, body)) - return "response" - - -def _transport() -> RecordingRollbackTransport: - secret = cast(dagger.Secret, object()) - return RecordingRollbackTransport(secret, secret, PROJECT) - - @pytest.mark.asyncio -async def test_should_build_only_documented_rollback_requests() -> None: +async def test_should_send_only_documented_requests_when_rolling_back() -> None: + # Given transport = _transport() + # When await transport.get_project() await transport.get_deployments() await transport.rollback(_id(PREVIOUS)) + # Then assert transport.requests == [ ("GET", "/pages/projects/edge-reco", ""), ("GET", "/pages/projects/edge-reco/deployments?env=production&per_page=10", ""), @@ -387,162 +526,123 @@ async def test_should_build_only_documented_rollback_requests() -> None: @pytest.mark.asyncio @pytest.mark.parametrize("value", ("", "../../projects/x", "ABC", "a" * 65, "a/b")) -async def test_should_refuse_malformed_rollback_id_before_transport(value: str) -> None: +async def test_should_refuse_before_transport_when_rollback_id_is_malformed(value: str) -> None: + # Given transport = _transport() + # When / Then with pytest.raises(CloudflarePolicyError, match="deployment id is malformed"): await transport.rollback(value) - assert transport.requests == [] -def test_should_refuse_malformed_project_before_transport() -> None: - secret = cast(dagger.Secret, object()) +def test_should_refuse_before_transport_when_project_name_is_malformed() -> None: + # Given + project = "../edge" + # When / Then with pytest.raises(CloudflarePolicyError, match="project name is malformed"): - CurlRollbackOperations(secret, secret, "../edge") + CurlRollbackOperations(_secret(), _secret(), project) @pytest.mark.asyncio @pytest.mark.parametrize("status", ("400", "403", "409", "500", "503")) -async def test_should_raise_api_error_for_rollback_http_failure( +async def test_should_raise_api_error_when_rollback_returns_http_failure( monkeypatch: pytest.MonkeyPatch, status: str ) -> None: + # Given container = FailingContainer(f'{status}\n{{"errors":[],"messages":[],"success":false}}') monkeypatch.setattr(main_module, "_curl_request_container", lambda *_: container) - secret = cast(dagger.Secret, object()) + operations = CurlRollbackOperations(_secret(), _secret(), PROJECT) + # When / Then with pytest.raises(CloudflareApiError, match="API request failed"): - await CurlRollbackOperations(secret, secret, PROJECT).rollback(_id(PREVIOUS)) - + await operations.rollback(_id(PREVIOUS)) -@dataclass -class FailingContainer: - output: str - async def stdout(self) -> str: - return self.output - - -def test_should_send_post_with_empty_json_body_and_no_retry() -> None: +def test_should_send_post_with_body_and_without_retry_when_method_writes() -> None: + # Given / When script = main_module._curl_script() + # Then assert 'if [ "$method" = PATCH ] || [ "$method" = POST ]; then' in script assert "retry = 0" in script -def _public_methods(*names: str) -> list[ast.AsyncFunctionDef]: - tree = ast.parse(Path(main_module.__file__).read_text()) - public = next( - node - for node in tree.body - if isinstance(node, ast.ClassDef) and node.name == "CloudflarePages" - ) - return [ - node - for node in public.body - if isinstance(node, ast.AsyncFunctionDef) and node.name in names - ] - - -def test_should_expose_rollback_functions_uncached_with_secret_credentials() -> None: - methods = _public_methods("rollback", "previous_production_deployment") - - assert {method.name for method in methods} == {"rollback", "previous_production_deployment"} - for method in methods: - decorator = next(item for item in method.decorator_list if isinstance(item, ast.Call)) - cache = next(item.value for item in decorator.keywords if item.arg == "cache") - assert isinstance(cache, ast.Constant) - assert cache.value == "never" - arguments = { - argument.arg: ast.unparse(argument.annotation or ast.Constant(None)) - for argument in method.args.args - } - credentials = (arguments["cloudflare_api_token"], arguments["cloudflare_account_id"]) - assert credentials == (SECRET_TYPE, SECRET_TYPE) - assert not set(arguments).intersection({"url", "origin", "command", "cmd", "script"}) - - -@pytest.mark.asyncio -async def test_should_raise_api_error_for_unsuccessful_rollback_body() -> None: - operations = UnsuccessfulRollbackOperations() - - with pytest.raises(CloudflareApiError, match="API request failed"): - await rollback_production(operations, PROJECT, _id(PREVIOUS)) - - -@dataclass -class UnsuccessfulRollbackOperations(FakeRollbackOperations): - async def rollback(self, deployment_id: str) -> str: - self.events.append(f"rollback:{deployment_id}") - return json.dumps({"errors": [], "messages": [], "success": False, "result": None}) +def test_should_expose_rollback_functions_as_uncached_when_published() -> None: + # Given + methods = _public_methods() + # When + cache_values = {method.name: _cache_value(method) for method in methods} -@pytest.mark.asyncio -async def test_should_accept_rollback_served_after_final_bounded_delay() -> None: - operations = LateOperations() + # Then + assert cache_values == dict.fromkeys(ROLLBACK_FUNCTIONS, "never") - evidence = await rollback_production(operations, PROJECT, _id(PREVIOUS)) - - assert evidence.live_deployment_id == _id(PREVIOUS) - assert [event for event in operations.events if event.startswith("sleep:")][-1] == "sleep:8" - - -@dataclass -class LateOperations(DelayedOperations): - """Production switches only during the final bounded delay.""" - async def sleep(self, seconds: int) -> None: - self.events.append(f"sleep:{seconds}") - if seconds == 8: - self.live = self.pending +def test_should_take_typed_secrets_without_escape_hatches_when_published() -> None: + # Given + methods = _public_methods() + # When + signatures = [_annotations(method) for method in methods] -def _patched_pages( - monkeypatch: pytest.MonkeyPatch, operations: FakeRollbackOperations -) -> main_module.CloudflarePages: - monkeypatch.setattr(main_module, "CurlRollbackOperations", lambda *_: operations) - return main_module.CloudflarePages() + # Then + assert len(signatures) == len(ROLLBACK_FUNCTIONS) + for arguments in signatures: + credentials = (arguments["cloudflare_api_token"], arguments["cloudflare_account_id"]) + assert credentials == (SECRET_TYPE, SECRET_TYPE) + assert ESCAPE_HATCHES.isdisjoint(arguments) @pytest.mark.asyncio -async def test_should_return_public_rollback_evidence(monkeypatch: pytest.MonkeyPatch) -> None: - secret = cast(dagger.Secret, object()) +async def test_should_return_public_evidence_when_rollback_function_succeeds( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Given pages = _patched_pages(monkeypatch, FakeRollbackOperations()) - evidence = await pages.rollback(secret, secret, PROJECT) + # When + evidence = await pages.rollback(_secret(), _secret(), PROJECT) - public = (evidence.project, evidence.from_deployment_id, evidence.to_deployment_id) - assert public == (PROJECT, _id(LIVE), _id(PREVIOUS)) + # Then + moved = (evidence.project, evidence.from_deployment_id, evidence.to_deployment_id) + assert moved == (PROJECT, _id(LIVE), _id(PREVIOUS)) live = (evidence.live_deployment_id, evidence.live_deployment_url) - assert live == (_id(PREVIOUS), f"https://{PREVIOUS}.{PROJECT}.pages.dev") + assert live == (_id(PREVIOUS), _url(PREVIOUS)) @pytest.mark.asyncio -async def test_should_return_public_recorded_target(monkeypatch: pytest.MonkeyPatch) -> None: - secret = cast(dagger.Secret, object()) +async def test_should_return_recorded_target_without_writing_when_helper_is_called( + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Given operations = FakeRollbackOperations() pages = _patched_pages(monkeypatch, operations) - recorded = await pages.previous_production_deployment(secret, secret, PROJECT) + # When + recorded = await pages.previous_production_deployment(_secret(), _secret(), PROJECT) + # Then assert (recorded.project, recorded.deployment_id) == (PROJECT, _id(LIVE)) - assert recorded.deployment_url == f"https://{LIVE}.{PROJECT}.pages.dev" + assert recorded.deployment_url == _url(LIVE) assert _rollbacks(operations) == [] @pytest.mark.asyncio -async def test_should_sleep_between_rollback_verification_reads( +async def test_should_use_async_sleep_when_waiting_between_verification_reads( monkeypatch: pytest.MonkeyPatch, ) -> None: + # Given delays: list[float] = [] async def record(seconds: float) -> None: delays.append(seconds) monkeypatch.setattr(asyncio, "sleep", record) - secret = cast(dagger.Secret, object()) - await CurlRollbackOperations(secret, secret, PROJECT).sleep(2) + # When + await CurlRollbackOperations(_secret(), _secret(), PROJECT).sleep(2) + # Then assert delays == [2]