From db896904e1b5491d2cdcedb01ff60ccb825af616 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Mon, 22 Sep 2025 11:28:32 +0530 Subject: [PATCH] PythonTfe-VariableSets --- examples/variable_sets_example.py | 506 ++++++++++++ src/tfe/client.py | 3 + src/tfe/models/registry_module_types.py | 8 +- src/tfe/resources/variable_sets.py | 969 +++++++++++++++++++++++ src/tfe/types.py | 162 +++- tests/units/test_variable_sets.py | 986 ++++++++++++++++++++++++ 6 files changed, 2625 insertions(+), 9 deletions(-) create mode 100644 examples/variable_sets_example.py create mode 100644 src/tfe/resources/variable_sets.py create mode 100644 tests/units/test_variable_sets.py diff --git a/examples/variable_sets_example.py b/examples/variable_sets_example.py new file mode 100644 index 0000000..df62102 --- /dev/null +++ b/examples/variable_sets_example.py @@ -0,0 +1,506 @@ +"""Example demonstrating Variable Set operations with the TFE Python SDK. + +This example shows how to: +1. Create a variable set +2. Create variables in the set +3. Apply the set to workspaces/projects +4. Update variables and sets +5. Clean up resources + +Make sure to set the following environment variables: +- TFE_TOKEN: Your Terraform Cloud/Enterprise API token +- TFE_ADDRESS: Your Terraform Cloud/Enterprise URL (optional, defaults to https://app.terraform.io) +- TFE_ORG: Your organization name +""" + +import os + +from tfe import TFEClient, TFEConfig +from tfe.types import ( + CategoryType, + Parent, + Project, + VariableSetApplyToProjectsOptions, + VariableSetApplyToWorkspacesOptions, + VariableSetCreateOptions, + VariableSetIncludeOpt, + VariableSetListOptions, + VariableSetRemoveFromProjectsOptions, + VariableSetRemoveFromWorkspacesOptions, + VariableSetUpdateOptions, + VariableSetVariableCreateOptions, + VariableSetVariableListOptions, + VariableSetVariableUpdateOptions, + Workspace, +) + + +def variable_set_example(): + """Demonstrate Variable Set operations.""" + + # Initialize client + token = os.getenv("TFE_TOKEN") + address = os.getenv("TFE_ADDRESS", "https://app.terraform.io") + org_name = os.getenv("TFE_ORG") + + if not token or not org_name: + print("Please set TFE_TOKEN and TFE_ORG environment variables") + return + + config = TFEConfig(token=token, address=address) + client = TFEClient(config=config) + + # Variable set and variable IDs for cleanup + created_variable_set_id = None + created_variable_ids = [] + + try: + print("=== Variable Set Operations Example ===\n") + + # 1. List existing variable sets + print("1. Listing existing variable sets...") + list_options = VariableSetListOptions( + page_size=10, include=[VariableSetIncludeOpt.WORKSPACES] + ) + variable_sets = client.variable_sets.list(org_name, list_options) + print(f"Found {len(variable_sets)} existing variable sets") + + for vs in variable_sets[:3]: # Show first 3 + print(f" - {vs.name} (ID: {vs.id}, Global: {vs.global_})") + print() + + # 2. Create a new variable set + print("2. Creating a new variable set...") + create_options = VariableSetCreateOptions.model_validate( + { + "name": "python-sdk-example-varset", + "description": "Example variable set created with Python SDK", + "global": False, # Not global, will apply to specific workspaces/projects + "priority": True, # High priority + } + ) + + new_variable_set = client.variable_sets.create(org_name, create_options) + created_variable_set_id = new_variable_set.id + print( + f"Created variable set: {new_variable_set.name} (ID: {new_variable_set.id})" + ) + print(f" Description: {new_variable_set.description}") + print(f" Global: {new_variable_set.global_}") + print(f" Priority: {new_variable_set.priority}") + print() + + # 3. Create variables in the variable set + print("3. Creating variables in the variable set...") + + # Create a Terraform variable + tf_var_options = VariableSetVariableCreateOptions( + key="environment", + value="production", + description="Environment name", + category=CategoryType.TERRAFORM, + hcl=False, + sensitive=False, + ) + + tf_variable = client.variable_set_variables.create( + created_variable_set_id, tf_var_options + ) + created_variable_ids.append(tf_variable.id) + print(f"Created Terraform variable: {tf_variable.key} = {tf_variable.value}") + + # Create an environment variable + env_var_options = VariableSetVariableCreateOptions( + key="DATABASE_URL", + value="postgres://prod-db:5432/myapp", + description="Production database connection string", + category=CategoryType.ENV, + hcl=False, + sensitive=True, # Mark as sensitive + ) + + env_variable = client.variable_set_variables.create( + created_variable_set_id, env_var_options + ) + created_variable_ids.append(env_variable.id) + print(f"Created environment variable: {env_variable.key} (sensitive)") + + # Create an HCL variable + hcl_var_options = VariableSetVariableCreateOptions( + key="instance_config", + value='{"type": "t3.medium", "count": 2}', + description="Instance configuration", + category=CategoryType.TERRAFORM, + hcl=True, # HCL formatted + sensitive=False, + ) + + hcl_variable = client.variable_set_variables.create( + created_variable_set_id, hcl_var_options + ) + created_variable_ids.append(hcl_variable.id) + print(f"Created HCL variable: {hcl_variable.key} (HCL format)") + print() + + # 4. List variables in the variable set + print("4. Listing variables in the variable set...") + var_list_options = VariableSetVariableListOptions(page_size=50) + variables = client.variable_set_variables.list( + created_variable_set_id, var_list_options + ) + print(f"Found {len(variables)} variables in the set:") + + for var in variables: + sensitive_note = " (sensitive)" if var.sensitive else "" + hcl_note = " (HCL)" if var.hcl else "" + print(f" - {var.key}: {var.category.value}{sensitive_note}{hcl_note}") + print(f" Description: {var.description}") + print() + + # 5. Update a variable + print("5. Updating a variable...") + update_var_options = VariableSetVariableUpdateOptions( + key="environment", + value="staging", + description="Updated to staging environment", + ) + + updated_variable = client.variable_set_variables.update( + created_variable_set_id, tf_variable.id, update_var_options + ) + print(f"Updated variable: {updated_variable.key} = {updated_variable.value}") + print(f" New description: {updated_variable.description}") + print() + + # 6. Update the variable set itself + print("6. Updating the variable set...") + update_set_options = VariableSetUpdateOptions( + name="python-sdk-updated-varset", + description="Updated variable set description", + priority=False, # Change priority + ) + + updated_variable_set = client.variable_sets.update( + created_variable_set_id, update_set_options + ) + print(f"Updated variable set: {updated_variable_set.name}") + print(f" New description: {updated_variable_set.description}") + print(f" Priority: {updated_variable_set.priority}") + print() + + # 7. Example: Apply to workspaces (if any exist) + print("7. Workspace operations example...") + try: + # List some workspaces first + from tfe.types import WorkspaceListOptions + + workspace_options = WorkspaceListOptions(page_size=5) + workspaces = list( + client.workspaces.list(org_name, options=workspace_options) + ) + if workspaces: + # Apply to first workspace as example + first_workspace = workspaces[0] + print(f"Applying variable set to workspace: {first_workspace.name}") + + apply_ws_options = VariableSetApplyToWorkspacesOptions( + workspaces=[Workspace(id=first_workspace.id)] + ) + client.variable_sets.apply_to_workspaces( + created_variable_set_id, apply_ws_options + ) + print("Successfully applied to workspace") + + # List variable sets for this workspace + workspace_varsets = client.variable_sets.list_for_workspace( + first_workspace.id + ) + print(f"Workspace now has {len(workspace_varsets)} variable sets") + + # Remove from workspace + remove_ws_options = VariableSetRemoveFromWorkspacesOptions( + workspaces=[Workspace(id=first_workspace.id)] + ) + client.variable_sets.remove_from_workspaces( + created_variable_set_id, remove_ws_options + ) + print("Successfully removed from workspace") + else: + print("No workspaces found to demonstrate workspace operations") + except Exception as e: + print(f"Workspace operations example failed: {e}") + print() + + # 8. Example: Apply to projects (if any exist) + print("8. Project operations example...") + try: + # List projects + projects = list(client.projects.list(org_name)) + if projects: + # Apply to first project as example + first_project = projects[0] + print(f"Applying variable set to project: {first_project.name}") + + apply_proj_options = VariableSetApplyToProjectsOptions( + projects=[Project(id=first_project.id)] + ) + client.variable_sets.apply_to_projects( + created_variable_set_id, apply_proj_options + ) + print("Successfully applied to project") + + # List variable sets for this project + project_varsets = client.variable_sets.list_for_project( + first_project.id + ) + print(f"Project now has {len(project_varsets)} variable sets") + + # Remove from project + remove_proj_options = VariableSetRemoveFromProjectsOptions( + projects=[Project(id=first_project.id)] + ) + client.variable_sets.remove_from_projects( + created_variable_set_id, remove_proj_options + ) + print("Successfully removed from project") + else: + print("No projects found to demonstrate project operations") + except Exception as e: + print(f"Project operations example failed: {e}") + print() + + # 9. Read the variable set with includes + print("9. Reading variable set with includes...") + from tfe.types import VariableSetReadOptions + + read_options = VariableSetReadOptions( + include=[VariableSetIncludeOpt.VARS, VariableSetIncludeOpt.WORKSPACES] + ) + + detailed_varset = client.variable_sets.read( + created_variable_set_id, read_options + ) + print(f"Variable set: {detailed_varset.name}") + print(f" Variables count: {len(detailed_varset.vars or [])}") + print(f" Workspaces count: {len(detailed_varset.workspaces or [])}") + print() + + print("=== Variable Set Operations Completed Successfully ===") + + except Exception as e: + print(f"Error during example execution: {e}") + raise + + finally: + # Cleanup: Delete created resources + print("\n=== Cleanup ===") + + if created_variable_ids and created_variable_set_id: + print("Cleaning up created variables...") + for var_id in created_variable_ids: + try: + client.variable_set_variables.delete( + created_variable_set_id, var_id + ) + print(f"Deleted variable: {var_id}") + except Exception as e: + print(f"Failed to delete variable {var_id}: {e}") + + if created_variable_set_id: + print("Cleaning up created variable set...") + try: + client.variable_sets.delete(created_variable_set_id) + print(f"Deleted variable set: {created_variable_set_id}") + except Exception as e: + print(f"Failed to delete variable set {created_variable_set_id}: {e}") + + print("Cleanup completed") + + +def global_variable_set_example(): + """Example of creating and managing a global variable set.""" + + token = os.getenv("TFE_TOKEN") + address = os.getenv("TFE_ADDRESS", "https://app.terraform.io") + org_name = os.getenv("TFE_ORG") + + if not token or not org_name: + print("Please set TFE_TOKEN and TFE_ORG environment variables") + return + + config = TFEConfig(token=token, address=address) + client = TFEClient(config=config) + created_variable_set_id = None + + try: + print("\n=== Global Variable Set Example ===\n") + + # Create a global variable set + print("Creating a global variable set...") + global_create_options = VariableSetCreateOptions.model_validate( + { + "name": "python-sdk-global-varset", + "description": "Global variable set for common settings", + "global": True, # Make it global + "priority": False, + } + ) + + global_varset = client.variable_sets.create(org_name, global_create_options) + created_variable_set_id = global_varset.id + print(f"Created global variable set: {global_varset.name}") + print(f" Global: {global_varset.global_}") + print(f" Priority: {global_varset.priority}") + + # Add some common variables + print("\nAdding common variables...") + + # Common Terraform variables + common_vars = [ + { + "key": "default_tags", + "value": '{"Environment": "shared", "ManagedBy": "terraform"}', + "description": "Default tags for all resources", + "category": CategoryType.TERRAFORM, + "hcl": True, + }, + { + "key": "TERRAFORM_VERSION", + "value": "1.5.0", + "description": "Terraform version requirement", + "category": CategoryType.ENV, + "hcl": False, + }, + ] + + for var_config in common_vars: + var_options = VariableSetVariableCreateOptions(**var_config) + variable = client.variable_set_variables.create( + created_variable_set_id, var_options + ) + print(f" Added {variable.category.value} variable: {variable.key}") + + print(f"\nGlobal variable set is now available to all workspaces in {org_name}") + + except Exception as e: + print(f"Error in global variable set example: {e}") + + finally: + # Cleanup + if created_variable_set_id: + try: + print("\nCleaning up global variable set...") + client.variable_sets.delete(created_variable_set_id) + print("Global variable set deleted") + except Exception as e: + print(f"Failed to delete global variable set: {e}") + + +def project_scoped_variable_set_example(): + """Example of creating a project-scoped variable set.""" + + token = os.getenv("TFE_TOKEN") + address = os.getenv("TFE_ADDRESS", "https://app.terraform.io") + org_name = os.getenv("TFE_ORG") + + if not token or not org_name: + print("Please set TFE_TOKEN and TFE_ORG environment variables") + return + + config = TFEConfig(token=token, address=address) + client = TFEClient(config=config) + created_variable_set_id = None + + try: + print("\n=== Project-Scoped Variable Set Example ===\n") + + # First, get a project to scope to + projects = list(client.projects.list(org_name)) + if not projects: + print( + "No projects found. Creating a project-scoped variable set requires an existing project." + ) + return + + target_project = projects[0] + print(f"Using project: {target_project.name} (ID: {target_project.id})") + + # Create a project-scoped variable set + print("Creating a project-scoped variable set...") + parent = Parent(project=Project(id=target_project.id)) + + project_create_options = VariableSetCreateOptions.model_validate( + { + "name": "python-sdk-project-varset", + "description": f"Project-specific variables for {target_project.name}", + "global": False, # Not global + "parent": parent.model_dump(), # Scope to specific project + } + ) + + project_varset = client.variable_sets.create(org_name, project_create_options) + created_variable_set_id = project_varset.id + print(f"Created project-scoped variable set: {project_varset.name}") + + # Add project-specific variables + project_vars = [ + { + "key": "PROJECT_NAME", + "value": target_project.name, + "description": "Project name", + "category": CategoryType.ENV, + "hcl": False, + }, + { + "key": "project_config", + "value": f'{{"name": "{target_project.name}", "id": "{target_project.id}"}}', + "description": "Project configuration", + "category": CategoryType.TERRAFORM, + "hcl": True, + }, + ] + + for var_config in project_vars: + var_options = VariableSetVariableCreateOptions(**var_config) + variable = client.variable_set_variables.create( + created_variable_set_id, var_options + ) + print(f" Added variable: {variable.key}") + + print( + f"\nProject-scoped variable set is available to workspaces in project: {target_project.name}" + ) + + except Exception as e: + print(f"Error in project-scoped variable set example: {e}") + + finally: + # Cleanup + if created_variable_set_id: + try: + print("\nCleaning up project-scoped variable set...") + client.variable_sets.delete(created_variable_set_id) + print("Project-scoped variable set deleted") + except Exception as e: + print(f"Failed to delete project-scoped variable set: {e}") + + +if __name__ == "__main__": + print("TFE Python SDK - Variable Set Examples") + print("=" * 50) + + try: + # Run the main example + variable_set_example() + + # Run additional examples + global_variable_set_example() + project_scoped_variable_set_example() + + except KeyboardInterrupt: + print("\nExample interrupted by user") + except Exception as e: + print(f"\nExample failed with error: {e}") + import traceback + + traceback.print_exc() diff --git a/src/tfe/client.py b/src/tfe/client.py index 1bea139..d316a05 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -10,6 +10,7 @@ from .resources.state_version_outputs import StateVersionOutputs from .resources.state_versions import StateVersions from .resources.variable import Variables +from .resources.variable_sets import VariableSets, VariableSetVariables from .resources.workspaces import Workspaces @@ -33,6 +34,8 @@ def __init__(self, config: TFEConfig | None = None): self.organizations = Organizations(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) + self.variable_sets = VariableSets(self._transport) + self.variable_set_variables = VariableSetVariables(self._transport) self.workspaces = Workspaces(self._transport) self.registry_modules = RegistryModules(self._transport) diff --git a/src/tfe/models/registry_module_types.py b/src/tfe/models/registry_module_types.py index fd4a721..3d20a88 100644 --- a/src/tfe/models/registry_module_types.py +++ b/src/tfe/models/registry_module_types.py @@ -5,7 +5,7 @@ from enum import Enum from typing import Any -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field # Load the main types.py file to get Organization and other main types # Path: from /src/tfe/types/registry_module_types.py to /src/tfe/types.py @@ -298,8 +298,7 @@ class RegistryModuleVCSRepoOptions(BaseModel): source_directory: str | None = Field(alias="source-directory", default=None) tag_prefix: str | None = Field(alias="tag-prefix", default=None) - class Config: - allow_population_by_field_name = True + model_config = ConfigDict(populate_by_name=True) class RegistryModuleVCSRepoUpdateOptions(BaseModel): @@ -333,8 +332,7 @@ class RegistryModuleCreateWithVCSConnectionOptions(BaseModel): default=None, description="Namespace for public modules" ) - class Config: - allow_population_by_field_name = True + model_config = ConfigDict(populate_by_name=True) class RegistryModuleUpdateOptions(BaseModel): diff --git a/src/tfe/resources/variable_sets.py b/src/tfe/resources/variable_sets.py new file mode 100644 index 0000000..bc0e0db --- /dev/null +++ b/src/tfe/resources/variable_sets.py @@ -0,0 +1,969 @@ +"""Variable Set resource implementation for the Python TFE SDK.""" + +import builtins +from typing import Any + +from tfe._http import HTTPTransport +from tfe.resources._base import _Service +from tfe.types import ( + VariableSet, + VariableSetApplyToProjectsOptions, + VariableSetApplyToWorkspacesOptions, + VariableSetCreateOptions, + VariableSetIncludeOpt, + VariableSetListOptions, + VariableSetReadOptions, + VariableSetRemoveFromProjectsOptions, + VariableSetRemoveFromWorkspacesOptions, + VariableSetUpdateOptions, + VariableSetUpdateWorkspacesOptions, + VariableSetVariable, + VariableSetVariableCreateOptions, + VariableSetVariableListOptions, + VariableSetVariableUpdateOptions, +) + + +class VariableSets(_Service): + """ + Variable Sets resource for managing Terraform Cloud/Enterprise Variable Sets. + + Variable Sets provide a way to define and manage collections of variables + that can be applied to multiple workspaces or projects, supporting both + global and scoped variable management. + + API Documentation: + https://developer.hashicorp.com/terraform/cloud-docs/api-docs/variable-sets + """ + + def __init__(self, transport: HTTPTransport): + """Initialize the Variable Sets resource. + + Args: + transport: HTTP transport instance for API communication + """ + super().__init__(transport) + + def list( + self, + organization: str, + options: VariableSetListOptions | None = None, + ) -> list[VariableSet]: + """List all variable sets within an organization. + + Args: + organization: Organization name + options: Optional parameters for filtering and pagination + + Returns: + List of VariableSet objects + + Raises: + ValueError: If organization name is invalid + TFEError: If API request fails + """ + if not organization or not isinstance(organization, str): + raise ValueError("Organization name is required and must be a string") + + path = f"/api/v2/organizations/{organization}/varsets" + params: dict[str, str] = {} + + if options: + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + if options.query: + params["q"] = options.query + if options.include: + params["include"] = ",".join([opt.value for opt in options.include]) + + response = self.t.request("GET", path, params=params) + data = response.json() + + return self._parse_variable_sets_response(data) + + def list_for_workspace( + self, + workspace_id: str, + options: VariableSetListOptions | None = None, + ) -> builtins.list[VariableSet]: + """List variable sets associated with a workspace. + + Args: + workspace_id: Workspace ID + options: Optional parameters for filtering and pagination + + Returns: + List of VariableSet objects associated with the workspace + + Raises: + ValueError: If workspace_id is invalid + TFEError: If API request fails + """ + if not workspace_id or not isinstance(workspace_id, str): + raise ValueError("Workspace ID is required and must be a string") + + path = f"/api/v2/workspaces/{workspace_id}/varsets" + params: dict[str, str] = {} + + if options: + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + if options.query: + params["q"] = options.query + if options.include: + params["include"] = ",".join([opt.value for opt in options.include]) + + response = self.t.request("GET", path, params=params) + data = response.json() + + return self._parse_variable_sets_response(data) + + def list_for_project( + self, + project_id: str, + options: VariableSetListOptions | None = None, + ) -> builtins.list[VariableSet]: + """List variable sets associated with a project. + + Args: + project_id: Project ID + options: Optional parameters for filtering and pagination + + Returns: + List of VariableSet objects associated with the project + + Raises: + ValueError: If project_id is invalid + TFEError: If API request fails + """ + if not project_id or not isinstance(project_id, str): + raise ValueError("Project ID is required and must be a string") + + path = f"/api/v2/projects/{project_id}/varsets" + params: dict[str, str] = {} + + if options: + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + if options.query: + params["q"] = options.query + if options.include: + params["include"] = ",".join([opt.value for opt in options.include]) + + response = self.t.request("GET", path, params=params) + data = response.json() + + return self._parse_variable_sets_response(data) + + def create( + self, + organization: str, + options: VariableSetCreateOptions, + ) -> VariableSet: + """Create a new variable set. + + Args: + organization: Organization name + options: Variable set creation options + + Returns: + Created VariableSet object + + Raises: + ValueError: If organization name or options are invalid + TFEError: If API request fails + """ + if not organization or not isinstance(organization, str): + raise ValueError("Organization name is required and must be a string") + + if not options or not isinstance(options, VariableSetCreateOptions): + raise ValueError( + "Options are required and must be VariableSetCreateOptions" + ) + + if not options.name: + raise ValueError("Variable set name is required") + + path = f"/api/v2/organizations/{organization}/varsets" + + payload: dict[str, Any] = { + "data": { + "type": "varsets", + "attributes": { + "name": options.name, + "global": options.global_, + }, + } + } + + attributes = payload["data"]["attributes"] + if options.description is not None: + attributes["description"] = options.description + + if options.priority is not None: + attributes["priority"] = options.priority + + # Handle parent relationship + if options.parent: + relationships: dict[str, Any] = {} + if options.parent.project and options.parent.project.id: + relationships["parent"] = { + "data": { + "type": "projects", + "id": options.parent.project.id, + } + } + elif options.parent.organization and options.parent.organization.id: + relationships["parent"] = { + "data": { + "type": "organizations", + "id": options.parent.organization.id, + } + } + if relationships: + payload["data"]["relationships"] = relationships + + response = self.t.request("POST", path, json_body=payload) + data = response.json() + + return self._parse_variable_set(data["data"]) + + def read( + self, + variable_set_id: str, + options: VariableSetReadOptions | None = None, + ) -> VariableSet: + """Read a variable set by its ID. + + Args: + variable_set_id: Variable set ID + options: Optional parameters for including related resources + + Returns: + VariableSet object + + Raises: + ValueError: If variable_set_id is invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + path = f"/api/v2/varsets/{variable_set_id}" + params: dict[str, str] = {} + + if options and options.include: + params["include"] = ",".join([opt.value for opt in options.include]) + + response = self.t.request("GET", path, params=params) + data = response.json() + + return self._parse_variable_set(data["data"]) + + def update( + self, + variable_set_id: str, + options: VariableSetUpdateOptions, + ) -> VariableSet: + """Update an existing variable set. + + Args: + variable_set_id: Variable set ID + options: Variable set update options + + Returns: + Updated VariableSet object + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance(options, VariableSetUpdateOptions): + raise ValueError( + "Options are required and must be VariableSetUpdateOptions" + ) + + path = f"/api/v2/varsets/{variable_set_id}" + + payload: dict[str, Any] = { + "data": { + "type": "varsets", + "id": variable_set_id, + "attributes": {}, + } + } + + attributes = payload["data"]["attributes"] + if options.name is not None: + attributes["name"] = options.name + + if options.description is not None: + attributes["description"] = options.description + + if options.global_ is not None: + attributes["global"] = options.global_ + + if options.priority is not None: + attributes["priority"] = options.priority + + response = self.t.request("PATCH", path, json_body=payload) + data = response.json() + + return self._parse_variable_set(data["data"]) + + def delete(self, variable_set_id: str) -> None: + """Delete a variable set by its ID. + + Args: + variable_set_id: Variable set ID + + Raises: + ValueError: If variable_set_id is invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + path = f"/api/v2/varsets/{variable_set_id}" + self.t.request("DELETE", path) + + def apply_to_workspaces( + self, + variable_set_id: str, + options: VariableSetApplyToWorkspacesOptions, + ) -> None: + """Apply variable set to workspaces. + + Note: This method will return an error if the variable set has global = true. + + Args: + variable_set_id: Variable set ID + options: Options specifying workspaces to apply to + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance(options, VariableSetApplyToWorkspacesOptions): + raise ValueError( + "Options are required and must be VariableSetApplyToWorkspacesOptions" + ) + + if not options.workspaces: + raise ValueError("At least one workspace is required") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/workspaces" + + # Build workspace relationships payload + workspace_data = [] + for workspace in options.workspaces: + if not workspace.id: + raise ValueError("All workspaces must have valid IDs") + workspace_data.append( + { + "type": "workspaces", + "id": workspace.id, + } + ) + + payload = {"data": workspace_data} + + self.t.request("POST", path, json_body=payload) + + def remove_from_workspaces( + self, + variable_set_id: str, + options: VariableSetRemoveFromWorkspacesOptions, + ) -> None: + """Remove variable set from workspaces. + + Note: This method will return an error if the variable set has global = true. + + Args: + variable_set_id: Variable set ID + options: Options specifying workspaces to remove from + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance( + options, VariableSetRemoveFromWorkspacesOptions + ): + raise ValueError( + "Options are required and must be VariableSetRemoveFromWorkspacesOptions" + ) + + if not options.workspaces: + raise ValueError("At least one workspace is required") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/workspaces" + + # Build workspace relationships payload + workspace_data = [] + for workspace in options.workspaces: + if not workspace.id: + raise ValueError("All workspaces must have valid IDs") + workspace_data.append( + { + "type": "workspaces", + "id": workspace.id, + } + ) + + payload = {"data": workspace_data} + + self.t.request("DELETE", path, json_body=payload) + + def apply_to_projects( + self, + variable_set_id: str, + options: VariableSetApplyToProjectsOptions, + ) -> None: + """Apply variable set to projects. + + This method will return an error if the variable set has global = true. + + Args: + variable_set_id: Variable set ID + options: Options specifying projects to apply to + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance(options, VariableSetApplyToProjectsOptions): + raise ValueError( + "Options are required and must be VariableSetApplyToProjectsOptions" + ) + + if not options.projects: + raise ValueError("At least one project is required") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/projects" + + # Build project relationships payload + project_data = [] + for project in options.projects: + if not project.id: + raise ValueError("All projects must have valid IDs") + project_data.append( + { + "type": "projects", + "id": project.id, + } + ) + + payload = {"data": project_data} + + self.t.request("POST", path, json_body=payload) + + def remove_from_projects( + self, + variable_set_id: str, + options: VariableSetRemoveFromProjectsOptions, + ) -> None: + """Remove variable set from projects. + + This method will return an error if the variable set has global = true. + + Args: + variable_set_id: Variable set ID + options: Options specifying projects to remove from + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance(options, VariableSetRemoveFromProjectsOptions): + raise ValueError( + "Options are required and must be VariableSetRemoveFromProjectsOptions" + ) + + if not options.projects: + raise ValueError("At least one project is required") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/projects" + + # Build project relationships payload + project_data = [] + for project in options.projects: + if not project.id: + raise ValueError("All projects must have valid IDs") + project_data.append( + { + "type": "projects", + "id": project.id, + } + ) + + payload = {"data": project_data} + + self.t.request("DELETE", path, json_body=payload) + + def update_workspaces( + self, + variable_set_id: str, + options: VariableSetUpdateWorkspacesOptions, + ) -> VariableSet: + """Update variable set to be applied to only the workspaces in the supplied list. + + Args: + variable_set_id: Variable set ID + options: Options specifying workspaces to apply to + + Returns: + Updated VariableSet object + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance(options, VariableSetUpdateWorkspacesOptions): + raise ValueError( + "Options are required and must be VariableSetUpdateWorkspacesOptions" + ) + + # Force inclusion of workspaces as that is the primary data + path = f"/api/v2/varsets/{variable_set_id}" + params: dict[str, str] = {"include": VariableSetIncludeOpt.WORKSPACES.value} + + payload = { + "data": { + "type": "varsets", + "id": variable_set_id, + "attributes": { + "global": False, # Force global to false when applying to workspaces + }, + "relationships": { + "workspaces": { + "data": [ + {"type": "workspaces", "id": ws.id} + for ws in options.workspaces + if ws.id + ] + } + }, + } + } + + response = self.t.request("PATCH", path, json_body=payload, params=params) + data = response.json() + + return self._parse_variable_set(data["data"]) + + def _parse_variable_sets_response( + self, data: dict[str, Any] + ) -> builtins.list[VariableSet]: + """Parse API response containing multiple variable sets. + + Args: + data: Raw API response data + + Returns: + List of VariableSet objects + """ + variable_sets = [] + for item in data.get("data", []): + variable_sets.append(self._parse_variable_set(item)) + return variable_sets + + def _parse_variable_set(self, data: dict[str, Any]) -> VariableSet: + """Parse a single variable set from API response data. + + Args: + data: Raw API response data for a single variable set + + Returns: + VariableSet object + """ + attrs = data.get("attributes", {}) + relationships = data.get("relationships", {}) + + # Build the data dict for Pydantic model + parsed_data = { + "id": data.get("id"), + "name": attrs.get("name", ""), + "description": attrs.get("description"), + "global": attrs.get( + "global", False + ), # Use "global" not "global_" for API data + "priority": attrs.get("priority"), + "created_at": attrs.get("created-at"), + "updated_at": attrs.get("updated-at"), + } + + # Build workspaces list - simplified to just contain minimal data + workspaces = [] + if "workspaces" in relationships: + ws_data = relationships["workspaces"].get("data", []) + if isinstance(ws_data, list): + for ws in ws_data: + if "id" in ws: + workspaces.append( + { + "id": ws["id"], + "name": f"workspace-{ws['id']}", # Placeholder name + "organization": "placeholder-org", # Placeholder organization + } + ) + parsed_data["workspaces"] = workspaces + + # Build projects list - simplified to just contain minimal data + projects = [] + if "projects" in relationships: + proj_data = relationships["projects"].get("data", []) + if isinstance(proj_data, list): + for proj in proj_data: + if "id" in proj: + projects.append( + { + "id": proj["id"], + "name": f"project-{proj['id']}", # Placeholder name + "organization": "placeholder-org", # Placeholder organization + } + ) + parsed_data["projects"] = projects + + # Build variables list - simplified to just contain minimal data + variables = [] + if "vars" in relationships: + vars_data = relationships["vars"].get("data", []) + if isinstance(vars_data, list): + for var in vars_data: + if "id" in var: + variables.append( + { + "id": var["id"], + "key": f"var-{var['id']}", # Placeholder key + "category": "terraform", # Default category + "variable_set": { + "id": data.get("id"), + "name": attrs.get("name", ""), + "global": attrs.get("global", False), + }, + } + ) + parsed_data["vars"] = variables + + # Handle parent relationship + parent = None + if "parent" in relationships: + parent_data = relationships["parent"].get("data") + if parent_data: + if parent_data.get("type") == "projects": + parent = { + "project": { + "id": parent_data["id"], + "name": f"project-{parent_data['id']}", + "organization": "placeholder-org", + } + } + elif parent_data.get("type") == "organizations": + parent = {"organization": {"id": parent_data["id"]}} + parsed_data["parent"] = parent + + # Use Pydantic model validation to handle aliases properly + return VariableSet.model_validate(parsed_data) + + +class VariableSetVariables(_Service): + """ + Variable Set Variables resource for managing variables within Variable Sets. + + This resource handles CRUD operations for individual variables within + Variable Sets, providing scoped variable management capabilities. + + API Documentation: + https://developer.hashicorp.com/terraform/cloud-docs/api-docs/variable-sets#variable-relationships + """ + + def __init__(self, transport: HTTPTransport): + """Initialize the Variable Set Variables resource. + + Args: + transport: HTTP transport instance for API communication + """ + super().__init__(transport) + + def list( + self, + variable_set_id: str, + options: VariableSetVariableListOptions | None = None, + ) -> list[VariableSetVariable]: + """List all variables in a variable set. + + Args: + variable_set_id: Variable set ID + options: Optional parameters for pagination + + Returns: + List of VariableSetVariable objects + + Raises: + ValueError: If variable_set_id is invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/vars" + params: dict[str, str] = {} + + if options: + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + + response = self.t.request("GET", path, params=params) + data = response.json() + + variables = [] + for item in data.get("data", []): + variables.append(self._parse_variable_set_variable(item)) + + return variables + + def create( + self, + variable_set_id: str, + options: VariableSetVariableCreateOptions, + ) -> VariableSetVariable: + """Create a new variable within a variable set. + + Args: + variable_set_id: Variable set ID + options: Variable creation options + + Returns: + Created VariableSetVariable object + + Raises: + ValueError: If variable_set_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not options or not isinstance(options, VariableSetVariableCreateOptions): + raise ValueError( + "Options are required and must be VariableSetVariableCreateOptions" + ) + + if not options.key: + raise ValueError("Variable key is required") + + if not options.category: + raise ValueError("Variable category is required") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/vars" + + payload: dict[str, Any] = { + "data": { + "type": "vars", + "attributes": { + "key": options.key, + "category": options.category.value, + }, + } + } + + attributes = payload["data"]["attributes"] + if options.value is not None: + attributes["value"] = options.value + + if options.description is not None: + attributes["description"] = options.description + + if options.hcl is not None: + attributes["hcl"] = options.hcl + + if options.sensitive is not None: + attributes["sensitive"] = options.sensitive + + response = self.t.request("POST", path, json_body=payload) + data = response.json() + + return self._parse_variable_set_variable(data["data"]) + + def read( + self, + variable_set_id: str, + variable_id: str, + ) -> VariableSetVariable: + """Read a variable by its ID. + + Args: + variable_set_id: Variable set ID + variable_id: Variable ID + + Returns: + VariableSetVariable object + + Raises: + ValueError: If variable_set_id or variable_id are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not variable_id or not isinstance(variable_id, str): + raise ValueError("Variable ID is required and must be a string") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/vars/{variable_id}" + + response = self.t.request("GET", path) + data = response.json() + + return self._parse_variable_set_variable(data["data"]) + + def update( + self, + variable_set_id: str, + variable_id: str, + options: VariableSetVariableUpdateOptions, + ) -> VariableSetVariable: + """Update an existing variable. + + Args: + variable_set_id: Variable set ID + variable_id: Variable ID + options: Variable update options + + Returns: + Updated VariableSetVariable object + + Raises: + ValueError: If variable_set_id, variable_id or options are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not variable_id or not isinstance(variable_id, str): + raise ValueError("Variable ID is required and must be a string") + + if not options or not isinstance(options, VariableSetVariableUpdateOptions): + raise ValueError( + "Options are required and must be VariableSetVariableUpdateOptions" + ) + + path = f"/api/v2/varsets/{variable_set_id}/relationships/vars/{variable_id}" + + payload: dict[str, Any] = { + "data": { + "type": "vars", + "id": variable_id, + "attributes": {}, + } + } + + attributes = payload["data"]["attributes"] + if options.key is not None: + attributes["key"] = options.key + + if options.value is not None: + attributes["value"] = options.value + + if options.description is not None: + attributes["description"] = options.description + + if options.hcl is not None: + attributes["hcl"] = options.hcl + + if options.sensitive is not None: + attributes["sensitive"] = options.sensitive + + response = self.t.request("PATCH", path, json_body=payload) + data = response.json() + + return self._parse_variable_set_variable(data["data"]) + + def delete( + self, + variable_set_id: str, + variable_id: str, + ) -> None: + """Delete a variable by its ID. + + Args: + variable_set_id: Variable set ID + variable_id: Variable ID + + Raises: + ValueError: If variable_set_id or variable_id are invalid + TFEError: If API request fails + """ + if not variable_set_id or not isinstance(variable_set_id, str): + raise ValueError("Variable set ID is required and must be a string") + + if not variable_id or not isinstance(variable_id, str): + raise ValueError("Variable ID is required and must be a string") + + path = f"/api/v2/varsets/{variable_set_id}/relationships/vars/{variable_id}" + + self.t.request("DELETE", path) + + def _parse_variable_set_variable(self, data: dict[str, Any]) -> VariableSetVariable: + """Parse a single variable set variable from API response data. + + Args: + data: Raw API response data for a single variable + + Returns: + VariableSetVariable object + """ + attrs = data.get("attributes", {}) + relationships = data.get("relationships", {}) + + # Build the data dict for Pydantic model + parsed_data = { + "id": data.get("id"), + "key": attrs.get("key", ""), + "value": attrs.get("value"), + "description": attrs.get("description"), + "category": attrs.get("category", "terraform"), + "hcl": attrs.get("hcl", False), + "sensitive": attrs.get("sensitive", False), + "version_id": attrs.get("version-id"), + } + + # Handle variable set relationship + variable_set = None + if "varset" in relationships: + vs_data = relationships["varset"].get("data") + if vs_data and "id" in vs_data: + variable_set = { + "id": vs_data["id"], + "name": f"varset-{vs_data['id']}", # Placeholder name + "global": False, # Placeholder global + } + parsed_data["variable_set"] = variable_set + + # Use Pydantic model validation + return VariableSetVariable.model_validate(parsed_data) diff --git a/src/tfe/types.py b/src/tfe/types.py index 3cfb372..bc115aa 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -106,9 +106,9 @@ class Project(BaseModel): """Project represents a Terraform Enterprise project""" id: str - name: str + name: str | None = None description: str = "" - organization: str + organization: str | None = None created_at: str = "" updated_at: str = "" workspace_count: int = 0 @@ -155,8 +155,8 @@ class ProjectAddTagBindingsOptions(BaseModel): class Workspace(BaseModel): id: str - name: str - organization: str + name: str | None = None + organization: str | None = None execution_mode: ExecutionMode | None = None project_id: str | None = None @@ -657,3 +657,157 @@ class WorkspaceAddTagBindingsOptions(BaseModel): """Options for adding tag bindings to a workspace.""" tag_bindings: list[TagBinding] = Field(default_factory=list) + + +# Variable Set related types + + +class VariableSetIncludeOpt(str, Enum): + """Include options for variable set operations.""" + + WORKSPACES = "workspaces" + PROJECTS = "projects" + VARS = "vars" + CURRENT_RUN = "current-run" + + +class Parent(BaseModel): + """Parent represents the variable set's parent (organizations and projects are supported).""" + + organization: Organization | None = None + project: Project | None = None + + +class VariableSet(BaseModel): + """Represents a Terraform Enterprise variable set.""" + + id: str | None = None + name: str | None = None + description: str | None = None + global_: bool | None = Field(default=None, alias="global") + priority: bool | None = None + created_at: datetime | None = None + updated_at: datetime | None = None + + # Relations + organization: Organization | None = None + workspaces: list[Workspace] = Field(default_factory=list) + projects: list[Project] = Field(default_factory=list) + vars: list[VariableSetVariable] = Field(default_factory=list) + parent: Parent | None = None + + +class VariableSetVariable(BaseModel): + """Represents a variable within a variable set.""" + + id: str | None = None + key: str + value: str | None = None + description: str | None = None + category: CategoryType + hcl: bool | None = None + sensitive: bool | None = None + version_id: str | None = None + + # Relations + variable_set: VariableSet | None = None + + +# Variable Set Options + + +class VariableSetListOptions(BaseModel): + """Options for listing variable sets.""" + + # Pagination options + page_number: int | None = None + page_size: int | None = None + include: list[VariableSetIncludeOpt] | None = None + query: str | None = None # Filter by name + + +class VariableSetCreateOptions(BaseModel): + """Options for creating a variable set.""" + + name: str + description: str | None = None + global_: bool = Field(alias="global") + priority: bool | None = None + parent: Parent | None = None + + +class VariableSetReadOptions(BaseModel): + """Options for reading a variable set.""" + + include: list[VariableSetIncludeOpt] | None = None + + +class VariableSetUpdateOptions(BaseModel): + """Options for updating a variable set.""" + + name: str | None = None + description: str | None = None + global_: bool | None = Field(alias="global", default=None) + priority: bool | None = None + + +class VariableSetApplyToWorkspacesOptions(BaseModel): + """Options for applying a variable set to workspaces.""" + + workspaces: list[Workspace] = Field(default_factory=list) + + +class VariableSetRemoveFromWorkspacesOptions(BaseModel): + """Options for removing a variable set from workspaces.""" + + workspaces: list[Workspace] = Field(default_factory=list) + + +class VariableSetApplyToProjectsOptions(BaseModel): + """Options for applying a variable set to projects.""" + + projects: list[Project] = Field(default_factory=list) + + +class VariableSetRemoveFromProjectsOptions(BaseModel): + """Options for removing a variable set from projects.""" + + projects: list[Project] = Field(default_factory=list) + + +class VariableSetUpdateWorkspacesOptions(BaseModel): + """Options for updating workspaces associated with a variable set.""" + + workspaces: list[Workspace] = Field(default_factory=list) + + +# Variable Set Variable Options + + +class VariableSetVariableListOptions(BaseModel): + """Options for listing variables in a variable set.""" + + # Pagination options + page_number: int | None = None + page_size: int | None = None + + +class VariableSetVariableCreateOptions(BaseModel): + """Options for creating a variable in a variable set.""" + + key: str + value: str | None = None + description: str | None = None + category: CategoryType + hcl: bool | None = None + sensitive: bool | None = None + + +class VariableSetVariableUpdateOptions(BaseModel): + """Options for updating a variable in a variable set.""" + + key: str | None = None + value: str | None = None + description: str | None = None + hcl: bool | None = None + sensitive: bool | None = None diff --git a/tests/units/test_variable_sets.py b/tests/units/test_variable_sets.py new file mode 100644 index 0000000..77a672d --- /dev/null +++ b/tests/units/test_variable_sets.py @@ -0,0 +1,986 @@ +"""Unit tests for Variable Set resources.""" + +from unittest.mock import Mock + +import pytest + +from tfe.resources.variable_sets import VariableSets, VariableSetVariables +from tfe.types import ( + CategoryType, + Parent, + Project, + VariableSet, + VariableSetApplyToProjectsOptions, + VariableSetApplyToWorkspacesOptions, + VariableSetCreateOptions, + VariableSetIncludeOpt, + VariableSetListOptions, + VariableSetReadOptions, + VariableSetRemoveFromProjectsOptions, + VariableSetRemoveFromWorkspacesOptions, + VariableSetUpdateOptions, + VariableSetUpdateWorkspacesOptions, + VariableSetVariable, + VariableSetVariableCreateOptions, + VariableSetVariableUpdateOptions, + Workspace, +) + + +class TestVariableSets: + """Test cases for Variable Sets resource.""" + + def setup_method(self): + """Setup method that runs before each test.""" + self.mock_transport = Mock() + self.variable_sets_service = VariableSets(self.mock_transport) + self.org_name = "test-org" + self.variable_set_id = "varset-test123" + self.workspace_id = "ws-test123" + self.project_id = "prj-test123" + + def test_variable_sets_service_init(self): + """Test Variable Sets service initialization.""" + assert self.variable_sets_service.t == self.mock_transport + + def test_list_variable_sets_success(self): + """Test successful listing of variable sets.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [ + { + "id": "varset-123", + "type": "varsets", + "attributes": { + "name": "test-varset", + "description": "Test variable set", + "global": False, + "priority": True, + "created-at": "2023-01-01T00:00:00.000Z", + "updated-at": "2023-01-01T00:00:00.000Z", + }, + "relationships": { + "workspaces": { + "data": [{"id": "ws-123", "type": "workspaces"}] + }, + "projects": {"data": [{"id": "prj-123", "type": "projects"}]}, + }, + } + ] + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.list(self.org_name) + + # Assertions + assert len(result) == 1 + assert isinstance(result[0], VariableSet) + assert result[0].id == "varset-123" + assert result[0].name == "test-varset" + assert result[0].description == "Test variable set" + assert result[0].global_ is False + assert result[0].priority is True + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/organizations/{self.org_name}/varsets", params={} + ) + + def test_list_variable_sets_with_options(self): + """Test listing variable sets with options.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = {"data": []} + self.mock_transport.request.return_value = mock_response + + # Create options + options = VariableSetListOptions( + page_number=2, + page_size=50, + query="test", + include=[VariableSetIncludeOpt.WORKSPACES, VariableSetIncludeOpt.PROJECTS], + ) + + # Call the method + result = self.variable_sets_service.list(self.org_name, options) + + # Verify the result + assert isinstance(result, list) + + # Verify API call with parameters + expected_params = { + "page[number]": "2", + "page[size]": "50", + "q": "test", + "include": "workspaces,projects", + } + self.mock_transport.request.assert_called_once_with( + "GET", + f"/api/v2/organizations/{self.org_name}/varsets", + params=expected_params, + ) + + def test_list_variable_sets_invalid_organization(self): + """Test listing variable sets with invalid organization.""" + with pytest.raises(ValueError, match="Organization name is required"): + self.variable_sets_service.list("") + + with pytest.raises(ValueError, match="Organization name is required"): + self.variable_sets_service.list(None) + + def test_list_for_workspace_success(self): + """Test successful listing of variable sets for workspace.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [ + { + "id": "varset-123", + "type": "varsets", + "attributes": { + "name": "workspace-varset", + "description": "Workspace variable set", + "global": False, + "priority": False, + }, + "relationships": {}, + } + ] + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.list_for_workspace(self.workspace_id) + + # Assertions + assert len(result) == 1 + assert result[0].name == "workspace-varset" + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/workspaces/{self.workspace_id}/varsets", params={} + ) + + def test_list_for_workspace_invalid_id(self): + """Test listing for workspace with invalid workspace ID.""" + with pytest.raises(ValueError, match="Workspace ID is required"): + self.variable_sets_service.list_for_workspace("") + + def test_list_for_project_success(self): + """Test successful listing of variable sets for project.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [ + { + "id": "varset-123", + "type": "varsets", + "attributes": { + "name": "project-varset", + "description": "Project variable set", + "global": False, + "priority": False, + }, + "relationships": {}, + } + ] + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.list_for_project(self.project_id) + + # Assertions + assert len(result) == 1 + assert result[0].name == "project-varset" + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/projects/{self.project_id}/varsets", params={} + ) + + def test_list_for_project_invalid_id(self): + """Test listing for project with invalid project ID.""" + with pytest.raises(ValueError, match="Project ID is required"): + self.variable_sets_service.list_for_project("") + + def test_create_variable_set_success(self): + """Test successful variable set creation.""" + # Prepare test data - using model_validate with alias + options = VariableSetCreateOptions.model_validate( + { + "name": "new-varset", + "description": "New variable set", + "global": False, # Use alias name + "priority": True, + } + ) + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "varset-new123", + "type": "varsets", + "attributes": { + "name": "new-varset", + "description": "New variable set", + "global": False, + "priority": True, + }, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.create(self.org_name, options) + + # Assertions + assert isinstance(result, VariableSet) + assert result.id == "varset-new123" + assert result.name == "new-varset" + assert result.description == "New variable set" + assert result.global_ is False + assert result.priority is True + + # Verify API call payload + expected_payload = { + "data": { + "type": "varsets", + "attributes": { + "name": "new-varset", + "description": "New variable set", + "global": False, + "priority": True, + }, + } + } + self.mock_transport.request.assert_called_once_with( + "POST", + f"/api/v2/organizations/{self.org_name}/varsets", + json_body=expected_payload, + ) + + def test_create_variable_set_with_parent_project(self): + """Test creating variable set with parent project.""" + # Prepare test data + project = Project(id="prj-parent123") + parent = Parent(project=project) + options = VariableSetCreateOptions.model_validate( + { + "name": "project-varset", + "description": "Project scoped variable set", + "global": False, # Use alias name + "parent": parent.model_dump(), + } + ) + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "varset-project123", + "type": "varsets", + "attributes": { + "name": "project-varset", + "description": "Project scoped variable set", + "global": False, + }, + "relationships": { + "parent": {"data": {"type": "projects", "id": "prj-parent123"}} + }, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.create(self.org_name, options) + + # Assertions + assert result.name == "project-varset" + + # Verify API call payload includes parent relationship + expected_payload = { + "data": { + "type": "varsets", + "attributes": { + "name": "project-varset", + "description": "Project scoped variable set", + "global": False, + }, + "relationships": { + "parent": { + "data": { + "type": "projects", + "id": "prj-parent123", + } + } + }, + } + } + self.mock_transport.request.assert_called_once_with( + "POST", + f"/api/v2/organizations/{self.org_name}/varsets", + json_body=expected_payload, + ) + + def test_create_variable_set_invalid_params(self): + """Test variable set creation with invalid parameters.""" + # Invalid organization + with pytest.raises(ValueError, match="Organization name is required"): + options = VariableSetCreateOptions.model_validate( + {"name": "test", "global": False} + ) + self.variable_sets_service.create("", options) + + # Invalid options + with pytest.raises(ValueError, match="Options are required"): + self.variable_sets_service.create(self.org_name, None) + + # Missing name + with pytest.raises(ValueError, match="Variable set name is required"): + options = VariableSetCreateOptions.model_validate( + {"name": "", "global": False} + ) + self.variable_sets_service.create(self.org_name, options) + + def test_read_variable_set_success(self): + """Test successful variable set read.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": self.variable_set_id, + "type": "varsets", + "attributes": { + "name": "read-varset", + "description": "Variable set for reading", + "global": True, + "priority": False, + }, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.read(self.variable_set_id) + + # Assertions + assert isinstance(result, VariableSet) + assert result.id == self.variable_set_id + assert result.name == "read-varset" + assert result.global_ is True + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/varsets/{self.variable_set_id}", params={} + ) + + def test_read_variable_set_with_include(self): + """Test reading variable set with include options.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": self.variable_set_id, + "type": "varsets", + "attributes": {"name": "test", "global": False}, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Create options + options = VariableSetReadOptions( + include=[VariableSetIncludeOpt.WORKSPACES, VariableSetIncludeOpt.VARS] + ) + + # Call the method + self.variable_sets_service.read(self.variable_set_id, options) + + # Verify API call with include parameters + expected_params = {"include": "workspaces,vars"} + self.mock_transport.request.assert_called_once_with( + "GET", f"/api/v2/varsets/{self.variable_set_id}", params=expected_params + ) + + def test_read_variable_set_invalid_id(self): + """Test reading variable set with invalid ID.""" + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variable_sets_service.read("") + + def test_update_variable_set_success(self): + """Test successful variable set update.""" + # Prepare test data + options = VariableSetUpdateOptions.model_validate( + { + "name": "updated-varset", + "description": "Updated variable set", + "global": True, # Use alias + "priority": False, + } + ) + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": self.variable_set_id, + "type": "varsets", + "attributes": { + "name": "updated-varset", + "description": "Updated variable set", + "global": True, + "priority": False, + }, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.update(self.variable_set_id, options) + + # Assertions + assert result.name == "updated-varset" + assert result.description == "Updated variable set" + assert result.global_ is True + + # Verify API call payload + expected_payload = { + "data": { + "type": "varsets", + "id": self.variable_set_id, + "attributes": { + "name": "updated-varset", + "description": "Updated variable set", + "global": True, + "priority": False, + }, + } + } + self.mock_transport.request.assert_called_once_with( + "PATCH", + f"/api/v2/varsets/{self.variable_set_id}", + json_body=expected_payload, + ) + + def test_update_variable_set_invalid_params(self): + """Test variable set update with invalid parameters.""" + options = VariableSetUpdateOptions(name="test") + + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variable_sets_service.update("", options) + + # Invalid options + with pytest.raises(ValueError, match="Options are required"): + self.variable_sets_service.update(self.variable_set_id, None) + + def test_delete_variable_set_success(self): + """Test successful variable set deletion.""" + # Call the method + self.variable_sets_service.delete(self.variable_set_id) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "DELETE", f"/api/v2/varsets/{self.variable_set_id}" + ) + + def test_delete_variable_set_invalid_id(self): + """Test variable set deletion with invalid ID.""" + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variable_sets_service.delete("") + + def test_apply_to_workspaces_success(self): + """Test successful application to workspaces.""" + # Prepare test data + workspaces = [Workspace(id="ws-1"), Workspace(id="ws-2")] + options = VariableSetApplyToWorkspacesOptions(workspaces=workspaces) + + # Call the method + self.variable_sets_service.apply_to_workspaces(self.variable_set_id, options) + + # Verify API call payload + expected_payload = { + "data": [ + {"type": "workspaces", "id": "ws-1"}, + {"type": "workspaces", "id": "ws-2"}, + ] + } + self.mock_transport.request.assert_called_once_with( + "POST", + f"/api/v2/varsets/{self.variable_set_id}/relationships/workspaces", + json_body=expected_payload, + ) + + def test_apply_to_workspaces_invalid_params(self): + """Test applying to workspaces with invalid parameters.""" + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variable_sets_service.apply_to_workspaces( + "", VariableSetApplyToWorkspacesOptions() + ) + + # Invalid options + with pytest.raises(ValueError, match="Options are required"): + self.variable_sets_service.apply_to_workspaces(self.variable_set_id, None) + + # Empty workspaces list + with pytest.raises(ValueError, match="At least one workspace is required"): + self.variable_sets_service.apply_to_workspaces( + self.variable_set_id, VariableSetApplyToWorkspacesOptions(workspaces=[]) + ) + + # Workspace without ID + workspaces = [Workspace(id="")] + options = VariableSetApplyToWorkspacesOptions(workspaces=workspaces) + with pytest.raises(ValueError, match="All workspaces must have valid IDs"): + self.variable_sets_service.apply_to_workspaces( + self.variable_set_id, options + ) + + def test_remove_from_workspaces_success(self): + """Test successful removal from workspaces.""" + # Prepare test data + workspaces = [Workspace(id="ws-1")] + options = VariableSetRemoveFromWorkspacesOptions(workspaces=workspaces) + + # Call the method + self.variable_sets_service.remove_from_workspaces(self.variable_set_id, options) + + # Verify API call payload + expected_payload = {"data": [{"type": "workspaces", "id": "ws-1"}]} + self.mock_transport.request.assert_called_once_with( + "DELETE", + f"/api/v2/varsets/{self.variable_set_id}/relationships/workspaces", + json_body=expected_payload, + ) + + def test_apply_to_projects_success(self): + """Test successful application to projects.""" + # Prepare test data + projects = [Project(id="prj-1"), Project(id="prj-2")] + options = VariableSetApplyToProjectsOptions(projects=projects) + + # Call the method + self.variable_sets_service.apply_to_projects(self.variable_set_id, options) + + # Verify API call payload + expected_payload = { + "data": [ + {"type": "projects", "id": "prj-1"}, + {"type": "projects", "id": "prj-2"}, + ] + } + self.mock_transport.request.assert_called_once_with( + "POST", + f"/api/v2/varsets/{self.variable_set_id}/relationships/projects", + json_body=expected_payload, + ) + + def test_apply_to_projects_invalid_params(self): + """Test applying to projects with invalid parameters.""" + # Empty projects list + with pytest.raises(ValueError, match="At least one project is required"): + self.variable_sets_service.apply_to_projects( + self.variable_set_id, VariableSetApplyToProjectsOptions(projects=[]) + ) + + # Project without ID + projects = [Project(id="")] + options = VariableSetApplyToProjectsOptions(projects=projects) + with pytest.raises(ValueError, match="All projects must have valid IDs"): + self.variable_sets_service.apply_to_projects(self.variable_set_id, options) + + def test_remove_from_projects_success(self): + """Test successful removal from projects.""" + # Prepare test data + projects = [Project(id="prj-1")] + options = VariableSetRemoveFromProjectsOptions(projects=projects) + + # Call the method + self.variable_sets_service.remove_from_projects(self.variable_set_id, options) + + # Verify API call payload + expected_payload = {"data": [{"type": "projects", "id": "prj-1"}]} + self.mock_transport.request.assert_called_once_with( + "DELETE", + f"/api/v2/varsets/{self.variable_set_id}/relationships/projects", + json_body=expected_payload, + ) + + def test_update_workspaces_success(self): + """Test successful workspace update.""" + # Prepare test data + workspaces = [Workspace(id="ws-1"), Workspace(id="ws-2")] + options = VariableSetUpdateWorkspacesOptions(workspaces=workspaces) + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": self.variable_set_id, + "type": "varsets", + "attributes": { + "name": "test-varset", + "global": False, + }, + "relationships": { + "workspaces": { + "data": [ + {"type": "workspaces", "id": "ws-1"}, + {"type": "workspaces", "id": "ws-2"}, + ] + } + }, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variable_sets_service.update_workspaces( + self.variable_set_id, options + ) + + # Assertions + assert isinstance(result, VariableSet) + assert len(result.workspaces) == 2 + + # Verify API call payload + expected_payload = { + "data": { + "type": "varsets", + "id": self.variable_set_id, + "attributes": {"global": False}, + "relationships": { + "workspaces": { + "data": [ + {"type": "workspaces", "id": "ws-1"}, + {"type": "workspaces", "id": "ws-2"}, + ] + } + }, + } + } + expected_params = {"include": "workspaces"} + self.mock_transport.request.assert_called_once_with( + "PATCH", + f"/api/v2/varsets/{self.variable_set_id}", + json_body=expected_payload, + params=expected_params, + ) + + def test_update_workspaces_invalid_params(self): + """Test updating workspaces with invalid parameters.""" + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variable_sets_service.update_workspaces( + "", VariableSetUpdateWorkspacesOptions() + ) + + # Invalid options + with pytest.raises(ValueError, match="Options are required"): + self.variable_sets_service.update_workspaces(self.variable_set_id, None) + + +class TestVariableSetVariables: + """Test cases for Variable Set Variables resource.""" + + def setup_method(self): + """Setup method that runs before each test.""" + self.mock_transport = Mock() + self.variables_service = VariableSetVariables(self.mock_transport) + self.variable_set_id = "varset-test123" + self.variable_id = "var-test123" + + def test_variable_set_variables_service_init(self): + """Test Variable Set Variables service initialization.""" + assert self.variables_service.t == self.mock_transport + + def test_list_variables_success(self): + """Test successful listing of variables.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": [ + { + "id": "var-123", + "type": "vars", + "attributes": { + "key": "TF_VAR_test", + "value": "test-value", + "description": "Test variable", + "category": "terraform", + "hcl": False, + "sensitive": False, + "version-id": "v1", + }, + "relationships": { + "varset": { + "data": {"id": self.variable_set_id, "type": "varsets"} + } + }, + } + ] + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variables_service.list(self.variable_set_id) + + # Assertions + assert len(result) == 1 + assert isinstance(result[0], VariableSetVariable) + assert result[0].id == "var-123" + assert result[0].key == "TF_VAR_test" + assert result[0].value == "test-value" + assert result[0].description == "Test variable" + assert result[0].category == CategoryType.TERRAFORM + assert result[0].hcl is False + assert result[0].sensitive is False + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", + f"/api/v2/varsets/{self.variable_set_id}/relationships/vars", + params={}, + ) + + def test_list_variables_invalid_varset_id(self): + """Test listing variables with invalid variable set ID.""" + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variables_service.list("") + + def test_create_variable_success(self): + """Test successful variable creation.""" + # Prepare test data + options = VariableSetVariableCreateOptions( + key="NEW_VAR", + value="new-value", + description="New variable", + category=CategoryType.ENV, + hcl=False, + sensitive=True, + ) + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "var-new123", + "type": "vars", + "attributes": { + "key": "NEW_VAR", + "value": "new-value", + "description": "New variable", + "category": "env", + "hcl": False, + "sensitive": True, + }, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variables_service.create(self.variable_set_id, options) + + # Assertions + assert isinstance(result, VariableSetVariable) + assert result.id == "var-new123" + assert result.key == "NEW_VAR" + assert result.value == "new-value" + assert result.category == CategoryType.ENV + assert result.sensitive is True + + # Verify API call payload + expected_payload = { + "data": { + "type": "vars", + "attributes": { + "key": "NEW_VAR", + "value": "new-value", + "description": "New variable", + "category": "env", + "hcl": False, + "sensitive": True, + }, + } + } + self.mock_transport.request.assert_called_once_with( + "POST", + f"/api/v2/varsets/{self.variable_set_id}/relationships/vars", + json_body=expected_payload, + ) + + def test_create_variable_invalid_params(self): + """Test variable creation with invalid parameters.""" + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variables_service.create( + "", + VariableSetVariableCreateOptions( + key="test", category=CategoryType.TERRAFORM + ), + ) + + # Invalid options + with pytest.raises(ValueError, match="Options are required"): + self.variables_service.create(self.variable_set_id, None) + + # Missing key + with pytest.raises(ValueError, match="Variable key is required"): + self.variables_service.create( + self.variable_set_id, + VariableSetVariableCreateOptions( + key="", category=CategoryType.TERRAFORM + ), + ) + + def test_read_variable_success(self): + """Test successful variable read.""" + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": self.variable_id, + "type": "vars", + "attributes": { + "key": "READ_VAR", + "value": "read-value", + "description": "Variable for reading", + "category": "terraform", + "hcl": True, + "sensitive": False, + }, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variables_service.read(self.variable_set_id, self.variable_id) + + # Assertions + assert isinstance(result, VariableSetVariable) + assert result.id == self.variable_id + assert result.key == "READ_VAR" + assert result.hcl is True + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "GET", + f"/api/v2/varsets/{self.variable_set_id}/relationships/vars/{self.variable_id}", + ) + + def test_read_variable_invalid_params(self): + """Test reading variable with invalid parameters.""" + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variables_service.read("", self.variable_id) + + # Invalid variable ID + with pytest.raises(ValueError, match="Variable ID is required"): + self.variables_service.read(self.variable_set_id, "") + + def test_update_variable_success(self): + """Test successful variable update.""" + # Prepare test data + options = VariableSetVariableUpdateOptions( + key="UPDATED_VAR", + value="updated-value", + description="Updated variable", + hcl=True, + sensitive=False, + ) + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": self.variable_id, + "type": "vars", + "attributes": { + "key": "UPDATED_VAR", + "value": "updated-value", + "description": "Updated variable", + "category": "terraform", + "hcl": True, + "sensitive": False, + }, + "relationships": {}, + } + } + self.mock_transport.request.return_value = mock_response + + # Call the method + result = self.variables_service.update( + self.variable_set_id, self.variable_id, options + ) + + # Assertions + assert result.key == "UPDATED_VAR" + assert result.value == "updated-value" + assert result.hcl is True + + # Verify API call payload + expected_payload = { + "data": { + "type": "vars", + "id": self.variable_id, + "attributes": { + "key": "UPDATED_VAR", + "value": "updated-value", + "description": "Updated variable", + "hcl": True, + "sensitive": False, + }, + } + } + self.mock_transport.request.assert_called_once_with( + "PATCH", + f"/api/v2/varsets/{self.variable_set_id}/relationships/vars/{self.variable_id}", + json_body=expected_payload, + ) + + def test_update_variable_invalid_params(self): + """Test variable update with invalid parameters.""" + options = VariableSetVariableUpdateOptions(key="test") + + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variables_service.update("", self.variable_id, options) + + # Invalid variable ID + with pytest.raises(ValueError, match="Variable ID is required"): + self.variables_service.update(self.variable_set_id, "", options) + + # Invalid options + with pytest.raises(ValueError, match="Options are required"): + self.variables_service.update(self.variable_set_id, self.variable_id, None) + + def test_delete_variable_success(self): + """Test successful variable deletion.""" + # Call the method + self.variables_service.delete(self.variable_set_id, self.variable_id) + + # Verify API call + self.mock_transport.request.assert_called_once_with( + "DELETE", + f"/api/v2/varsets/{self.variable_set_id}/relationships/vars/{self.variable_id}", + ) + + def test_delete_variable_invalid_params(self): + """Test variable deletion with invalid parameters.""" + # Invalid variable set ID + with pytest.raises(ValueError, match="Variable set ID is required"): + self.variables_service.delete("", self.variable_id) + + # Invalid variable ID + with pytest.raises(ValueError, match="Variable ID is required"): + self.variables_service.delete(self.variable_set_id, "")