diff --git a/examples/stack_configuration.py b/examples/stack_configuration.py new file mode 100644 index 00000000..d51ace60 --- /dev/null +++ b/examples/stack_configuration.py @@ -0,0 +1,116 @@ +# 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 ( + StackConfigurationCreateOptions, + StackConfigurationListOptions, +) + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Stack Configurations 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("--stack-id", required=True, help="Stack ID (e.g. st-xxxxx)") + parser.add_argument( + "--page-size", + type=int, + default=100, + help="Page size for listing configurations", + ) + parser.add_argument( + "--create", action="store_true", help="Create a new stack configuration" + ) + parser.add_argument( + "--speculative", + action="store_true", + help="Mark created configuration as speculative", + ) + parser.add_argument( + "--read", action="store_true", help="Read a specific stack configuration" + ) + parser.add_argument( + "--upload-url", + action="store_true", + help="Fetch the upload URL for a stack configuration", + ) + parser.add_argument( + "--fetch-from-vcs", + action="store_true", + help="Trigger fetch of latest config from VCS", + ) + parser.add_argument("--id", help="Stack configuration ID (e.g. stc-xxxxx)") + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) Always list existing stack configurations + _print_header(f"Listing stack configurations for stack: {args.stack_id}") + options = StackConfigurationListOptions(page_size=args.page_size) + config_count = 0 + for config in client.stack_configurations.list( + stack_id=args.stack_id, options=options + ): + config_count += 1 + print(f"- ID: {config.id}") + print(f" Status: {config.status.value if config.status else None}") + print(f" Sequence: {config.sequence_number}") + print(f" Speculative: {config.speculative}") + print(f" Created: {config.created_at}") + print(f" Updated: {config.updated_at}") + print() + + if config_count == 0: + print("No stack configurations found.") + else: + print(f"Total: {config_count} stack configurations") + + # 2) Create a new stack configuration + if args.create: + _print_header("Creating a new stack configuration") + create_opts = StackConfigurationCreateOptions( + speculative_enabled=args.speculative + ) + config = client.stack_configurations.create( + stack_id=args.stack_id, options=create_opts + ) + print(f"Created stack configuration: {config.id}") + print(f" Status: {config.status.value if config.status else None}") + print(f" Speculative: {config.speculative}") + print(f" Sequence: {config.sequence_number}") + print(f" Created: {config.created_at}") + + # 3) Read a specific stack configuration + if args.read: + if not args.id: + print("--id is required for --read") + else: + _print_header(f"Reading stack configuration: {args.id}") + config = client.stack_configurations.read(stack_configuration_id=args.id) + print(f"ID: {config.id}") + print(f"Status: {config.status.value if config.status else None}") + print(f"Sequence: {config.sequence_number}") + print(f"Speculative: {config.speculative}") + print(f"Created: {config.created_at}") + print(f"Updated: {config.updated_at}") + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 0e88e1ab..4642d9a8 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -35,8 +35,12 @@ 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.stack_configuration import StackConfigurations from .resources.state_version_outputs import StateVersionOutputs from .resources.state_versions import StateVersions +from .resources.team import Teams +from .resources.team_project_access import TeamProjectAccesses from .resources.user import Users from .resources.variable import Variables from .resources.variable_sets import VariableSets, VariableSetVariables @@ -88,6 +92,10 @@ def __init__(self, config: TFEConfig | None = None): self.registry_provider_versions = RegistryProviderVersions(self._transport) self.registry_provider_platforms = RegistryProviderPlatforms(self._transport) + # Stack resources + self.stacks = Stacks(self._transport) + self.stack_configurations = StackConfigurations(self._transport) + # State and execution resources self.state_versions = StateVersions(self._transport) self.state_version_outputs = StateVersionOutputs(self._transport) @@ -107,6 +115,10 @@ def __init__(self, config: TFEConfig | None = None): # SSH Keys self.ssh_keys = SSHKeys(self._transport) + # Team project access + self.teams = Teams(self._transport) + self.team_project_accesses = TeamProjectAccesses(self._transport) + # Reserved Tag Key self.reserved_tag_key = ReservedTagKeys(self._transport) diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index f2340af3..113dee9a 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -627,3 +627,18 @@ def __init__( message: str = "invalid value for registry-name. It must be either private or public", ): super().__init__(message) + + +# Stack Configuration errors +class InvalidStackIDError(InvalidValues): + """Raised when an invalid stack ID is provided.""" + + def __init__(self, message: str = "invalid value for stack ID"): + super().__init__(message) + + +class InvalidStackConfigurationIDError(InvalidValues): + """Raised when an invalid stack configuration ID is provided.""" + + def __init__(self, message: str = "invalid value for stack configuration ID"): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index c0d6c773..d2e8648c 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -313,6 +313,16 @@ SSHKeyListOptions, SSHKeyUpdateOptions, ) +from .stack_configuration import ( + StackComponent, + StackConfiguration, + StackConfigurationCreateOptions, + StackConfigurationIncludeOps, + StackConfigurationListOptions, + StackConfigurationReadOptions, + StackConfigurationSource, + StackConfigurationStatus, +) from .state_version import ( StateVersion, StateVersionCreateOptions, @@ -513,6 +523,15 @@ "RegistryProviderPlatformID", "RegistryProviderPlatformListOptions", "RegistryProviderPlatformPermissions", + # Stack Configuration + "StackComponent", + "StackConfiguration", + "StackConfigurationCreateOptions", + "StackConfigurationIncludeOps", + "StackConfigurationListOptions", + "StackConfigurationReadOptions", + "StackConfigurationSource", + "StackConfigurationStatus", # Query runs "QueryRun", "QueryRunActions", diff --git a/src/pytfe/models/stack_configuration.py b/src/pytfe/models/stack_configuration.py new file mode 100644 index 00000000..b92931a9 --- /dev/null +++ b/src/pytfe/models/stack_configuration.py @@ -0,0 +1,100 @@ +# 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 + +from .configuration_version import IngressAttributes +from .stack import Stack + + +class StackConfigurationStatus(str, Enum): + """StackConfigurationStatus represents the status of a stack configuration.""" + + PENDING = "pending" + QUEUED = "queued" + PREPARING = "preparing" + COMPLETED = "completed" + FAILED = "failed" + + +class StackComponent(BaseModel): + """StackComponent represents a stack component, specified by configuration""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + name: str = Field(alias="name", default="") + correlator: str = Field(alias="correlator", default="") + expanded: bool | None = Field(alias="expanded", default=None) + removed: bool | None = Field(alias="removed", default=None) + + +class StackConfigurationSource(str, Enum): + """StackConfigurationSource controls how configuration content is sourced.""" + + MANUAL = "manual" + FETCH = "fetch" + REUSE = "reuse" + + +class StackConfigurationIncludeOps(str, Enum): + """StackConfigurationIncludeOps represents include options for stack configuration endpoints.""" + + INGRESS_ATTRIBUTES = "ingress_attributes" + STACK_DIAGNOSTICS = "stack_diagnostics" + + +class StackConfiguration(BaseModel): + """StackConfiguration represents a snapshot of a stack's configuration.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + status: StackConfigurationStatus | None = Field(default=None, alias="status") + sequence_number: int | None = Field(default=None, alias="sequence-number") + components: list[StackComponent] = Field(default_factory=list, alias="components") + preparing_event_stream_url: str = Field( + default="", alias="preparing-event-stream-url" + ) + created_at: datetime | None = Field(default=None, alias="created-at") + updated_at: datetime | None = Field(default=None, alias="updated-at") + speculative: bool | None = Field(default=None, alias="speculative") + + # Relations + stack: Stack | None = Field(default=None, alias="stack") + ingress_attributes: IngressAttributes | None = Field( + default=None, alias="ingress-attributes" + ) + + +class StackConfigurationCreateOptions(BaseModel): + """Options for creating a stack configuration.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + speculative_enabled: bool = Field(default=False, alias="speculative") + destroy_all: bool = Field(default=False, alias="destroy-all") + selected_deployments: list[str] | None = Field( + default=None, alias="selected-deployments" + ) + + +class StackConfigurationListOptions(BaseModel): + """Options for listing stack configurations.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_size: int | None = Field(default=None, alias="page[size]") + include: list[StackConfigurationIncludeOps] | None = None + + +class StackConfigurationReadOptions(BaseModel): + """Options for reading a stack configuration.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + include: list[StackConfigurationIncludeOps] | None = None diff --git a/src/pytfe/resources/stack.py b/src/pytfe/resources/stack.py index dac0b1ca..f683fd17 100644 --- a/src/pytfe/resources/stack.py +++ b/src/pytfe/resources/stack.py @@ -117,6 +117,13 @@ def force_delete(self, stack_id: str) -> None: ) return None + def fetch_latest_from_vcs(self, stack_id: str) -> Stack: + """FetchLatestFromVcs updates the configuration of a stack, triggering stack preparation.""" + path = f"/api/v2/stacks/{stack_id}/fetch-latest-from-vcs" + r = self.t.request("POST", path=path) + data = r.json().get("data", {}) + return self._stack_from(data) + def _stack_from(self, data: dict) -> Stack: attrs = data.get("attributes", {}) attrs["id"] = data.get("id") diff --git a/src/pytfe/resources/stack_configuration.py b/src/pytfe/resources/stack_configuration.py new file mode 100644 index 00000000..3b672b23 --- /dev/null +++ b/src/pytfe/resources/stack_configuration.py @@ -0,0 +1,98 @@ +# 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 pytfe.models.configuration_version import IngressAttributes + +from ..models.stack import Stack +from ..models.stack_configuration import ( + StackConfiguration, + StackConfigurationCreateOptions, + StackConfigurationListOptions, + StackConfigurationReadOptions, + StackConfigurationSource, +) +from ._base import _Service + + +class StackConfigurations(_Service): + """Service for managing Terraform stack configurations.""" + + def create( + self, + stack_id: str, + options: StackConfigurationCreateOptions | None = None, + source: StackConfigurationSource = StackConfigurationSource.MANUAL, + ) -> StackConfiguration: + """Create a stack configuration for the given stack.""" + path = f"/api/v2/stacks/{stack_id}/stack-configurations" + params: dict[str, str] = {} + if source != StackConfigurationSource.MANUAL: + params["source"] = source.value + + attributes: dict[str, Any] = {} + if options: + attributes = options.model_dump(by_alias=True, exclude_none=True) + + payload = { + "data": { + "type": "stack-configurations", + "attributes": attributes, + } + } + r = self.t.request("POST", path=path, json_body=payload, params=params) + data = r.json().get("data", {}) + return self._stack_configuration_from(data) + + def list( + self, + stack_id: str, + options: StackConfigurationListOptions | None = None, + ) -> Iterator[StackConfiguration]: + """List stack configurations for the given stack.""" + path = f"/api/v2/stacks/{stack_id}/stack-configurations" + params: dict[str, Any] = {} + if options: + if options.page_size is not None: + params["page[size]"] = options.page_size + if options.include: + params["include"] = ",".join([i.value for i in options.include]) + for item in self._list(path=path, params=params): + yield self._stack_configuration_from(item) + + def read( + self, + stack_configuration_id: str, + options: StackConfigurationReadOptions | None = None, + ) -> StackConfiguration: + """Read a stack configuration by its ID.""" + path = f"/api/v2/stack-configurations/{stack_configuration_id}" + params: dict[str, str] = {} + if options and options.include: + params["include"] = ",".join([i.value for i in options.include]) + r = self.t.request("GET", path=path, params=params) + data = r.json().get("data", {}) + return self._stack_configuration_from(data) + + def _stack_configuration_from(self, data: dict[str, Any]) -> StackConfiguration: + """Parse a StackConfiguration from API response data.""" + attrs = dict(data.get("attributes", {})) + attrs["id"] = data.get("id") + relationships = data.get("relationships", {}) + + stack_data = relationships.get("stack", {}).get("data") + if stack_data and stack_data.get("id"): + attrs["stack"] = Stack.model_validate({"id": stack_data["id"]}) + ingress_attributes_data = relationships.get("ingress-attributes", {}).get( + "data" + ) + if ingress_attributes_data and ingress_attributes_data.get("id"): + attrs["ingress_attributes"] = IngressAttributes.model_validate( + {"id": ingress_attributes_data["id"]} + ) + + return StackConfiguration.model_validate(attrs) diff --git a/tests/units/test_stack.py b/tests/units/test_stack.py index 03c9d28f..40f57c2a 100644 --- a/tests/units/test_stack.py +++ b/tests/units/test_stack.py @@ -265,3 +265,75 @@ def test_stack_from_handles_missing_relationships(self, stacks_service): assert result.id == "st-789" assert result.project is None assert result.agent_pool is None + + def test_fetch_latest_from_vcs_success( + self, stacks_service, mock_transport, stack_response_data + ): + """Test successful fetch-latest-from-vcs operation.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_response_data} + mock_transport.request.return_value = mock_response + + result = stacks_service.fetch_latest_from_vcs("st-123") + + mock_transport.request.assert_called_once_with( + "POST", + path="/api/v2/stacks/st-123/fetch-latest-from-vcs", + ) + assert isinstance(result, Stack) + assert result.id == "st-123" + + def test_create_stack_invalid_name(self): + """StackCreateOptions raises when name is empty.""" + with pytest.raises(ValueError): + StackCreateOptions( + name="", + project=Project(id="prj-123"), + ) + + def test_create_stack_invalid_project_id(self): + """StackCreateOptions raises when project id is empty.""" + with pytest.raises(ValueError): + StackCreateOptions( + name="demo-stack", + project=Project(id=""), + ) + + def test_list_stacks_no_options(self, stacks_service): + """list() works correctly with minimal options (no filter/sort).""" + stacks_service._list = Mock(return_value=[]) + + results = list( + stacks_service.list( + "org-123", + StackListOptions(), + ) + ) + + stacks_service._list.assert_called_once_with( + "/api/v2/organizations/org-123/stacks", + params={}, + ) + assert results == [] + + def test_stack_from_with_vcs_repo(self, stacks_service): + """_stack_from parses vcs-repo fields correctly.""" + data = { + "id": "st-vcs", + "attributes": { + "name": "vcs-stack", + "vcs-repo": { + "identifier": "hashicorp/terraform", + "branch": "main", + "oauth-token-id": "ot-abc", + }, + }, + "relationships": {}, + } + + result = stacks_service._stack_from(data) + + assert result.vcs_repo is not None + assert result.vcs_repo.identifier == "hashicorp/terraform" + assert result.vcs_repo.branch == "main" + assert result.vcs_repo.oauth_token_id == "ot-abc" diff --git a/tests/units/test_stack_configuration.py b/tests/units/test_stack_configuration.py new file mode 100644 index 00000000..f547bbef --- /dev/null +++ b/tests/units/test_stack_configuration.py @@ -0,0 +1,355 @@ +# Copyright IBM Corp. 2025, 2026 +# SPDX-License-Identifier: MPL-2.0 + +"""Unit tests for the stack_configuration module.""" + +from unittest.mock import Mock + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.models.configuration_version import IngressAttributes +from pytfe.models.stack import Stack +from pytfe.models.stack_configuration import ( + StackComponent, + StackConfiguration, + StackConfigurationCreateOptions, + StackConfigurationIncludeOps, + StackConfigurationListOptions, + StackConfigurationReadOptions, + StackConfigurationSource, + StackConfigurationStatus, +) +from pytfe.resources.stack_configuration import StackConfigurations + + +class TestStackConfigurations: + """Test the StackConfigurations service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def service(self, mock_transport): + """Create a StackConfigurations service with mocked transport.""" + return StackConfigurations(mock_transport) + + @pytest.fixture + def stack_configuration_api_data(self): + """Typical API response for a single stack configuration.""" + return { + "id": "stc-abc123", + "type": "stack-configurations", + "attributes": { + "status": "completed", + "sequence-number": 3, + "speculative": False, + "destroy-all": False, + "preparing-event-stream-url": "https://example.com/stream", + "created-at": "2026-05-07T11:32:17.031000+00:00", + "updated-at": "2026-05-07T11:32:50.500000+00:00", + "components": [ + { + "name": "simple_default", + "correlator": "simple_default", + "expanded": True, + "removed": False, + } + ], + }, + "relationships": { + "stack": {"data": {"id": "st-xyz789", "type": "stacks"}}, + "ingress-attributes": { + "data": {"id": "ia-111", "type": "ingress-attributes"} + }, + }, + } + + # ── Model tests ────────────────────────────────────────────────────────── + + def test_stack_component_defaults(self): + """StackComponent can be constructed with defaults.""" + comp = StackComponent() + assert comp.name == "" + assert comp.correlator == "" + assert comp.expanded is None + assert comp.removed is None + + def test_stack_component_full(self): + """StackComponent parses all fields.""" + comp = StackComponent.model_validate( + { + "name": "my_stack", + "correlator": "corr-1", + "expanded": True, + "removed": False, + } + ) + assert comp.name == "my_stack" + assert comp.correlator == "corr-1" + assert comp.expanded is True + assert comp.removed is False + + def test_stack_configuration_status_enum(self): + """StackConfigurationStatus values are correct.""" + assert StackConfigurationStatus.PENDING == "pending" + assert StackConfigurationStatus.QUEUED == "queued" + assert StackConfigurationStatus.PREPARING == "preparing" + assert StackConfigurationStatus.COMPLETED == "completed" + assert StackConfigurationStatus.FAILED == "failed" + + def test_stack_configuration_source_enum(self): + """StackConfigurationSource values are correct.""" + assert StackConfigurationSource.MANUAL == "manual" + assert StackConfigurationSource.FETCH == "fetch" + assert StackConfigurationSource.REUSE == "reuse" + + def test_stack_configuration_include_enum(self): + """StackConfigurationIncludeOps values are correct.""" + assert StackConfigurationIncludeOps.INGRESS_ATTRIBUTES == "ingress_attributes" + assert StackConfigurationIncludeOps.STACK_DIAGNOSTICS == "stack_diagnostics" + + def test_create_options_defaults(self): + """StackConfigurationCreateOptions has sane defaults.""" + opts = StackConfigurationCreateOptions() + assert opts.speculative_enabled is False + assert opts.destroy_all is False + assert opts.selected_deployments is None + + def test_create_options_serializes_with_aliases(self): + """StackConfigurationCreateOptions serialises with API aliases.""" + opts = StackConfigurationCreateOptions( + speculative_enabled=True, + destroy_all=True, + selected_deployments=["dep-a", "dep-b"], + ) + dumped = opts.model_dump(by_alias=True, exclude_none=True) + assert dumped["speculative"] is True + assert dumped["destroy-all"] is True + assert dumped["selected-deployments"] == ["dep-a", "dep-b"] + + def test_list_options_serialization(self): + """StackConfigurationListOptions serialises page[size] alias.""" + opts = StackConfigurationListOptions( + page_size=50, + include=[StackConfigurationIncludeOps.INGRESS_ATTRIBUTES], + ) + assert opts.page_size == 50 + assert opts.include == [StackConfigurationIncludeOps.INGRESS_ATTRIBUTES] + + def test_read_options(self): + """StackConfigurationReadOptions stores include list.""" + opts = StackConfigurationReadOptions( + include=[ + StackConfigurationIncludeOps.INGRESS_ATTRIBUTES, + StackConfigurationIncludeOps.STACK_DIAGNOSTICS, + ] + ) + assert len(opts.include) == 2 + + # ── Parser tests ───────────────────────────────────────────────────────── + + def test_stack_configuration_from_full_data( + self, service, stack_configuration_api_data + ): + """_stack_configuration_from parses all attributes and relations.""" + result = service._stack_configuration_from(stack_configuration_api_data) + + assert isinstance(result, StackConfiguration) + assert result.id == "stc-abc123" + assert result.status == StackConfigurationStatus.COMPLETED + assert result.sequence_number == 3 + assert result.speculative is False + assert result.preparing_event_stream_url == "https://example.com/stream" + assert result.created_at is not None + assert result.updated_at is not None + + # Components + assert len(result.components) == 1 + assert result.components[0].name == "simple_default" + assert result.components[0].expanded is True + + # Relations + assert isinstance(result.stack, Stack) + assert result.stack.id == "st-xyz789" + assert isinstance(result.ingress_attributes, IngressAttributes) + + def test_stack_configuration_from_no_relationships(self, service): + """_stack_configuration_from handles missing relationship data gracefully.""" + data = { + "id": "stc-min", + "attributes": { + "status": "pending", + "sequence-number": 1, + }, + "relationships": {}, + } + result = service._stack_configuration_from(data) + + assert result.id == "stc-min" + assert result.status == StackConfigurationStatus.PENDING + assert result.stack is None + assert result.ingress_attributes is None + + def test_stack_configuration_from_null_relationship_data(self, service): + """_stack_configuration_from handles null data inside relationship.""" + data = { + "id": "stc-null", + "attributes": {"status": "queued"}, + "relationships": { + "stack": {"data": None}, + "ingress-attributes": {"data": None}, + }, + } + result = service._stack_configuration_from(data) + + assert result.id == "stc-null" + assert result.stack is None + assert result.ingress_attributes is None + + def test_stack_configuration_from_empty_components(self, service): + """_stack_configuration_from handles empty components list.""" + data = { + "id": "stc-empty", + "attributes": {"status": "completed", "components": []}, + "relationships": {}, + } + result = service._stack_configuration_from(data) + assert result.components == [] + + # ── Resource method tests ───────────────────────────────────────────────── + + def test_create_success( + self, service, mock_transport, stack_configuration_api_data + ): + """create() POSTs the correct payload and returns a StackConfiguration.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_configuration_api_data} + mock_transport.request.return_value = mock_response + + opts = StackConfigurationCreateOptions(speculative_enabled=True) + result = service.create(stack_id="st-xyz789", options=opts) + + mock_transport.request.assert_called_once_with( + "POST", + path="/api/v2/stacks/st-xyz789/stack-configurations", + json_body={ + "data": { + "type": "stack-configurations", + "attributes": {"speculative": True, "destroy-all": False}, + } + }, + params={}, + ) + assert isinstance(result, StackConfiguration) + assert result.id == "stc-abc123" + + def test_create_with_fetch_source( + self, service, mock_transport, stack_configuration_api_data + ): + """create() passes source param when not MANUAL.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_configuration_api_data} + mock_transport.request.return_value = mock_response + + service.create(stack_id="st-xyz789", source=StackConfigurationSource.FETCH) + + _, kwargs = mock_transport.request.call_args + assert kwargs["params"] == {"source": "fetch"} + + def test_create_no_options( + self, service, mock_transport, stack_configuration_api_data + ): + """create() sends empty attributes when no options provided.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_configuration_api_data} + mock_transport.request.return_value = mock_response + + service.create(stack_id="st-xyz789") + + _, kwargs = mock_transport.request.call_args + assert kwargs["json_body"]["data"]["attributes"] == {} + + def test_list_success(self, service, stack_configuration_api_data): + """list() yields StackConfiguration objects from paginated results.""" + service._list = Mock(return_value=[stack_configuration_api_data]) + + opts = StackConfigurationListOptions(page_size=20) + results = list(service.list(stack_id="st-xyz789", options=opts)) + + service._list.assert_called_once_with( + path="/api/v2/stacks/st-xyz789/stack-configurations", + params={"page[size]": 20}, + ) + assert len(results) == 1 + assert isinstance(results[0], StackConfiguration) + assert results[0].id == "stc-abc123" + + def test_list_with_include(self, service, stack_configuration_api_data): + """list() passes include param as comma-separated string.""" + service._list = Mock(return_value=[stack_configuration_api_data]) + + opts = StackConfigurationListOptions( + include=[StackConfigurationIncludeOps.INGRESS_ATTRIBUTES] + ) + list(service.list(stack_id="st-xyz789", options=opts)) + + _, kwargs = service._list.call_args + assert kwargs["params"]["include"] == "ingress_attributes" + + def test_list_empty(self, service): + """list() returns empty iterator when no items returned.""" + service._list = Mock(return_value=[]) + + results = list(service.list(stack_id="st-xyz789")) + assert results == [] + + def test_list_no_options(self, service, stack_configuration_api_data): + """list() works correctly when no options are given.""" + service._list = Mock(return_value=[stack_configuration_api_data]) + + results = list(service.list(stack_id="st-xyz789")) + + service._list.assert_called_once_with( + path="/api/v2/stacks/st-xyz789/stack-configurations", + params={}, + ) + assert len(results) == 1 + + def test_read_success(self, service, mock_transport, stack_configuration_api_data): + """read() GETs the correct path and returns a StackConfiguration.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_configuration_api_data} + mock_transport.request.return_value = mock_response + + result = service.read(stack_configuration_id="stc-abc123") + + mock_transport.request.assert_called_once_with( + "GET", + path="/api/v2/stack-configurations/stc-abc123", + params={}, + ) + assert isinstance(result, StackConfiguration) + assert result.id == "stc-abc123" + assert result.status == StackConfigurationStatus.COMPLETED + + def test_read_with_include( + self, service, mock_transport, stack_configuration_api_data + ): + """read() appends include query param.""" + mock_response = Mock() + mock_response.json.return_value = {"data": stack_configuration_api_data} + mock_transport.request.return_value = mock_response + + opts = StackConfigurationReadOptions( + include=[ + StackConfigurationIncludeOps.INGRESS_ATTRIBUTES, + StackConfigurationIncludeOps.STACK_DIAGNOSTICS, + ] + ) + service.read(stack_configuration_id="stc-abc123", options=opts) + + _, kwargs = mock_transport.request.call_args + assert kwargs["params"]["include"] == "ingress_attributes,stack_diagnostics"