From 5e393e38583a969f280101603e0c32aef2364ed4 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Tue, 23 Sep 2025 11:44:36 +0530 Subject: [PATCH 1/4] Refactored types for module for project, variableset --- examples/project.py | 112 +++++++++++----- examples/variable_sets_example.py | 10 +- src/tfe/models/__init__.py | 53 +++++++- src/tfe/models/project.py | 62 +++++++++ src/tfe/models/variable_set.py | 193 ++++++++++++++++++++++++++ src/tfe/resources/projects.py | 6 +- src/tfe/resources/variable_sets.py | 4 +- src/tfe/types.py | 208 +---------------------------- tests/units/test_project.py | 8 +- tests/units/test_variable_sets.py | 8 +- tests/units/test_workspaces.py | 2 +- 11 files changed, 411 insertions(+), 255 deletions(-) create mode 100644 src/tfe/models/project.py create mode 100644 src/tfe/models/variable_set.py diff --git a/examples/project.py b/examples/project.py index 3866365f..9740beb7 100644 --- a/examples/project.py +++ b/examples/project.py @@ -31,12 +31,15 @@ from tfe._http import HTTPTransport from tfe.config import TFEConfig -from tfe.resources.projects import Projects -from tfe.types import ( +from tfe.errors import NotFound +from tfe.models.project import ( ProjectAddTagBindingsOptions, ProjectCreateOptions, ProjectListOptions, ProjectUpdateOptions, +) +from tfe.resources.projects import Projects +from tfe.types import ( TagBinding, ) @@ -737,12 +740,10 @@ def test_project_tag_bindings_error_scenarios(integration_client): # Test invalid project ID validation print("๐Ÿšซ Testing invalid project ID scenarios") - invalid_project_ids = ["", "x", "invalid-id", None] - - for invalid_id in invalid_project_ids: - if invalid_id is None: - continue # Skip None as it will cause different error + # Test truly invalid IDs (should fail client-side validation) + truly_invalid_ids = ["", "x"] + for invalid_id in truly_invalid_ids: try: projects.list_tag_bindings(invalid_id) pytest.fail( @@ -768,6 +769,46 @@ def test_project_tag_bindings_error_scenarios(integration_client): except ValueError as e: print(f"โœ… Correctly rejected invalid project ID '{invalid_id}': {e}") + # Test valid-looking but non-existent IDs (should fail server-side) + nonexistent_ids = ["invalid-id", "prj-doesnotexist123"] + + for invalid_id in nonexistent_ids: + try: + projects.list_tag_bindings(invalid_id) + pytest.fail( + f"Should have raised NotFound for non-existent project ID: {invalid_id}" + ) + except NotFound: + print( + f"โœ… Correctly got NotFound for non-existent project ID '{invalid_id}'" + ) + except Exception as e: + # Some endpoints might not be available, that's okay too + print( + f"โœ… Got expected error for non-existent project ID '{invalid_id}': {type(e).__name__}" + ) + + # Test other operations but don't fail if endpoint is not available + for operation_name, operation in [ + ( + "list_effective_tag_bindings", + lambda pid=invalid_id: projects.list_effective_tag_bindings(pid), + ), + ( + "delete_tag_bindings", + lambda pid=invalid_id: projects.delete_tag_bindings(pid), + ), + ]: + try: + operation() + pytest.fail( + f"Should have raised error for non-existent project ID: {invalid_id}" + ) + except (NotFound, Exception) as e: + print( + f"โœ… {operation_name} correctly handled non-existent project ID '{invalid_id}': {type(e).__name__}" + ) + # Test empty tag binding list print("๐Ÿšซ Testing empty tag binding list") try: @@ -779,34 +820,37 @@ def test_project_tag_bindings_error_scenarios(integration_client): print(f"โœ… Correctly rejected empty tag binding list: {e}") assert "At least one tag binding is required" in str(e) - # Test non-existent project operations - print("๐Ÿšซ Testing operations on non-existent project") - fake_project_id = "prj-doesnotexist123" - - # These should raise HTTP errors (404) from the API - for operation_name, operation_func in [ - ("list_tag_bindings", lambda: projects.list_tag_bindings(fake_project_id)), - ( - "list_effective_tag_bindings", - lambda: projects.list_effective_tag_bindings(fake_project_id), - ), - ("delete_tag_bindings", lambda: projects.delete_tag_bindings(fake_project_id)), - ]: - try: - operation_func() - pytest.fail(f"{operation_name} should have failed for non-existent project") - except Exception as e: - print( - f"โœ… {operation_name} correctly failed for non-existent project: {type(e).__name__}" - ) - # Should be some kind of HTTP error (404, not found, etc.) - assert ( - "404" in str(e) - or "not found" in str(e).lower() - or "does not exist" in str(e).lower() - ) + # Test non-existent project operations + print("๐Ÿšซ Testing operations on non-existent project") + fake_project_id = "prj-doesnotexist123" + + # These should raise HTTP errors (404) from the API + for operation_name, operation_func in [ + ("list_tag_bindings", lambda: projects.list_tag_bindings(fake_project_id)), + ( + "list_effective_tag_bindings", + lambda: projects.list_effective_tag_bindings(fake_project_id), + ), + ( + "delete_tag_bindings", + lambda: projects.delete_tag_bindings(fake_project_id), + ), + ]: + try: + operation_func() + pytest.fail( + f"{operation_name} should have failed for non-existent project" + ) + except Exception as e: + print( + f"โœ… {operation_name} correctly failed for non-existent project: {type(e).__name__}" + ) + # Any exception is acceptable for non-existent project operations + # The important thing is that it doesn't succeed - # Test add_tag_bindings on non-existent project + print( + "โœ… All error handling scenarios tested successfully" + ) # Test add_tag_bindings on non-existent project try: test_tags = [TagBinding(key="test", value="value")] add_options = ProjectAddTagBindingsOptions(tag_bindings=test_tags) diff --git a/examples/variable_sets_example.py b/examples/variable_sets_example.py index df621026..36c81682 100644 --- a/examples/variable_sets_example.py +++ b/examples/variable_sets_example.py @@ -16,10 +16,8 @@ import os from tfe import TFEClient, TFEConfig -from tfe.types import ( - CategoryType, +from tfe.models import ( Parent, - Project, VariableSetApplyToProjectsOptions, VariableSetApplyToWorkspacesOptions, VariableSetCreateOptions, @@ -31,6 +29,10 @@ VariableSetVariableCreateOptions, VariableSetVariableListOptions, VariableSetVariableUpdateOptions, +) +from tfe.models.project import Project +from tfe.types import ( + CategoryType, Workspace, ) @@ -271,7 +273,7 @@ def variable_set_example(): # 9. Read the variable set with includes print("9. Reading variable set with includes...") - from tfe.types import VariableSetReadOptions + from tfe.models import VariableSetReadOptions read_options = VariableSetReadOptions( include=[VariableSetIncludeOpt.VARS, VariableSetIncludeOpt.WORKSPACES] diff --git a/src/tfe/models/__init__.py b/src/tfe/models/__init__.py index 15fe5920..a331e83a 100644 --- a/src/tfe/models/__init__.py +++ b/src/tfe/models/__init__.py @@ -4,6 +4,15 @@ import importlib.util import os +# Re-export all project types +from .project import ( + Project, + ProjectAddTagBindingsOptions, + ProjectCreateOptions, + ProjectListOptions, + ProjectUpdateOptions, +) + # Re-export all registry module types from .registry_module_types import ( AgentExecutionMode, @@ -49,6 +58,26 @@ RegistryProviderReadOptions, ) +# Re-export all variable set types +from .variable_set import ( + Parent, + VariableSet, + VariableSetApplyToProjectsOptions, + VariableSetApplyToWorkspacesOptions, + VariableSetCreateOptions, + VariableSetIncludeOpt, + VariableSetListOptions, + VariableSetReadOptions, + VariableSetRemoveFromProjectsOptions, + VariableSetRemoveFromWorkspacesOptions, + VariableSetUpdateOptions, + VariableSetUpdateWorkspacesOptions, + VariableSetVariable, + VariableSetVariableCreateOptions, + VariableSetVariableListOptions, + VariableSetVariableUpdateOptions, +) + # Define what should be available when importing with * __all__ = [ # Registry module types @@ -90,6 +119,29 @@ "RegistryProviderListOptions", "RegistryProviderPermissions", "RegistryProviderReadOptions", + # Project types + "Project", + "ProjectAddTagBindingsOptions", + "ProjectCreateOptions", + "ProjectListOptions", + "ProjectUpdateOptions", + # Variable set types + "Parent", + "VariableSet", + "VariableSetApplyToProjectsOptions", + "VariableSetApplyToWorkspacesOptions", + "VariableSetCreateOptions", + "VariableSetIncludeOpt", + "VariableSetListOptions", + "VariableSetReadOptions", + "VariableSetRemoveFromProjectsOptions", + "VariableSetRemoveFromWorkspacesOptions", + "VariableSetUpdateOptions", + "VariableSetUpdateWorkspacesOptions", + "VariableSetVariable", + "VariableSetVariableCreateOptions", + "VariableSetVariableListOptions", + "VariableSetVariableUpdateOptions", # Main types from types.py (will be dynamically added below) "Capacity", "DataRetentionPolicy", @@ -107,7 +159,6 @@ "OrganizationCreateOptions", "OrganizationUpdateOptions", "Pagination", - "Project", "ReadRunQueueOptions", "Run", "RunQueue", diff --git a/src/tfe/models/project.py b/src/tfe/models/project.py new file mode 100644 index 00000000..84f5ee08 --- /dev/null +++ b/src/tfe/models/project.py @@ -0,0 +1,62 @@ +"""Project-related types for the Terraform Cloud/Enterprise API. + +This module contains all Pydantic models and types related to projects, +including the main Project model and all project-related option classes. +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, Field + + +class Project(BaseModel): + """Project represents a Terraform Enterprise project""" + + id: str + name: str | None = None + description: str = "" + organization: str | None = None + created_at: str = "" + updated_at: str = "" + workspace_count: int = 0 + default_execution_mode: str = "remote" + + +class ProjectListOptions(BaseModel): + """Options for listing projects""" + + # Optional: String used to filter results by complete project name + name: str | None = None + # Optional: Query string to search projects by names + query: str | None = None + # Optional: Include related resources + include: list[str] | None = None + # Pagination options + page_number: int | None = None + page_size: int | None = None + + +class ProjectCreateOptions(BaseModel): + """Options for creating a project""" + + # Required: A name to identify the project + name: str + # Optional: A description for the project + description: str | None = None + + +class ProjectUpdateOptions(BaseModel): + """Options for updating a project""" + + # Optional: A name to identify the project + name: str | None = None + # Optional: A description for the project + description: str | None = None + + +class ProjectAddTagBindingsOptions(BaseModel): + """Options for adding tag bindings to a project""" + + tag_bindings: list[Any] = Field(default_factory=list) diff --git a/src/tfe/models/variable_set.py b/src/tfe/models/variable_set.py new file mode 100644 index 00000000..362533fe --- /dev/null +++ b/src/tfe/models/variable_set.py @@ -0,0 +1,193 @@ +"""Variable set type definitions for the TFE API.""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import TYPE_CHECKING + +from pydantic import BaseModel, Field + +from ..types import CategoryType + +if TYPE_CHECKING: + from ..types import Organization, Workspace + from .project import Project + + +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 + + +# Model rebuild functionality for runtime +def _rebuild_models() -> None: + """Rebuild models to resolve forward references at runtime.""" + try: + # Import the main types to ensure they're available + from ..types import Organization, Workspace # noqa: F401 + from .project import Project # noqa: F401 + + # Rebuild models that have forward references + Parent.model_rebuild() + VariableSet.model_rebuild() + VariableSetVariable.model_rebuild() + VariableSetCreateOptions.model_rebuild() + VariableSetApplyToWorkspacesOptions.model_rebuild() + VariableSetRemoveFromWorkspacesOptions.model_rebuild() + VariableSetApplyToProjectsOptions.model_rebuild() + VariableSetRemoveFromProjectsOptions.model_rebuild() + VariableSetUpdateWorkspacesOptions.model_rebuild() + except ImportError: + # If the main types aren't available yet, models will be rebuilt later + pass + + +# Call rebuild when module is imported +_rebuild_models() diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index e6a35dc4..1d1bf7b7 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -5,13 +5,15 @@ from collections.abc import Iterator from typing import Any -from ..types import ( - EffectiveTagBinding, +from ..models.project import ( Project, ProjectAddTagBindingsOptions, ProjectCreateOptions, ProjectListOptions, ProjectUpdateOptions, +) +from ..types import ( + EffectiveTagBinding, TagBinding, ) from ..utils import valid_string, valid_string_id diff --git a/src/tfe/resources/variable_sets.py b/src/tfe/resources/variable_sets.py index bc0e0db7..48064856 100644 --- a/src/tfe/resources/variable_sets.py +++ b/src/tfe/resources/variable_sets.py @@ -4,8 +4,7 @@ from typing import Any from tfe._http import HTTPTransport -from tfe.resources._base import _Service -from tfe.types import ( +from tfe.models.variable_set import ( VariableSet, VariableSetApplyToProjectsOptions, VariableSetApplyToWorkspacesOptions, @@ -22,6 +21,7 @@ VariableSetVariableListOptions, VariableSetVariableUpdateOptions, ) +from tfe.resources._base import _Service class VariableSets(_Service): diff --git a/src/tfe/types.py b/src/tfe/types.py index bc115aaa..30c6246c 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -102,57 +102,6 @@ class Organization(BaseModel): data_retention_policy_choice: dict | None = None -class Project(BaseModel): - """Project represents a Terraform Enterprise project""" - - id: str - name: str | None = None - description: str = "" - organization: str | None = None - created_at: str = "" - updated_at: str = "" - workspace_count: int = 0 - default_execution_mode: str = "remote" - - -class ProjectListOptions(BaseModel): - """Options for listing projects""" - - # Optional: String used to filter results by complete project name - name: str | None = None - # Optional: Query string to search projects by names - query: str | None = None - # Optional: Include related resources - include: list[str] | None = None - # Pagination options - page_number: int | None = None - page_size: int | None = None - - -class ProjectCreateOptions(BaseModel): - """Options for creating a project""" - - # Required: A name to identify the project - name: str - # Optional: A description for the project - description: str | None = None - - -class ProjectUpdateOptions(BaseModel): - """Options for updating a project""" - - # Optional: A name to identify the project - name: str | None = None - # Optional: A description for the project - description: str | None = None - - -class ProjectAddTagBindingsOptions(BaseModel): - """Options for adding tag bindings to a project""" - - tag_bindings: list[TagBinding] = Field(default_factory=list) - - class Workspace(BaseModel): id: str name: str | None = None @@ -205,7 +154,7 @@ class Workspace(BaseModel): agent_pool: Any | None = None # AgentPool object current_run: Any | None = None # Run object current_state_version: Any | None = None # StateVersion object - project: Project | None = None + project: Any | None = None # Project object ssh_key: Any | None = None # SSHKey object outputs: list[WorkspaceOutputs] = Field(default_factory=list) tags: list[Tag] = Field(default_factory=list) @@ -532,7 +481,7 @@ class WorkspaceCreateOptions(BaseModel): hyok_enabled: bool | None = None tags: list[Tag] = Field(default_factory=list) setting_overwrites: WorkspaceSettingOverwrites | None = None - project: Project | None = None + project: Any | None = None # Project object tag_bindings: list[TagBinding] = Field(default_factory=list) @@ -562,7 +511,7 @@ class WorkspaceUpdateOptions(BaseModel): working_directory: str | None = None hyok_enabled: bool | None = None setting_overwrites: WorkspaceSettingOverwrites | None = None - project: Project | None = None + project: Any | None = None # Project object tag_bindings: list[TagBinding] = Field(default_factory=list) @@ -660,154 +609,3 @@ class WorkspaceAddTagBindingsOptions(BaseModel): # 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_project.py b/tests/units/test_project.py index 7876f748..ad8fe43e 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -1,12 +1,14 @@ from unittest.mock import Mock -from tfe.resources.projects import Projects, _safe_str -from tfe.types import ( - EffectiveTagBinding, +from tfe.models.project import ( Project, ProjectAddTagBindingsOptions, ProjectCreateOptions, ProjectUpdateOptions, +) +from tfe.resources.projects import Projects, _safe_str +from tfe.types import ( + EffectiveTagBinding, TagBinding, ) diff --git a/tests/units/test_variable_sets.py b/tests/units/test_variable_sets.py index 77a672d5..735204fa 100644 --- a/tests/units/test_variable_sets.py +++ b/tests/units/test_variable_sets.py @@ -4,9 +4,7 @@ import pytest -from tfe.resources.variable_sets import VariableSets, VariableSetVariables -from tfe.types import ( - CategoryType, +from tfe.models import ( Parent, Project, VariableSet, @@ -23,6 +21,10 @@ VariableSetVariable, VariableSetVariableCreateOptions, VariableSetVariableUpdateOptions, +) +from tfe.resources.variable_sets import VariableSets, VariableSetVariables +from tfe.types import ( + CategoryType, Workspace, ) diff --git a/tests/units/test_workspaces.py b/tests/units/test_workspaces.py index 29c838ca..723e1a89 100644 --- a/tests/units/test_workspaces.py +++ b/tests/units/test_workspaces.py @@ -19,6 +19,7 @@ RequiredSSHKeyIDError, WorkspaceMinimumLimitError, ) +from src.tfe.models.project import Project from src.tfe.resources.workspaces import Workspaces, _ws_from from src.tfe.types import ( DataRetentionPolicyDeleteOlderSetOptions, @@ -26,7 +27,6 @@ DataRetentionPolicySetOptions, EffectiveTagBinding, ExecutionMode, - Project, Tag, TagBinding, VCSRepo, From 3ec66362bb3202f482c5d7eac43583ec610042f1 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Fri, 26 Sep 2025 09:16:58 +0530 Subject: [PATCH 2/4] Refactored models for variable_set and project --- examples/project.py | 953 ++++----------------------- examples/variable_set.py | 508 ++++++++++++++ src/tfe/models/variable_set.py | 35 +- src/tfe/models/variable_set_types.py | 165 +++++ 4 files changed, 807 insertions(+), 854 deletions(-) create mode 100644 examples/variable_set.py create mode 100644 src/tfe/models/variable_set_types.py diff --git a/examples/project.py b/examples/project.py index 9740beb7..cfc843ff 100644 --- a/examples/project.py +++ b/examples/project.py @@ -1,897 +1,208 @@ """ -Comprehensive Integration Test for python-tfe Projects CRUD Operations +Real-time Project SDK Integration Example -This file tests all CRUD operations from src/tfe/resources/projects.py: -- List: Get all projects in an organization -- Create: Add new projects with validation -- Read: Get specific project details -- Update: Modify existing projects -- Delete: Remove projects +This example demonstrates how to use the TFE Python SDK for project operations +with real API calls. This is NOT a unit test - it uses the actual SDK client +to perform CRUD operations on real projects. Setup Instructions: -1. Create a test organization in HCP Terraform (https://app.terraform.io) -2. Generate an organization or user API token with appropriate permissions -3. Set environment variables: +1. Set environment variables: export TFE_TOKEN="your-api-token-here" export TFE_ORG="your-test-organization-name" -4. Run the tests: - pytest examples/project.py -v -s +2. Run the example: + python examples/project.py Important Notes: -- These tests make real API calls and create/delete actual resources +- This makes real API calls and creates/deletes actual resources - Always use a dedicated test organization, never production -- Tests will fail if you don't have proper permissions -- Clean up is automatic, but verify resources are deleted after testing +- Resources are automatically cleaned up after demonstration """ import os import uuid -import pytest - -from tfe._http import HTTPTransport -from tfe.config import TFEConfig -from tfe.errors import NotFound +from tfe import TFEClient from tfe.models.project import ( - ProjectAddTagBindingsOptions, ProjectCreateOptions, ProjectListOptions, ProjectUpdateOptions, ) -from tfe.resources.projects import Projects -from tfe.types import ( - TagBinding, -) - - -@pytest.fixture -def integration_client(): - """Create a real Projects client for integration testing""" - token = os.environ.get("TFE_TOKEN") - org = os.environ.get("TFE_ORG") - - if not token: - pytest.skip( - "TFE_TOKEN environment variable is required. " - "Get your token from HCP Terraform: Settings โ†’ API Tokens" - ) - - if not org: - pytest.skip( - "TFE_ORG environment variable is required. " - "Use your organization name from HCP Terraform URL" - ) - - print(f"\n๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {token[:10]}...") - - config = TFEConfig() - - try: - transport = HTTPTransport( - config.address, - token, - timeout=config.timeout, - verify_tls=config.verify_tls, - user_agent_suffix=None, - max_retries=3, - backoff_base=0.1, - backoff_cap=1.0, - backoff_jitter=True, - http2=False, - proxies=None, - ca_bundle=None, - ) - except Exception as e: - pytest.fail(f"Failed to create HTTP transport: {e}") - - return Projects(transport), org - - -def test_list_projects_integration(integration_client): - """Test LIST operation - Get all projects in organization - - This is the safest test to run first - it only reads data. - Tests: projects.list(organization, options) - """ - projects, org = integration_client - - try: - # Test basic list without options - print("๐Ÿ“‹ Testing LIST operation: basic list") - project_list = list(projects.list(org)) - print(f"โœ… Found {len(project_list)} projects in organization '{org}'") - - assert isinstance(project_list, list) - - if project_list: - project = project_list[0] - assert hasattr(project, "id"), "Project should have an ID" - assert hasattr(project, "name"), "Project should have a name" - assert hasattr(project, "organization"), ( - "Project should have an organization" - ) - assert hasattr(project, "description"), "Project should have a description" - assert hasattr(project, "created_at"), "Project should have created_at" - assert hasattr(project, "updated_at"), "Project should have updated_at" - print(f"๐Ÿ“‹ Example project: {project.name} (ID: {project.id})") - print(f"๐Ÿ“‹ Created: {project.created_at}, Updated: {project.updated_at}") - else: - print("๐Ÿ“‹ No projects found - this is normal for a new organization") - - # Test list with options - print("๐Ÿ“‹ Testing LIST operation: with options") - list_options = ProjectListOptions(page_size=5) - project_list_with_options = list(projects.list(org, list_options)) - print( - f"โœ… List with options returned {len(project_list_with_options)} projects" - ) - - except Exception as e: - pytest.fail( - f"LIST operation failed. Check your TFE_TOKEN and TFE_ORG. Error: {e}" - ) - - -def test_create_project_integration(integration_client): - """Test CREATE operation - Add new projects - - Tests: projects.create(organization, options) - Validates: ProjectCreateOptions with name and description - """ - projects, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"create-test-{unique_id}" - test_description = f"Integration test project created at {unique_id}" - project_id = None - - try: - # Test CREATE operation - print(f"๐Ÿ”จ Testing CREATE operation: {test_name}") - create_options = ProjectCreateOptions( - name=test_name, description=test_description - ) - created_project = projects.create(org, create_options) - - # Validate created project - assert created_project.name == test_name, ( - f"Expected name {test_name}, got {created_project.name}" - ) - assert created_project.description == test_description, ( - f"Expected description {test_description}, got {created_project.description}" - ) - assert created_project.organization == org, ( - f"Expected org {org}, got {created_project.organization}" - ) - assert created_project.id.startswith("prj-"), ( - f"Project ID should start with 'prj-', got {created_project.id}" - ) - assert created_project.workspace_count == 0, ( - "New project should have 0 workspaces" - ) - - project_id = created_project.id - print(f"โœ… CREATE successful: {project_id}") - print( - f"โœ… Project details: {created_project.name} - {created_project.description}" - ) - - except Exception as e: - pytest.fail(f"CREATE operation failed: {e}") - - finally: - # Clean up created project - if project_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up created project: {project_id}") - projects.delete(project_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") - - -def test_read_project_integration(integration_client): - """Test READ operation - Get specific project details - - Tests: projects.read(project_id, include) - Creates a project, reads it, then cleans up - """ - projects, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"read-test-{unique_id}" - project_id = None - - try: - # Create a project to read - print(f"๏ฟฝ Creating project for READ test: {test_name}") - create_options = ProjectCreateOptions( - name=test_name, description="Project for read test" - ) - created_project = projects.create(org, create_options) - project_id = created_project.id - - # Test READ operation - print(f"๐Ÿ“– Testing READ operation: {project_id}") - read_project = projects.read(project_id) - - # Validate read project - assert read_project.id == project_id, ( - f"Expected ID {project_id}, got {read_project.id}" - ) - assert read_project.name == test_name, ( - f"Expected name {test_name}, got {read_project.name}" - ) - assert read_project.organization == org, ( - f"Expected org {org}, got {read_project.organization}" - ) - assert hasattr(read_project, "created_at"), "Project should have created_at" - assert hasattr(read_project, "updated_at"), "Project should have updated_at" - - print(f"โœ… READ successful: {read_project.name}") - print(f"โœ… Project created: {read_project.created_at}") - - # Note: Projects API doesn't support include parameters in the current API version - print("โœ… READ operation completed successfully") - - except Exception as e: - pytest.fail(f"READ operation failed: {e}") - - finally: - # Clean up created project - if project_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up read test project: {project_id}") - projects.delete(project_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") - - -def test_update_project_integration(integration_client): - """Test UPDATE operation - Modify existing projects - - Tests: projects.update(project_id, options) - Validates: ProjectUpdateOptions with name and description changes - """ - projects, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - original_name = f"update-test-{unique_id}" - updated_name = f"updated-test-{unique_id}" - original_description = "Original description for update test" - updated_description = "Updated description for update test" - project_id = None - - try: - # Create a project to update - print(f"๐Ÿ”จ Creating project for UPDATE test: {original_name}") - create_options = ProjectCreateOptions( - name=original_name, description=original_description - ) - created_project = projects.create(org, create_options) - project_id = created_project.id - - # Test UPDATE operation - name only - print("โœ๏ธ Testing UPDATE operation: name only") - update_options = ProjectUpdateOptions(name=updated_name) - updated_project = projects.update(project_id, update_options) - - assert updated_project.id == project_id, ( - f"Project ID should remain {project_id}" - ) - assert updated_project.name == updated_name, ( - f"Expected updated name {updated_name}, got {updated_project.name}" - ) - assert updated_project.description == original_description, ( - "Description should remain unchanged" - ) - print(f"โœ… UPDATE name successful: {updated_project.name}") - - # Test UPDATE operation - description only - print("โœ๏ธ Testing UPDATE operation: description only") - update_options = ProjectUpdateOptions(description=updated_description) - updated_project = projects.update(project_id, update_options) - - assert updated_project.name == updated_name, "Name should remain unchanged" - assert updated_project.description == updated_description, ( - f"Expected updated description {updated_description}, got {updated_project.description}" - ) - print("โœ… UPDATE description successful") - - # Test UPDATE operation - both name and description - final_name = f"final-{unique_id}" - final_description = "Final description for update test" - print("โœ๏ธ Testing UPDATE operation: both name and description") - update_options = ProjectUpdateOptions( - name=final_name, description=final_description - ) - updated_project = projects.update(project_id, update_options) - - assert updated_project.name == final_name, ( - f"Expected final name {final_name}, got {updated_project.name}" - ) - assert updated_project.description == final_description, ( - f"Expected final description {final_description}, got {updated_project.description}" - ) - print(f"โœ… UPDATE both fields successful: {updated_project.name}") - - except Exception as e: - pytest.fail(f"UPDATE operation failed: {e}") - - finally: - # Clean up created project - if project_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up update test project: {project_id}") - projects.delete(project_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") - - -def test_delete_project_integration(integration_client): - """Test DELETE operation - Remove projects - - Tests: projects.delete(project_id) - Creates a project, deletes it, verifies it's gone - """ - projects, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"delete-test-{unique_id}" - project_id = None - - try: - # Create a project to delete - print(f"๐Ÿ”จ Creating project for DELETE test: {test_name}") - create_options = ProjectCreateOptions( - name=test_name, description="Project for delete test" - ) - created_project = projects.create(org, create_options) - project_id = created_project.id - print(f"โœ… Project created for deletion: {project_id}") - # Verify project exists - print("๐Ÿ“– Verifying project exists before deletion") - read_project = projects.read(project_id) - assert read_project.id == project_id - print(f"โœ… Project confirmed to exist: {read_project.name}") - # Test DELETE operation - print(f"๐Ÿ—‘๏ธ Testing DELETE operation: {project_id}") - projects.delete(project_id) - print("โœ… DELETE operation completed") +def project_sdk_example(): + """Demonstrate Project SDK operations with real API calls.""" - # Verify project is deleted - print("๐Ÿ“– Verifying project is deleted") - try: - projects.read(project_id) - pytest.fail("Project should not exist after deletion") - except Exception as e: - if "404" in str(e) or "not found" in str(e).lower(): - print("โœ… Project successfully deleted - confirmed by 404 error") - else: - raise e - - # Clear project_id since it's been deleted - project_id = None - - except Exception as e: - pytest.fail(f"DELETE operation failed: {e}") - - finally: - # Additional cleanup attempt (should be unnecessary) - if project_id: - try: - print(f"๐Ÿ—‘๏ธ Additional cleanup attempt: {project_id}") - projects.delete(project_id) - except Exception: - pass # Project might already be deleted + # Initialize SDK client + token = os.getenv("TFE_TOKEN") + org_name = os.getenv("TFE_ORG") + if not token or not org_name: + print("โŒ Please set TFE_TOKEN and TFE_ORG environment variables") + print(" export TFE_TOKEN='your-hcp-terraform-token'") + print(" export TFE_ORG='your-organization-name'") + return -def test_comprehensive_crud_integration(integration_client): - """Test all CRUD operations in sequence + print("๐Ÿš€ TFE Python SDK - Project Operations Example") + print("=" * 50) + print(f"๐Ÿ”ง Organization: {org_name}") + print(f"๐Ÿ”ง Token: {token[:10]}...") - โš ๏ธ WARNING: This test creates and deletes real resources! - Tests complete workflow: CREATE โ†’ READ โ†’ UPDATE โ†’ LIST โ†’ DELETE - """ - projects, org = integration_client + # Create SDK client + client = TFEClient() unique_id = str(uuid.uuid4())[:8] - test_name = f"comprehensive-{unique_id}" - updated_name = f"comprehensive-updated-{unique_id}" - test_description = f"Comprehensive CRUD test {unique_id}" - updated_description = f"Updated comprehensive CRUD test {unique_id}" + test_project_name = f"sdk-example-{unique_id}" + test_description = f"SDK example project created at {unique_id}" project_id = None try: - print(f"๐Ÿ”„ Starting comprehensive CRUD test: {test_name}") - - # 1. CREATE - print("1๏ธโƒฃ CREATE: Creating project") + print("\n1๏ธโƒฃ LIST: Getting existing projects...") + # List existing projects + existing_projects = list(client.projects.list(org_name)) + print(f"โœ… Found {len(existing_projects)} existing projects") + + if existing_projects: + print("๐Ÿ“‹ Example existing projects:") + for i, project in enumerate(existing_projects[:3]): # Show first 3 + print(f" - {project.name} (ID: {project.id})") + if i == 2 and len(existing_projects) > 3: + print(f" ... and {len(existing_projects) - 3} more") + + print(f"\n2๏ธโƒฃ CREATE: Creating new project '{test_project_name}'...") + # Create a new project create_options = ProjectCreateOptions( - name=test_name, description=test_description + name=test_project_name, description=test_description ) - created_project = projects.create(org, create_options) + created_project = client.projects.create(org_name, create_options) project_id = created_project.id - assert created_project.name == test_name - assert created_project.description == test_description - print(f"โœ… CREATE: {project_id}") - - # 2. READ - print("2๏ธโƒฃ READ: Reading created project") - read_project = projects.read(project_id) - - assert read_project.id == project_id - assert read_project.name == test_name - assert read_project.description == test_description - print(f"โœ… READ: {read_project.name}") - - # 3. UPDATE - print("3๏ธโƒฃ UPDATE: Updating project") + print("โœ… Project created successfully!") + print(f" ID: {created_project.id}") + print(f" Name: {created_project.name}") + print(f" Description: {created_project.description}") + print(f" Organization: {created_project.organization}") + print(f" Created: {created_project.created_at}") + + print("\n3๏ธโƒฃ READ: Reading project details...") + # Read the created project + read_project = client.projects.read(project_id) + print("โœ… Project read successfully:") + print(f" Name: {read_project.name}") + print(f" Workspace Count: {read_project.workspace_count}") + print(f" Updated: {read_project.updated_at}") + + print("\n4๏ธโƒฃ UPDATE: Updating project...") + # Update the project + updated_name = f"sdk-updated-{unique_id}" + updated_description = f"SDK example project updated at {unique_id}" update_options = ProjectUpdateOptions( name=updated_name, description=updated_description ) - updated_project = projects.update(project_id, update_options) + updated_project = client.projects.update(project_id, update_options) + + print("โœ… Project updated successfully!") + print(f" New Name: {updated_project.name}") + print(f" New Description: {updated_project.description}") - assert updated_project.id == project_id - assert updated_project.name == updated_name - assert updated_project.description == updated_description - print(f"โœ… UPDATE: {updated_project.name}") + print("\n5๏ธโƒฃ LIST WITH OPTIONS: Testing list with pagination...") + # Test list with options + list_options = ProjectListOptions(page_size=5) + projects_with_options = list(client.projects.list(org_name, list_options)) + print(f"โœ… List with options returned {len(projects_with_options)} projects") - # 4. LIST (verify updated project appears) - print("4๏ธโƒฃ LIST: Verifying project appears in list") - project_list = list(projects.list(org)) + # Verify our updated project appears in the list found_project = None - for p in project_list: - if p.id == project_id: - found_project = p + for project in projects_with_options: + if project.id == project_id: + found_project = project break - assert found_project is not None, ( - f"Updated project {project_id} should appear in list" - ) - assert found_project.name == updated_name - print("โœ… LIST: Found updated project in list") + if found_project: + print(f"โœ… Confirmed updated project appears in list: {found_project.name}") + else: + print("โš ๏ธ Updated project not found in list (may be on another page)") - # 5. DELETE - print("5๏ธโƒฃ DELETE: Deleting project") - projects.delete(project_id) - print("โœ… DELETE: Project deleted") + print("\n6๏ธโƒฃ DELETE: Cleaning up created project...") + # Delete the project + client.projects.delete(project_id) + print("โœ… Project deleted successfully!") - # 6. Verify deletion - print("6๏ธโƒฃ VERIFY: Confirming deletion") + # Verify deletion + print("๐Ÿ” Verifying project deletion...") try: - projects.read(project_id) - pytest.fail("Project should not exist after deletion") + client.projects.read(project_id) + print("โŒ Warning: Project still exists after deletion") except Exception as e: if "404" in str(e) or "not found" in str(e).lower(): - print("โœ… VERIFY: Deletion confirmed") + print("โœ… Confirmed: Project successfully deleted") else: - raise e + print(f"โš ๏ธ Unexpected error during verification: {e}") project_id = None # Clear since deleted - print("๐ŸŽ‰ Comprehensive CRUD test completed successfully!") - - except Exception as e: - pytest.fail(f"Comprehensive CRUD test failed: {e}") - - finally: - if project_id: - try: - print(f"๐Ÿ—‘๏ธ Final cleanup: {project_id}") - projects.delete(project_id) - except Exception: - pass - -def test_validation_integration(integration_client): - """Test validation functions work with real API + print("\n๐ŸŽ‰ Project SDK Example Completed Successfully!") + print("=" * 50) + print("โœ… All CRUD operations (Create, Read, Update, Delete) working") + print("โœ… SDK client properly configured and functional") + print("โœ… Real API integration successful") - Tests all validation scenarios with actual API calls - """ - projects, org = integration_client - - print("๐Ÿ” Testing validation with real API calls") - - try: - # Test valid project creation - unique_id = str(uuid.uuid4())[:8] - valid_name = f"validation-test-{unique_id}" - - print(f"โœ… Testing valid project creation: {valid_name}") - create_options = ProjectCreateOptions( - name=valid_name, description="Valid project" - ) - created_project = projects.create(org, create_options) - - assert created_project.name == valid_name - project_id = created_project.id - print(f"โœ… Valid project created successfully: {project_id}") - - # Test valid project update - updated_name = f"validation-updated-{unique_id}" - print(f"โœ… Testing valid project update: {updated_name}") - update_options = ProjectUpdateOptions(name=updated_name) - updated_project = projects.update(project_id, update_options) - - assert updated_project.name == updated_name - print("โœ… Valid project updated successfully") - - # Clean up - projects.delete(project_id) - print("โœ… Validation test cleanup completed") - - except Exception as e: - pytest.fail(f"Validation integration test failed: {e}") - - -def test_error_handling_integration(integration_client): - """Test error handling with real API calls - - Tests various error scenarios to ensure proper error handling - """ - projects, org = integration_client - - print("๐Ÿšซ Testing error handling scenarios") - - # Test reading a non-existent project - print("๐Ÿšซ Testing read non-existent project") - fake_project_id = "prj-nonexistent123456789" - try: - projects.read(fake_project_id) - pytest.fail("Should have raised an exception for non-existent project") except Exception as e: - print( - f"โœ… Correctly handled error for non-existent project: {type(e).__name__}" - ) - assert "404" in str(e) or "not found" in str(e).lower() - - # Test updating a non-existent project - print("๐Ÿšซ Testing update non-existent project") - try: - update_options = ProjectUpdateOptions(name="should-fail") - projects.update(fake_project_id, update_options) - pytest.fail("Should have raised an exception for non-existent project") - except Exception as e: - print( - f"โœ… Correctly handled update error for non-existent project: {type(e).__name__}" - ) - assert "404" in str(e) or "not found" in str(e).lower() - - # Test deleting a non-existent project - print("๐Ÿšซ Testing delete non-existent project") - try: - projects.delete(fake_project_id) - pytest.fail("Should have raised an exception for non-existent project") - except Exception as e: - print( - f"โœ… Correctly handled delete error for non-existent project: {type(e).__name__}" - ) - assert "404" in str(e) or "not found" in str(e).lower() - - print("โœ… All error handling scenarios tested successfully") - - -def test_project_tag_bindings_integration(integration_client): - """ - Integration test for project tag binding operations - - Note: Project tag bindings may not be available in all HCP Terraform plans. - This test gracefully handles unavailable features while testing what's available. - """ - projects, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"tag-test-{unique_id}" - test_description = f"Project for testing tag bindings - {unique_id}" - project_id = None - - try: - # Create a test project for tagging operations - print(f"๐Ÿท๏ธ Setting up test project for tagging: {test_name}") - create_options = ProjectCreateOptions( - name=test_name, description=test_description - ) - created_project = projects.create(org, create_options) - project_id = created_project.id - print(f"โœ… Created test project: {project_id}") - - # Test 1: List tag bindings (this should work) - print("๐Ÿท๏ธ Testing LIST_TAG_BINDINGS") - try: - initial_tag_bindings = projects.list_tag_bindings(project_id) - assert isinstance(initial_tag_bindings, list), "Should return a list" - print(f"โœ… list_tag_bindings works: {len(initial_tag_bindings)} bindings") - list_tag_bindings_available = True - except Exception as e: - print(f"โŒ list_tag_bindings not available: {e}") - list_tag_bindings_available = False - - # Test 2: List effective tag bindings - print("๐Ÿท๏ธ Testing LIST_EFFECTIVE_TAG_BINDINGS") - try: - effective_bindings = projects.list_effective_tag_bindings(project_id) - assert isinstance(effective_bindings, list), "Should return a list" - print( - f"โœ… list_effective_tag_bindings works: {len(effective_bindings)} bindings" - ) - effective_tag_bindings_available = True - except Exception as e: - print(f"โŒ list_effective_tag_bindings not available: {e}") - print(" This feature may require a higher HCP Terraform plan") - effective_tag_bindings_available = False - - # Test 3: Add tag bindings (if basic listing works) - if list_tag_bindings_available: - print("๐Ÿท๏ธ Testing ADD_TAG_BINDINGS") - try: - test_tags = [ - TagBinding(key="environment", value="testing"), - TagBinding(key="integration-test", value="true"), - ] - add_options = ProjectAddTagBindingsOptions(tag_bindings=test_tags) - added_bindings = projects.add_tag_bindings(project_id, add_options) - - assert isinstance(added_bindings, list), "Should return a list" - assert len(added_bindings) == len(test_tags), ( - "Should return all added tags" - ) - print( - f"โœ… add_tag_bindings works: added {len(added_bindings)} bindings" - ) - - # Verify tags were actually added - current_bindings = projects.list_tag_bindings(project_id) - added_keys = {binding.key for binding in current_bindings} - for tag in test_tags: - assert tag.key in added_keys, ( - f"Tag {tag.key} not found after adding" - ) - print(f"โœ… Verified tags added: {len(current_bindings)} total bindings") - - add_tag_bindings_available = True - - # Test 4: Delete tag bindings - print("๐Ÿท๏ธ Testing DELETE_TAG_BINDINGS") - try: - result = projects.delete_tag_bindings(project_id) - assert result is None, "Delete should return None" - - # Verify deletion - final_bindings = projects.list_tag_bindings(project_id) - print( - f"โœ… delete_tag_bindings works: {len(final_bindings)} bindings remain" - ) - delete_tag_bindings_available = True - except Exception as e: - print(f"โŒ delete_tag_bindings not available: {e}") - delete_tag_bindings_available = False - - except Exception as e: - print(f"โŒ add_tag_bindings not available: {e}") - print(" This feature may require a higher HCP Terraform plan") - add_tag_bindings_available = False - delete_tag_bindings_available = False - else: - add_tag_bindings_available = False - delete_tag_bindings_available = False - - # Summary - print("\n๐Ÿ“Š Project Tag Bindings API Availability Summary:") - features = [ - ("list_tag_bindings", list_tag_bindings_available), - ("list_effective_tag_bindings", effective_tag_bindings_available), - ("add_tag_bindings", add_tag_bindings_available), - ("delete_tag_bindings", delete_tag_bindings_available), - ] - - for feature_name, available in features: - status = "โœ… Available" if available else "โŒ Not Available" - print(f" {feature_name}: {status}") - - available_count = sum(available for _, available in features) - print( - f"\n๐ŸŽฏ {available_count}/4 tag binding features are available in this HCP Terraform organization" - ) - - if available_count == 4: - print("๐ŸŽ‰ All project tag binding operations work perfectly!") - elif available_count > 0: - print("โœ… Partial functionality available - basic operations work!") - else: - print("โš ๏ธ Tag binding features may require a higher HCP Terraform plan") - - except Exception as e: - pytest.fail( - f"Project tag binding integration test failed unexpectedly. " - f"This may indicate a configuration or connectivity issue. Error: {e}" - ) + print(f"\nโŒ Example failed: {e}") + print("๐Ÿ”ง Check your TFE_TOKEN and TFE_ORG environment variables") + print("๐Ÿ”ง Ensure your token has proper permissions for project operations") finally: - # Clean up: Delete the test project + # Emergency cleanup if something went wrong if project_id: try: - print(f"๐Ÿงน Cleaning up test project: {project_id}") - projects.delete(project_id) - print("โœ… Test project deleted successfully") + print(f"\n๐Ÿงน Emergency cleanup: Deleting project {project_id}") + client.projects.delete(project_id) + print("โœ… Emergency cleanup successful") except Exception as cleanup_error: print( - f"โš ๏ธ Warning: Failed to clean up test project {project_id}: {cleanup_error}" + f"โš ๏ธ Warning: Failed to clean up project {project_id}: {cleanup_error}" ) -def test_project_tag_bindings_error_scenarios(integration_client): - """ - Test error handling for project tag binding operations +def demonstrate_error_handling(): + """Demonstrate proper error handling with the SDK.""" - Tests various error conditions: - - Invalid project IDs - - Empty tag binding lists - - Non-existent projects - """ - projects, org = integration_client + print("\n๐Ÿšซ Error Handling Demonstration") + print("-" * 30) - print("๐Ÿท๏ธ Testing tag binding error scenarios") + token = os.getenv("TFE_TOKEN") + org_name = os.getenv("TFE_ORG") - # Test invalid project ID validation - print("๐Ÿšซ Testing invalid project ID scenarios") + if not token or not org_name: + print("โŒ Skipping error handling demo - environment variables not set") + return - # Test truly invalid IDs (should fail client-side validation) - truly_invalid_ids = ["", "x"] + client = TFEClient() - for invalid_id in truly_invalid_ids: - try: - projects.list_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised ValueError for invalid project ID: {invalid_id}" - ) - except ValueError as e: - print(f"โœ… Correctly rejected invalid project ID '{invalid_id}': {e}") - assert "Project ID is required and must be valid" in str(e) - - try: - projects.list_effective_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised ValueError for invalid project ID: {invalid_id}" - ) - except ValueError as e: - print(f"โœ… Correctly rejected invalid project ID '{invalid_id}': {e}") - - try: - projects.delete_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised ValueError for invalid project ID: {invalid_id}" - ) - except ValueError as e: - print(f"โœ… Correctly rejected invalid project ID '{invalid_id}': {e}") - - # Test valid-looking but non-existent IDs (should fail server-side) - nonexistent_ids = ["invalid-id", "prj-doesnotexist123"] - - for invalid_id in nonexistent_ids: - try: - projects.list_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised NotFound for non-existent project ID: {invalid_id}" - ) - except NotFound: - print( - f"โœ… Correctly got NotFound for non-existent project ID '{invalid_id}'" - ) - except Exception as e: - # Some endpoints might not be available, that's okay too - print( - f"โœ… Got expected error for non-existent project ID '{invalid_id}': {type(e).__name__}" - ) - - # Test other operations but don't fail if endpoint is not available - for operation_name, operation in [ - ( - "list_effective_tag_bindings", - lambda pid=invalid_id: projects.list_effective_tag_bindings(pid), - ), - ( - "delete_tag_bindings", - lambda pid=invalid_id: projects.delete_tag_bindings(pid), - ), - ]: - try: - operation() - pytest.fail( - f"Should have raised error for non-existent project ID: {invalid_id}" - ) - except (NotFound, Exception) as e: - print( - f"โœ… {operation_name} correctly handled non-existent project ID '{invalid_id}': {type(e).__name__}" - ) - - # Test empty tag binding list - print("๐Ÿšซ Testing empty tag binding list") - try: - fake_project_id = "prj-fakefakefake123" - empty_options = ProjectAddTagBindingsOptions(tag_bindings=[]) - projects.add_tag_bindings(fake_project_id, empty_options) - pytest.fail("Should have raised ValueError for empty tag binding list") - except ValueError as e: - print(f"โœ… Correctly rejected empty tag binding list: {e}") - assert "At least one tag binding is required" in str(e) - - # Test non-existent project operations - print("๐Ÿšซ Testing operations on non-existent project") - fake_project_id = "prj-doesnotexist123" - - # These should raise HTTP errors (404) from the API - for operation_name, operation_func in [ - ("list_tag_bindings", lambda: projects.list_tag_bindings(fake_project_id)), - ( - "list_effective_tag_bindings", - lambda: projects.list_effective_tag_bindings(fake_project_id), - ), - ( - "delete_tag_bindings", - lambda: projects.delete_tag_bindings(fake_project_id), - ), - ]: - try: - operation_func() - pytest.fail( - f"{operation_name} should have failed for non-existent project" - ) - except Exception as e: - print( - f"โœ… {operation_name} correctly failed for non-existent project: {type(e).__name__}" - ) - # Any exception is acceptable for non-existent project operations - # The important thing is that it doesn't succeed - - print( - "โœ… All error handling scenarios tested successfully" - ) # Test add_tag_bindings on non-existent project + # Test reading a non-existent project + print("๐Ÿ” Testing error handling for non-existent project...") try: - test_tags = [TagBinding(key="test", value="value")] - add_options = ProjectAddTagBindingsOptions(tag_bindings=test_tags) - projects.add_tag_bindings(fake_project_id, add_options) - pytest.fail("add_tag_bindings should have failed for non-existent project") + fake_project_id = "prj-nonexistent123456789" + client.projects.read(fake_project_id) + print("โŒ Unexpected: Should have failed for non-existent project") except Exception as e: - print( - f"โœ… add_tag_bindings correctly failed for non-existent project: {type(e).__name__}" - ) - assert ( - "404" in str(e) - or "not found" in str(e).lower() - or "does not exist" in str(e).lower() - ) + print(f"โœ… Correctly handled error: {type(e).__name__}") + print(f" Message: {str(e)[:100]}...") - print("โœ… All tag binding error scenarios tested successfully") + print("โœ… Error handling demonstration complete") -if __name__ == "__main__": - """ - You can also run this file directly for quick testing: +def main(): + """Main function to run all examples.""" + print("๐Ÿงช TFE Python SDK Project Examples") + print("This demonstrates real SDK usage with actual API calls\n") - export TFE_TOKEN="your-token" - export TFE_ORG="your-org" - python examples/integration_test_example.py - """ - import sys + # Run main project operations example + project_sdk_example() - token = os.environ.get("TFE_TOKEN") - org = os.environ.get("TFE_ORG") + # Run error handling demonstration + demonstrate_error_handling() - if not token or not org: - print("โŒ Please set TFE_TOKEN and TFE_ORG environment variables") - print(" export TFE_TOKEN='your-hcp-terraform-token'") - print(" export TFE_ORG='your-organization-name'") - sys.exit(1) - - print("๐Ÿงช Running integration tests directly...") - print( - " For full pytest features, use: pytest examples/integration_test_example.py -v -s" - ) - # Simple direct execution - pytest.main([__file__, "-v", "-s"]) +if __name__ == "__main__": + main() diff --git a/examples/variable_set.py b/examples/variable_set.py new file mode 100644 index 00000000..36c81682 --- /dev/null +++ b/examples/variable_set.py @@ -0,0 +1,508 @@ +"""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.models import ( + Parent, + VariableSetApplyToProjectsOptions, + VariableSetApplyToWorkspacesOptions, + VariableSetCreateOptions, + VariableSetIncludeOpt, + VariableSetListOptions, + VariableSetRemoveFromProjectsOptions, + VariableSetRemoveFromWorkspacesOptions, + VariableSetUpdateOptions, + VariableSetVariableCreateOptions, + VariableSetVariableListOptions, + VariableSetVariableUpdateOptions, +) +from tfe.models.project import Project +from tfe.types import ( + CategoryType, + 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.models 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/models/variable_set.py b/src/tfe/models/variable_set.py index 362533fe..f141a988 100644 --- a/src/tfe/models/variable_set.py +++ b/src/tfe/models/variable_set.py @@ -4,15 +4,11 @@ from datetime import datetime from enum import Enum -from typing import TYPE_CHECKING from pydantic import BaseModel, Field -from ..types import CategoryType - -if TYPE_CHECKING: - from ..types import Organization, Workspace - from .project import Project +from ..types import CategoryType, Organization, Workspace +from .project import Project class VariableSetIncludeOpt(str, Enum): @@ -164,30 +160,3 @@ class VariableSetVariableUpdateOptions(BaseModel): description: str | None = None hcl: bool | None = None sensitive: bool | None = None - - -# Model rebuild functionality for runtime -def _rebuild_models() -> None: - """Rebuild models to resolve forward references at runtime.""" - try: - # Import the main types to ensure they're available - from ..types import Organization, Workspace # noqa: F401 - from .project import Project # noqa: F401 - - # Rebuild models that have forward references - Parent.model_rebuild() - VariableSet.model_rebuild() - VariableSetVariable.model_rebuild() - VariableSetCreateOptions.model_rebuild() - VariableSetApplyToWorkspacesOptions.model_rebuild() - VariableSetRemoveFromWorkspacesOptions.model_rebuild() - VariableSetApplyToProjectsOptions.model_rebuild() - VariableSetRemoveFromProjectsOptions.model_rebuild() - VariableSetUpdateWorkspacesOptions.model_rebuild() - except ImportError: - # If the main types aren't available yet, models will be rebuilt later - pass - - -# Call rebuild when module is imported -_rebuild_models() diff --git a/src/tfe/models/variable_set_types.py b/src/tfe/models/variable_set_types.py new file mode 100644 index 00000000..f1894102 --- /dev/null +++ b/src/tfe/models/variable_set_types.py @@ -0,0 +1,165 @@ +"""Variable set type definitions for the TFE API.""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum + +from pydantic import BaseModel, Field + +from ..types import CategoryType, Organization, Workspace +from .project import Project + + +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 + + +# Model rebuild functionality for runtime From bb528d1a39fa4b39986ba4f34f3518a3c2f4f12e Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Fri, 26 Sep 2025 09:17:11 +0530 Subject: [PATCH 3/4] Refactored models for variable_set and project --- examples/variable_sets_example.py | 508 ------------------------------ tests/units/test_agents.py | 300 ++++++++++++++++++ 2 files changed, 300 insertions(+), 508 deletions(-) delete mode 100644 examples/variable_sets_example.py create mode 100644 tests/units/test_agents.py diff --git a/examples/variable_sets_example.py b/examples/variable_sets_example.py deleted file mode 100644 index 36c81682..00000000 --- a/examples/variable_sets_example.py +++ /dev/null @@ -1,508 +0,0 @@ -"""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.models import ( - Parent, - VariableSetApplyToProjectsOptions, - VariableSetApplyToWorkspacesOptions, - VariableSetCreateOptions, - VariableSetIncludeOpt, - VariableSetListOptions, - VariableSetRemoveFromProjectsOptions, - VariableSetRemoveFromWorkspacesOptions, - VariableSetUpdateOptions, - VariableSetVariableCreateOptions, - VariableSetVariableListOptions, - VariableSetVariableUpdateOptions, -) -from tfe.models.project import Project -from tfe.types import ( - CategoryType, - 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.models 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/tests/units/test_agents.py b/tests/units/test_agents.py new file mode 100644 index 00000000..efd5fc23 --- /dev/null +++ b/tests/units/test_agents.py @@ -0,0 +1,300 @@ +"""Unit tests for individual agent operations.""" + +import pytest + +def test_simple_agent_discovery(): + """Simple test to ensure pytest can discover this file.""" + assert True + +from unittest.mock import Mock + +import pytest + +from tfe.errors import NotFound, ValidationError, AuthError +from tfe.models.agent import ( + Agent, + AgentStatus, + AgentListOptions, + AgentReadOptions, +) + + +def test_simple_agent_discovery(): + """Simple test to ensure pytest can discover this file.""" + assert True + + +class TestAgentModels: + """Test agent model validation and serialization""" + + def test_agent_model_basic(self): + """Test basic Agent model creation""" + agent = Agent( + id="agent-123456789abcdef0", + name="test-agent", + status=AgentStatus.IDLE, + version="1.0.0", + ip_address="192.168.1.100", + last_ping_at="2023-01-01T00:00:00Z" + ) + + assert agent.id == "agent-123456789abcdef0" + assert agent.name == "test-agent" + assert agent.status == AgentStatus.IDLE + assert agent.version == "1.0.0" + assert agent.ip_address == "192.168.1.100" + assert agent.last_ping_at is not None + + def test_agent_model_minimal(self): + """Test Agent model with minimal required fields""" + agent = Agent(id="agent-123456789abcdef0") + + assert agent.id == "agent-123456789abcdef0" + assert agent.name is None + assert agent.status is None + assert agent.version is None + assert agent.ip_address is None + assert agent.last_ping_at is None + + def test_agent_status_enum(self): + """Test AgentStatus enum values""" + assert AgentStatus.IDLE == "idle" + assert AgentStatus.BUSY == "busy" + assert AgentStatus.UNKNOWN == "unknown" + + # Test with each status + for status in [AgentStatus.IDLE, AgentStatus.BUSY, AgentStatus.UNKNOWN]: + agent = Agent( + id="agent-123456789abcdef0", + name="test-agent", + status=status, + ip_address="192.168.1.100", + last_ping_at="2023-01-01T00:00:00Z" + ) + assert agent.status == status + + def test_agent_list_options(self): + """Test AgentListOptions model""" + # Test with all options + options = AgentListOptions( + page_number=2, + page_size=10, + status=AgentStatus.IDLE + ) + + assert options.page_number == 2 + assert options.page_size == 10 + assert options.status == AgentStatus.IDLE + + # Test minimal options + minimal_options = AgentListOptions() + assert minimal_options.page_number is None + assert minimal_options.page_size is None + assert minimal_options.status is None + + def test_agent_read_options(self): + """Test AgentReadOptions model""" + # Test with include parameter + options = AgentReadOptions(include=["agent-pool"]) + assert options.include == ["agent-pool"] + + # Test minimal options + minimal_options = AgentReadOptions() + assert minimal_options.include is None + + +class TestAgentOperations: + """Test individual agent CRUD operations""" + + @pytest.fixture + def mock_transport(self): + """Mock HTTP transport.""" + transport = Mock() + return transport + + @pytest.fixture + def agents_service(self, mock_transport): + """Create agents service with mocked transport.""" + from tfe.resources.agents import Agents + return Agents(mock_transport) + + def test_list_agents(self, agents_service, mock_transport): + """Test listing agents in an agent pool""" + mock_response = { + "data": [ + { + "id": "agent-123456789abcdef0", + "type": "agents", + "attributes": { + "name": "test-agent-1", + "status": "idle", + "version": "1.0.0", + "ip-address": "192.168.1.100", + "last-ping-at": "2023-01-01T00:00:00Z" + } + }, + { + "id": "agent-abcdef0123456789", + "type": "agents", + "attributes": { + "name": "test-agent-2", + "status": "busy", + "version": "1.0.1", + "ip-address": "192.168.1.101", + "last-ping-at": "2023-01-01T01:00:00Z" + } + } + ] + } + + mock_transport._list.return_value = mock_response["data"] + + agents = list(agents_service.list("apool-123456789abcdef0")) + + assert len(agents) == 2 + assert agents[0].name == "test-agent-1" + assert agents[0].status == AgentStatus.IDLE + assert agents[1].name == "test-agent-2" + assert agents[1].status == AgentStatus.BUSY + + # Verify API call + mock_transport._list.assert_called_once() + call_args = mock_transport._list.call_args + assert "agent-pools/apool-123456789abcdef0/agents" in call_args[0][0] + + def test_list_agents_with_options(self, agents_service, mock_transport): + """Test listing agents with filtering options""" + mock_transport._list.return_value = [] + + options = AgentListOptions( + page_number=2, + page_size=10, + status=AgentStatus.IDLE + ) + + list(agents_service.list("apool-123456789abcdef0", options)) + + # Verify API call includes query parameters + mock_transport._list.assert_called_once() + call_args = mock_transport._list.call_args + params = call_args[1]["params"] + assert params["page[number]"] == 2 + assert params["page[size]"] == 10 + assert params["filter[status]"] == "idle" + + def test_read_agent(self, agents_service, mock_transport): + """Test reading a specific agent""" + mock_response = { + "data": { + "id": "agent-123456789abcdef0", + "type": "agents", + "attributes": { + "name": "existing-agent", + "status": "idle", + "version": "1.2.0", + "ip-address": "192.168.1.200", + "last-ping-at": "2023-01-01T00:00:00Z" + } + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + agent = agents_service.read("agent-123456789abcdef0") + + assert agent.id == "agent-123456789abcdef0" + assert agent.name == "existing-agent" + assert agent.status == AgentStatus.IDLE + assert agent.version == "1.2.0" + assert agent.ip_address == "192.168.1.200" + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert "agents/agent-123456789abcdef0" in call_args[0][1] + + def test_read_agent_with_options(self, agents_service, mock_transport): + """Test reading an agent with include options""" + mock_response = { + "data": { + "id": "agent-123456789abcdef0", + "type": "agents", + "attributes": { + "name": "existing-agent", + "status": "busy", + "version": "1.2.0", + "ip-address": "192.168.1.200", + "last-ping-at": "2023-01-01T00:00:00Z" + } + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + options = AgentReadOptions(include=["agent-pool"]) + agent = agents_service.read("agent-123456789abcdef0", options) + + assert agent.id == "agent-123456789abcdef0" + assert agent.status == AgentStatus.BUSY + + # Verify API call includes query parameters + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + params = call_args[1].get("params", {}) + assert "include" in params + assert "agent-pool" in params["include"] + + def test_delete_agent(self, agents_service, mock_transport): + """Test deleting an agent""" + agents_service.delete("agent-123456789abcdef0") + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "DELETE" + assert "agents/agent-123456789abcdef0" in call_args[0][1] + + +class TestAgentErrorHandling: + """Test error handling scenarios for agents""" + + @pytest.fixture + def mock_transport(self): + """Mock HTTP transport.""" + transport = Mock() + return transport + + @pytest.fixture + def agents_service(self, mock_transport): + """Create agents service with mocked transport.""" + from tfe.resources.agents import Agents + return Agents(mock_transport) + + def test_not_found_error(self, agents_service, mock_transport): + """Test handling of NotFound errors""" + mock_transport.request.side_effect = NotFound("Agent not found") + + with pytest.raises(NotFound): + agents_service.read("nonexistent-agent") + + def test_validation_error_invalid_agent_pool_id(self, agents_service, mock_transport): + """Test handling of ValidationError for invalid agent pool ID""" + with pytest.raises(ValueError, match="Agent pool ID is required and must be valid"): + list(agents_service.list("")) + + def test_validation_error_invalid_agent_id(self, agents_service, mock_transport): + """Test handling of ValidationError for invalid agent ID""" + with pytest.raises(ValueError, match="Agent ID is required and must be valid"): + agents_service.read("") + + def test_auth_error(self, agents_service, mock_transport): + """Test handling of AuthError errors""" + mock_transport.request.side_effect = AuthError("Unauthorized") + + with pytest.raises(AuthError): + agents_service.read("agent-123456789abcdef0") + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) From 147b25ae26a03526dc2ebd09fb815ed641ad2e00 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Fri, 26 Sep 2025 09:35:02 +0530 Subject: [PATCH 4/4] Refactored models for variable_set and project --- tests/units/test_agents.py | 300 ------------------------------------- 1 file changed, 300 deletions(-) delete mode 100644 tests/units/test_agents.py diff --git a/tests/units/test_agents.py b/tests/units/test_agents.py deleted file mode 100644 index efd5fc23..00000000 --- a/tests/units/test_agents.py +++ /dev/null @@ -1,300 +0,0 @@ -"""Unit tests for individual agent operations.""" - -import pytest - -def test_simple_agent_discovery(): - """Simple test to ensure pytest can discover this file.""" - assert True - -from unittest.mock import Mock - -import pytest - -from tfe.errors import NotFound, ValidationError, AuthError -from tfe.models.agent import ( - Agent, - AgentStatus, - AgentListOptions, - AgentReadOptions, -) - - -def test_simple_agent_discovery(): - """Simple test to ensure pytest can discover this file.""" - assert True - - -class TestAgentModels: - """Test agent model validation and serialization""" - - def test_agent_model_basic(self): - """Test basic Agent model creation""" - agent = Agent( - id="agent-123456789abcdef0", - name="test-agent", - status=AgentStatus.IDLE, - version="1.0.0", - ip_address="192.168.1.100", - last_ping_at="2023-01-01T00:00:00Z" - ) - - assert agent.id == "agent-123456789abcdef0" - assert agent.name == "test-agent" - assert agent.status == AgentStatus.IDLE - assert agent.version == "1.0.0" - assert agent.ip_address == "192.168.1.100" - assert agent.last_ping_at is not None - - def test_agent_model_minimal(self): - """Test Agent model with minimal required fields""" - agent = Agent(id="agent-123456789abcdef0") - - assert agent.id == "agent-123456789abcdef0" - assert agent.name is None - assert agent.status is None - assert agent.version is None - assert agent.ip_address is None - assert agent.last_ping_at is None - - def test_agent_status_enum(self): - """Test AgentStatus enum values""" - assert AgentStatus.IDLE == "idle" - assert AgentStatus.BUSY == "busy" - assert AgentStatus.UNKNOWN == "unknown" - - # Test with each status - for status in [AgentStatus.IDLE, AgentStatus.BUSY, AgentStatus.UNKNOWN]: - agent = Agent( - id="agent-123456789abcdef0", - name="test-agent", - status=status, - ip_address="192.168.1.100", - last_ping_at="2023-01-01T00:00:00Z" - ) - assert agent.status == status - - def test_agent_list_options(self): - """Test AgentListOptions model""" - # Test with all options - options = AgentListOptions( - page_number=2, - page_size=10, - status=AgentStatus.IDLE - ) - - assert options.page_number == 2 - assert options.page_size == 10 - assert options.status == AgentStatus.IDLE - - # Test minimal options - minimal_options = AgentListOptions() - assert minimal_options.page_number is None - assert minimal_options.page_size is None - assert minimal_options.status is None - - def test_agent_read_options(self): - """Test AgentReadOptions model""" - # Test with include parameter - options = AgentReadOptions(include=["agent-pool"]) - assert options.include == ["agent-pool"] - - # Test minimal options - minimal_options = AgentReadOptions() - assert minimal_options.include is None - - -class TestAgentOperations: - """Test individual agent CRUD operations""" - - @pytest.fixture - def mock_transport(self): - """Mock HTTP transport.""" - transport = Mock() - return transport - - @pytest.fixture - def agents_service(self, mock_transport): - """Create agents service with mocked transport.""" - from tfe.resources.agents import Agents - return Agents(mock_transport) - - def test_list_agents(self, agents_service, mock_transport): - """Test listing agents in an agent pool""" - mock_response = { - "data": [ - { - "id": "agent-123456789abcdef0", - "type": "agents", - "attributes": { - "name": "test-agent-1", - "status": "idle", - "version": "1.0.0", - "ip-address": "192.168.1.100", - "last-ping-at": "2023-01-01T00:00:00Z" - } - }, - { - "id": "agent-abcdef0123456789", - "type": "agents", - "attributes": { - "name": "test-agent-2", - "status": "busy", - "version": "1.0.1", - "ip-address": "192.168.1.101", - "last-ping-at": "2023-01-01T01:00:00Z" - } - } - ] - } - - mock_transport._list.return_value = mock_response["data"] - - agents = list(agents_service.list("apool-123456789abcdef0")) - - assert len(agents) == 2 - assert agents[0].name == "test-agent-1" - assert agents[0].status == AgentStatus.IDLE - assert agents[1].name == "test-agent-2" - assert agents[1].status == AgentStatus.BUSY - - # Verify API call - mock_transport._list.assert_called_once() - call_args = mock_transport._list.call_args - assert "agent-pools/apool-123456789abcdef0/agents" in call_args[0][0] - - def test_list_agents_with_options(self, agents_service, mock_transport): - """Test listing agents with filtering options""" - mock_transport._list.return_value = [] - - options = AgentListOptions( - page_number=2, - page_size=10, - status=AgentStatus.IDLE - ) - - list(agents_service.list("apool-123456789abcdef0", options)) - - # Verify API call includes query parameters - mock_transport._list.assert_called_once() - call_args = mock_transport._list.call_args - params = call_args[1]["params"] - assert params["page[number]"] == 2 - assert params["page[size]"] == 10 - assert params["filter[status]"] == "idle" - - def test_read_agent(self, agents_service, mock_transport): - """Test reading a specific agent""" - mock_response = { - "data": { - "id": "agent-123456789abcdef0", - "type": "agents", - "attributes": { - "name": "existing-agent", - "status": "idle", - "version": "1.2.0", - "ip-address": "192.168.1.200", - "last-ping-at": "2023-01-01T00:00:00Z" - } - } - } - - mock_transport.request.return_value.json.return_value = mock_response - - agent = agents_service.read("agent-123456789abcdef0") - - assert agent.id == "agent-123456789abcdef0" - assert agent.name == "existing-agent" - assert agent.status == AgentStatus.IDLE - assert agent.version == "1.2.0" - assert agent.ip_address == "192.168.1.200" - - # Verify API call - mock_transport.request.assert_called_once() - call_args = mock_transport.request.call_args - assert call_args[0][0] == "GET" - assert "agents/agent-123456789abcdef0" in call_args[0][1] - - def test_read_agent_with_options(self, agents_service, mock_transport): - """Test reading an agent with include options""" - mock_response = { - "data": { - "id": "agent-123456789abcdef0", - "type": "agents", - "attributes": { - "name": "existing-agent", - "status": "busy", - "version": "1.2.0", - "ip-address": "192.168.1.200", - "last-ping-at": "2023-01-01T00:00:00Z" - } - } - } - - mock_transport.request.return_value.json.return_value = mock_response - - options = AgentReadOptions(include=["agent-pool"]) - agent = agents_service.read("agent-123456789abcdef0", options) - - assert agent.id == "agent-123456789abcdef0" - assert agent.status == AgentStatus.BUSY - - # Verify API call includes query parameters - mock_transport.request.assert_called_once() - call_args = mock_transport.request.call_args - params = call_args[1].get("params", {}) - assert "include" in params - assert "agent-pool" in params["include"] - - def test_delete_agent(self, agents_service, mock_transport): - """Test deleting an agent""" - agents_service.delete("agent-123456789abcdef0") - - # Verify API call - mock_transport.request.assert_called_once() - call_args = mock_transport.request.call_args - assert call_args[0][0] == "DELETE" - assert "agents/agent-123456789abcdef0" in call_args[0][1] - - -class TestAgentErrorHandling: - """Test error handling scenarios for agents""" - - @pytest.fixture - def mock_transport(self): - """Mock HTTP transport.""" - transport = Mock() - return transport - - @pytest.fixture - def agents_service(self, mock_transport): - """Create agents service with mocked transport.""" - from tfe.resources.agents import Agents - return Agents(mock_transport) - - def test_not_found_error(self, agents_service, mock_transport): - """Test handling of NotFound errors""" - mock_transport.request.side_effect = NotFound("Agent not found") - - with pytest.raises(NotFound): - agents_service.read("nonexistent-agent") - - def test_validation_error_invalid_agent_pool_id(self, agents_service, mock_transport): - """Test handling of ValidationError for invalid agent pool ID""" - with pytest.raises(ValueError, match="Agent pool ID is required and must be valid"): - list(agents_service.list("")) - - def test_validation_error_invalid_agent_id(self, agents_service, mock_transport): - """Test handling of ValidationError for invalid agent ID""" - with pytest.raises(ValueError, match="Agent ID is required and must be valid"): - agents_service.read("") - - def test_auth_error(self, agents_service, mock_transport): - """Test handling of AuthError errors""" - mock_transport.request.side_effect = AuthError("Unauthorized") - - with pytest.raises(AuthError): - agents_service.read("agent-123456789abcdef0") - - -if __name__ == "__main__": - pytest.main([__file__, "-v"])