diff --git a/examples/workspace_resources.py b/examples/workspace_resources.py new file mode 100644 index 0000000..15fbf1b --- /dev/null +++ b/examples/workspace_resources.py @@ -0,0 +1,118 @@ +"""Example script for working with workspace resources in Terraform Enterprise. + +This script demonstrates how to list resources within a workspace. +""" + +import argparse +import sys + +from pytfe import TFEClient +from pytfe.models import WorkspaceResourceListOptions + + +def list_workspace_resources( + client: TFEClient, + workspace_id: str, + page_number: int | None = None, + page_size: int | None = None, +) -> None: + """List all resources in a workspace.""" + try: + print(f"Listing resources for workspace: {workspace_id}") + + # Prepare list options + options = None + if page_number or page_size: + options = WorkspaceResourceListOptions() + if page_number: + options.page_number = page_number + if page_size: + options.page_size = page_size + + # List workspace resources (returns an iterator) + resources = list(client.workspace_resources.list(workspace_id, options)) + + if not resources: + print("No resources found in this workspace.") + return + + print(f"\nFound {len(resources)} resource(s):") + print("-" * 80) + + for resource in resources: + print(f"ID: {resource.id}") + print(f"Address: {resource.address}") + print(f"Name: {resource.name}") + print(f"Module: {resource.module}") + print(f"Provider: {resource.provider}") + print(f"Provider Type: {resource.provider_type}") + print(f"Created At: {resource.created_at}") + print(f"Updated At: {resource.updated_at}") + print(f"Modified By State Version: {resource.modified_by_state_version_id}") + if resource.name_index: + print(f"Name Index: {resource.name_index}") + print("-" * 80) + + except Exception as e: + print(f"Error listing workspace resources: {e}", file=sys.stderr) + sys.exit(1) + + +def main(): + """Main function to handle command line arguments and execute operations.""" + parser = argparse.ArgumentParser( + description="Manage workspace resources in Terraform Enterprise", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + # List all resources in a workspace + python workspace_resources.py --list --workspace-id ws-abc123 + + # List with pagination + python workspace_resources.py --list --workspace-id ws-abc123 --page-number 2 --page-size 50 + +Environment variables: + TFE_TOKEN: Your Terraform Enterprise API token + TFE_URL: Your Terraform Enterprise URL (default: https://app.terraform.io) + TFE_ORG: Your Terraform Enterprise organization name + """, + ) + + # Add command flags + parser.add_argument("--list", action="store_true", help="List workspace resources") + parser.add_argument( + "--workspace-id", + required=True, + help="ID of the workspace (required, e.g., ws-abc123)", + ) + parser.add_argument("--page-number", type=int, help="Page number for pagination") + parser.add_argument("--page-size", type=int, help="Page size for pagination") + + args = parser.parse_args() + + if not args.list: + parser.print_help() + sys.exit(1) + + # Initialize TFE client + try: + client = TFEClient() + except Exception as e: + print(f"Error initializing TFE client: {e}", file=sys.stderr) + print( + "Make sure TFE_TOKEN and TFE_URL environment variables are set.", + file=sys.stderr, + ) + sys.exit(1) + + # Execute the list command + list_workspace_resources( + client, + args.workspace_id, + args.page_number, + args.page_size, + ) + + +if __name__ == "__main__": + main() diff --git a/src/pytfe/client.py b/src/pytfe/client.py index 7aa1bd8..f50cdfe 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -33,6 +33,7 @@ from .resources.state_versions import StateVersions from .resources.variable import Variables from .resources.variable_sets import VariableSets, VariableSetVariables +from .resources.workspace_resources import WorkspaceResourcesService from .resources.workspaces import Workspaces @@ -72,6 +73,7 @@ def __init__(self, config: TFEConfig | None = None): self.variable_sets = VariableSets(self._transport) self.variable_set_variables = VariableSetVariables(self._transport) self.workspaces = Workspaces(self._transport) + self.workspace_resources = WorkspaceResourcesService(self._transport) self.registry_modules = RegistryModules(self._transport) self.registry_providers = RegistryProviders(self._transport) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index a3cc71f..c70dd05 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -353,6 +353,12 @@ WorkspaceUpdateRemoteStateConsumersOptions, ) +# ── Workspace Resources ─────────────────────────────────────────────────────── +from .workspace_resource import ( + WorkspaceResource, + WorkspaceResourceListOptions, +) + # ── Public surface ──────────────────────────────────────────────────────────── __all__ = [ # OAuth @@ -524,6 +530,9 @@ "WorkspaceTagListOptions", "WorkspaceUpdateOptions", "WorkspaceUpdateRemoteStateConsumersOptions", + # Workspace Resources + "WorkspaceResource", + "WorkspaceResourceListOptions", "RunQueue", "ReadRunQueueOptions", # Runs diff --git a/src/pytfe/models/workspace_resource.py b/src/pytfe/models/workspace_resource.py new file mode 100644 index 0000000..78eaa40 --- /dev/null +++ b/src/pytfe/models/workspace_resource.py @@ -0,0 +1,29 @@ +"""Workspace resources models for Terraform Enterprise.""" + +from pydantic import BaseModel + + +class WorkspaceResource(BaseModel): + """Represents a Terraform Enterprise workspace resource. + + These are resources managed by Terraform in a workspace's state. + """ + + id: str + address: str + name: str + created_at: str + updated_at: str + module: str + provider: str + provider_type: str + modified_by_state_version_id: str + name_index: str | None = None + + +class WorkspaceResourceListOptions(BaseModel): + """Options for listing workspace resources.""" + + # Pagination + page_number: int | None = None + page_size: int | None = None diff --git a/src/pytfe/resources/workspace_resources.py b/src/pytfe/resources/workspace_resources.py new file mode 100644 index 0000000..617b5f7 --- /dev/null +++ b/src/pytfe/resources/workspace_resources.py @@ -0,0 +1,66 @@ +"""Workspace resources service for Terraform Enterprise.""" + +from collections.abc import Iterator +from typing import Any + +from pytfe.models import ( + WorkspaceResource, + WorkspaceResourceListOptions, +) + +from ._base import _Service + + +def _workspace_resource_from(data: dict[str, Any]) -> WorkspaceResource: + """Convert API response data to WorkspaceResource model.""" + attributes = data.get("attributes", {}) + + return WorkspaceResource( + id=data.get("id", ""), + address=attributes.get("address", ""), + name=attributes.get("name", ""), + created_at=attributes.get("created-at", ""), + updated_at=attributes.get("updated-at", ""), + module=attributes.get("module", ""), + provider=attributes.get("provider", ""), + provider_type=attributes.get("provider-type", ""), + modified_by_state_version_id=attributes.get("modified-by-state-version-id", ""), + name_index=attributes.get("name-index"), + ) + + +class WorkspaceResourcesService(_Service): + """Service for managing workspace resources in Terraform Enterprise. + + Workspace resources represent the infrastructure resources + managed by Terraform in a workspace's state file. + """ + + def list( + self, workspace_id: str, options: WorkspaceResourceListOptions | None = None + ) -> Iterator[WorkspaceResource]: + """List workspace resources for a given workspace. + + Args: + workspace_id: The ID of the workspace to list resources for + options: Optional query parameters for filtering and pagination + + Yields: + WorkspaceResource objects + """ + if not workspace_id or not workspace_id.strip(): + raise ValueError("workspace_id is required") + + url = f"/api/v2/workspaces/{workspace_id}/resources" + + # Handle parameters + params: dict[str, int] = {} + if options: + if options.page_number is not None: + params["page[number]"] = options.page_number + if options.page_size is not None: + params["page[size]"] = options.page_size + + # Use the _list method from base service to handle pagination + for item in self._list(url, params=params): + yield _workspace_resource_from(item) diff --git a/tests/units/test_workspace_resources.py b/tests/units/test_workspace_resources.py new file mode 100644 index 0000000..f685d84 --- /dev/null +++ b/tests/units/test_workspace_resources.py @@ -0,0 +1,278 @@ +"""Unit tests for workspace resources service.""" + +from unittest.mock import Mock + +import pytest + +from pytfe.models.workspace_resource import ( + WorkspaceResource, + WorkspaceResourceListOptions, +) +from pytfe.resources.workspace_resources import WorkspaceResourcesService + + +class TestWorkspaceResourcesService: + """Test suite for WorkspaceResourcesService.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock transport for testing.""" + return Mock() + + @pytest.fixture + def service(self, mock_transport): + """Create a WorkspaceResourcesService instance for testing.""" + return WorkspaceResourcesService(mock_transport) + + @pytest.fixture + def sample_workspace_resource_response(self): + """Sample API response for workspace resources list.""" + return { + "data": [ + { + "id": "resource-1", + "type": "resources", + "attributes": { + "address": "media_bucket.aws_s3_bucket_public_access_block.this[0]", + "name": "this", + "created-at": "2023-01-01T00:00:00Z", + "updated-at": "2023-01-01T00:00:00Z", + "module": "media_bucket", + "provider": "hashicorp/aws", + "provider-type": "aws", + "modified-by-state-version-id": "sv-abc123", + "name-index": "0", + }, + }, + { + "id": "resource-2", + "type": "resources", + "attributes": { + "address": "aws_instance.example", + "name": "example", + "created-at": "2023-01-02T00:00:00Z", + "updated-at": "2023-01-02T00:00:00Z", + "module": "root", + "provider": "hashicorp/aws", + "provider-type": "aws", + "modified-by-state-version-id": "sv-def456", + "name-index": None, + }, + }, + ], + "meta": { + "pagination": { + "current_page": 1, + "total_pages": 1, + "total_count": 2, + "page_size": 20, + } + }, + } + + @pytest.fixture + def sample_empty_response(self): + """Sample API response for empty workspace resources list.""" + return { + "data": [], + "meta": { + "pagination": { + "current_page": 1, + "total_pages": 1, + "total_count": 0, + "page_size": 20, + } + }, + } + + def test_list_workspace_resources_success( + self, service, mock_transport, sample_workspace_resource_response + ): + """Test successful listing of workspace resources.""" + # Mock the transport response + mock_response = Mock() + mock_response.json.return_value = sample_workspace_resource_response + mock_transport.request.return_value = mock_response + + # Call the service + result = list(service.list("ws-abc123")) + + # Verify request was made correctly + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/workspaces/ws-abc123/resources", + params={"page[number]": 1, "page[size]": 100}, + ) + + # Verify response parsing + assert isinstance(result, list) + assert len(result) == 2 + + # Check first resource + resource1 = result[0] + assert isinstance(resource1, WorkspaceResource) + assert resource1.id == "resource-1" + assert ( + resource1.address + == "media_bucket.aws_s3_bucket_public_access_block.this[0]" + ) + assert resource1.name == "this" + assert resource1.module == "media_bucket" + assert resource1.provider == "hashicorp/aws" + assert resource1.provider_type == "aws" + assert resource1.modified_by_state_version_id == "sv-abc123" + assert resource1.name_index == "0" + assert resource1.created_at == "2023-01-01T00:00:00Z" + assert resource1.updated_at == "2023-01-01T00:00:00Z" + + # Check second resource + resource2 = result[1] + assert resource2.id == "resource-2" + assert resource2.address == "aws_instance.example" + assert resource2.name == "example" + assert resource2.module == "root" + assert resource2.name_index is None + + def test_list_workspace_resources_with_options( + self, service, mock_transport, sample_workspace_resource_response + ): + """Test listing workspace resources with pagination options.""" + # Mock the transport response + mock_response = Mock() + mock_response.json.return_value = sample_workspace_resource_response + mock_transport.request.return_value = mock_response + + # Create options + options = WorkspaceResourceListOptions(page_number=2, page_size=50) + + # Call the service + result = list(service.list("ws-abc123", options)) + + # Verify request was made correctly + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/workspaces/ws-abc123/resources", + params={"page[number]": 2, "page[size]": 50}, + ) + + # Verify response + assert isinstance(result, list) + assert len(result) == 2 + + def test_list_workspace_resources_empty( + self, service, mock_transport, sample_empty_response + ): + """Test listing workspace resources when no resources exist.""" + # Mock the transport response + mock_response = Mock() + mock_response.json.return_value = sample_empty_response + mock_transport.request.return_value = mock_response + + # Call the service + result = list(service.list("ws-abc123")) + + # Verify request was made correctly + mock_transport.request.assert_called_once_with( + "GET", + "/api/v2/workspaces/ws-abc123/resources", + params={"page[number]": 1, "page[size]": 100}, + ) + + # Verify response + assert isinstance(result, list) + assert len(result) == 0 + + def test_list_workspace_resources_invalid_workspace_id(self, service): + """Test listing workspace resources with invalid workspace ID.""" + with pytest.raises(ValueError, match="workspace_id is required"): + list(service.list("")) + + with pytest.raises(ValueError, match="workspace_id is required"): + list(service.list(None)) + + def test_list_workspace_resources_malformed_response(self, service, mock_transport): + """Test handling of malformed API response.""" + # Mock malformed response + mock_response = Mock() + mock_response.json.return_value = {"invalid": "response"} + mock_transport.request.return_value = mock_response + + # Call the service + result = list(service.list("ws-abc123")) + + # Should handle gracefully and return empty list + assert isinstance(result, list) + assert len(result) == 0 + + def test_list_workspace_resources_api_error(self, service, mock_transport): + """Test handling of API errors.""" + # Mock API error + mock_transport.request.side_effect = Exception("API Error") + + # Should propagate the exception + with pytest.raises(Exception, match="API Error"): + list(service.list("ws-abc123")) + + +class TestWorkspaceResourceModel: + """Test suite for WorkspaceResource model.""" + + def test_workspace_resource_creation(self): + """Test creating a WorkspaceResource instance.""" + resource = WorkspaceResource( + id="resource-1", + address="aws_instance.example", + name="example", + created_at="2023-01-01T00:00:00Z", + updated_at="2023-01-01T00:00:00Z", + module="root", + provider="hashicorp/aws", + provider_type="aws", + modified_by_state_version_id="sv-abc123", + name_index="0", + ) + + assert resource.id == "resource-1" + assert resource.address == "aws_instance.example" + assert resource.name == "example" + assert resource.module == "root" + assert resource.provider == "hashicorp/aws" + assert resource.provider_type == "aws" + assert resource.modified_by_state_version_id == "sv-abc123" + assert resource.name_index == "0" + + def test_workspace_resource_optional_fields(self): + """Test WorkspaceResource with optional fields.""" + resource = WorkspaceResource( + id="resource-1", + address="aws_instance.example", + name="example", + created_at="2023-01-01T00:00:00Z", + updated_at="2023-01-01T00:00:00Z", + module="root", + provider="hashicorp/aws", + provider_type="aws", + modified_by_state_version_id="sv-abc123", + # name_index is optional + ) + + assert resource.name_index is None + + +class TestWorkspaceResourceListOptions: + """Test suite for WorkspaceResourceListOptions model.""" + + def test_workspace_resource_list_options_creation(self): + """Test creating WorkspaceResourceListOptions.""" + options = WorkspaceResourceListOptions(page_number=2, page_size=50) + + assert options.page_number == 2 + assert options.page_size == 50 + + def test_workspace_resource_list_options_defaults(self): + """Test WorkspaceResourceListOptions with defaults.""" + options = WorkspaceResourceListOptions() + + # Should use default values from BaseListOptions + assert options.page_number is None + assert options.page_size is None