diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 00000000..792fdf3d --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,136 @@ +# AGENTS.md — guide for AI agents working in this repository + +If you are an AI coding agent (Claude Code, Codex, Cursor, GitHub Copilot Workspace, etc.) about to make changes to this repository, read this file first. It will save you from generating code that diverges from the codebase's conventions. + +If you are a human contributor, the same conventions apply to you — but the more comprehensive [`docs/CONTRIBUTING.md`](docs/CONTRIBUTING.md) and the deep-dive references linked below are written for you specifically. + +## What this repo is + +`pytfe` is the official Python SDK for the HCP Terraform and Terraform Enterprise V2 API. It wraps roughly 50 resource services (workspaces, runs, policies, teams, agents, …) and is consumed by downstream projects. Source layout: + +``` +src/pytfe/ + client.py # TFEClient — composition root, wires every resource + config.py # TFEConfig — auth, timeout, retry, proxy settings + _http.py # HTTPTransport — request, retry, redirects, auth + _jsonapi.py # JSON:API envelope helpers + errors.py # Typed exception hierarchy (TFEError + ~80 subclasses) + utils.py # Validation + small helpers + models/ # Pydantic v2 models, one file per resource + resources/ # Service classes, one file per resource + +tests/units/ # Pytest unit tests with mocked transport, one file per resource +examples/ # Runnable CLI demos, one file per resource (or extended) +docs/ # Internal reference (see below) +``` + +## Required reading before generating code + +These three documents define the patterns this codebase already uses. Generating code without consulting them will produce inconsistent output: + +| Topic | Doc | +|---|---| +| `list_*` methods, pagination, iterator vs list, the `_list` helper | [`docs/ITERATORS.md`](docs/ITERATORS.md) | +| Pydantic model conventions: `ConfigDict`, aliases, validators, relationships, exporting | [`docs/MODELS.md`](docs/MODELS.md) | +| Resource service patterns: method shape, JSON:API envelopes, client wiring, examples | [`docs/RESOURCE.md`](docs/RESOURCE.md) | + +Each doc ends with a checklist. Use those checklists; they encode the rules a reviewer will look for. + +## Source verification for API shape + +Before adding a new resource, endpoint, enum, or non-obvious response parser, verify the wire contract against primary sources: + +- Official HCP Terraform API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs +- go-tfe implementation: https://github.com/hashicorp/go-tfe +- OpenAPI specs or live probes when the public docs/go-tfe are missing, beta, or ambiguous + +Use the official docs and go-tfe as the first sources of truth. OpenAPI and live probes are supporting evidence, especially for endpoints that are newly released or not fully documented yet. When behavior is surprising, note the source you checked in the PR description, test name, example header, or a short code comment. + +## The cardinal rules + +A handful of conventions are pervasive enough that you'll regret breaking them. In rough order of "how loudly it breaks at review time": + +1. **`list_*` methods return `Iterator[X]`.** Not `list[X]`, not `Iterable[X]`, not a custom `Pager`. Use `for item in self._list(path): yield ...` inside the method body. ([ITERATORS.md](docs/ITERATORS.md)) + +2. **JSON:API attribute names go through `Field(alias="...")`.** The API sends `created-at`, Python uses `created_at`. Pair with `model_config = ConfigDict(populate_by_name=True, validate_by_name=True)` on the model. ([MODELS.md](docs/MODELS.md)) + +3. **`model_dump(by_alias=True, exclude_none=True)` for write payloads.** Without `by_alias=True` you'll send snake_case to the API and it will silently drop the fields. Add `mode="json"` if the options contain enums. + +4. **For new public APIs, prefer typed `TFEError` subclasses.** The error hierarchy in `errors.py` is part of the public API, and downstream consumers often `except TFEError:` once. Existing methods still expose many `ValueError` paths; do not change those established exceptions unless the breaking-change impact is explicitly accepted. + +5. **Validate IDs at the top of every method.** Use `valid_string_id` from `utils.py`. New methods should prefer typed `InvalidIDError` errors; existing resources may already use `ValueError` and should keep that public behavior unless a breaking change is intentional. + +6. **Wire every new resource into `client.py`.** A resource not added to `TFEClient.__init__` is unreachable. Same for new models in `models/__init__.py`. + +7. **Use the standard verb names: `list`, `read`, `create`, `update`, `delete`.** Plus `add_*` / `remove_*` for relationship modifications. Argument order is always *identifiers first, options last*. + +## Things that look reasonable but are actually wrong here + +These are mistakes a competent Python developer would make if they hadn't read the conventions. Avoid them: + +- **Don't catch `httpx` errors directly.** The transport already translates them into `TFEError` subclasses. Catching `httpx.HTTPError` in a resource means the typed error never propagates. +- **Always send the bearer token, even to absolute URLs returned by the API.** Endpoints like `hosted_state_download_url`, `hosted_state_upload_url`, plan `json-output`, and apply `errored-state` redirect to `archivist.terraform.io` — which is HashiCorp infrastructure that *requires* the bearer. go-tfe does the same (see `state_version.go::Download` + `tfe.go::NewRequest`). Stripping the bearer breaks downstream consumers (notably the Ansible collection's statefile + dynamic-inventory flows). `HTTPTransport.request` accepts `include_auth=False` only as an opt-out for the hypothetical case of calling a genuinely non-HashiCorp host; do not use it for Archivist URLs. +- **Don't write a custom page loop.** `self._list(path, params=...)` handles pagination + non-paginated endpoints transparently. Rolling your own loop will diverge from the rest of the codebase. +- **Don't reuse generators.** Iterators returned by `list_*` are single-use. If you need to traverse twice, `materialized = list(client.foo.list_bars(...))` first. +- **Don't add features beyond what was asked.** This codebase is approaching v1.0.0. Adding "while I'm here" refactors or speculative abstractions slows reviews and risks breaking the Ansible collection. +- **Don't assume every successful response is `{"data": ...}`.** Check the docs/go-tfe/spec for each endpoint: some return a JSON:API envelope, some return a bare resource object, `204 No Content`, `null`, raw bytes, or a redirect to a blob URL. Add tests for non-standard shapes. +- **Don't use bare `list[...]` annotations inside a resource class after defining `def list(...)`.** In class scope, mypy can resolve `list` to the method instead of the builtin. Use `builtins.list[...]`, `Sequence[...]`, or another unshadowed type. + +## Known cross-dependencies you should not break + +| Consumer | What they depend on | +|---|---| +| `hashicorp/terraform-ansible-collection` | The pytfe public API — resource methods, model fields, exception classes. Any signature change here is a breaking change. In particular, `client.projects.list_tag_bindings` is consumed with an `isinstance(response, list)` check; it intentionally still returns `list[TagBinding]` (see [ITERATORS.md](docs/ITERATORS.md) — Known exceptions). | +| Downstream user code generally | Method signatures, return types, model fields, and exception types. New errors should subclass an existing parent so `except TFEError:` continues to work, but existing `ValueError` behavior should not be changed casually. | + +When in doubt about whether a change is breaking: check `gh search code '' --owner hashicorp` to see if the Ansible repo uses it. + +## How to make a change + +This is the workflow that produces low-friction reviews. Follow it. + +1. **Understand the scope first.** If the task is "add resource X", read `docs/RESOURCE.md` end-to-end. If it's "fix bug in Y", read `Y`'s current implementation and tests before touching anything. +2. **Check official API docs and go-tfe for the canonical API shape.** `pytfe` mirrors the HCP Terraform API and often follows go-tfe's surface. URL paths, method names, payload shapes, response shapes, enum values, and redirect behavior should be verified against https://developer.hashicorp.com/terraform/cloud-docs/api-docs and https://github.com/hashicorp/go-tfe before designing anything. +3. **Add models first** (`src/pytfe/models/.py`), then the resource (`src/pytfe/resources/.py`), then wire both into the respective `__init__.py` / `client.py`. +4. **Write tests.** Mock `HTTPTransport`. One test per method, plus an invalid-id case for every public method. See `tests/units/test_comment.py` as a small reference. +5. **Run `make test` and `make lint`.** Both must pass. `pytest tests/units/` runs the suite directly; it should be < 2 seconds. +6. **Add or extend an example.** Real engineers will copy-paste it; make it work end-to-end. Use env vars (`TFE_TOKEN`, `TFE_ORG`) for auth, never hard-code credentials. +7. **If the change is non-trivial, verify live.** The repo doesn't run integration tests in CI, so the only way to catch a wrong URL or a typo in an attribute alias is to run the example against a real organization. + +## Things to never do + +- **Never put a token, password, or other credential in any file.** Use environment variables. The user will rotate them after; you don't need to know them. +- **Never use `git push --force` or `git reset --hard` without explicit instruction.** Same for `--no-verify`, force-push to `main`, or rebasing public commits. +- **Never commit `.env`, `credentials.json`, `*.tfstate`, or anything with secrets.** Match against the existing `.gitignore` if unsure. +- **Never bypass pre-commit hooks.** If a hook fails, fix the underlying issue. +- **Never run an example that creates real resources against production without explicit user confirmation.** Sandbox orgs are safe; user's actual workspace is not. + +## Style + +The codebase uses [ruff](https://docs.astral.sh/ruff/) for both formatting and linting and [mypy](https://mypy.readthedocs.io/) for type checking. Type hints are required on every public method's signature. Docstrings are required on every public method — keep them to one or two lines unless the behavior is genuinely non-obvious. + +Comments are minimal by design. A comment should explain *why* something non-obvious is true, not *what* the code does. The names and types should be enough to convey "what". + +```python +# ❌ Don't +# Increment the counter by 1 +counter += 1 + +# ✅ Do (only when the why is non-obvious) +# Run task stages are wire values, not Python names. If the API/go-tfe says +# "pre-plan", keep the hyphen; do not "normalize" it to snake_case. +stage_value = raw_value +``` + +## When you're done + +A reasonable PR includes: + +- Code (resource + models) +- Tests covering every public method +- An updated or new example +- A short `CHANGELOG.md` entry under `# v.0 (Unreleased)` describing the user-visible change + +Open the PR with a description that explains *why* the change is needed, links to any HCP Terraform API docs or go-tfe code referenced, and notes any behavior changes a downstream consumer might see. + +The reviewer's checklist will be the union of the checklists in [`docs/ITERATORS.md`](docs/ITERATORS.md), [`docs/MODELS.md`](docs/MODELS.md), and [`docs/RESOURCE.md`](docs/RESOURCE.md). Pre-running them yourself is the fastest way to a merge. diff --git a/README.md b/README.md index 45c869f0..2a507001 100644 --- a/README.md +++ b/README.md @@ -36,8 +36,7 @@ config = TFEConfig( client = TFEClient(config) -orgs = client.organizations.list() -for org in orgs.items: +for org in client.organizations.list(): print(org.name) ``` @@ -57,8 +56,7 @@ from pytfe import TFEClient, TFEConfig # Equivalent to providing no values; falls back to env vars if set. client = TFEClient(TFEConfig()) -orgs = client.organizations.list() -for org in orgs.items: +for org in client.organizations.list(): print(org.name) ``` @@ -69,11 +67,32 @@ from pytfe import TFEClient, TFEConfig config = TFEConfig(address="", token="") client = TFEClient(config) -orgs = client.organizations.list() -for org in orgs.items: +for org in client.organizations.list(): print(org.name) ``` +## Listing resources + +Anything named `list` or `list_*` on a resource service returns an **iterator**, not a Python `list`. Pagination is handled for you under the hood — the iterator keeps fetching pages from the API until there are no more. This mirrors the underlying HCP Terraform API, where every list endpoint is paginated (`page[number]` / `page[size]`), and keeps memory flat even when an organization has thousands of workspaces or runs. + +You'll use it one of two ways: + +```python +# Stream — handy when you might break early or when results are large +for ws in client.workspaces.list("my-org"): + if ws.name.startswith("prod-"): + print(ws.id, ws.name) + +# Materialize — when you actually want a list to index, len(), or pass around +workspaces = list(client.workspaces.list("my-org")) +print(f"found {len(workspaces)} workspaces") +``` + +A couple of things worth knowing: + +- The iterator is **single-use**. Once you've walked it, iterating again gives you nothing. Capture it with `list(...)` first if you need to reuse the result. +- Filters and page size live on the `*ListOptions` model for each resource — e.g. `WorkspaceListOptions(search="prod", page_size=50)`. Pagination still happens transparently; `page_size` only controls how big each underlying API page is. + ## Documentation - API reference and guides (SDK): **coming soon** diff --git a/docs/ITERATORS.md b/docs/ITERATORS.md new file mode 100644 index 00000000..46634a24 --- /dev/null +++ b/docs/ITERATORS.md @@ -0,0 +1,187 @@ +# Iterators and pagination in pyTFE + +This is internal reference for anyone — human or AI — adding a new resource to the SDK or auditing existing ones. Keep this file up to date as the conventions evolve. + +## The one-line rule + +> Any method named `list` or `list_*` on a resource service returns `Iterator[X]`. Never `list[X]`, never `LazyList`, never a custom `*Pager`. Just `Iterator[X]`. + +The matching import is `from collections.abc import Iterator`, not `typing.Iterator` (which is deprecated alias since 3.9). + +This rule applies to public resource service methods. Private parsing helpers may return concrete lists when they are just internal implementation details and are not part of the SDK public surface. + +## Why iterators + +The HCP Terraform API paginates **every** list endpoint. The contract is uniform: pass `page[number]` and `page[size]`, read pagination metadata out of the response envelope, and follow links until there are no more. Materialising the entire result set up front would mean fetching every page synchronously before the caller sees the first row — fine for ten workspaces, painful for ten thousand. Iterators let the SDK do the right thing by default: lazy under the hood, simple at the call site. + +Even when an endpoint isn't actually paginated (some single-shot relationship reads like `effective-tag-bindings`), we still use the iterator signature. The reason is purely consistency — a future contributor (or an LLM generating new resources by analogy) should never have to think about which list method is which shape. If it's named `list_*`, it returns `Iterator[X]`. + +## How callers use them + +There are two idioms. Both are normal and expected. + +```python +# 1. Stream — handy when results are large, or you can break early. +for ws in client.workspaces.list("my-org"): + if ws.name.startswith("prod-"): + print(ws.id) + break + +# 2. Materialize — when you actually want a list to len(), index, or hand off. +workspaces = list(client.workspaces.list("my-org")) +print(f"{len(workspaces)} workspaces") +``` + +Two things every caller needs to know: + +- **Iterators are single-use.** Iterating an already-walked iterator yields nothing. If you need to traverse the same result more than once, capture it with `list(...)` first. +- **Iterators are always truthy.** `if iterator:` is True even when the iterator is empty. Use `materialized = list(...); if materialized:` if you need a non-empty check. + +## How to implement a new `list_*` method + +There is **one canonical pattern** in the codebase, and it works for both paginated and non-paginated endpoints. Use it unless you have a specific reason not to. + +### The canonical pattern: `self._list(...)` + `yield` + +```python +def list( + self, organization: str, options: WorkspaceListOptions | None = None +) -> Iterator[Workspace]: + if not valid_string_id(organization): + raise InvalidOrgError() + + params = options.model_dump(by_alias=True, exclude_none=True, mode="json") if options else {} + path = f"/api/v2/organizations/{organization}/workspaces" + for item in self._list(path, params=params): + yield self._workspace_from(item) +``` + +That's it. `self._list()` lives in `_base.py` and handles `page[number]` / `page[size]` and follow-through automatically. It is also robust to endpoints that **don't paginate** — if the response has no pagination metadata and the returned data is smaller than the requested page size, the helper just breaks after one round-trip. So you do not need a different code path for relationship reads like `GET /workspaces/{id}/tag-bindings` (single response) versus list endpoints like `GET /organizations/{org}/workspaces` (paginated). The same `for item in self._list(path): yield ...` works for both. + + +### Note on lazy validation + +A Python generator function defers its entire body until the caller calls `next()`. That means `if not valid_string_id(...): raise ...` only fires on first iteration, not at call time. Practically, callers iterate immediately so this is fine — but tests need to materialize before asserting that a `ValueError` is raised: + +```python +# tests/units/test_*.py — invalid-id case +with pytest.raises(InvalidOrgError): + list(client.workspaces.list("")) # wrap with list() to force iteration +``` + +If you genuinely need eager validation (raised from the call expression itself, not the first `for` loop), use the wrapper pattern: + +```python +def list(self, organization: str, ...) -> Iterator[Workspace]: + if not valid_string_id(organization): + raise InvalidOrgError() # eager + + params = ... + path = ... + def _gen() -> Iterator[Workspace]: + for item in self._list(path, params=params): + yield self._workspace_from(item) + return _gen() +``` + +Both forms appear in the codebase (`policy_set.list` uses the wrapper; most others don't). Pick the wrapper only when eager-error behavior is important to a specific resource. + +### Mypy note: `def list` shadows `list[...]` + +Inside a class that defines a method named `list`, mypy can resolve later bare annotations like `list[str]` to the method instead of the builtin type. If the same resource class has helper methods after `def list(...)`, avoid bare `list[...]` in those later signatures or annotations. Use one of these instead: + +```python +import builtins +from collections.abc import Sequence + + +def add_users(self, team_id: str, usernames: builtins.list[str]) -> None: ... +def add_users(self, team_id: str, usernames: Sequence[str]) -> None: ... +``` + +### The one deviation: `return iter(list)` for endpoints with fallback logic + +There is exactly one situation where neither the plain generator nor the wrapper pattern works well: when the method has a **try/except fallback that calls a different endpoint** after the primary one fails. A pure generator could yield items from the primary endpoint, fail partway through, then switch to the fallback and yield duplicates. + +For this case — and this case only — fetch eagerly and return `iter(materialized_list)`: + +```python +def list_versions( + self, module_id: RegistryModuleID +) -> Iterator[RegistryModuleVersion]: + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + try: + versions = [...] # primary endpoint + return iter(versions) + except Exception: + try: + versions = [...] # fallback: different endpoint + return iter(versions) + except Exception: + return iter([]) +``` + +`registry_module.list_versions` is the only method in the codebase that does this. Add a docstring note explaining the reason if you find yourself reaching for this pattern, so future readers don't mistake it for something to copy. + +Do **not** reach for `iter(list)` just because the endpoint is non-paginated. Use `self._list()` for those — that's the convention. + +### Shape that does **not** match the convention (don't do this) + +```python +# ❌ Returns Iterable instead of Iterator — looks similar, isn't. +def list(...) -> Iterable[Workspace]: ... + +# ❌ Returns Pager / LazyList / custom wrapper. +def list(...) -> WorkspaceList: ... + +# ❌ Returns concrete list. The type is a public contract; consumers will +# rely on len(), indexing, and isinstance(result, list). See "Known +# exceptions" below for the one method where this is documented. +def list_widgets(...) -> list[Widget]: ... +``` + +## Known exceptions (and why) + +A handful of methods deliberately diverge from the convention. They're tracked here so future audits don't try to "fix" them and silently break a downstream consumer. + +| Method | Returns | Why we left it | +|---|---|---| +| `projects.list_tag_bindings` | `list[TagBinding]` | Downstream code checks `isinstance(response, list)`, so changing this would be a breaking change. | +| `registry_module.list_commits` | `CommitList` | The endpoint returns a typed envelope with metadata fields beyond just the list of commits. A custom return type is appropriate here. | +| `registry_module.list_versions` | `Iterator[X]` via `iter(list)` | Has a try/except fallback path that calls a different endpoint on failure. See the deviation pattern above. | + +Anything public and not in that table should follow the canonical pattern. If you find one that doesn't, either fix it or add it to the table with the compatibility reason. + +## How to test a `list_*` method + +Mock the transport, then materialize with `list(...)` to assert: + +```python +def test_list_workspaces(self): + mock_response = Mock() + mock_response.json.return_value = { + "data": [{"id": "ws-1", "attributes": {"name": "first"}}], + "meta": {"pagination": {"current-page": 1, "total-pages": 1}}, + } + self.mock_transport.request.return_value = mock_response + + result = list(self.workspaces_service.list("my-org")) + + assert len(result) == 1 + assert result[0].id == "ws-1" +``` + +Don't assert `isinstance(result, list)` against the raw return — that asserts on the *idiom* the caller chose, not on the SDK contract. If you want to assert iterator semantics, use `isinstance(result, Iterator)` from `collections.abc`. + +## Quick checklist when reviewing a new resource PR + +- [ ] Every `list*` method returns `Iterator[X]`, not `list[X]` or `Iterable[X]` +- [ ] `Iterator` is imported from `collections.abc`, not `typing` +- [ ] The body uses the canonical `for item in self._list(path, params=params): yield ...` pattern — including for non-paginated single-shot endpoints +- [ ] Hand-rolled `iter(materialized_list)` only appears if the method has a try/except fallback to a different endpoint (extremely rare — has a docstring note explaining why) +- [ ] If a class defines `def list(...)`, later annotations in that class avoid bare `list[...]` so mypy does not resolve `list` to the method +- [ ] Examples that call the method use `list(client.foo.list_bars(...))` (or stream with a `for` loop) — never assume list semantics on the bare return +- [ ] Unit tests materialize with `list(...)` before asserting length/indexing; invalid-id tests also wrap with `list(...)` to force iteration +- [ ] The README's `## Listing resources` section is still accurate after your change diff --git a/docs/MODELS.md b/docs/MODELS.md new file mode 100644 index 00000000..2c036090 --- /dev/null +++ b/docs/MODELS.md @@ -0,0 +1,263 @@ +# Models — Pydantic conventions in pyTFE + +This is internal reference for adding or editing Pydantic models in `src/pytfe/models/`. The patterns below are what the codebase already does; follow them so new resources line up with what's there. + +All models inherit from `pydantic.BaseModel` and target Pydantic v2. The `from __future__ import annotations` line is at the top of every model file so forward references and type hints work without runtime imports. + +## Layout of a model file + +One model file per resource, named after the resource (`workspace.py`, `agent.py`, `team.py`). Each file usually contains: + +1. **Enums** for fixed string sets the API uses (status, type, kind). +2. **The main resource model** (the thing you get back from a `read`/`list` — e.g. `Workspace`, `Team`). +3. **`*CreateOptions`** for `POST` requests. +4. **`*UpdateOptions`** for `PATCH` requests. +5. **`*ListOptions`** for `GET` collection requests (filters, pagination, includes). +6. **`*ReadOptions`** for `GET` single-resource requests that take `include[]` (only when needed). + +Use a single file unless the model surface is large enough that splitting helps. There's no "package per resource" pattern here — one file is the default. + +## `ConfigDict` + +New or touched `BaseModel` classes should set `model_config = ConfigDict(...)` unless you are deliberately preserving a local legacy pattern. Several older models predate this convention; do not mass-refactor them just to satisfy this rule because changing validation/coercion behavior can be a public API change. The conventions for new work are: + +| Setting | When to use | +|---|---| +| `populate_by_name=True` | **Always.** Lets callers pass either the field name (`created_at=...`) or the alias (`{"created-at": ...}`) when constructing. | +| `validate_by_name=True` | Use on models that are parsed *from* API responses **or** constructed by callers via field names. Pair with `populate_by_name=True`. | +| `extra="forbid"` | Use on `*CreateOptions` / `*UpdateOptions` / option models where you want a typo (`workspce_id=...`) to fail loudly instead of being silently dropped. Don't put it on response models — the API can add fields and we don't want that to break parsing. | +| `arbitrary_types_allowed=True` | Only when you genuinely have a non-Pydantic type in a field (rare). | + +The standard line you'll write 90% of the time: + +```python +class Foo(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + ... +``` + +## Field aliases: JSON:API hyphens → Python snake_case + +HCP Terraform speaks JSON:API, which uses hyphenated attribute names (`created-at`, `auto-apply`, `state-versions`). Python uses snake_case. Bridge with `Field(alias=...)`: + +```python +from datetime import datetime +from pydantic import BaseModel, ConfigDict, Field + + +class Run(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + has_changes: bool | None = Field(None, alias="has-changes") + is_destroy: bool | None = Field(None, alias="is-destroy") + auto_apply: bool | None = Field(None, alias="auto-apply") + created_at: datetime | None = Field(None, alias="created-at") + canceled_at: datetime | None = Field(None, alias="canceled-at") +``` + +Rules: + +- **Every multi-word JSON:API attribute** gets an alias. Don't try to invent a snake_case-to-hyphen mapper — be explicit per field. +- **Page params** use the JSON:API square-bracket form: `Field(None, alias="page[number]")`, `Field(None, alias="page[size]")`. +- **Filter params** use the same convention: `Field(None, alias="filter[workspace][name]")`. +- **`include`** is a comma-separated string on the wire but exposed as `list[SomeEnum] | None` in Python; the resource layer dumps options with `mode="json"` and joins the resulting values (`",".join(params["include"])`). See the `policy_set.read_with_options` pattern. + +## Optional vs required vs default fields + +The codebase is conservative about which fields are required. The pattern: + +- **Resource models** (parsed from API responses): almost everything except `id` is `field: T | None = Field(None, alias="...")`. The API may omit fields depending on permissions or include params, so being permissive avoids brittle parsing. +- **`*CreateOptions`**: required fields use `field: T = Field(..., description="...")` (Pydantic's "required" sentinel). Optional fields use `field: T | None = None`. +- **`*UpdateOptions`**: **everything** is optional (`field: T | None = None`). `PATCH` semantics — only set fields are sent. +- **Collection fields**: prefer `default_factory=list` over `= []` (avoids the mutable-default trap). For maps, `default_factory=dict`. + +Example: + +```python +class WorkspaceCreateOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + name: str = Field(..., description="Workspace name") + description: str | None = None + auto_apply: bool | None = Field(None, alias="auto-apply") + project: dict | None = None # relationship — see "Relationships" below + + +class WorkspaceUpdateOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + name: str | None = None + description: str | None = None + auto_apply: bool | None = Field(None, alias="auto-apply") +``` + +## Enums + +String enums with explicit string values, mirroring what the API returns: + +```python +from enum import Enum + + +class RunStatus(str, Enum): + PENDING = "pending" + PLANNING = "planning" + PLANNED = "planned" + APPLIED = "applied" + ERRORED = "errored" + DISCARDED = "discarded" +``` + +A few conventions: + +- **`str, Enum`** so the value is JSON-serialisable without `.value` indirection (pydantic handles this with `mode="json"` on `model_dump`). +- **`SCREAMING_SNAKE`** member names. Values mirror the wire string exactly — usually lowercase, sometimes with underscores. Don't change the wire value to "look nicer". +- When the API uses hyphenated values (`"pre-plan"`, `"post-plan"`), keep the hyphens in the value string. Verify enum values against the official HCP Terraform API docs, go-tfe, or live API if unsure — there have been past bugs where underscore values diverged from what the server actually returns. +- Put enums **above** the model that uses them in the same file. + +## Validators + +Two flavours, both Pydantic v2: + +### `model_validator(mode="after")` for option models + +Use on `*CreateOptions` / `*UpdateOptions` to enforce required-name / valid-ID rules at construction time. For new public APIs, prefer a typed `TFEError` subclass from `pytfe.errors`. For existing option models that already raise `ValueError`, preserve that behavior unless the breaking-change impact is explicitly accepted: + +```python +from pydantic import model_validator +from ..errors import InvalidNameError, RequiredNameError +from ..utils import valid_string, valid_string_id + + +class AgentPoolCreateOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + name: str = Field(..., alias="name") + + @model_validator(mode="after") + def valid(self) -> AgentPoolCreateOptions: + if not valid_string(self.name): + raise RequiredNameError() + if not valid_string_id(self.name): + raise InvalidNameError() + return self +``` + +### `field_validator` for per-field coercion or normalisation + +Use sparingly — only when you need to massage input before Pydantic's default coercion, or when a single field has a non-trivial rule: + +```python +from pydantic import field_validator + + +class NotificationConfiguration(BaseModel): + @field_validator("triggers", mode="before") + @classmethod + def _coerce_triggers(cls, v): + ... +``` + +`mode="before"` runs on the raw input; `mode="after"` runs on the already-validated value. Default to `"after"` unless you need pre-validation cleanup. + +## Relationships + +JSON:API responses include a `relationships` block separate from `attributes`. Two ways to model relationship references on the resource: + +### Option 1 — ID stub on the related model + +When you only need the related id, use a typed stub. The resource layer fills it in via `Model.model_construct(id=...)`: + +```python +class TaskStage(BaseModel): + model_config = ConfigDict(populate_by_name=True) + id: str + run: Run | None = Field(None, alias="run") # only .id populated + task_results: list[TaskResult] | None = Field(None, alias="task-results") +``` + +Use `model_construct` (not `model_validate`) in the resource for these stubs — it skips validation, which is correct because you only have `{id, type}`: + +```python +attributes["run"] = Run.model_construct(id=run_data["id"]) +``` + +### Option 2 — Flat `*_id` field + +When the relationship is "owned" by this resource and just one id matters, expose it as a flat `*_id` field (with hyphen alias if needed). Less plumbing, fine when you don't need the related model object: + +```python +class TeamWorkspaceAccess(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + id: str + team_id: str | None = Field(default=None, alias="team-id") + workspace_id: str | None = Field(default=None, alias="workspace-id") +``` + +The resource layer reads from `relationships.team.data.id` and stuffs it into `attributes["team-id"]` before calling `model_validate`. See `resources/team_workspace_access.py:_parse`. + +Pick Option 1 when callers may want to traverse the relationship further (e.g. `task_stage.run.id`). Pick Option 2 when the id is all you'll ever need. + +## Forward references and `model_rebuild` + +If a model A references model B and B references A (or A is defined before B), Pydantic can't resolve the forward ref at class-definition time. The fix: leave the annotation as a string in the model file, then call `Model.model_rebuild()` from `models/__init__.py` once everything is imported. + +The block at the bottom of `models/__init__.py` is where this happens: + +```python +Run.model_rebuild( + raise_errors=False, + _types_namespace={"TaskStage": TaskStage}, +) +Workspace.model_rebuild( + raise_errors=False, + _types_namespace={"AgentPool": AgentPool, "Run": Run, "TaskStage": TaskStage}, +) +``` + +`raise_errors=False` is the project default — failure to resolve a forward ref shouldn't crash the SDK at import time. Add your new model's rebuild call there if it has forward-referenced relations. + +## Exporting + +Two things to update when you add a model: + +1. **Imports** at the top of `models/__init__.py` — add your new classes alphabetically within their section. +2. **`__all__`** at the bottom — add the names that should be importable as `from pytfe.models import Foo`. + +Don't forget option models, enums, and any include-opt enums. The `__all__` list is what users see in `pytfe.models` — if it's not there, it's not part of the public API. + +## What NOT to do + +```python +# ❌ Don't use bare strings for the alias when the field has multiple words. +created_at: datetime | None = None # parses "created_at", misses "created-at" + +# ❌ Don't reach for arbitrary_types_allowed unless you actually have one. + +# ❌ Don't use mutable default values directly. +tags: list[str] = [] # all instances share the same list +tags: list[str] = Field(default_factory=list) # ✅ + +# Prefer a typed TFEError subclass for new public APIs. +raise ValueError("name required") # existing APIs may still do this +raise RequiredNameError() # preferred for new APIs + +# ❌ Don't model relationships as raw dicts when there's a typed stub option. +workspace: dict | None = None # loses type information +workspace: Workspace | None = None # ✅ (filled via model_construct in resource) +``` + +## Checklist when adding a new model + +- [ ] `from __future__ import annotations` at the top +- [ ] New or touched classes use `model_config = ConfigDict(populate_by_name=True, validate_by_name=True)` unless preserving a local legacy pattern +- [ ] Hyphenated JSON:API attribute names → `Field(alias="...")` +- [ ] Response model fields default to `T | None = Field(None, alias="...")` +- [ ] `*CreateOptions` uses `Field(...)` for required fields, `T | None = None` for optional +- [ ] `*UpdateOptions` is fully optional +- [ ] Enums are `str, Enum` with SCREAMING_SNAKE member names and wire-faithful values +- [ ] New validators prefer typed `TFEError` subclasses; existing `ValueError` behavior is not changed without an explicit compatibility decision +- [ ] Collections use `default_factory=list` / `default_factory=dict` +- [ ] Added to `models/__init__.py` imports + `__all__` +- [ ] If you used forward references, added a `model_rebuild()` call at the bottom of `models/__init__.py` diff --git a/docs/RESOURCE.md b/docs/RESOURCE.md new file mode 100644 index 00000000..b3be73cd --- /dev/null +++ b/docs/RESOURCE.md @@ -0,0 +1,469 @@ +# Resources — adding a new resource to pyTFE + +This is internal reference for adding or editing resource services in `src/pytfe/resources/`. A "resource" here is a service class like `Workspaces`, `Comments`, `TeamWorkspaceAccesses` — it wraps a related set of HCP Terraform API endpoints. The patterns below reflect what the codebase already does. Follow them. + +Companion docs you'll need alongside this one: + +- [MODELS.md](MODELS.md) — how to define the Pydantic models the resource takes and returns +- [ITERATORS.md](ITERATORS.md) — how `list_*` methods are shaped +- The [examples/](../examples) directory — runnable demos for each resource + +## What a resource file looks like + +One file per resource in `src/pytfe/resources/`, named after the resource (`workspaces.py`, `policy_set.py`, `team_workspace_access.py`). The class inside is the plural form (`Workspaces`, `PolicySets`, `TeamWorkspaceAccesses`). + +Every resource class inherits from `_Service` (in `_base.py`), which gives it: + +- `self.t` — the `HTTPTransport` for making requests +- `self._list(path, params=...)` — the paginated iterator helper (see [ITERATORS.md](ITERATORS.md)) + +Standard file scaffolding: + +```python +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import InvalidWorkspaceIDError, InvalidOrgError +from ..models.widget import Widget, WidgetCreateOptions, WidgetListOptions +from ..utils import valid_string_id +from ._base import _Service + + +class Widgets(_Service): + """Service for managing widgets.""" + + def list(...) -> Iterator[Widget]: ... + def read(...) -> Widget: ... + def create(...) -> Widget: ... + def update(...) -> Widget: ... + def delete(...) -> None: ... + + def _widget_from(self, data: dict[str, Any]) -> Widget: ... +``` + +The private `_widget_from(data)` helper at the bottom is convention — every resource that returns a model has one, used to translate a JSON:API resource object into the Pydantic model. + +## Method conventions + +The verbs are stable across the codebase. Use these names exactly. + +| Verb | Signature | HTTP | Returns | +|---|---|---|---| +| `list(...)` | `(parent_id, options=None)` | `GET` collection endpoint | `Iterator[Widget]` | +| `read(id)` | `(widget_id)` | `GET /widgets/{id}` | `Widget` | +| `read_with_options(id, options)` | `(widget_id, options)` | `GET /widgets/{id}?include=...` | `Widget` | +| `create(parent_id, options)` | parent first, options last | `POST` | `Widget` | +| `update(id, options)` | `(widget_id, options)` | `PATCH /widgets/{id}` | `Widget` | +| `delete(id)` | `(widget_id)` | `DELETE /widgets/{id}` | `None` | + +Argument-order rule: **identifiers first, options last**. `create(organization, options)`, `update(widget_id, options)`. Never reverse this; downstream callers rely on it. + +For relationship endpoints (`POST /widgets/{id}/relationships/foos`), use verbs like: + +- `add_*(id, options)` / `remove_*(id, options)` — modifies an unordered set +- `update_*(id, options)` — replaces the entire set +- `attach_*` / `detach_*` — pair-style operations +- `assign_*` — when there's a single relation being set (e.g. `assign_ssh_key`) + +Pick the verb that mirrors what go-tfe and the API docs use — consistency across SDKs matters when users are reading both. + +## Source verification + +Before implementing a new endpoint, check the primary sources for the exact contract: + +- Official HCP Terraform API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs +- go-tfe implementation: https://github.com/hashicorp/go-tfe +- OpenAPI specs or live probes when docs/go-tfe are missing, beta, or ambiguous + +Verify URL path, HTTP method, request envelope, enum values, response shape, redirects, and whether the feature is generally available. If the implementation depends on a surprising behavior, record the source in the PR description, test name, example header, or a short code comment. + +## Validation: eager, with typed errors + +Validate every ID/name argument at the top of every method. Two helpers from `utils.py`: + +- `valid_string(s)` — non-empty string +- `valid_string_id(s)` — non-empty string with no `/` or whitespace (the JSON:API id contract) + +For new public APIs, prefer a typed `TFEError` subclass from `pytfe/errors.py`. Existing resources still expose many `ValueError` validation paths; do not change those established exception types unless the breaking-change impact is explicitly accepted. + +```python +from ..errors import InvalidWidgetIDError +from ..utils import valid_string_id + +def read(self, widget_id: str) -> Widget: + if not valid_string_id(widget_id): + raise InvalidWidgetIDError() + r = self.t.request("GET", f"/api/v2/widgets/{widget_id}") + return self._widget_from(r.json().get("data", {})) +``` + +If the typed error class you need doesn't exist yet, add it to `errors.py`. Follow the existing naming: + +- `InvalidIDError(InvalidValues)` — the id is missing or malformed +- `RequiredError(InvalidValues)` — a field that must be set wasn't +- `NotFoundError(NotFound)` — the API returned 404 (use sparingly; usually the transport raises `NotFound` already) + +Subclass from a sensible parent (`InvalidValues`, `WorkspaceValidationError`, etc.) so consumers can `except TFEError:` once and catch new errors. When touching existing methods, preserve their historical exception behavior unless the change is intentionally breaking. + +## Building JSON:API request payloads + +Most resource write requests use the JSON:API envelope: + +```python +payload = { + "data": { + "type": "widgets", + "attributes": options.model_dump(by_alias=True, exclude_none=True), + } +} +self.t.request("POST", "/api/v2/...", json_body=payload) +``` + +Key arguments to `model_dump`: + +- **`by_alias=True`** — emit the hyphenated JSON:API attribute names, not the Python snake_case field names. Without this you'll send `auto_apply` instead of `auto-apply` and the API will silently ignore it. +- **`exclude_none=True`** — don't send fields the caller didn't set. `PATCH` semantics depend on this. +- **`mode="json"`** — when your options contain enums, including query params such as `include`. Without `mode="json"`, an enum field serialises as `EnumClass.MEMBER` (the repr) instead of the wire value. The bug is silent — the API returns 400 with "Invalid parameter". + +So for option models that contain enum fields: + +```python +params = options.model_dump(by_alias=True, exclude_none=True, mode="json") +if isinstance(params.get("include"), list): + params["include"] = ",".join(params["include"]) +``` + +For relationships, use the JSON:API identifier-object shape: + +```python +payload = { + "data": { + "type": "team-workspaces", + "attributes": attrs, + "relationships": { + "team": {"data": {"type": "teams", "id": team_id}}, + "workspace": {"data": {"type": "workspaces", "id": workspace_id}}, + }, + } +} +``` + +For *replace-many* relationships, pass an array of identifiers: + +```python +payload = { + "data": [ + {"type": "workspaces", "id": wid} for wid in workspace_ids + ] +} +self.t.request("POST", f"/api/v2/projects/{project_id}/relationships/workspaces", json_body=payload) +``` + +## Parsing responses + +Do not assume every successful response is a JSON:API envelope. Check the official docs/go-tfe/spec before writing the parser. Common shapes in this SDK include: + +- JSON:API envelope: `{"data": {...}}` or `{"data": [{...}]}` +- Bare resource object with top-level `attributes` +- `204 No Content` +- `null` +- Raw bytes +- `3xx` redirect to a presigned blob URL + +Add unit tests for every non-standard shape a method supports. The common `_widget_from(data)` helper takes a single JSON:API `data` object (already unwrapped from the envelope by the caller) and returns the Pydantic model: + +```python +def _widget_from(self, data: dict[str, Any]) -> Widget: + attrs = dict(data.get("attributes") or {}) + attrs["id"] = data.get("id") + return Widget.model_validate(attrs) +``` + +If the model has relationships, pull them from `data["relationships"]` and either: + +1. **Embed an id-stub** using `Model.model_construct(id=...)` — use this when the model defines the relation as `OtherModel | None`. `model_construct` skips validation, which is correct for partial `{id, type}` data: + + ```python + relationships = data.get("relationships", {}) + run_data = relationships.get("run", {}).get("data") + if run_data: + attributes["run"] = Run.model_construct(id=run_data["id"]) + ``` + +2. **Flatten to `*_id`** when the model exposes a flat `team_id: str | None` field: + + ```python + team_data = (relationships.get("team") or {}).get("data") or {} + if team_data.get("id"): + attributes["team-id"] = team_data["id"] + ``` + +Always defensively coalesce with `or {}` — relationships may be missing from sparse responses. + +## Presigned URLs and redirects + +The TFE bearer token must not be forwarded to Archivist, S3, or other presigned blob hosts. Signed upload/download URLs already carry their own credentials. + +- Direct signed URL: `self.t.request("GET", url, include_auth=False)` +- API endpoint that returns a redirect: call the API path with `allow_redirects=False`, read the `Location` header, then fetch that URL with `include_auth=False` +- Add a unit test that asserts the blob URL call uses `include_auth=False` + +This applies to state upload/download, plan JSON output/schema, apply errored state, and any future blob-backed endpoint. + +## Pagination — use `self._list`, don't roll your own + +`_Service._list(path, params=...)` is the universal helper. It yields raw `dict` items from the `data` array, transparently following pagination. It gracefully handles single-shot non-paginated endpoints too — see [ITERATORS.md](ITERATORS.md) for the full breakdown. + +Don't write your own page loop. If you find yourself doing it, you're solving a problem `_list()` already handled. + +## URL paths + +Always start with `/api/v2/...`. The base URL is set on the transport, but the path includes the API version prefix: + +```python +"/api/v2/organizations/{organization}/widgets" # collection scoped to org +"/api/v2/widgets/{widget_id}" # single resource +"/api/v2/widgets/{widget_id}/relationships/foos" # JSON:API relationship route +"/api/v2/widgets/{widget_id}/actions/lock" # action endpoint +``` + +Use f-strings to interpolate ids — they've been validated by `valid_string_id` above. URL-quote organization names with `urllib.parse.quote` only when the API explicitly requires it (most don't). + +## Errors raised by the transport + +`HTTPTransport.request` raises typed errors from `pytfe.errors` based on status code: + +- `AuthError` for 401/403 +- `NotFound` for 404 +- `RateLimited` for 429 (with `.retry_after`) +- `ServerError` for 5xx +- `TFEError` for everything else 4xx + +You usually don't need to catch these — let them propagate to the caller. Catch only when: + +- You want to translate to a more specific error (`except TFEError as e: if "rate-limit" in str(e): raise ...`) +- The "error" is actually an expected outcome — like a `NotFound` meaning "no current assessment yet": + + ```python + try: + r = self.t.request("GET", f"/api/v2/workspaces/{ws_id}/current-assessment-result") + except NotFound: + return None + ``` + +## Wiring into the client + +Two places to update: + +### `src/pytfe/client.py` + +1. Add the import alphabetically within its section. +2. Add `self.widgets = Widgets(self._transport)` to `TFEClient.__init__`, grouped with related resources. + +```python +from .resources.widget import Widgets +... +self.widgets = Widgets(self._transport) +``` + +The attribute name on the client is **plural snake_case** (`workspaces`, `team_tokens`, `team_workspace_accesses`). It must match the class name's lowercased plural. + +## Typing gotcha: `def list` shadows `list[...]` + +If a resource class defines `def list(...)`, mypy can resolve later annotations in the same class like `list[str]` to the method instead of the builtin type. For helper methods defined after `list`, avoid bare `list[...]`. Use `builtins.list[...]`, `Sequence[...]`, or another unshadowed collection type. + +### `src/pytfe/models/__init__.py` + +If you added new models (almost always yes), wire them through: + +1. Import them alphabetically in the right section block. +2. Add their names to the `__all__` list at the bottom. + +If your models use forward references, add a `Model.model_rebuild(...)` call at the bottom — see [MODELS.md](MODELS.md). + +## Tests + +One test file per resource: `tests/units/test_widget.py`. Mirror the structure of `tests/units/test_comment.py` (small and clean) or `tests/units/test_workspaces.py` (large). + +The structure: + +```python +import pytest +from unittest.mock import Mock + +from pytfe._http import HTTPTransport +from pytfe.errors import InvalidWidgetIDError +from pytfe.models.widget import Widget, WidgetCreateOptions +from pytfe.resources.widget import Widgets + + +class TestWidgets: + @pytest.fixture + def mock_transport(self): + return Mock(spec=HTTPTransport) + + @pytest.fixture + def service(self, mock_transport): + return Widgets(mock_transport) + + def test_read_widget_invalid_id(self, service): + with pytest.raises(InvalidWidgetIDError): + service.read("") + + def test_list_widgets(self, service, mock_transport): + mock_response = Mock() + mock_response.json.return_value = { + "data": [{"id": "wid-1", "type": "widgets", "attributes": {...}}], + "meta": {"pagination": {"current-page": 1, "total-pages": 1}}, + } + mock_transport.request.return_value = mock_response + + result = list(service.list("my-org")) # materialize Iterator + + assert len(result) == 1 + assert result[0].id == "wid-1" +``` + +Two things test reviewers always check: + +- **Invalid-ID tests** for list methods wrap with `list(...)` to force iteration — generator-based methods defer validation. See [ITERATORS.md](ITERATORS.md). +- **JSON:API path assertions** match exactly. If you're constructing `/api/v2/widgets/{id}` with f-string interpolation, the test asserts the exact path. Don't be tempted to leave wildcards. + +`make test` runs everything; `make lint` runs ruff + mypy. Both must pass. + +## Examples — when and how + +New resources should get an example file in `examples/`, or an existing example should be extended if that is the natural home. The purpose isn't comprehensive coverage — it's "a real engineer landing on this repo can copy-paste this and have a working demo in 60 seconds". Prefer the current style for new examples, but do not churn older examples only to rename helpers or match prose. + +### When to make a new example file vs extend an existing one + +- **New resource, no existing example for the parent surface** → new file (`examples/widget.py`) +- **New method on an existing resource** → extend the existing example file, gate the new section behind a flag like `--demo-foo` or `--show-bar` +- **Single related feature spread across multiple resources** → pick the most natural home; don't duplicate + +We've already consolidated some examples into existing ones — see `examples/apply.py --recover-errored-state` for the pattern. When in doubt, extend rather than fragment. + +### Example file structure + +```python +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import WidgetCreateOptions + + +def _print_header(title: str) -> None: + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Widgets demo for python-tfe SDK") + parser.add_argument( + "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) + parser.add_argument("--organization", default=os.getenv("TFE_ORG", "")) + parser.add_argument("--widget-id", help="Widget id for read/update/delete") + parser.add_argument("--list", action="store_true", help="List widgets") + parser.add_argument("--create", action="store_true", help="Create a widget") + args = parser.parse_args() + + if not args.token: + print("TFE_TOKEN is not set") + return 2 + + client = TFEClient(TFEConfig(address=args.address, token=args.token)) + + if args.list: + _print_header(f"Listing widgets for {args.organization}") + for w in client.widgets.list(args.organization): + print(f" - {w.id} {w.name}") + + if args.create: + _print_header("Creating a widget") + w = client.widgets.create( + args.organization, WidgetCreateOptions(name="example") + ) + print(f" created {w.id}") + # If the example creates resources, also clean them up at the end. + try: + client.widgets.delete(w.id) + print(f" cleaned up {w.id}") + except Exception as e: + print(f" WARN: cleanup failed: {e}") + + client.close() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) +``` + +### Rules examples follow + +- **Auth via env vars by default** — `TFE_TOKEN`, `TFE_ADDRESS`, `TFE_ORG`. Never hard-code a token. +- **`argparse` for CLI**, with sensible `--` flags so users can opt into specific demos. +- **`_print_header(title)` helper** for visual separation between sections in new examples. Older examples vary; do not churn them only for naming consistency. +- **Cleanup what you create.** If the example creates scratch resources, delete them in a `try/finally`. Warn but don't fail on cleanup errors. +- **Exit codes**: `return 0` on success, `return 2` for missing config, raise on unexpected SDK errors so the user sees the traceback. +- **`client.close()` at the end** when you opened a client. + +## What NOT to do + +```python +# ❌ Reaching directly through self.t for paginated endpoints — use self._list +data = self.t.request("GET", path).json()["data"] # one page only, no pagination +for item in self._list(path): ... # ✅ + +# ❌ Letting the JSON:API envelope leak out +return r.json() # raw dict with "data"/"included"/etc. +return self._widget_from(r.json()["data"]) # ✅ + +# ❌ Sending snake_case attrs to the API +options.model_dump(exclude_none=True) # workspace_id → wrong on the wire +options.model_dump(by_alias=True, exclude_none=True) # ✅ + +# ❌ Forgetting mode="json" with enums — silent 400 from the API +options.model_dump(by_alias=True, exclude_none=True) # enum becomes 'EnumClass.MEMBER' +options.model_dump(by_alias=True, exclude_none=True, mode="json") # ✅ + +# Prefer typed errors for new public APIs. Preserve existing ValueError +# behavior unless the compatibility impact is explicitly accepted. +raise ValueError("invalid widget id") # existing APIs may do this +raise InvalidWidgetIDError() # preferred for new APIs + +# ❌ Forgetting to wire the resource into the client +# Just add `self.widgets = Widgets(self._transport)` in client.py — the +# resource is otherwise unreachable from TFEClient. +``` + +## Checklist when adding a new resource + +- [ ] New file `src/pytfe/resources/widget.py` with `class Widgets(_Service)` +- [ ] Standard verbs (`list`, `read`, `create`, `update`, `delete`) with the standard signatures +- [ ] Every method validates IDs; new public APIs prefer typed `TFEError` subclasses, while established `ValueError` behavior is preserved unless intentionally changed +- [ ] Write requests use the JSON:API envelope; `model_dump` uses `by_alias=True, exclude_none=True`, plus `mode="json"` if there are enums +- [ ] `list*` returns `Iterator[X]` via `self._list(...)` (see [ITERATORS.md](ITERATORS.md)) +- [ ] Response parsing helper `_widget_from(data)` translates JSON:API → Pydantic +- [ ] Non-standard response shapes (`204`, `null`, bare resources, raw bytes, redirects) are verified against docs/go-tfe/spec and covered by tests +- [ ] Presigned upload/download/blob URLs are fetched with `include_auth=False` +- [ ] Classes with `def list(...)` avoid later bare `list[...]` annotations +- [ ] Models added per [MODELS.md](MODELS.md), wired in `models/__init__.py` +- [ ] Resource wired into `client.py` (import + `self.widgets = Widgets(...)`) +- [ ] Unit tests in `tests/units/test_widget.py`, including invalid-id cases and at least one happy-path per method +- [ ] Example in `examples/widget.py` (or extension to existing file), with env-var auth, cleanup, and `_print_header` +- [ ] `make test` and `make lint` both pass diff --git a/examples/apply.py b/examples/apply.py index ea72dfa8..8c32dc07 100644 --- a/examples/apply.py +++ b/examples/apply.py @@ -22,6 +22,15 @@ def main(): ) parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) parser.add_argument("--apply-id", required=True, help="Apply ID to work with") + parser.add_argument( + "--recover-errored-state", + action="store_true", + help="Fetch the failed-upload state via /applies/{id}/errored-state", + ) + parser.add_argument( + "--out", + help="When recovering errored state, write the bytes to this path", + ) args = parser.parse_args() cfg = TFEConfig(address=args.address, token=args.token) @@ -49,6 +58,27 @@ def main(): print(f"Error reading apply: {e}") return 1 + if args.recover_errored_state: + from pytfe.errors import NotFound + + _print_header("Recovering errored state (GET /applies/{id}/errored-state)") + try: + data = client.applies.errored_state(args.apply_id) + except NotFound: + print( + "No errored state available — apply did not fail during state " + "upload, or storage retention has elapsed." + ) + else: + print(f"Recovered {len(data)} bytes of errored state") + if args.out: + with open(args.out, "wb") as f: + f.write(data) + print(f"Wrote {args.out}") + else: + preview = data[:256].decode("utf-8", errors="replace") + print(f"--- preview ---\n{preview}\n--- end preview ---") + print("\n" + "=" * 80) print("Apply demo completed successfully!") print("=" * 80) diff --git a/examples/configuration_version.py b/examples/configuration_version.py index b894c116..5488a58b 100644 --- a/examples/configuration_version.py +++ b/examples/configuration_version.py @@ -859,6 +859,42 @@ def main(): print("Functions 10: Enterprise backing data operations") print("=" * 80) + # ===================================================== + # TEST 11: INGRESS ATTRIBUTES (VCS metadata) + # ===================================================== + print("\n11. Testing ingress_attributes() function:") + cv_for_ingress = uploadable_cv_id or created_cv_id + if cv_for_ingress: + try: + ingress = client.configuration_versions.ingress_attributes(cv_for_ingress) + if ingress is None: + print( + f"CV {cv_for_ingress} has no ingress attributes " + "(non-VCS-backed configuration version)." + ) + else: + print(f"Ingress attributes for {cv_for_ingress}:") + for field in ( + "branch", + "clone_url", + "commit_sha", + "commit_message", + "commit_url", + "identifier", + "is_pull_request", + "pull_request_number", + "pull_request_title", + "tag", + "sender_username", + ): + value = getattr(ingress, field, None) + if value is not None: + print(f" {field}: {value}") + except Exception as e: + print(f"Failed to read ingress attributes: {e}") + else: + print("Skipped — no CV was created in this run.") + # Close client client.close() diff --git a/examples/plan.py b/examples/plan.py index 8910bd07..a95e318c 100644 --- a/examples/plan.py +++ b/examples/plan.py @@ -22,13 +22,27 @@ def main(): "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") ) parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) - parser.add_argument("--plan-id", required=True, help="Plan ID to work with") + parser.add_argument("--plan-id", required=False, help="Plan ID to work with") + parser.add_argument( + "--run-id", + help="Run ID — fetches the plan and JSON output via the run instead " + "of needing the plan id", + ) parser.add_argument("--save-json", help="Path to save JSON output") args = parser.parse_args() + if not args.plan_id and not args.run_id: + parser.error("provide --plan-id and/or --run-id") cfg = TFEConfig(address=args.address, token=args.token) client = TFEClient(cfg) + # If we were given only --run-id, resolve the plan via the run. + if not args.plan_id and args.run_id: + _print_header(f"Reading plan for run {args.run_id}") + plan_for_run = client.plans.read_for_run(args.run_id) + args.plan_id = plan_for_run.id + print(f"Resolved plan id: {args.plan_id}") + # 1) Read the plan details _print_header("Reading Plan Details") try: @@ -81,6 +95,33 @@ def main(): except Exception as e: print(f"Error reading JSON output: {e}") + # 3) Run-id-based endpoints + if args.run_id: + _print_header(f"Reading JSON output via run id ({args.run_id})") + try: + json_for_run = client.plans.read_json_output_for_run(args.run_id) + if json_for_run is None: + print("Plan has not yet completed (HTTP 204).") + else: + print( + f"JSON keys: {sorted(json_for_run.keys())[:8]} " + f"(total {len(json_for_run)})" + ) + except Exception as e: + print(f"Error: {e}") + + _print_header(f"Reading provider JSON schema via run id ({args.run_id})") + try: + schema = client.plans.read_json_schema_for_run(args.run_id) + if schema is None: + print("Plan has not yet completed (HTTP 204).") + elif isinstance(schema, dict): + print(f"Schema keys: {sorted(schema.keys())[:8]}") + else: + print(f"Schema type: {type(schema).__name__}") + except Exception as e: + print(f"Error: {e}") + print("\n" + "=" * 80) print("Plan demo completed successfully!") print("=" * 80) diff --git a/examples/policy_set.py b/examples/policy_set.py index 69ca88e2..9acdc30b 100644 --- a/examples/policy_set.py +++ b/examples/policy_set.py @@ -11,15 +11,20 @@ Policy, PolicyKind, PolicySetAddPoliciesOptions, + PolicySetAddProjectExclusionsOptions, PolicySetAddProjectsOptions, PolicySetAddWorkspacesOptions, PolicySetCreateOptions, + PolicySetIncludeOpt, PolicySetListOptions, + PolicySetReadOptions, PolicySetRemovePoliciesOptions, + PolicySetRemoveProjectExclusionsOptions, PolicySetRemoveProjectsOptions, PolicySetRemoveWorkspacesOptions, PolicySetUpdateOptions, Project, + ProjectCreateOptions, Workspace, ) @@ -133,6 +138,12 @@ def main(): parser.add_argument("--search", help="Search policy sets by name") parser.add_argument("--page", type=int, default=1) parser.add_argument("--page-size", type=int, default=20) + parser.add_argument( + "--demo-project-exclusions", + action="store_true", + help="End-to-end demo: create a scratch global policy set + project, " + "add the project to exclusions, then remove it and clean up.", + ) args = parser.parse_args() if not args.token: @@ -443,6 +454,79 @@ def main(): except Exception as e: print(f"Error deleting policy set: {e}") + # 12) Demo: project-exclusions lifecycle (creates scratch resources) + if args.demo_project_exclusions: + import time + + _print_header("Project-exclusions lifecycle demo (scratch resources)") + stamp = int(time.time()) + created_ps_id = None + created_proj_id = None + try: + ps = client.policy_sets.create( + args.org, + PolicySetCreateOptions(name=f"pytfe-pe-{stamp}", Global=True), + ) + created_ps_id = ps.id + print(f"created policy set: {ps.id} ({ps.name}, global=True)") + + proj = client.projects.create( + args.org, + ProjectCreateOptions(name=f"pytfe-pe-proj-{stamp}"), + ) + created_proj_id = proj.id + print(f"created project: {proj.id} ({proj.name})") + + print(f"\nadding project {proj.id} to exclusions of {ps.id}") + client.policy_sets.add_project_exclusions( + ps.id, + PolicySetAddProjectExclusionsOptions( + project_exclusions=[Project(id=proj.id)] + ), + ) + print("added") + + ps_after = client.policy_sets.read_with_options( + ps.id, + PolicySetReadOptions( + include=[PolicySetIncludeOpt.POLICY_SET_PROJECT_EXCLUSIONS] + ), + ) + excluded_ids = [p.id for p in (ps_after.project_exclusions or [])] + print(f"current excluded projects: {excluded_ids}") + + print(f"\nremoving project {proj.id} from exclusions") + client.policy_sets.remove_project_exclusions( + ps.id, + PolicySetRemoveProjectExclusionsOptions( + project_exclusions=[Project(id=proj.id)] + ), + ) + print("removed") + ps_final = client.policy_sets.read_with_options( + ps.id, + PolicySetReadOptions( + include=[PolicySetIncludeOpt.POLICY_SET_PROJECT_EXCLUSIONS] + ), + ) + print( + "final excluded projects: " + f"{[p.id for p in (ps_final.project_exclusions or [])]}" + ) + finally: + if created_proj_id: + try: + client.projects.delete(created_proj_id) + print(f"cleaned up project {created_proj_id}") + except Exception as e: + print(f"WARN: could not clean up project: {e}") + if created_ps_id: + try: + client.policy_sets.delete(created_ps_id) + print(f"cleaned up policy set {created_ps_id}") + except Exception as e: + print(f"WARN: could not clean up policy set: {e}") + if __name__ == "__main__": main() diff --git a/examples/project.py b/examples/project.py index f6f500bf..83db6119 100644 --- a/examples/project.py +++ b/examples/project.py @@ -15,6 +15,7 @@ ProjectSettingOverwrites, ProjectUpdateOptions, TagBinding, + WorkspaceCreateOptions, ) @@ -125,6 +126,19 @@ def main() -> None: action="store_true", help="Append a short random suffix to --name for create", ) + parser.add_argument( + "--move-workspace-id", + action="append", + default=[], + help="Workspace id to move into --project-id (repeatable). Requires " + "--project-id.", + ) + parser.add_argument( + "--demo-move", + action="store_true", + help="End-to-end demo: create scratch projects + workspace, move the " + "workspace between projects, clean up.", + ) args = parser.parse_args() @@ -278,7 +292,7 @@ def main() -> None: # 7) List effective tag bindings if args.list_effective_tag_bindings: _print_header(f"Listing effective tag bindings for project: {args.project_id}") - bindings = client.projects.list_effective_tag_bindings(args.project_id) + bindings = list(client.projects.list_effective_tag_bindings(args.project_id)) if not bindings: print("No effective tag bindings found.") @@ -308,6 +322,66 @@ def main() -> None: client.projects.delete_tag_bindings(args.project_id) print("Deleted all project tag bindings") + # 10) Move workspaces into the given project (additive, not destructive) + if args.move_workspace_id: + if not args.project_id: + raise SystemExit("--project-id is required for --move-workspace-id") + _print_header( + f"Moving {len(args.move_workspace_id)} workspace(s) into " + f"project {args.project_id}" + ) + client.projects.move_workspaces(args.project_id, args.move_workspace_id) + print("done") + + # 11) End-to-end demo: create scratch resources, move, cleanup + if args.demo_move: + import time + + _print_header("project.move_workspaces end-to-end demo (scratch resources)") + stamp = int(time.time()) + created: dict[str, str] = {} + try: + src = client.projects.create( + args.organization, + ProjectCreateOptions(name=f"pytfe-move-src-{stamp}"), + ) + created["src_project"] = src.id + print(f"created source project: {src.id} ({src.name})") + dst = client.projects.create( + args.organization, + ProjectCreateOptions(name=f"pytfe-move-dst-{stamp}"), + ) + created["dst_project"] = dst.id + print(f"created target project: {dst.id} ({dst.name})") + ws = client.workspaces.create( + args.organization, + WorkspaceCreateOptions( + name=f"pytfe-move-ws-{stamp}", project={"id": src.id} + ), + ) + created["workspace"] = ws.id + print(f"created workspace: {ws.id} in {src.id}") + client.projects.move_workspaces(dst.id, [ws.id]) + ws2 = client.workspaces.read_by_id(ws.id) + moved = ws2.project.id if ws2.project else "?" + print(f"workspace now belongs to: {moved}") + assert moved == dst.id + print("OK") + finally: + if "workspace" in created: + try: + client.workspaces.delete_by_id(created["workspace"]) + print(f"cleaned up workspace {created['workspace']}") + except Exception as e: + print(f"WARN: workspace cleanup failed: {e}") + for key in ("dst_project", "src_project"): + if key in created: + try: + client.projects.delete(created[key]) + print(f"cleaned up project {created[key]}") + except Exception as e: + print(f"WARN: project {key} cleanup failed: {e}") + if __name__ == "__main__": main() diff --git a/examples/registry_module.py b/examples/registry_module.py index bf8e7ab6..8a7bc088 100644 --- a/examples/registry_module.py +++ b/examples/registry_module.py @@ -304,8 +304,7 @@ def main(): registry_name=RegistryName.PRIVATE, ) - versions = client.registry_modules.list_versions(module_id) - versions_list = list(versions) if hasattr(versions, "__iter__") else [] + versions_list = list(client.registry_modules.list_versions(module_id)) print(f"Found {len(versions_list)} versions") for i, version in enumerate(versions_list[:3], 1): diff --git a/examples/state_versions.py b/examples/state_versions.py index e9a98d3d..dcafcd00 100644 --- a/examples/state_versions.py +++ b/examples/state_versions.py @@ -40,6 +40,16 @@ def main(): parser.add_argument("--download", help="Path to save downloaded current state") parser.add_argument("--upload", help="Path to a .tfstate (or JSON state) to upload") parser.add_argument("--page-size", type=int, default=10) + parser.add_argument( + "--rollback-to", + help="State version id to roll the workspace back to. The workspace " + "will be locked, rolled back, then unlocked.", + ) + parser.add_argument( + "--rollback-dry-run", + action="store_true", + help="With --rollback-to, print the plan without performing the rollback.", + ) args = parser.parse_args() cfg = TFEConfig(address=args.address, token=args.token) @@ -161,6 +171,31 @@ def main(): # Some older/self-hosted versions don’t support direct upload print(f"Upload not supported on this server: {e}") + # 6) (Optional) Roll back to a previous state version + if args.rollback_to: + _print_header( + f"Rolling {args.workspace_id} back to state version {args.rollback_to}" + ) + if args.rollback_dry_run: + print("--rollback-dry-run set; not locking or rolling back") + else: + print("locking workspace ...") + client.workspaces.lock( + args.workspace_id, + WorkspaceLockOptions(reason="python-tfe rollback demo"), + ) + try: + new_sv = client.state_versions.rollback( + args.workspace_id, args.rollback_to + ) + print( + f"rollback succeeded — new state version: {new_sv.id} " + f"(serial={new_sv.serial})" + ) + finally: + print("unlocking workspace ...") + client.workspaces.unlock(args.workspace_id) + if __name__ == "__main__": main() diff --git a/examples/team.py b/examples/team.py index 5615a8b8..2be9df2f 100644 --- a/examples/team.py +++ b/examples/team.py @@ -101,6 +101,35 @@ def main(): default=None, help="Team ID for read/update/delete operation", ) + parser.add_argument( + "--add-user", + action="append", + default=[], + help="HCP Terraform username to add to --team-id (repeatable)", + ) + parser.add_argument( + "--remove-user", + action="append", + default=[], + help="HCP Terraform username to remove from --team-id (repeatable)", + ) + parser.add_argument( + "--add-ou", + action="append", + default=[], + help="Organization membership id (ou-…) to add to --team-id (repeatable)", + ) + parser.add_argument( + "--remove-ou", + action="append", + default=[], + help="Organization membership id to remove from --team-id (repeatable)", + ) + parser.add_argument( + "--list-members", + action="store_true", + help="List the team's current users and organization memberships", + ) args = parser.parse_args() cfg = TFEConfig(address=args.address, token=args.token) @@ -179,6 +208,43 @@ def main(): ) print() + # Team membership management (runs before the list output below) + membership_requested = ( + args.add_user + or args.remove_user + or args.add_ou + or args.remove_ou + or args.list_members + ) + if membership_requested: + if not args.team_id: + print("Error: --team-id is required for membership operations") + return + if args.add_user: + _print_header(f"Adding users to {args.team_id}: {args.add_user}") + client.teams.add_users(args.team_id, args.add_user) + if args.remove_user: + _print_header(f"Removing users from {args.team_id}: {args.remove_user}") + client.teams.remove_users(args.team_id, args.remove_user) + if args.add_ou: + _print_header(f"Adding org memberships to {args.team_id}: {args.add_ou}") + client.teams.add_organization_memberships(args.team_id, args.add_ou) + if args.remove_ou: + _print_header( + f"Removing org memberships from {args.team_id}: {args.remove_ou}" + ) + client.teams.remove_organization_memberships(args.team_id, args.remove_ou) + if args.list_members: + _print_header(f"Listing members of team {args.team_id}") + users = list(client.teams.list_users(args.team_id)) + print(f"users ({len(users)}):") + for u in users: + print(f" - {u.id} {getattr(u, 'username', '')}") + ous = list(client.teams.list_organization_memberships(args.team_id)) + print(f"organization memberships ({len(ous)}):") + for m in ous: + print(f" - {m.id} {getattr(m, 'email', '')}") + if args.delete: if not args.team_id: print("Error: --team-id is required when using --delete") diff --git a/examples/team_workspace_access.py b/examples/team_workspace_access.py new file mode 100644 index 00000000..81b0e7c9 --- /dev/null +++ b/examples/team_workspace_access.py @@ -0,0 +1,147 @@ +#!/usr/bin/env python3 +"""Team-workspace access example. + +Demonstrates the new ``client.team_workspace_accesses`` resource (the +go-tfe equivalent is ``TeamAccesses`` — ``/api/v2/team-workspaces``):: + + client.team_workspace_accesses.add(options) + client.team_workspace_accesses.list(workspace_id) + client.team_workspace_accesses.read(team_workspace_access_id) + client.team_workspace_accesses.update(id, options) + client.team_workspace_accesses.remove(id) + +By default the script creates a scratch team and a scratch workspace, +grants the team read access on the workspace, escalates the grant to +``custom`` (and tweaks the per-resource permissions), then removes the +grant and tears down the scratch resources. + +Usage:: + + TFE_TOKEN=... TFE_ORG=prab-sandbox02 \\ + python examples/team_workspace_access.py +""" + +from __future__ import annotations + +import argparse +import os +import time + +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + TeamCreateOptions, + TeamWorkspaceAccessAddOptions, + TeamWorkspaceAccessType, + TeamWorkspaceAccessUpdateOptions, + TeamWorkspaceRunsPermission, + TeamWorkspaceStateVersionsPermission, + TeamWorkspaceVariablesPermission, + WorkspaceCreateOptions, +) + + +def main() -> int: + p = argparse.ArgumentParser() + p.add_argument( + "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + p.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) + p.add_argument("--organization", default=os.getenv("TFE_ORG", "")) + p.add_argument("--team-id") + p.add_argument("--workspace-id") + args = p.parse_args() + if not args.token or not args.organization: + print("set TFE_TOKEN and TFE_ORG") + return 2 + + client = TFEClient(TFEConfig(address=args.address, token=args.token)) + + created: dict[str, str] = {} + grant_id: str | None = None + try: + team_id = args.team_id + workspace_id = args.workspace_id + + if not team_id: + stamp = int(time.time()) + t = client.teams.create( + args.organization, + TeamCreateOptions(name=f"pytfe-twa-{stamp}", visibility="secret"), + ) + created["team"] = t.id + team_id = t.id + print(f"created team: {t.id} ({t.name})") + + if not workspace_id: + stamp = int(time.time()) + ws = client.workspaces.create( + args.organization, + WorkspaceCreateOptions(name=f"pytfe-twa-ws-{stamp}"), + ) + created["workspace"] = ws.id + workspace_id = ws.id + print(f"created workspace: {ws.id} ({ws.name})") + + print(f"\nlisting existing grants on workspace {workspace_id} ...") + existing = list(client.team_workspace_accesses.list(workspace_id)) + print(f" {len(existing)} existing grant(s)") + for g in existing: + print(f" - {g.id} team-access={g.access}") + + print(f"\ngranting team {team_id} READ access on workspace {workspace_id}") + grant = client.team_workspace_accesses.add( + TeamWorkspaceAccessAddOptions( + team_id=team_id, + workspace_id=workspace_id, + access=TeamWorkspaceAccessType.READ, + ) + ) + grant_id = grant.id + print(f" created grant {grant.id} access={grant.access}") + + print("\nreading grant back") + readback = client.team_workspace_accesses.read(grant.id) + print(f" access={readback.access}") + + print("\nupgrading grant to CUSTOM (apply runs + write vars + write state)") + updated = client.team_workspace_accesses.update( + grant.id, + TeamWorkspaceAccessUpdateOptions( + access=TeamWorkspaceAccessType.CUSTOM, + runs=TeamWorkspaceRunsPermission.APPLY, + variables=TeamWorkspaceVariablesPermission.WRITE, + state_versions=TeamWorkspaceStateVersionsPermission.WRITE, + workspace_locking=True, + ), + ) + print( + f" access={updated.access} runs={updated.runs} " + f"variables={updated.variables} state_versions={updated.state_versions} " + f"workspace_locking={updated.workspace_locking}" + ) + + return 0 + finally: + if grant_id: + try: + client.team_workspace_accesses.remove(grant_id) + print(f"cleaned up grant {grant_id}") + except Exception as e: + print(f"WARN: could not remove grant: {e}") + if "workspace" in created: + try: + client.workspaces.delete_by_id(created["workspace"]) + print(f"cleaned up workspace {created['workspace']}") + except Exception as e: + print(f"WARN: could not clean up workspace: {e}") + if "team" in created: + try: + client.teams.delete(created["team"]) + print(f"cleaned up team {created['team']}") + except Exception as e: + print(f"WARN: could not clean up team: {e}") + client.close() + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/examples/workspace.py b/examples/workspace.py index 54f68715..b12e7b2a 100644 --- a/examples/workspace.py +++ b/examples/workspace.py @@ -131,6 +131,12 @@ def main(): ) parser.add_argument("--wildcard-name", help="Filter by wildcard name matching") parser.add_argument("--project-id", help="Filter by project ID") + parser.add_argument( + "--show-assessment", + action="store_true", + help="Show the workspace's current health-assessment result and the " + "variable sets applicable to it", + ) args = parser.parse_args() cfg = TFEConfig(address=args.address, token=args.token) @@ -497,7 +503,51 @@ def main(): print(f"readme result: {e}") print("(Expected if workspace has no README)") - # 16) Delete workspace if requested (should be last operation) + # 16) Show health assessment + applicable variable sets (read-only) + if args.show_assessment: + if not args.workspace_id: + print( + "--show-assessment requires --workspace-id (uses workspace-id " + "based endpoints)" + ) + else: + _print_header(f"Current assessment result for {args.workspace_id}") + result = client.workspaces.current_assessment_result(args.workspace_id) + if result is None: + print( + "no assessment result yet — assessments may be disabled, " + "or none have run." + ) + else: + for field in ( + "id", + "succeeded", + "all_checks_succeeded", + "drifted", + "resources_drifted", + "resources_undrifted", + "checks_passed", + "checks_failed", + "checks_errored", + "created_at", + "error_message", + ): + value = getattr(result, field, None) + if value is not None: + print(f" {field}: {value}") + + _print_header(f"Applicable variable sets for {args.workspace_id}") + count = 0 + for vs in client.workspaces.list_applicable_varsets(args.workspace_id): + count += 1 + print( + f" - {vs.get('id'):<24} {vs.get('name'):<30} " + f"global={vs.get('global')} vars={vs.get('var-count')}" + ) + if count == 0: + print(" (none)") + + # 17) Delete workspace if requested (should be last operation) if args.delete and args.workspace: _print_header(f"Deleting workspace: {args.workspace}") diff --git a/src/pytfe/_http.py b/src/pytfe/_http.py index 22b29b3f..ad1e6fd2 100644 --- a/src/pytfe/_http.py +++ b/src/pytfe/_http.py @@ -110,7 +110,11 @@ def request( self._sleep(attempt, retry_after) attempt += 1 continue - # print(resp) + # When the caller explicitly opted out of redirect-following, + # surface 3xx responses to them (so they can read Location) + # rather than treating them as errors. + if not allow_redirects and 300 <= resp.status_code < 400: + return resp self._raise_if_error(resp) return resp diff --git a/src/pytfe/client.py b/src/pytfe/client.py index f011ee0f..2655c1fc 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -49,6 +49,7 @@ from .resources.team import Teams from .resources.team_project_access import TeamProjectAccesses from .resources.team_token import TeamTokens +from .resources.team_workspace_access import TeamWorkspaceAccesses from .resources.user import Users from .resources.variable import Variables from .resources.variable_sets import VariableSets, VariableSetVariables @@ -141,6 +142,7 @@ def __init__(self, config: TFEConfig | None = None): self.teams = Teams(self._transport) self.team_project_accesses = TeamProjectAccesses(self._transport) self.team_tokens = TeamTokens(self._transport) + self.team_workspace_accesses = TeamWorkspaceAccesses(self._transport) # Reserved Tag Key self.reserved_tag_key = ReservedTagKeys(self._transport) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 730e2068..a75e978c 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -21,6 +21,7 @@ AgentTokenCreateOptions, AgentTokenListOptions, ) +from .assessment_result import AssessmentResult from .comment import ( Comment, CommentCreateOptions, @@ -164,6 +165,7 @@ from .policy_set import ( PolicySet, PolicySetAddPoliciesOptions, + PolicySetAddProjectExclusionsOptions, PolicySetAddProjectsOptions, PolicySetAddWorkspaceExclusionsOptions, PolicySetAddWorkspacesOptions, @@ -173,6 +175,7 @@ PolicySetListOptions, PolicySetReadOptions, PolicySetRemovePoliciesOptions, + PolicySetRemoveProjectExclusionsOptions, PolicySetRemoveProjectsOptions, PolicySetRemoveWorkspaceExclusionsOptions, PolicySetRemoveWorkspacesOptions, @@ -398,6 +401,16 @@ TeamTokenCreateOptions, TeamTokenListOptions, ) +from .team_workspace_access import ( + TeamWorkspaceAccess, + TeamWorkspaceAccessAddOptions, + TeamWorkspaceAccessType, + TeamWorkspaceAccessUpdateOptions, + TeamWorkspaceRunsPermission, + TeamWorkspaceSentinelMocksPermission, + TeamWorkspaceStateVersionsPermission, + TeamWorkspaceVariablesPermission, +) # Variables from .variable import ( @@ -646,6 +659,17 @@ "TeamToken", "TeamTokenCreateOptions", "TeamTokenListOptions", + # Team Workspace Access + "TeamWorkspaceAccess", + "TeamWorkspaceAccessAddOptions", + "TeamWorkspaceAccessType", + "TeamWorkspaceAccessUpdateOptions", + "TeamWorkspaceRunsPermission", + "TeamWorkspaceSentinelMocksPermission", + "TeamWorkspaceStateVersionsPermission", + "TeamWorkspaceVariablesPermission", + # Assessment Result + "AssessmentResult", "Project", "ProjectAddTagBindingsOptions", "ProjectCreateOptions", @@ -801,6 +825,7 @@ "PolicySetAddProjectsOptions", "PolicySetAddWorkspacesOptions", "PolicySetAddWorkspaceExclusionsOptions", + "PolicySetAddProjectExclusionsOptions", "PolicySetCreateOptions", "PolicySetListOptions", "PolicySetReadOptions", @@ -808,6 +833,7 @@ "PolicySetRemoveWorkspacesOptions", "PolicySetRemoveWorkspaceExclusionsOptions", "PolicySetRemoveProjectsOptions", + "PolicySetRemoveProjectExclusionsOptions", "PolicySetUpdateOptions", # Policy Set Parameters "PolicySetParameter", diff --git a/src/pytfe/models/assessment_result.py b/src/pytfe/models/assessment_result.py new file mode 100644 index 00000000..0303bbf4 --- /dev/null +++ b/src/pytfe/models/assessment_result.py @@ -0,0 +1,29 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict, Field + + +class AssessmentResult(BaseModel): + """Result of a workspace health assessment (drift detection).""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + succeeded: bool | None = Field(default=None, alias="succeeded") + all_checks_succeeded: bool | None = Field( + default=None, alias="all-checks-succeeded" + ) + checks_errored: int | None = Field(default=None, alias="checks-errored") + checks_failed: int | None = Field(default=None, alias="checks-failed") + checks_passed: int | None = Field(default=None, alias="checks-passed") + checks_unknown: int | None = Field(default=None, alias="checks-unknown") + created_at: datetime | None = Field(default=None, alias="created-at") + drifted: bool | None = Field(default=None, alias="drifted") + error_message: str | None = Field(default=None, alias="error-message") + resources_drifted: int | None = Field(default=None, alias="resources-drifted") + resources_undrifted: int | None = Field(default=None, alias="resources-undrifted") diff --git a/src/pytfe/models/policy_set.py b/src/pytfe/models/policy_set.py index 6bd7ca0c..e35257d8 100644 --- a/src/pytfe/models/policy_set.py +++ b/src/pytfe/models/policy_set.py @@ -23,6 +23,7 @@ class PolicySetIncludeOpt(str, Enum): POLICY_SET_NEWEST_VERSION = "newest_version" POLICY_SET_CURRENT_VERSION = "current_version" POLICY_SET_WORKSPACE_EXCLUSIONS = "workspace_exclusions" + POLICY_SET_PROJECT_EXCLUSIONS = "project_exclusions" class PolicySet(BaseModel): @@ -56,6 +57,9 @@ class PolicySet(BaseModel): workspace_exclusions: list[Workspace] = Field( default_factory=list, alias="workspace-exclusions" ) + project_exclusions: list[Project] = Field( + default_factory=list, alias="project-exclusions" + ) class PolicySetList(BaseModel): @@ -165,3 +169,15 @@ class PolicySetRemoveProjectsOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) projects: list[Project] = Field(default_factory=list) + + +class PolicySetAddProjectExclusionsOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + project_exclusions: list[Project] = Field(default_factory=list) + + +class PolicySetRemoveProjectExclusionsOptions(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + project_exclusions: list[Project] = Field(default_factory=list) diff --git a/src/pytfe/models/run_task_integration.py b/src/pytfe/models/run_task_integration.py index ee3b4c00..6eecf5bc 100644 --- a/src/pytfe/models/run_task_integration.py +++ b/src/pytfe/models/run_task_integration.py @@ -12,10 +12,7 @@ class TaskResultStatus(str, Enum): - """Statuses accepted by the Run Task callback endpoint. - - Mirrors the Go SDK's accepted callback statuses (passed, failed, running). - """ + """Statuses accepted by the Run Task callback endpoint.""" passed = "passed" failed = "failed" diff --git a/src/pytfe/models/team_workspace_access.py b/src/pytfe/models/team_workspace_access.py new file mode 100644 index 00000000..8e1593c0 --- /dev/null +++ b/src/pytfe/models/team_workspace_access.py @@ -0,0 +1,96 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field + + +class TeamWorkspaceAccessType(str, Enum): + READ = "read" + PLAN = "plan" + WRITE = "write" + ADMIN = "admin" + CUSTOM = "custom" + + +class TeamWorkspaceRunsPermission(str, Enum): + READ = "read" + PLAN = "plan" + APPLY = "apply" + + +class TeamWorkspaceVariablesPermission(str, Enum): + NONE = "none" + READ = "read" + WRITE = "write" + + +class TeamWorkspaceStateVersionsPermission(str, Enum): + NONE = "none" + READ_OUTPUTS = "read-outputs" + READ = "read" + WRITE = "write" + + +class TeamWorkspaceSentinelMocksPermission(str, Enum): + NONE = "none" + READ = "read" + + +class TeamWorkspaceAccess(BaseModel): + """A team's access grant on a workspace (`/api/v2/team-workspaces/{id}`).""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + access: TeamWorkspaceAccessType | None = None + runs: TeamWorkspaceRunsPermission | None = None + variables: TeamWorkspaceVariablesPermission | None = None + state_versions: TeamWorkspaceStateVersionsPermission | None = Field( + default=None, alias="state-versions" + ) + sentinel_mocks: TeamWorkspaceSentinelMocksPermission | None = Field( + default=None, alias="sentinel-mocks" + ) + workspace_locking: bool | None = Field(default=None, alias="workspace-locking") + run_tasks: bool | None = Field(default=None, alias="run-tasks") + policy_overrides: bool | None = Field(default=None, alias="policy-overrides") + + # Relationships (populated from the JSON:API ``relationships`` block). + team_id: str | None = Field(default=None, alias="team-id") + workspace_id: str | None = Field(default=None, alias="workspace-id") + + +class TeamWorkspaceAccessAddOptions(BaseModel): + """Options for adding a team access grant on a workspace.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + team_id: str + workspace_id: str + access: TeamWorkspaceAccessType + runs: TeamWorkspaceRunsPermission | None = None + variables: TeamWorkspaceVariablesPermission | None = None + state_versions: TeamWorkspaceStateVersionsPermission | None = None + sentinel_mocks: TeamWorkspaceSentinelMocksPermission | None = None + workspace_locking: bool | None = None + run_tasks: bool | None = None + policy_overrides: bool | None = None + + +class TeamWorkspaceAccessUpdateOptions(BaseModel): + """Options for updating an existing team access grant.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + access: TeamWorkspaceAccessType | None = None + runs: TeamWorkspaceRunsPermission | None = None + variables: TeamWorkspaceVariablesPermission | None = None + state_versions: TeamWorkspaceStateVersionsPermission | None = None + sentinel_mocks: TeamWorkspaceSentinelMocksPermission | None = None + workspace_locking: bool | None = None + run_tasks: bool | None = None + policy_overrides: bool | None = None diff --git a/src/pytfe/resources/apply.py b/src/pytfe/resources/apply.py index 621a8181..42747d1b 100644 --- a/src/pytfe/resources/apply.py +++ b/src/pytfe/resources/apply.py @@ -53,3 +53,35 @@ def _done(self, apply_id: str) -> tuple[bool, Exception | None]: return is_complete, None except Exception as e: return False, e + + def errored_state(self, apply_id: str) -> bytes: + """Recover the raw state bytes from an apply that failed during state upload. + + The TFE endpoint returns a 307 redirect to a signed object-storage URL. + We follow it manually so the API bearer token is not forwarded to the + third-party blob host. + + Raises NotFound if the apply has no recoverable errored state. + """ + if not valid_string_id(apply_id): + raise InvalidApplyIDError() + + # Do not auto-follow: the redirect target is presigned and must not + # receive our Authorization header. + resp = self.t.request( + "GET", + f"/api/v2/applies/{apply_id}/errored-state", + allow_redirects=False, + ) + if resp.status_code in (301, 302, 303, 307, 308): + location = resp.headers.get("Location") or resp.headers.get("location") + if not location: + from ..errors import TFEError + + raise TFEError( + "errored-state redirect did not include a Location header" + ) + blob = self.t.request("GET", location) + return blob.content + # 2xx body case (some servers may return inline); honour it + return resp.content diff --git a/src/pytfe/resources/configuration_version.py b/src/pytfe/resources/configuration_version.py index 6a7d0477..c64e693b 100644 --- a/src/pytfe/resources/configuration_version.py +++ b/src/pytfe/resources/configuration_version.py @@ -20,6 +20,7 @@ ConfigurationVersionCreateOptions, ConfigurationVersionListOptions, ConfigurationVersionReadOptions, + IngressAttributes, ) from ..utils import pack_contents, valid_string_id from ._base import _Service @@ -198,6 +199,39 @@ def download(self, cv_id: str) -> bytes: response = self.t.request("GET", path) return response.content + def ingress_attributes(self, cv_id: str) -> IngressAttributes | None: + """Get the VCS ingress attributes for a configuration version. + + Returns ``None`` if the configuration version was not created from a + VCS connection (so has no ingress data). The API responds with + ``null`` for API-driven CVs and with 404 for some older TFE + instances. + """ + if not valid_string_id(cv_id): + raise ValueError(ERR_INVALID_CONFIG_VERSION_ID) + try: + response = self.t.request( + "GET", + f"/api/v2/configuration-versions/{cv_id}/ingress-attributes", + ) + except NotFound: + return None + body = response.json() + if body is None: + return None + if not isinstance(body, dict): + return None + # The OpenAPI spec describes the response as the bare + # `ingress-attributes` resource, but the live API wraps it in the + # standard JSON:API envelope. Accept both shapes. + data = body.get("data", body) + if not isinstance(data, dict) or not data: + return None + attributes = data.get("attributes") + if not isinstance(attributes, dict): + return None + return IngressAttributes.model_validate(attributes) + def soft_delete_backing_data(self, cv_id: str) -> None: """Soft delete backing data for a configuration version (Enterprise only).""" self._manage_backing_data(cv_id, "soft_delete_backing_data") diff --git a/src/pytfe/resources/plan.py b/src/pytfe/resources/plan.py index 7f7d39a8..6e3f982c 100644 --- a/src/pytfe/resources/plan.py +++ b/src/pytfe/resources/plan.py @@ -5,7 +5,7 @@ from typing import Any -from ..errors import InvalidPlanIDError +from ..errors import InvalidPlanIDError, InvalidRunIDError from ..models.plan import ( Plan, PlanStatus, @@ -14,6 +14,15 @@ from ._base import _Service +def _plan_from_jsonapi(d: dict[str, Any]) -> Plan: + attr = d.get("attributes", {}) or {} + plan_id = str(d.get("id") or "") + return Plan( + id=plan_id, + **{k.replace("-", "_"): v for k, v in attr.items()}, + ) + + class Plans(_Service): def read(self, plan_id: str) -> Plan: """Read a specific plan by its ID.""" @@ -24,12 +33,14 @@ def read(self, plan_id: str) -> Plan: "GET", f"/api/v2/plans/{plan_id}", ) - d = r.json()["data"] - attr = d.get("attributes", {}) or {} - return Plan( - id=d.get("id"), - **{k.replace("-", "_"): v for k, v in attr.items()}, - ) + return _plan_from_jsonapi(r.json()["data"]) + + def read_for_run(self, run_id: str) -> Plan: + """Read the plan belonging to a run, via the run id.""" + if not valid_string_id(run_id): + raise InvalidRunIDError() + r = self.t.request("GET", f"/api/v2/runs/{run_id}/plan") + return _plan_from_jsonapi(r.json()["data"]) def logs(self, plan_id: str) -> str: """Get logs for a specific plan. @@ -54,29 +65,70 @@ def logs(self, plan_id: str) -> str: # Placeholder implementation - in future this would stream logs return "" - def read_json_output(self, plan_id: str) -> dict[str, Any]: + def _follow_json_output_redirect(self, path: str) -> dict[str, Any] | None: + """Fetch a json-output endpoint that returns 307 → presigned blob URL. + + The redirect target is a presigned object-storage URL; the API bearer + token must not be forwarded to it. + + Returns ``None`` if the API responds with 204 ("plan JSON supported, + but plan has not yet completed"). Callers should check the plan's + ``status`` before retrying. + """ + resp = self.t.request("GET", path, allow_redirects=False) + if resp.status_code == 204: + return None + if resp.status_code in (301, 302, 303, 307, 308): + location = resp.headers.get("Location") or resp.headers.get("location") + if not location: + from ..errors import TFEError + + raise TFEError("json-output redirect did not include a Location header") + blob = self.t.request("GET", location) + data = blob.json() + else: + # Defensive: 2xx body case (some servers may return inline) + try: + data = resp.json() + except Exception: + return None + if data is None: + return None + if isinstance(data, dict): + return data + return {"data": data} + + def read_json_output(self, plan_id: str) -> dict[str, Any] | None: """Get the JSON execution plan for a specific plan by its ID. Returns the JSON representation of the Terraform execution plan, - which includes detailed information about planned changes. + or ``None`` if the plan has not yet completed (HTTP 204). """ if not valid_string_id(plan_id): raise InvalidPlanIDError() + return self._follow_json_output_redirect(f"/api/v2/plans/{plan_id}/json-output") - r = self.t.request( - "GET", - f"/api/v2/plans/{plan_id}/json-output", + def read_json_output_for_run(self, run_id: str) -> dict[str, Any] | None: + """Get the JSON execution plan for a run, via the run id. + + Returns ``None`` if the plan has not yet completed (HTTP 204). + """ + if not valid_string_id(run_id): + raise InvalidRunIDError() + return self._follow_json_output_redirect( + f"/api/v2/runs/{run_id}/plan/json-output" ) - # Return the raw JSON data - this endpoint returns JSON directly - # not wrapped in a JSON:API format - json_data = r.json() - # Ensure we return a dictionary, not Any - if isinstance(json_data, dict): - return json_data - else: - # If somehow the response isn't a dict, wrap it - return {"data": json_data} + def read_json_schema_for_run(self, run_id: str) -> dict[str, Any] | None: + """Get the provider JSON schema corresponding to a plan, via the run id. + + Returns ``None`` if the plan has not yet completed (HTTP 204). + """ + if not valid_string_id(run_id): + raise InvalidRunIDError() + return self._follow_json_output_redirect( + f"/api/v2/runs/{run_id}/plan/json-schema" + ) def _done(self, plan_id: str) -> bool: """Create a done function for plan log reading.""" diff --git a/src/pytfe/resources/policy_set.py b/src/pytfe/resources/policy_set.py index 64d2ea0d..222cc7b9 100644 --- a/src/pytfe/resources/policy_set.py +++ b/src/pytfe/resources/policy_set.py @@ -4,6 +4,7 @@ from __future__ import annotations from collections.abc import Iterator +from typing import Any from ..errors import ( InvalidNameError, @@ -18,6 +19,7 @@ from ..models.policy_set import ( PolicySet, PolicySetAddPoliciesOptions, + PolicySetAddProjectExclusionsOptions, PolicySetAddProjectsOptions, PolicySetAddWorkspaceExclusionsOptions, PolicySetAddWorkspacesOptions, @@ -25,6 +27,7 @@ PolicySetListOptions, PolicySetReadOptions, PolicySetRemovePoliciesOptions, + PolicySetRemoveProjectExclusionsOptions, PolicySetRemoveProjectsOptions, PolicySetRemoveWorkspaceExclusionsOptions, PolicySetRemoveWorkspacesOptions, @@ -48,8 +51,16 @@ def list( raise InvalidOrgError() # Build params from options but do not pass page[number] — let _list handle pagination. - params = options.model_dump(by_alias=True, exclude_none=True) if options else {} + # mode="json" ensures enums (e.g. PolicySetIncludeOpt) serialize to + # their string values rather than `'PolicySetIncludeOpt.FOO'` reprs. + params = ( + options.model_dump(by_alias=True, exclude_none=True, mode="json") + if options + else {} + ) params.pop("page[number]", None) + if isinstance(params.get("include"), list): + params["include"] = ",".join(params["include"]) path = f"/api/v2/organizations/{organization}/policy-sets" @@ -65,6 +76,11 @@ def _gen() -> Iterator[PolicySet]: .get("workspace-exclusions", {}) .get("data", []) ) + attrs["project_exclusions"] = ( + d.get("relationships", {}) + .get("project-exclusions", {}) + .get("data", []) + ) attrs["workspaces"] = ( d.get("relationships", {}).get("workspaces", {}).get("data", []) ) @@ -143,6 +159,9 @@ def create(self, organization: str, options: PolicySetCreateOptions) -> PolicySe attrs["workspace_exclusions"] = relationships_data.get( "workspace-exclusions", {} ).get("data", []) + attrs["project_exclusions"] = relationships_data.get( + "project-exclusions", {} + ).get("data", []) attrs["workspaces"] = relationships_data.get("workspaces", {}).get("data", []) attrs["projects"] = relationships_data.get("projects", {}).get("data", []) attrs["policies"] = relationships_data.get("policies", {}).get("data", []) @@ -160,9 +179,11 @@ def read_with_options( if not valid_string_id(policy_set_id): raise InvalidPolicySetIDError() - params = ( - options.model_dump(by_alias=True, exclude_none=True) if options else None - ) + params: dict[str, Any] | None = None + if options is not None: + params = options.model_dump(by_alias=True, exclude_none=True, mode="json") + if isinstance(params.get("include"), list): + params["include"] = ",".join(params["include"]) r = self.t.request( "GET", @@ -180,6 +201,9 @@ def read_with_options( attrs["workspace_exclusions"] = relationships_data.get( "workspace-exclusions", {} ).get("data", []) + attrs["project_exclusions"] = relationships_data.get( + "project-exclusions", {} + ).get("data", []) attrs["workspaces"] = relationships_data.get("workspaces", {}).get("data", []) attrs["projects"] = relationships_data.get("projects", {}).get("data", []) attrs["policies"] = relationships_data.get("policies", {}).get("data", []) @@ -219,6 +243,9 @@ def update(self, policy_set_id: str, options: PolicySetUpdateOptions) -> PolicyS attrs["workspace_exclusions"] = relationships_data.get( "workspace-exclusions", {} ).get("data", []) + attrs["project_exclusions"] = relationships_data.get( + "project-exclusions", {} + ).get("data", []) attrs["workspaces"] = relationships_data.get("workspaces", {}).get("data", []) attrs["projects"] = relationships_data.get("projects", {}).get("data", []) attrs["policies"] = relationships_data.get("policies", {}).get("data", []) @@ -385,6 +412,52 @@ def remove_workspace_exclusions( ) return None + def add_project_exclusions( + self, + policy_set_id: str, + options: PolicySetAddProjectExclusionsOptions, + ) -> None: + """Add project exclusions to a policy set.""" + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + if not options.project_exclusions: + raise ValueError("project_exclusions is required") + payload = { + "data": [ + {"id": project.id, "type": "projects"} + for project in options.project_exclusions + ] + } + self.t.request( + "POST", + f"/api/v2/policy-sets/{policy_set_id}/relationships/project-exclusions", + json_body=payload, + ) + return None + + def remove_project_exclusions( + self, + policy_set_id: str, + options: PolicySetRemoveProjectExclusionsOptions, + ) -> None: + """Remove project exclusions from a policy set.""" + if not valid_string_id(policy_set_id): + raise InvalidPolicySetIDError() + if not options.project_exclusions: + raise ValueError("project_exclusions is required") + payload = { + "data": [ + {"id": project.id, "type": "projects"} + for project in options.project_exclusions + ] + } + self.t.request( + "DELETE", + f"/api/v2/policy-sets/{policy_set_id}/relationships/project-exclusions", + json_body=payload, + ) + return None + def add_projects( self, policy_set_id: str, options: PolicySetAddProjectsOptions ) -> None: diff --git a/src/pytfe/resources/projects.py b/src/pytfe/resources/projects.py index 335b1314..44fb63a1 100644 --- a/src/pytfe/resources/projects.py +++ b/src/pytfe/resources/projects.py @@ -268,6 +268,29 @@ def delete(self, project_id: str) -> None: path = f"/api/v2/projects/{project_id}" self.t.request("DELETE", path) + def move_workspaces( + self, project_id: str, workspace_ids: builtins.list[str] + ) -> None: + """Move one or more workspaces into a project. + + The caller must have permission to move each workspace out of its + current project and into the target project. + """ + if not valid_string_id(project_id): + raise ValueError("Project ID is required and must be valid") + if not workspace_ids: + raise ValueError("at least one workspace id is required") + for wid in workspace_ids: + if not valid_string_id(wid): + raise ValueError(f"invalid workspace id: {wid!r}") + payload = {"data": [{"id": wid, "type": "workspaces"} for wid in workspace_ids]} + self.t.request( + "POST", + f"/api/v2/projects/{project_id}/relationships/workspaces", + json_body=payload, + ) + return None + def list_tag_bindings(self, project_id: str) -> builtins.list[TagBinding]: """List tag bindings for a project""" # Validate inputs @@ -292,29 +315,21 @@ def list_tag_bindings(self, project_id: str) -> builtins.list[TagBinding]: def list_effective_tag_bindings( self, project_id: str - ) -> builtins.list[EffectiveTagBinding]: - """List effective tag bindings for a project""" - # Validate inputs + ) -> Iterator[EffectiveTagBinding]: + """List effective tag bindings for a project.""" if not valid_string_id(project_id): raise ValueError("Project ID is required and must be valid") path = f"/api/v2/projects/{project_id}/effective-tag-bindings" - response = self.t.request("GET", path) - data = response.json()["data"] - - effective_tag_bindings = [] - for item in data: + for item in self._list(path): attr = item.get("attributes", {}) or {} links = item.get("links", {}) or {} - effective_tag_binding = EffectiveTagBinding( + yield EffectiveTagBinding( id=_safe_str(item.get("id")), key=_safe_str(attr.get("key")), value=_safe_str(attr.get("value")), links=links, ) - effective_tag_bindings.append(effective_tag_binding) - - return effective_tag_bindings def add_tag_bindings( self, project_id: str, options: ProjectAddTagBindingsOptions diff --git a/src/pytfe/resources/registry_module.py b/src/pytfe/resources/registry_module.py index 651471fb..3945f8a4 100644 --- a/src/pytfe/resources/registry_module.py +++ b/src/pytfe/resources/registry_module.py @@ -217,8 +217,23 @@ def read_version( return self._parse_registry_module_version(data) - def list_versions(self, module_id: RegistryModuleID) -> list[RegistryModuleVersion]: # type: ignore[valid-type] - """List all versions of a registry module.""" + def list_versions( + self, module_id: RegistryModuleID + ) -> Iterator[RegistryModuleVersion]: + """List all versions of a registry module. + + This method intentionally fetches eagerly and returns ``iter(list)`` + instead of the canonical ``for x in self._list(...): yield ...`` + pattern used elsewhere in the SDK. The reason is the fallback path: + if the primary ``/versions`` endpoint is unavailable, we fall back + to reading the module and extracting versions from + ``version_statuses``. A pure generator could yield items from the + primary endpoint, fail partway, then switch to the fallback and + yield duplicates. Eager materialization avoids that risk. + + See ``docs/ITERATORS.md`` for the convention and when it's OK to + deviate from it. + """ if not self._validate_module_id(module_id): raise ValueError("Invalid module ID") @@ -241,12 +256,12 @@ def list_versions(self, module_id: RegistryModuleID) -> list[RegistryModuleVersi # Handle the case where data might be None or empty data = response_data.get("data", []) if response_data else [] - versions = [] + versions: list[RegistryModuleVersion] = [] for item in data: if item: # Skip None items versions.append(self._parse_registry_module_version(item)) - return versions + return iter(versions) except Exception: # Fallback: If the API endpoint doesn't exist, try to get versions from the module itself @@ -270,9 +285,9 @@ def list_versions(self, module_id: RegistryModuleID) -> list[RegistryModuleVersi } versions.append(self._parse_registry_module_version(version_data)) - return versions + return iter(versions) except Exception: - return [] # Return empty list if all methods fail + return iter([]) # Return empty iterator if all methods fail def read_terraform_registry_module( self, module_id: RegistryModuleID, version: str diff --git a/src/pytfe/resources/state_versions.py b/src/pytfe/resources/state_versions.py index 260ec5a1..3f7ec760 100644 --- a/src/pytfe/resources/state_versions.py +++ b/src/pytfe/resources/state_versions.py @@ -236,7 +236,6 @@ def upload( sv.hosted_state_upload_url, data=raw_state, headers={"Content-Type": "application/octet-stream"}, - include_auth=False, ) if raw_json_state is not None: @@ -249,7 +248,6 @@ def upload( sv.hosted_json_state_upload_url, data=raw_json_state, headers={"Content-Type": "application/octet-stream"}, - include_auth=False, ) return self.read(sv.id) @@ -273,10 +271,14 @@ def download(self, state_version_id: str) -> bytes: raise NotFound("download url not available for this state version") - # Download the bytes from the signed Archivist URL (follow redirects). - # Avoid JSON:API headers here; Accept */* is fine. + # Download the bytes from the signed Archivist URL. The presigned URL + # already carries its own credentials, so the TFE bearer token must + # NOT be forwarded. resp = self.t.request( - "GET", url, allow_redirects=True, headers={"Accept": "application/json"} + "GET", + url, + allow_redirects=True, + headers={"Accept": "*/*"}, ) return resp.content @@ -292,7 +294,10 @@ def download_current(self, workspace_id: str) -> bytes: raise NotFound("download url not available for current state") resp = self.t.request( - "GET", url, allow_redirects=True, headers={"Accept": "*/*"} + "GET", + url, + allow_redirects=True, + headers={"Accept": "*/*"}, ) return resp.content @@ -353,3 +358,41 @@ def permanently_delete_backing_data(self, state_version_id: str) -> None: f"/api/v2/state-versions/{state_version_id}/actions/permanently_delete_backing_data", ) return None + + def rollback( + self, + workspace_id: str, + rollback_state_version_id: str, + ) -> StateVersion: + """Roll a workspace back to a previous state version. + + Duplicates the named state version and sets the copy as the workspace's + current state version. The workspace must be locked by the caller + before invoking this operation, otherwise the API returns 409. + """ + if not valid_string_id(workspace_id): + raise ValueError("invalid workspace id") + if not valid_string_id(rollback_state_version_id): + raise ValueError("invalid rollback state version id") + body = { + "data": { + "type": "state-versions", + "relationships": { + "rollback-state-version": { + "data": { + "type": "state-versions", + "id": rollback_state_version_id, + } + } + }, + } + } + resp = self.t.request( + "PATCH", + f"/api/v2/workspaces/{workspace_id}/state-versions", + json_body=body, + ) + data = (resp.json() or {}).get("data") or {} + attributes = dict(data.get("attributes") or {}) + attributes["id"] = data.get("id", "") + return StateVersion.model_validate(attributes) diff --git a/src/pytfe/resources/team.py b/src/pytfe/resources/team.py index 37df9876..9996bcff 100644 --- a/src/pytfe/resources/team.py +++ b/src/pytfe/resources/team.py @@ -1,5 +1,6 @@ from __future__ import annotations +import builtins from collections.abc import Iterator from ..errors import ( @@ -106,3 +107,142 @@ def delete(self, team_id: str) -> None: path=f"/api/v2/teams/{team_id}", ) return None + + # ------------------------------------------------------------------ + # Team membership management + # ------------------------------------------------------------------ + + def add_users(self, team_id: str, usernames: builtins.list[str]) -> None: + """Add users to a team by username.""" + if not valid_string_id(team_id): + raise InvalidTeamIDError() + if not usernames: + raise ValueError("at least one username is required") + if any(not isinstance(u, str) or not u.strip() for u in usernames): + raise ValueError("usernames must be non-empty strings") + payload = {"data": [{"type": "users", "id": u} for u in usernames]} + self.t.request( + "POST", + path=f"/api/v2/teams/{team_id}/relationships/users", + json_body=payload, + ) + return None + + def remove_users(self, team_id: str, usernames: builtins.list[str]) -> None: + """Remove users from a team by username.""" + if not valid_string_id(team_id): + raise InvalidTeamIDError() + if not usernames: + raise ValueError("at least one username is required") + if any(not isinstance(u, str) or not u.strip() for u in usernames): + raise ValueError("usernames must be non-empty strings") + payload = {"data": [{"type": "users", "id": u} for u in usernames]} + self.t.request( + "DELETE", + path=f"/api/v2/teams/{team_id}/relationships/users", + json_body=payload, + ) + return None + + def add_organization_memberships( + self, team_id: str, organization_membership_ids: builtins.list[str] + ) -> None: + """Add users to a team by organization membership id.""" + if not valid_string_id(team_id): + raise InvalidTeamIDError() + if not organization_membership_ids: + raise ValueError("at least one organization membership id is required") + if any(not valid_string_id(i) for i in organization_membership_ids): + raise ValueError("invalid organization membership id") + payload = { + "data": [ + {"type": "organization-memberships", "id": i} + for i in organization_membership_ids + ] + } + self.t.request( + "POST", + path=f"/api/v2/teams/{team_id}/relationships/organization-memberships", + json_body=payload, + ) + return None + + def remove_organization_memberships( + self, team_id: str, organization_membership_ids: builtins.list[str] + ) -> None: + """Remove users from a team by organization membership id.""" + if not valid_string_id(team_id): + raise InvalidTeamIDError() + if not organization_membership_ids: + raise ValueError("at least one organization membership id is required") + if any(not valid_string_id(i) for i in organization_membership_ids): + raise ValueError("invalid organization membership id") + payload = { + "data": [ + {"type": "organization-memberships", "id": i} + for i in organization_membership_ids + ] + } + self.t.request( + "DELETE", + path=f"/api/v2/teams/{team_id}/relationships/organization-memberships", + json_body=payload, + ) + return None + + def list_users(self, team_id: str) -> Iterator[User]: + """List the users that belong to a team. + + Implemented via ``GET /teams/{id}?include=users`` — the API has no + dedicated paginated endpoint for team users, so all results arrive + in a single response. The signature still returns an iterator to + stay consistent with the other ``list_*`` methods in the SDK; wrap + the result in ``list(...)`` if you need a materialized list. + """ + if not valid_string_id(team_id): + raise InvalidTeamIDError() + r = self.t.request( + "GET", + path=f"/api/v2/teams/{team_id}", + params={"include": "users"}, + ) + payload = r.json() or {} + included = payload.get("included") or [] + for inc in included: + if inc.get("type") != "users": + continue + attrs = dict(inc.get("attributes") or {}) + attrs["id"] = inc.get("id") + yield User.model_validate(attrs) + + def list_organization_memberships( + self, + team_id: str, + *, + status: str | None = None, + is_service_account: bool | None = None, + sort: str | None = None, + ) -> Iterator[OrganizationMembership]: + """List the organization memberships that belong to a team. + + Uses the dedicated paginated endpoint + ``GET /teams/{id}/relationships/organization-memberships`` so + callers get server-side pagination, filtering by status / + service-account flag, and sort. + """ + if not valid_string_id(team_id): + raise InvalidTeamIDError() + params: dict[str, str] = {} + if status is not None: + params["filter[status]"] = status + if is_service_account is not None: + params["filter[is_service_account]"] = ( + "true" if is_service_account else "false" + ) + if sort is not None: + params["sort"] = sort + path = f"/api/v2/teams/{team_id}/relationships/organization-memberships" + for item in self._list(path, params=params): + attrs = dict(item.get("attributes") or {}) + attrs["id"] = item.get("id") + yield OrganizationMembership.model_validate(attrs) diff --git a/src/pytfe/resources/team_workspace_access.py b/src/pytfe/resources/team_workspace_access.py new file mode 100644 index 00000000..8f37c21b --- /dev/null +++ b/src/pytfe/resources/team_workspace_access.py @@ -0,0 +1,125 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import InvalidTeamIDError, InvalidWorkspaceIDError, TFEError +from ..models.team_workspace_access import ( + TeamWorkspaceAccess, + TeamWorkspaceAccessAddOptions, + TeamWorkspaceAccessUpdateOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class InvalidTeamWorkspaceAccessIDError(TFEError): + """Raised when a team-workspace access id is missing or malformed.""" + + def __init__(self, message: str = "invalid team workspace access id"): + super().__init__(message) + + +def _parse(data: dict[str, Any]) -> TeamWorkspaceAccess: + attributes = dict(data.get("attributes") or {}) + attributes["id"] = data.get("id", "") + relationships = data.get("relationships") or {} + team_data = (relationships.get("team") or {}).get("data") or {} + workspace_data = (relationships.get("workspace") or {}).get("data") or {} + if team_data.get("id"): + attributes["team-id"] = team_data["id"] + if workspace_data.get("id"): + attributes["workspace-id"] = workspace_data["id"] + return TeamWorkspaceAccess.model_validate(attributes) + + +def _attributes_payload(model_dict: dict[str, Any]) -> dict[str, Any]: + """Hyphenate snake_case attribute keys for JSON:API.""" + return {k.replace("_", "-"): v for k, v in model_dict.items() if v is not None} + + +class TeamWorkspaceAccesses(_Service): + """Manage team access grants on workspaces (`/api/v2/team-workspaces`).""" + + def list(self, workspace_id: str) -> Iterator[TeamWorkspaceAccess]: + """List team access grants for a workspace.""" + if not valid_string_id(workspace_id): + raise InvalidWorkspaceIDError() + path = "/api/v2/team-workspaces" + params = {"filter[workspace][id]": workspace_id} + for item in self._list(path, params=params): + yield _parse(item) + + def read(self, team_workspace_access_id: str) -> TeamWorkspaceAccess: + """Read a single team-workspace access grant by id.""" + if not valid_string_id(team_workspace_access_id): + raise InvalidTeamWorkspaceAccessIDError() + r = self.t.request("GET", f"/api/v2/team-workspaces/{team_workspace_access_id}") + return _parse((r.json() or {}).get("data") or {}) + + def add(self, options: TeamWorkspaceAccessAddOptions) -> TeamWorkspaceAccess: + """Add a team access grant to a workspace.""" + if not valid_string_id(options.team_id): + raise InvalidTeamIDError() + if not valid_string_id(options.workspace_id): + raise InvalidWorkspaceIDError() + attrs = _attributes_payload( + options.model_dump( + by_alias=False, + exclude={"team_id", "workspace_id"}, + exclude_none=True, + mode="json", + ) + ) + payload = { + "data": { + "type": "team-workspaces", + "attributes": attrs, + "relationships": { + "team": {"data": {"type": "teams", "id": options.team_id}}, + "workspace": { + "data": {"type": "workspaces", "id": options.workspace_id} + }, + }, + } + } + r = self.t.request("POST", "/api/v2/team-workspaces", json_body=payload) + return _parse((r.json() or {}).get("data") or {}) + + def update( + self, + team_workspace_access_id: str, + options: TeamWorkspaceAccessUpdateOptions, + ) -> TeamWorkspaceAccess: + """Update an existing team-workspace access grant.""" + if not valid_string_id(team_workspace_access_id): + raise InvalidTeamWorkspaceAccessIDError() + attrs = _attributes_payload( + options.model_dump(by_alias=False, exclude_none=True, mode="json") + ) + payload = { + "data": { + "type": "team-workspaces", + "id": team_workspace_access_id, + "attributes": attrs, + } + } + r = self.t.request( + "PATCH", + f"/api/v2/team-workspaces/{team_workspace_access_id}", + json_body=payload, + ) + return _parse((r.json() or {}).get("data") or {}) + + def remove(self, team_workspace_access_id: str) -> None: + """Remove (delete) a team-workspace access grant.""" + if not valid_string_id(team_workspace_access_id): + raise InvalidTeamWorkspaceAccessIDError() + self.t.request( + "DELETE", + f"/api/v2/team-workspaces/{team_workspace_access_id}", + ) + return None diff --git a/src/pytfe/resources/workspaces.py b/src/pytfe/resources/workspaces.py index 1a6ac6cb..a2d419ec 100644 --- a/src/pytfe/resources/workspaces.py +++ b/src/pytfe/resources/workspaces.py @@ -22,6 +22,7 @@ WorkspaceRequiredError, ) from ..models.agent import AgentPool +from ..models.assessment_result import AssessmentResult from ..models.common import ( EffectiveTagBinding, Tag, @@ -1015,3 +1016,44 @@ def readme(self, workspace_id: str) -> str | None: return (inc.get("attributes") or {}).get("raw-markdown") return None + + def current_assessment_result(self, workspace_id: str) -> AssessmentResult | None: + """Get the current health-assessment (drift detection) result for a workspace. + + Returns ``None`` if the workspace has no assessment result yet (assessments + may be disabled, or no assessment has run). + """ + if not valid_string_id(workspace_id): + raise InvalidWorkspaceIDError() + try: + r = self.t.request( + "GET", + f"/api/v2/workspaces/{workspace_id}/current-assessment-result", + ) + except Exception as exc: + from ..errors import NotFound + + if isinstance(exc, NotFound): + return None + raise + data = (r.json() or {}).get("data") or {} + attributes = dict(data.get("attributes") or {}) + attributes["id"] = data.get("id", "") + return AssessmentResult.model_validate(attributes) + + def list_applicable_varsets(self, workspace_id: str) -> Iterator[dict[str, Any]]: + """List variable sets that apply to a workspace, including inherited ones. + + Returns raw varset attribute dicts (id/name/global/var-count/etc.). The + endpoint summarises varsets rather than returning the full relationship + graph, so it is exposed as plain dicts to avoid the heavier + ``VariableSet`` parsing path. Callers wanting the full model can pass + each ``id`` to ``client.variable_sets.read``. + """ + if not valid_string_id(workspace_id): + raise InvalidWorkspaceIDError() + path = f"/api/v2/workspaces/{workspace_id}/applicable-varsets" + for item in self._list(path): + attrs = dict(item.get("attributes") or {}) + attrs["id"] = item.get("id", "") + yield attrs diff --git a/tests/units/test_plan.py b/tests/units/test_plan.py index 36c3f7ee..cbfe111c 100644 --- a/tests/units/test_plan.py +++ b/tests/units/test_plan.py @@ -101,7 +101,7 @@ def test_logs_success(self, plans_service): assert result == "" def test_read_json_output_success(self, plans_service): - """Test successful read_json_output operation.""" + """Test successful read_json_output operation (200 response).""" mock_json_data = { "format_version": "1.1", @@ -124,19 +124,47 @@ def test_read_json_output_success(self, plans_service): with patch.object(plans_service, "t") as mock_transport: mock_response = Mock() + mock_response.status_code = 200 mock_response.json.return_value = mock_json_data mock_transport.request.return_value = mock_response result = plans_service.read_json_output("plan-123") - # Verify request was made correctly mock_transport.request.assert_called_once_with( - "GET", "/api/v2/plans/plan-123/json-output" + "GET", "/api/v2/plans/plan-123/json-output", allow_redirects=False ) - # Verify JSON data is returned assert result == mock_json_data assert result["format_version"] == "1.1" assert result["terraform_version"] == "1.5.0" assert len(result["resource_changes"]) == 1 assert result["resource_changes"][0]["change"]["actions"] == ["create"] + + def test_read_json_output_follows_redirect(self, plans_service): + """The 307 redirect target is followed manually and its body returned.""" + mock_json_data = {"format_version": "1.1"} + + with patch.object(plans_service, "t") as mock_transport: + redirect_resp = Mock() + redirect_resp.status_code = 307 + redirect_resp.headers = { + "Location": "https://archivist.example/blob?sig=abc" + } + blob_resp = Mock() + blob_resp.status_code = 200 + blob_resp.json.return_value = mock_json_data + mock_transport.request.side_effect = [redirect_resp, blob_resp] + + result = plans_service.read_json_output("plan-123") + + assert result == mock_json_data + assert mock_transport.request.call_count == 2 + first_call = mock_transport.request.call_args_list[0] + second_call = mock_transport.request.call_args_list[1] + assert first_call.args == ("GET", "/api/v2/plans/plan-123/json-output") + assert first_call.kwargs == {"allow_redirects": False} + assert second_call.args == ( + "GET", + "https://archivist.example/blob?sig=abc", + ) + assert second_call.kwargs == {} diff --git a/tests/units/test_project.py b/tests/units/test_project.py index c42fb3af..2f4a3321 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -364,7 +364,7 @@ def test_list_tag_bindings_invalid_project_id(self): def test_list_effective_tag_bindings_success(self): """Test successful listing of effective tag bindings""" - # Mock API response + # Mock API response — single page, no pagination metadata. mock_response = Mock() mock_response.json.return_value = { "data": [ @@ -380,8 +380,10 @@ def test_list_effective_tag_bindings_success(self): } self.mock_transport.request.return_value = mock_response - # Call the method - result = self.projects_service.list_effective_tag_bindings(self.project_id) + # Call the method (returns Iterator — materialize to assert) + result = list( + self.projects_service.list_effective_tag_bindings(self.project_id) + ) # Assertions assert len(result) == 1 @@ -392,19 +394,23 @@ def test_list_effective_tag_bindings_success(self): assert result[0].value == "production" assert "self" in result[0].links - # Verify API call - self.mock_transport.request.assert_called_once_with( - "GET", f"/api/v2/projects/{self.project_id}/effective-tag-bindings" + # Verify the request was issued against the right path (params include + # page[number]/page[size] from _list — assert on path only). + call = self.mock_transport.request.call_args + assert call.args == ( + "GET", + f"/api/v2/projects/{self.project_id}/effective-tag-bindings", ) def test_list_effective_tag_bindings_invalid_project_id(self): """Test listing effective tag bindings with invalid project ID""" import pytest + # Generator-based list methods validate on first iteration. with pytest.raises( ValueError, match="Project ID is required and must be valid" ): - self.projects_service.list_effective_tag_bindings(None) + list(self.projects_service.list_effective_tag_bindings(None)) def test_add_tag_bindings_success(self): """Test successful addition of tag bindings""" diff --git a/tests/units/test_state_version.py b/tests/units/test_state_version.py index a3e3659d..f9e3cd9b 100644 --- a/tests/units/test_state_version.py +++ b/tests/units/test_state_version.py @@ -2,7 +2,6 @@ from unittest.mock import Mock, patch -import httpx import pytest from pytfe._http import HTTPTransport @@ -284,54 +283,14 @@ def test_upload_state_version_success(self, state_versions_service, mock_transpo "https://example.com/upload-raw", data=b"raw-state", headers={"Content-Type": "application/octet-stream"}, - include_auth=False, ) mock_transport.request.assert_any_call( "PUT", "https://example.com/upload-json", data=b"json-state", headers={"Content-Type": "application/octet-stream"}, - include_auth=False, ) - def test_upload_state_version_presigned_put_omits_authorization_header(self): - """Test upload() does not send the TFE token to presigned upload URLs.""" - seen_authorization_headers: list[str | None] = [] - - def handler(request: httpx.Request) -> httpx.Response: - seen_authorization_headers.append(request.headers.get("authorization")) - return httpx.Response(200) - - transport = HTTPTransport( - "https://app.terraform.io", - "secret-token", - timeout=5, - verify_tls=True, - user_agent_suffix=None, - max_retries=0, - backoff_base=0, - backoff_cap=0, - backoff_jitter=False, - http2=False, - proxies=None, - ca_bundle=None, - ) - transport._sync = httpx.Client(transport=httpx.MockTransport(handler)) - service = StateVersions(transport) - created_sv = StateVersion( - id="sv-upload-1", - status=StateVersionStatus.PENDING, - hosted_state_upload_url="https://archivist.terraform.io/upload-raw", - ) - final_sv = StateVersion(id="sv-upload-1", status=StateVersionStatus.FINALIZED) - options = StateVersionCreateOptions(serial=10, md5="abc123") - - with patch.object(service, "create", return_value=created_sv): - with patch.object(service, "read", return_value=final_sv): - service.upload("ws-123", raw_state=b"raw-state", options=options) - - assert seen_authorization_headers == [None] - def test_upload_state_version_unsupported_on_create_error( self, state_versions_service ): @@ -406,7 +365,7 @@ def test_download_state_version_success( "GET", "https://example.com/signed-download", allow_redirects=True, - headers={"Accept": "application/json"}, + headers={"Accept": "*/*"}, ) assert result == b"{}"