diff --git a/examples/stack.py b/examples/stack.py new file mode 100644 index 00000000..6e8546e8 --- /dev/null +++ b/examples/stack.py @@ -0,0 +1,224 @@ +# 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.agent import AgentPool +from pytfe.models.project import Project +from pytfe.models.stack import ( + StackCreateOptions, + StackListOptions, + StackSortColumn, + StackUpdateOptions, + StackVcsRepoOptions, +) + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def _print_stack(item): + print(f"- id: {item.id}") + print(f"- name: {item.name}") + print(f"- description: {item.description}") + print(f"- created_at: {item.created_at}") + print(f"- updated_at: {item.updated_at}") + print(f"- speculation_enabled: {item.speculation_enabled}") + print(f"- project_id: {item.project.id if item.project else None}") + print(f"- agent_pool_id: {item.agent_pool.id if item.agent_pool else None}") + + if item.vcs_repo: + print("- vcs_repo:") + print(f" identifier={item.vcs_repo.identifier}") + print(f" branch={item.vcs_repo.branch}") + print(f" github_app_installation_id={item.vcs_repo.gha_installation_id}") + print(f" oauth_token_id={item.vcs_repo.oauth_token_id}") + + +def _build_vcs_repo_options(args) -> StackVcsRepoOptions | None: + if not args.vcs_identifier: + return None + + return StackVcsRepoOptions( + identifier=args.vcs_identifier, + branch=args.vcs_branch, + gha_installation_id=args.vcs_github_app_installation_id, + oauth_token_id=args.vcs_oauth_token_id, + ) + + +def main(): + parser = argparse.ArgumentParser( + description="Stacks operations demo for python-tfe" + ) + 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", help="Organization name (required for list)") + parser.add_argument( + "--operation", + required=True, + choices=["create", "read", "update", "list", "delete", "force-delete"], + help="Operation to execute", + ) + + parser.add_argument( + "--stack-id", help="Stack ID (required for read/update/delete/force-delete)" + ) + + parser.add_argument("--name", help="Stack name (required for create)") + parser.add_argument("--description", help="Stack description") + parser.add_argument( + "--speculation-enabled", + type=lambda v: str(v).lower() in ("1", "true", "yes", "y"), + default=None, + help="Enable speculation (true/false)", + ) + + parser.add_argument( + "--project-id", + help="Project ID (required for create, optional for list filter)", + ) + parser.add_argument( + "--agent-pool-id", help="Agent pool ID (optional for create/update)" + ) + + parser.add_argument( + "--vcs-identifier", + help="VCS repo identifier (e.g. org/repo), optional for create/update", + ) + parser.add_argument("--vcs-branch", help="VCS branch") + parser.add_argument( + "--vcs-github-app-installation-id", + help="GitHub App installation ID for VCS repo", + ) + parser.add_argument("--vcs-oauth-token-id", help="OAuth token ID for VCS repo") + + parser.add_argument("--page-size", type=int, default=20, help="Page size for list") + parser.add_argument( + "--sort", + choices=[item.value for item in StackSortColumn], + default=None, + help="Sort column for list", + ) + parser.add_argument( + "--search-name", + default=None, + help="Search stacks by name", + ) + + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + if args.operation == "create": + if not args.name: + parser.error("--name is required for operation=create") + if not args.project_id: + parser.error("--project-id is required for operation=create") + + _print_header("Creating stack") + options = StackCreateOptions( + name=args.name, + description=args.description, + speculation_enabled=args.speculation_enabled, + vcs_repo=_build_vcs_repo_options(args), + project=Project(id=args.project_id), + agent_pool=AgentPool(id=args.agent_pool_id) if args.agent_pool_id else None, + ) + result = client.stacks.create(options) + print("Created stack") + _print_stack(result) + return + + if args.operation == "read": + if not args.stack_id: + parser.error("--stack-id is required for operation=read") + + _print_header("Reading stack") + result = client.stacks.read(args.stack_id) + print("Retrieved stack") + _print_stack(result) + return + + if args.operation == "update": + if not args.stack_id: + parser.error("--stack-id is required for operation=update") + if not any( + [ + args.name, + args.description, + args.speculation_enabled is not None, + args.agent_pool_id, + args.vcs_identifier, + args.vcs_branch, + args.vcs_github_app_installation_id, + args.vcs_oauth_token_id, + args.project_id, + ] + ): + parser.error("Provide at least one field to update") + + _print_header("Updating stack") + options = StackUpdateOptions( + name=args.name, + description=args.description, + speculation_enabled=args.speculation_enabled, + vcs_repo=_build_vcs_repo_options(args), + agent_pool=AgentPool(id=args.agent_pool_id) if args.agent_pool_id else None, + project=Project(id=args.project_id) if args.project_id else None, + ) + result = client.stacks.update(args.stack_id, options) + print("Updated stack") + _print_stack(result) + return + + if args.operation == "list": + if not args.organization: + parser.error("--organization is required for operation=list") + + _print_header("Listing stacks") + list_options = StackListOptions( + page_size=args.page_size, + project_id=args.project_id, + sort=StackSortColumn(args.sort) if args.sort else None, + search_by_name=args.search_name, + ) + + items = list(client.stacks.list(args.organization, list_options)) + print(f"Found {len(items)} stacks") + for item in items: + print("-") + _print_stack(item) + return + + if args.operation == "delete": + if not args.stack_id: + parser.error("--stack-id is required for operation=delete") + + _print_header("Deleting stack") + client.stacks.delete(args.stack_id) + print(f"Deleted stack: {args.stack_id}") + return + + if args.operation == "force-delete": + if not args.stack_id: + parser.error("--stack-id is required for operation=force-delete") + + _print_header("Force deleting stack") + client.stacks.force_delete(args.stack_id) + print(f"Force deleted stack: {args.stack_id}") + return + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 30b506b9..f9f8648c 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -33,6 +33,7 @@ from .resources.run_task import RunTasks from .resources.run_trigger import RunTriggers from .resources.ssh_keys import SSHKeys +from .resources.stack import Stacks from .resources.state_version_outputs import StateVersionOutputs from .resources.state_versions import StateVersions from .resources.variable import Variables @@ -104,6 +105,7 @@ def __init__(self, config: TFEConfig | None = None): # Reserved Tag Key self.reserved_tag_key = ReservedTagKeys(self._transport) + self.stacks = Stacks(self._transport) def close(self) -> None: try: diff --git a/src/pytfe/models/stack.py b/src/pytfe/models/stack.py new file mode 100644 index 00000000..c6378f17 --- /dev/null +++ b/src/pytfe/models/stack.py @@ -0,0 +1,120 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from ..errors import ERR_REQUIRED_NAME, ERR_REQUIRED_PROJECT +from .agent import AgentPool +from .project import Project + + +class StackSortColumn(str, Enum): + """StackSortColumn represents a string that can be used to sort items when using the List method.""" + + STACK_SORT_BY_NAME = "name" + STACK_SORT_BY_UPDATED_AT = "updated-at" + STACK_SORT_BY_NAME_DESC = "-name" + STACK_SORT_BY_UPDATED_AT_DESC = "-updated-at" + + +class StackVcsRepo(BaseModel): + """StackVCSRepo represents the version control system repository for a stack.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + identifier: str = Field(alias="identifier") + branch: str | None = Field(default=None, alias="branch") + gha_installation_id: str | None = Field( + default=None, alias="github-app-installation-id" + ) + oauth_token_id: str | None = Field(default=None, alias="oauth-token-id") + + +class StackVcsRepoOptions(BaseModel): + """StackVCSRepoOptions represents the options for the version control system repository for a stack.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + identifier: str = Field(alias="identifier") + branch: str | None = Field(default=None, alias="branch") + gha_installation_id: str | None = Field( + default=None, alias="github-app-installation-id" + ) + oauth_token_id: str | None = Field(default=None, alias="oauth-token-id") + + +class Stack(BaseModel): + """Stack represents a stack in Terraform Cloud.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + name: str | None = Field(default=None, alias="name") + description: str | None = Field(default=None, alias="description") + created_at: datetime | None = Field(default=None, alias="created-at") + updated_at: datetime | None = Field(default=None, alias="updated-at") + vcs_repo: StackVcsRepo | None = Field(default=None, alias="vcs-repo") + speculation_enabled: bool | None = Field(default=None, alias="speculation-enabled") + upstream_count: int | None = Field(default=None, alias="upstream-count") + downstream_count: int | None = Field(default=None, alias="downstream-count") + inputs_count: int | None = Field(default=None, alias="inputs-count") + outputs_count: int | None = Field(default=None, alias="outputs-count") + creation_source: str | None = Field(default=None, alias="creation-source") + + # Relations + project: Project | None = Field(default=None, alias="project") + agent_pool: AgentPool | None = Field(default=None, alias="agent-pool") + # latest_stack_configuration: dict[str, Any] | None = Field(default=None, alias="latest-stack-configuration") + + +class StackListOptions(BaseModel): + """StackListOptions represents the options for listing stacks.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_size: int | None = Field(default=None, alias="page[size]") + project_id: str | None = Field(default=None, alias="filter[project][id]") + sort: StackSortColumn | None = Field(default=None, alias="sort") + search_by_name: str | None = Field(default=None, alias="search[name]") + + +class StackCreateOptions(BaseModel): + """StackCreateOptions represents the options for creating a stack.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + name: str = Field(alias="name") + migration: bool | None = Field(default=None, alias="migration") + description: str | None = Field(default=None, alias="description") + speculation_enabled: bool | None = Field(default=None, alias="speculation-enabled") + vcs_repo: StackVcsRepoOptions | None = Field(default=None, alias="vcs-repo") + project: Project = Field(alias="project") + agent_pool: AgentPool | None = Field(default=None, alias="agent-pool") + + @model_validator(mode="after") + def valid(self) -> StackCreateOptions: + if self.name == "": + raise ValueError(ERR_REQUIRED_NAME) + + if self.project and self.project.id == "": + raise ValueError(ERR_REQUIRED_PROJECT) + + return self + + +class StackUpdateOptions(BaseModel): + """StackUpdateOptions represents the options for updating a stack.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + name: str | None = Field(default=None, alias="name") + description: str | None = Field(default=None, alias="description") + speculation_enabled: bool | None = Field(default=None, alias="speculation-enabled") + vcs_repo: StackVcsRepoOptions | None = Field(default=None, alias="vcs-repo") + project: Project | None = Field(default=None, alias="project") + agent_pool: AgentPool | None = Field(default=None, alias="agent-pool") diff --git a/src/pytfe/resources/stack.py b/src/pytfe/resources/stack.py new file mode 100644 index 00000000..dac0b1ca --- /dev/null +++ b/src/pytfe/resources/stack.py @@ -0,0 +1,135 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +from __future__ import annotations + +from collections.abc import Iterator + +from pytfe.models import ( + AgentPool, + Project, +) + +from ..models.stack import ( + Stack, + StackCreateOptions, + StackListOptions, + StackUpdateOptions, + StackVcsRepo, +) +from ._base import _Service + + +class Stacks(_Service): + def create(self, options: StackCreateOptions) -> Stack: + """Create a new stack within a project.""" + payload = { + "data": { + "attributes": options.model_dump( + by_alias=True, exclude_none=True, exclude={"project", "agent_pool"} + ), + "type": "stacks", + "relationships": {}, + } + } + relationships = {} + if options.project: + relationships["project"] = { + "data": {"id": options.project.id, "type": "projects"} + } + if options.agent_pool: + relationships["agent-pool"] = { + "data": {"id": options.agent_pool.id, "type": "agent-pools"} + } + payload["data"]["relationships"] = relationships + r = self.t.request( + "POST", + path="/api/v2/stacks", + json_body=payload, + ) + data = r.json().get("data", {}) + return self._stack_from(data) + + def update(self, stack_id: str, options: StackUpdateOptions) -> Stack: + """Update an existing stack.""" + payload = { + "data": { + "attributes": options.model_dump( + by_alias=True, + exclude_none=True, + exclude={"agent_pool", "project"}, + ), + "type": "stacks", + "relationships": {}, + } + } + relationships = {} + if options.project: + relationships.update( + {"project": {"data": {"id": options.project.id, "type": "projects"}}} + ) + if options.agent_pool: + relationships.update( + { + "agent-pool": { + "data": {"id": options.agent_pool.id, "type": "agent-pools"} + } + } + ) + payload["data"]["relationships"] = relationships + r = self.t.request( + "PATCH", + path=f"/api/v2/stacks/{stack_id}", + json_body=payload, + ) + data = r.json().get("data", {}) + return self._stack_from(data) + + def list(self, organization: str, options: StackListOptions) -> Iterator[Stack]: + """List stacks within an organization, with optional filtering by project.""" + params = options.model_dump(by_alias=True, exclude_none=True) + path = f"/api/v2/organizations/{organization}/stacks" + for item in self._list(path, params=params): + yield self._stack_from(item) + + def read(self, stack_id: str) -> Stack: + """Read a stack by ID.""" + r = self.t.request( + "GET", + path=f"/api/v2/stacks/{stack_id}", + ) + data = r.json().get("data", {}) + return self._stack_from(data) + + def delete(self, stack_id: str) -> None: + """Delete a stack by ID.""" + self.t.request( + "DELETE", + path=f"/api/v2/stacks/{stack_id}", + ) + return None + + def force_delete(self, stack_id: str) -> None: + """ForceDelete deletes a stack that still has deployments.""" + self.t.request( + "DELETE", + path=f"/api/v2/stacks/{stack_id}?force=true", + ) + return None + + def _stack_from(self, data: dict) -> Stack: + attrs = data.get("attributes", {}) + attrs["id"] = data.get("id") + relationships = data.get("relationships", {}) + vcs_repo_raw = attrs.get("vcs-repo") + if vcs_repo_raw: + attrs["vcs_repo"] = StackVcsRepo.model_validate(vcs_repo_raw) + else: + attrs["vcs_repo"] = None + project_data = relationships.get("project", {}).get("data", {}) + agent_pool_data = relationships.get("agent-pool", {}).get("data", {}) + if isinstance(project_data, dict) and project_data.get("id"): + attrs["project"] = Project(id=project_data["id"]) + if isinstance(agent_pool_data, dict) and agent_pool_data.get("id"): + attrs["agent_pool"] = AgentPool(id=agent_pool_data["id"]) + return Stack.model_validate(attrs) diff --git a/tests/units/test_stack.py b/tests/units/test_stack.py new file mode 100644 index 00000000..03c9d28f --- /dev/null +++ b/tests/units/test_stack.py @@ -0,0 +1,267 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +"""Unit tests for the stack module.""" + +from unittest.mock import Mock + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.models.agent import AgentPool +from pytfe.models.project import Project +from pytfe.models.stack import ( + Stack, + StackCreateOptions, + StackListOptions, + StackSortColumn, + StackUpdateOptions, + StackVcsRepoOptions, +) +from pytfe.resources.stack import Stacks + + +class TestStacks: + """Test the Stacks service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def stacks_service(self, mock_transport): + """Create a Stacks service with mocked transport.""" + return Stacks(mock_transport) + + @pytest.fixture + def stack_response_data(self): + """Return sample API response data for a stack.""" + return { + "id": "st-123", + "attributes": { + "name": "demo-stack", + "description": "Stack description", + "speculation-enabled": True, + "vcs-repo": { + "identifier": "hashicorp/terraform", + "branch": "main", + "oauth-token-id": "ot-123", + }, + }, + "relationships": { + "project": {"data": {"id": "prj-123", "type": "projects"}}, + "agent-pool": {"data": {"id": "apool-123", "type": "agent-pools"}}, + }, + } + + def test_list_stacks_success(self, stacks_service, stack_response_data): + """Test successful list operation.""" + stacks_service._list = Mock(return_value=[stack_response_data]) + + options = StackListOptions( + page_size=10, + project_id="prj-123", + sort=StackSortColumn.STACK_SORT_BY_NAME, + search_by_name="demo", + ) + + result_iter = stacks_service.list("org-123", options) + items = list(result_iter) + + stacks_service._list.assert_called_once_with( + "/api/v2/organizations/org-123/stacks", + params={ + "page[size]": 10, + "filter[project][id]": "prj-123", + "sort": "name", + "search[name]": "demo", + }, + ) + + assert len(items) == 1 + assert isinstance(items[0], Stack) + assert items[0].id == "st-123" + assert items[0].name == "demo-stack" + + def test_create_stack_success( + self, stacks_service, mock_transport, stack_response_data + ): + """Test successful create operation.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_response_data} + mock_transport.request.return_value = mock_response + + options = StackCreateOptions( + name="demo-stack", + description="Stack description", + speculation_enabled=True, + vcs_repo=StackVcsRepoOptions( + identifier="hashicorp/terraform", + branch="main", + oauth_token_id="ot-123", + ), + project=Project(id="prj-123"), + agent_pool=AgentPool(id="apool-123"), + ) + + result = stacks_service.create(options) + + mock_transport.request.assert_called_once_with( + "POST", + path="/api/v2/stacks", + json_body={ + "data": { + "attributes": { + "name": "demo-stack", + "description": "Stack description", + "speculation-enabled": True, + "vcs-repo": { + "identifier": "hashicorp/terraform", + "branch": "main", + "oauth-token-id": "ot-123", + }, + }, + "type": "stacks", + "relationships": { + "project": {"data": {"id": "prj-123", "type": "projects"}}, + "agent-pool": { + "data": {"id": "apool-123", "type": "agent-pools"} + }, + }, + } + }, + ) + + assert isinstance(result, Stack) + assert result.id == "st-123" + assert result.project.id == "prj-123" + assert result.agent_pool.id == "apool-123" + + def test_update_stack_success( + self, stacks_service, mock_transport, stack_response_data + ): + """Test successful update operation.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_response_data} + mock_transport.request.return_value = mock_response + + options = StackUpdateOptions( + description="Updated description", + vcs_repo=StackVcsRepoOptions( + identifier="hashicorp/terraform", + branch="main", + ), + project=Project(id="prj-123"), + agent_pool=AgentPool(id="apool-123"), + ) + + result = stacks_service.update("st-123", options) + + mock_transport.request.assert_called_once_with( + "PATCH", + path="/api/v2/stacks/st-123", + json_body={ + "data": { + "attributes": { + "description": "Updated description", + "vcs-repo": { + "identifier": "hashicorp/terraform", + "branch": "main", + }, + }, + "type": "stacks", + "relationships": { + "project": {"data": {"id": "prj-123", "type": "projects"}}, + "agent-pool": { + "data": {"id": "apool-123", "type": "agent-pools"} + }, + }, + } + }, + ) + + assert isinstance(result, Stack) + assert result.id == "st-123" + + def test_read_stack_success( + self, stacks_service, mock_transport, stack_response_data + ): + """Test successful read operation.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_response_data} + mock_transport.request.return_value = mock_response + + result = stacks_service.read("st-123") + + mock_transport.request.assert_called_once_with( + "GET", + path="/api/v2/stacks/st-123", + ) + + assert isinstance(result, Stack) + assert result.id == "st-123" + assert result.name == "demo-stack" + + def test_delete_stack_success(self, stacks_service, mock_transport): + """Test successful delete operation.""" + result = stacks_service.delete("st-123") + + mock_transport.request.assert_called_once_with( + "DELETE", + path="/api/v2/stacks/st-123", + ) + assert result is None + + def test_force_delete_stack_success(self, stacks_service, mock_transport): + """Test successful force-delete operation.""" + result = stacks_service.force_delete("st-123") + + mock_transport.request.assert_called_once_with( + "DELETE", + path="/api/v2/stacks/st-123?force=true", + ) + assert result is None + + def test_stack_from_handles_null_vcs_repo(self, stacks_service): + """Test parsing stack data when vcs-repo is null.""" + data = { + "id": "st-456", + "attributes": { + "name": "no-vcs-stack", + "vcs-repo": None, + }, + "relationships": { + "project": {"data": {"id": "prj-999", "type": "projects"}}, + }, + } + + result = stacks_service._stack_from(data) + + assert isinstance(result, Stack) + assert result.id == "st-456" + assert result.vcs_repo is None + assert result.project is not None + assert result.project.id == "prj-999" + assert result.agent_pool is None + + def test_stack_from_handles_missing_relationships(self, stacks_service): + """Test parsing stack data when relationship data is missing.""" + data = { + "id": "st-789", + "attributes": { + "name": "minimal-stack", + "vcs-repo": None, + }, + "relationships": { + "project": {"data": None}, + "agent-pool": {"data": None}, + }, + } + + result = stacks_service._stack_from(data) + + assert isinstance(result, Stack) + assert result.id == "st-789" + assert result.project is None + assert result.agent_pool is None