diff --git a/examples/agent.py b/examples/agent.py new file mode 100644 index 0000000..01daf88 --- /dev/null +++ b/examples/agent.py @@ -0,0 +1,127 @@ +"""Simple Individual Agent operations example with the TFE Python SDK. + +This example demonstrates: +1. Listing agents within agent pools +2. Reading individual agent details +3. Agent status monitoring +4. Using the organization SDK client + +Note: Individual agents are created by running the agent binary, not through the API. +This example shows how to manage agents that have already connected to agent pools. + +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 + +Usage: + export TFE_TOKEN="your-token-here" + export TFE_ORG="your-organization" + python examples/agent_simple.py +""" + +import os + +from tfe.client import TFEClient +from tfe.config import TFEConfig +from tfe.errors import NotFound +from tfe.models.agent import AgentListOptions + + +def main(): + """Main function demonstrating agent operations.""" + # Get environment variables + token = os.environ.get("TFE_TOKEN") + org = os.environ.get("TFE_ORG") + address = os.environ.get("TFE_ADDRESS", "https://app.terraform.io") + + if not token: + print("โŒ TFE_TOKEN environment variable is required") + return 1 + + if not org: + print("โŒ TFE_ORG environment variable is required") + return 1 + + # Create TFE client + config = TFEConfig(token=token, address=address) + client = TFEClient(config=config) + + print(f"๐Ÿ”— Connected to: {address}") + print(f"๐Ÿข Organization: {org}") + + try: + # Example 1: Find agent pools to demonstrate agent operations + print("\n๐Ÿ“‹ Finding agent pools...") + agent_pools = client.agent_pools.list(org) + + # Convert to list to check if empty and get count + pool_list = list(agent_pools) + if not pool_list: + print("โš ๏ธ No agent pools found. Create an agent pool first.") + return 1 + + print(f"Found {len(pool_list)} agent pools:") + for pool in pool_list: + print(f" - {pool.name} (ID: {pool.id}, Agents: {pool.agent_count})") + + # Example 2: List agents in each pool + print("\n๐Ÿค– Listing agents in each pool...") + total_agents = 0 + + for pool in pool_list: + print(f"\n๐Ÿ“‚ Agents in pool '{pool.name}':") + + # Use optional parameters for listing + list_options = AgentListOptions(page_size=10) # Optional parameter + agents = client.agents.list(pool.id, options=list_options) + + # Convert to list to check if empty and iterate + agent_list = list(agents) + if agent_list: + total_agents += len(agent_list) + for agent in agent_list: + print(f" - Agent {agent.id}") + print(f" Name: {agent.name or 'Unnamed'}") + print(f" Status: {agent.status}") + print(f" Version: {agent.version or 'Unknown'}") + print(f" IP: {agent.ip_address or 'Unknown'}") + print(f" Last Ping: {agent.last_ping_at or 'Never'}") + + # Example 3: Read detailed agent information + try: + agent_details = client.agents.read(agent.id) + print(" โœ… Agent details retrieved successfully") + print(f" Full name: {agent_details.name or 'Unnamed'}") + print(f" Current status: {agent_details.status}") + except NotFound: + print(" โš ๏ธ Agent details not accessible") + except Exception as e: + print(f" โŒ Error reading agent details: {e}") + + print("") + else: + print(" No agents found in this pool") + + if total_agents == 0: + print("\nโš ๏ธ No agents found in any pools.") + print("To see agents in action:") + print("1. Create an agent pool") + print("2. Run a Terraform Enterprise agent binary connected to the pool") + print("3. Run this example again") + else: + print(f"\n๐Ÿ“Š Total agents found across all pools: {total_agents}") + + print("\n๐ŸŽ‰ Agent operations completed successfully!") + return 0 + + except NotFound as e: + print(f"โŒ Resource not found: {e}") + return 1 + except Exception as e: + print(f"โŒ Error: {e}") + return 1 + + +if __name__ == "__main__": + exit(main()) diff --git a/examples/agent_pool.py b/examples/agent_pool.py new file mode 100644 index 0000000..d9a37df --- /dev/null +++ b/examples/agent_pool.py @@ -0,0 +1,140 @@ +"""Simple Agent Pool operations example with the TFE Python SDK. + +This example demonstrates: +1. Agent Pool CRUD operations (Create, Read, Update, Delete) +2. Agent token creation and management +3. Using the organization SDK client +4. Proper error handling + +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 + +Usage: + export TFE_TOKEN="your-token-here" + export TFE_ORG="your-organization" + python examples/agent_pool_simple.py +""" + +import os +import uuid + +from tfe import TFEClient, TFEConfig +from tfe.errors import NotFound +from tfe.models.agent import ( + AgentPoolAllowedWorkspacePolicy, + AgentPoolCreateOptions, + AgentPoolListOptions, + AgentPoolUpdateOptions, + AgentTokenCreateOptions, +) + + +def main(): + """Main function demonstrating agent pool operations.""" + # Get environment variables + token = os.environ.get("TFE_TOKEN") + org = os.environ.get("TFE_ORG") + address = os.environ.get("TFE_ADDRESS", "https://app.terraform.io") + + if not token: + print("โŒ TFE_TOKEN environment variable is required") + return 1 + + if not org: + print("โŒ TFE_ORG environment variable is required") + return 1 + + # Create TFE client + config = TFEConfig(token=token, address=address) + client = TFEClient(config=config) + + print(f"๐Ÿ”— Connected to: {address}") + print(f"๐Ÿข Organization: {org}") + + try: + # Example 1: List existing agent pools + print("\n๐Ÿ“‹ Listing existing agent pools...") + list_options = AgentPoolListOptions(page_size=10) # Optional parameters + agent_pools = client.agent_pools.list(org, options=list_options) + + # Convert to list to get count and iterate + pool_list = list(agent_pools) + print(f"Found {len(pool_list)} agent pools:") + for pool in pool_list: + print(f" - {pool.name} (ID: {pool.id}, Agents: {pool.agent_count})") + + # Example 2: Create a new agent pool + print("\n๐Ÿ†• Creating a new agent pool...") + unique_name = f"sdk-example-pool-{uuid.uuid4().hex[:8]}" + + create_options = AgentPoolCreateOptions( + name=unique_name, + organization_scoped=True, # Optional parameter + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, # Optional + ) + + new_pool = client.agent_pools.create(org, create_options) + print(f"โœ… Created agent pool: {new_pool.name} (ID: {new_pool.id})") + + # Example 3: Read the agent pool + print("\n๐Ÿ“– Reading agent pool details...") + pool_details = client.agent_pools.read(new_pool.id) + print(f" Name: {pool_details.name}") + print(f" Organization Scoped: {pool_details.organization_scoped}") + print(f" Policy: {pool_details.allowed_workspace_policy}") + print(f" Agent Count: {pool_details.agent_count}") + + # Example 4: Update the agent pool + print("\nโœ๏ธ Updating agent pool...") + update_options = AgentPoolUpdateOptions( + name=f"{unique_name}-updated", + organization_scoped=False, # Making this optional parameter different + ) + + updated_pool = client.agent_pools.update(new_pool.id, update_options) + print(f"โœ… Updated agent pool name to: {updated_pool.name}") + + # Example 5: Create an agent token + print("\n๐Ÿ”‘ Creating agent token...") + token_options = AgentTokenCreateOptions( + description="SDK example token" # Optional description + ) + + agent_token = client.agent_tokens.create(new_pool.id, token_options) + print(f"โœ… Created agent token: {agent_token.id}") + if agent_token.token: + print(f" Token (first 10 chars): {agent_token.token[:10]}...") + + # Example 6: List agent tokens + print("\n๐Ÿ“ Listing agent tokens...") + tokens = client.agent_tokens.list(new_pool.id) + + # Convert to list to get count and iterate + token_list = list(tokens) + print(f"Found {len(token_list)} tokens:") + for token in token_list: + print(f" - {token.description or 'No description'} (ID: {token.id})") + + # Example 7: Clean up - delete the token and pool + print("\n๐Ÿงน Cleaning up...") + client.agent_tokens.delete(agent_token.id) + print("โœ… Deleted agent token") + + client.agent_pools.delete(new_pool.id) + print("โœ… Deleted agent pool") + + print("\n๐ŸŽ‰ Agent pool operations completed successfully!") + return 0 + + except NotFound as e: + print(f"โŒ Resource not found: {e}") + return 1 + except Exception as e: + print(f"โŒ Error: {e}") + return 1 + + +if __name__ == "__main__": + exit(main()) diff --git a/src/tfe/client.py b/src/tfe/client.py index ab6c04c..822644c 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -2,6 +2,8 @@ from ._http import HTTPTransport from .config import TFEConfig +from .resources.agent_pools import AgentPools +from .resources.agents import Agents, AgentTokens from .resources.apply import Applies from .resources.configuration_version import ConfigurationVersions from .resources.organizations import Organizations @@ -36,6 +38,12 @@ def __init__(self, config: TFEConfig | None = None): proxies=cfg.proxies, ca_bundle=cfg.ca_bundle, ) + # Agent resources + self.agent_pools = AgentPools(self._transport) + self.agents = Agents(self._transport) + self.agent_tokens = AgentTokens(self._transport) + + # Core resources self.configuration_versions = ConfigurationVersions(self._transport) self.applies = Applies(self._transport) self.plans = Plans(self._transport) @@ -48,6 +56,7 @@ def __init__(self, config: TFEConfig | None = None): self.registry_modules = RegistryModules(self._transport) self.registry_providers = RegistryProviders(self._transport) + # State and execution resources self.state_versions = StateVersions(self._transport) self.state_version_outputs = StateVersionOutputs(self._transport) self.run_tasks = RunTasks(self._transport) diff --git a/src/tfe/models/__init__.py b/src/tfe/models/__init__.py index 865b9e9..e56676b 100644 --- a/src/tfe/models/__init__.py +++ b/src/tfe/models/__init__.py @@ -4,6 +4,25 @@ import importlib.util import os +# Re-export all agent and agent pool types +from .agent import ( + Agent, + AgentListOptions, + AgentPool, + AgentPoolAllowedWorkspacePolicy, + AgentPoolAssignToWorkspacesOptions, + AgentPoolCreateOptions, + AgentPoolListOptions, + AgentPoolReadOptions, + AgentPoolRemoveFromWorkspacesOptions, + AgentPoolUpdateOptions, + AgentReadOptions, + AgentStatus, + AgentToken, + AgentTokenCreateOptions, + AgentTokenListOptions, +) + # Re-export all configuration version types from .configuration_version_types import ( ConfigurationSource, @@ -65,6 +84,22 @@ # Define what should be available when importing with * __all__ = [ + # Agent and agent pool types + "Agent", + "AgentPool", + "AgentPoolAllowedWorkspacePolicy", + "AgentPoolAssignToWorkspacesOptions", + "AgentPoolCreateOptions", + "AgentPoolListOptions", + "AgentPoolReadOptions", + "AgentPoolRemoveFromWorkspacesOptions", + "AgentPoolUpdateOptions", + "AgentStatus", + "AgentListOptions", + "AgentReadOptions", + "AgentToken", + "AgentTokenCreateOptions", + "AgentTokenListOptions", # Configuration version types "ConfigurationSource", "ConfigurationStatus", diff --git a/src/tfe/models/agent.py b/src/tfe/models/agent.py new file mode 100644 index 0000000..48c3055 --- /dev/null +++ b/src/tfe/models/agent.py @@ -0,0 +1,168 @@ +"""Agent and Agent Pool models for the Python TFE SDK. + +This module contains Pydantic models for Terraform Enterprise/Cloud agents and agent pools, +including all necessary option classes for CRUD operations. +""" + +from __future__ import annotations + +from datetime import datetime +from enum import Enum +from typing import Any + +from pydantic import BaseModel, Field + + +class AgentStatus(str, Enum): + """Agent status enumeration.""" + + IDLE = "idle" + BUSY = "busy" + UNKNOWN = "unknown" + + +class AgentPoolAllowedWorkspacePolicy(str, Enum): + """Agent pool allowed workspace policy enumeration.""" + + ALL_WORKSPACES = "all-workspaces" + SPECIFIC_WORKSPACES = "specific-workspaces" + + +class Agent(BaseModel): + """Agent represents a Terraform Enterprise agent.""" + + id: str + name: str | None = None + status: AgentStatus | None = None + version: str | None = None + last_ping_at: datetime | None = None + ip_address: str | None = None + + # Relations + agent_pool: AgentPool | None = None + + +class AgentPool(BaseModel): + """Agent Pool represents a Terraform Enterprise agent pool.""" + + id: str + name: str | None = None + created_at: datetime | None = None + organization_scoped: bool | None = None + allowed_workspace_policy: AgentPoolAllowedWorkspacePolicy | None = None + agent_count: int = 0 + + # Relations + organization: Any | None = None # Organization type from main types + workspaces: list[Any] = Field(default_factory=list) # Workspace types + agents: list[Agent] = Field(default_factory=list) + + +# Agent Pool Options + + +class AgentPoolListOptions(BaseModel): + """Options for listing agent pools.""" + + # Pagination options + page_number: int | None = None + page_size: int | None = None + # Optional: Include related resources + include: list[str] | None = None + # Optional: Filter by allowed workspace policy + allowed_workspace_policy: AgentPoolAllowedWorkspacePolicy | None = None + + +class AgentPoolCreateOptions(BaseModel): + """Options for creating an agent pool.""" + + # Required: A name to identify the agent pool + name: str + # Optional: Whether the agent pool is organization scoped + organization_scoped: bool | None = None + # Optional: Allowed workspace policy + allowed_workspace_policy: AgentPoolAllowedWorkspacePolicy | None = None + + +class AgentPoolUpdateOptions(BaseModel): + """Options for updating an agent pool.""" + + # Optional: A name to identify the agent pool + name: str | None = None + # Optional: Whether the agent pool is organization scoped + organization_scoped: bool | None = None + # Optional: Allowed workspace policy + allowed_workspace_policy: AgentPoolAllowedWorkspacePolicy | None = None + + +class AgentPoolReadOptions(BaseModel): + """Options for reading an agent pool.""" + + # Optional: Include related resources + include: list[str] | None = None + + +# Agent Pool Workspace Assignment Options + + +class AgentPoolAssignToWorkspacesOptions(BaseModel): + """Options for assigning an agent pool to workspaces.""" + + workspace_ids: list[str] = Field(default_factory=list) + + +class AgentPoolRemoveFromWorkspacesOptions(BaseModel): + """Options for removing an agent pool from workspaces.""" + + workspace_ids: list[str] = Field(default_factory=list) + + +# Agent Options + + +class AgentListOptions(BaseModel): + """Options for listing agents.""" + + # Pagination options + page_number: int | None = None + page_size: int | None = None + # Optional: Filter by status + status: AgentStatus | None = None + + +class AgentReadOptions(BaseModel): + """Options for reading an agent.""" + + # Optional: Include related resources + include: list[str] | None = None + + +# Agent Token Options + + +class AgentTokenCreateOptions(BaseModel): + """Options for creating an agent token.""" + + # Required: A description for the token + description: str + + +class AgentToken(BaseModel): + """Agent Token represents an authentication token for agents.""" + + id: str + description: str | None = None + created_at: datetime | None = None + last_used_at: datetime | None = None + token: str | None = None # Only returned on creation + + # Relations + agent_pool: AgentPool | None = None + + +class AgentTokenListOptions(BaseModel): + """Options for listing agent tokens.""" + + # Pagination options + page_number: int | None = None + page_size: int | None = None diff --git a/src/tfe/models/agentpool.py b/src/tfe/models/agentpool.py deleted file mode 100644 index 2c1780c..0000000 --- a/src/tfe/models/agentpool.py +++ /dev/null @@ -1,7 +0,0 @@ -from __future__ import annotations - -from pydantic import BaseModel - - -class AgentPool(BaseModel): - id: str diff --git a/src/tfe/models/run_task.py b/src/tfe/models/run_task.py index afb83b3..2fde05a 100644 --- a/src/tfe/models/run_task.py +++ b/src/tfe/models/run_task.py @@ -5,7 +5,7 @@ from pydantic import BaseModel, Field from ..types import Pagination -from .agentpool import AgentPool +from .agent import AgentPool from .organization import Organization from .workspace_run_task import WorkspaceRunTask diff --git a/src/tfe/project.py b/src/tfe/project.py new file mode 100644 index 0000000..9fcfa71 --- /dev/null +++ b/src/tfe/project.py @@ -0,0 +1,83 @@ +"""Project-specific utility functions and validation.""" + +import re +from typing import Any + +from .utils import valid_string, valid_string_id + + +def valid_project_name(name: str) -> bool: + """Validate project name format""" + if not valid_string(name): + return False + # Project names can contain letters, numbers, spaces, hyphens, underscores, and periods + # Must be between 1 and 90 characters + if len(name) > 90: + return False + # Allow most printable characters except some special ones + # Based on Terraform Cloud API documentation + pattern = re.compile(r"^[a-zA-Z0-9\s._-]+$") + return bool(pattern.match(name)) + + +def valid_organization_name(org_name: str) -> bool: + """Validate organization name format""" + if not valid_string(org_name): + return False + # Organization names must be valid identifiers + return valid_string_id(org_name) + + +def validate_project_create_options( + organization: str, name: str, description: str | None = None +) -> None: + """Validate project creation parameters""" + if not valid_organization_name(organization): + raise ValueError("Organization name is required and must be valid") + + if not valid_string(name): + raise ValueError("Project name is required") + + if not valid_project_name(name): + raise ValueError("Project name contains invalid characters or is too long") + + if description is not None and not valid_string(description): + raise ValueError("Description must be a valid string") + + +def validate_project_update_options( + project_id: str, name: str | None = None, description: str | None = None +) -> None: + """Validate project update parameters""" + if not valid_string_id(project_id): + raise ValueError("Project ID is required") + + if name is not None: + if not valid_string(name): + raise ValueError("Project name cannot be empty") + if not valid_project_name(name): + raise ValueError("Project name contains invalid characters or is too long") + + if description is not None and not valid_string(description): + raise ValueError("Description must be a valid string") + + +def validate_project_list_options( + organization: str, query: str | None = None, name: str | None = None +) -> None: + """Validate project list options.""" + if not valid_organization_name(organization): + raise ValueError("Organization name is required and must be valid") + + if query and not valid_string(query): + raise ValueError("Query must be a valid string") + + if name and not valid_project_name(name): + raise ValueError("Project name must be valid") + + +def _safe_str(value: Any, default: str = "") -> str: + """Safely convert a value to string with optional default.""" + if value is None: + return default + return str(value) diff --git a/src/tfe/resources/agent_pools.py b/src/tfe/resources/agent_pools.py new file mode 100644 index 0000000..e0ff776 --- /dev/null +++ b/src/tfe/resources/agent_pools.py @@ -0,0 +1,432 @@ +"""Agent Pool resource implementation for the Python TFE SDK. + +This module provides the AgentPools service for managing Terraform Enterprise/Cloud +agent pools, including CRUD operations and workspace assignments. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any, cast + +from ..models.agent import ( + AgentPool, + AgentPoolAllowedWorkspacePolicy, + AgentPoolAssignToWorkspacesOptions, + AgentPoolCreateOptions, + AgentPoolListOptions, + AgentPoolReadOptions, + AgentPoolRemoveFromWorkspacesOptions, + AgentPoolUpdateOptions, +) +from ..utils import valid_string, valid_string_id +from ._base import _Service + + +def valid_agent_pool_name(name: str) -> bool: + """Validate agent pool name format.""" + if not valid_string(name): + return False + # Agent pool names must be between 1 and 90 characters + # and can contain letters, numbers, spaces, hyphens, and underscores + if len(name) > 90: + return False + return True + + +def validate_agent_pool_create_options(organization: str, name: str) -> None: + """Validate agent pool creation parameters.""" + if not valid_string(organization): + raise ValueError("Organization name is required and must be valid") + + if not valid_string(name): + raise ValueError("Agent pool name is required") + + if not valid_agent_pool_name(name): + raise ValueError("Agent pool name contains invalid characters or is too long") + + +def validate_agent_pool_update_options( + agent_pool_id: str, name: str | None = None +) -> None: + """Validate agent pool update parameters.""" + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + if name is not None: + if not valid_string(name): + raise ValueError("Agent pool name must be a valid string") + if not valid_agent_pool_name(name): + raise ValueError( + "Agent pool name contains invalid characters or is too long" + ) + + +def _safe_str(value: Any, default: str = "") -> str: + """Safely convert a value to string with optional default.""" + if value is None: + return default + return str(value) + + +def _safe_int(value: Any, default: int = 0) -> int: + """Safely convert a value to an integer.""" + if value is None: + return default + if isinstance(value, int): + return value + try: + return int(value) + except (ValueError, TypeError): + return default + + +def _safe_bool(value: Any) -> bool | None: + """Safely convert a value to a boolean.""" + if value is None: + return None + if isinstance(value, bool): + return value + if isinstance(value, str): + return value.lower() in ("true", "1", "yes", "on") + return bool(value) + + +def _safe_workspace_policy(value: Any) -> AgentPoolAllowedWorkspacePolicy | None: + """Safely convert a value to an AgentPoolAllowedWorkspacePolicy enum.""" + if value is None: + return None + if isinstance(value, AgentPoolAllowedWorkspacePolicy): + return value + try: + return AgentPoolAllowedWorkspacePolicy(str(value)) + except (ValueError, TypeError): + return None + + +class AgentPools(_Service): + """Agent Pools service for managing Terraform Enterprise agent pools.""" + + def list( + self, organization: str, options: AgentPoolListOptions | None = None + ) -> Iterator[AgentPool]: + """List agent pools in an organization. + + Args: + organization: Organization name + options: Optional parameters for filtering and pagination + + Returns: + Iterator of AgentPool objects + + Raises: + ValueError: If organization name is invalid + TFEError: If API request fails + """ + if not valid_string(organization): + raise ValueError("Organization name is required and must be valid") + + path = f"/api/v2/organizations/{organization}/agent-pools" + params: dict[str, str | int] = {} + + if options: + if options.page_number is not None: + params["page[number]"] = options.page_number + if options.page_size is not None: + params["page[size]"] = options.page_size + if options.include: + params["include"] = ",".join(options.include) + if options.allowed_workspace_policy: + params["filter[allowed_workspace_policy]"] = ( + options.allowed_workspace_policy.value + ) + + items_iter = self._list(path, params=params) + + for item in items_iter: + # Extract agent pool data from API response + attr = item.get("attributes", {}) or {} + relationships = item.get("relationships", {}) or {} + + # Note: organization and workspace relationships available but not currently used + + # Extract agents from relationships + agents_data = relationships.get("agents", {}).get("data", []) + agent_count = ( + len(agents_data) if agents_data else attr.get("agent-count", 0) + ) + + agent_pool_data = { + "id": _safe_str(item.get("id")), + "name": _safe_str(attr.get("name")), + "created_at": attr.get("created-at"), + "organization_scoped": attr.get("organization-scoped"), + "allowed_workspace_policy": attr.get("allowed-workspace-policy"), + "agent_count": agent_count, + } + + yield AgentPool( + id=_safe_str(agent_pool_data["id"]) or "", + name=_safe_str(agent_pool_data["name"]), + created_at=cast(Any, agent_pool_data["created_at"]), + organization_scoped=_safe_bool(agent_pool_data["organization_scoped"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), + agent_count=_safe_int(agent_pool_data["agent_count"]), + ) + + def create(self, organization: str, options: AgentPoolCreateOptions) -> AgentPool: + """Create a new agent pool in an organization. + + Args: + organization: Organization name + options: Agent pool creation options + + Returns: + Created AgentPool object + + Raises: + ValueError: If parameters are invalid + TFEError: If API request fails + """ + validate_agent_pool_create_options(organization, options.name) + + path = f"/api/v2/organizations/{organization}/agent-pools" + attributes: dict[str, Any] = {"name": options.name} + + if options.organization_scoped is not None: + attributes["organization-scoped"] = options.organization_scoped + + if options.allowed_workspace_policy is not None: + attributes["allowed-workspace-policy"] = ( + options.allowed_workspace_policy.value + ) + + payload = {"data": {"type": "agent-pools", "attributes": attributes}} + + response = self.t.request("POST", path, json_body=payload) + data = response.json()["data"] + + # Extract agent pool data from response + attr = data.get("attributes", {}) or {} + agent_pool_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "created_at": attr.get("created-at"), + "organization_scoped": attr.get("organization-scoped"), + "allowed_workspace_policy": attr.get("allowed-workspace-policy"), + "agent_count": attr.get("agent-count", 0), + } + + return AgentPool( + id=_safe_str(agent_pool_data["id"]) or "", + name=_safe_str(agent_pool_data["name"]), + created_at=cast(Any, agent_pool_data["created_at"]), + organization_scoped=_safe_bool(agent_pool_data["organization_scoped"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), + agent_count=_safe_int(agent_pool_data["agent_count"]), + ) + + def read( + self, agent_pool_id: str, options: AgentPoolReadOptions | None = None + ) -> AgentPool: + """Get a specific agent pool by ID. + + Args: + agent_pool_id: Agent pool ID + options: Optional parameters for including related resources + + Returns: + AgentPool object + + Raises: + ValueError: If agent_pool_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + path = f"/api/v2/agent-pools/{agent_pool_id}" + params: dict[str, str] = {} + + if options and options.include: + params["include"] = ",".join(options.include) + + if params: + response = self.t.request("GET", path, params=params) + else: + response = self.t.request("GET", path) + + data = response.json()["data"] + + # Extract agent pool data from response + attr = data.get("attributes", {}) or {} + relationships = data.get("relationships", {}) or {} + + # Extract agents count + agents_data = relationships.get("agents", {}).get("data", []) + agent_count = len(agents_data) if agents_data else attr.get("agent-count", 0) + + agent_pool_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "created_at": attr.get("created-at"), + "organization_scoped": attr.get("organization-scoped"), + "allowed_workspace_policy": attr.get("allowed-workspace-policy"), + "agent_count": agent_count, + } + + return AgentPool( + id=_safe_str(agent_pool_data["id"]) or "", + name=_safe_str(agent_pool_data["name"]), + created_at=cast(Any, agent_pool_data["created_at"]), + organization_scoped=_safe_bool(agent_pool_data["organization_scoped"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), + agent_count=_safe_int(agent_pool_data["agent_count"]), + ) + + def update(self, agent_pool_id: str, options: AgentPoolUpdateOptions) -> AgentPool: + """Update an agent pool's properties. + + Args: + agent_pool_id: Agent pool ID + options: Agent pool update options + + Returns: + Updated AgentPool object + + Raises: + ValueError: If parameters are invalid + TFEError: If API request fails + """ + validate_agent_pool_update_options(agent_pool_id, options.name) + + path = f"/api/v2/agent-pools/{agent_pool_id}" + attributes: dict[str, Any] = {} + + if options.name is not None: + attributes["name"] = options.name + + if options.organization_scoped is not None: + attributes["organization-scoped"] = options.organization_scoped + + if options.allowed_workspace_policy is not None: + attributes["allowed-workspace-policy"] = ( + options.allowed_workspace_policy.value + ) + + payload = { + "data": { + "type": "agent-pools", + "id": agent_pool_id, + "attributes": attributes, + } + } + + response = self.t.request("PATCH", path, json_body=payload) + data = response.json()["data"] + + # Extract agent pool data from response + attr = data.get("attributes", {}) or {} + agent_pool_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "created_at": attr.get("created-at"), + "organization_scoped": attr.get("organization-scoped"), + "allowed_workspace_policy": attr.get("allowed-workspace-policy"), + "agent_count": attr.get("agent-count", 0), + } + + return AgentPool( + id=_safe_str(agent_pool_data["id"]) or "", + name=_safe_str(agent_pool_data["name"]), + created_at=cast(Any, agent_pool_data["created_at"]), + organization_scoped=_safe_bool(agent_pool_data["organization_scoped"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), + agent_count=_safe_int(agent_pool_data["agent_count"]), + ) + + def delete(self, agent_pool_id: str) -> None: + """Delete an agent pool. + + Args: + agent_pool_id: Agent pool ID + + Raises: + ValueError: If agent_pool_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + path = f"/api/v2/agent-pools/{agent_pool_id}" + self.t.request("DELETE", path) + + def assign_to_workspaces( + self, agent_pool_id: str, options: AgentPoolAssignToWorkspacesOptions + ) -> None: + """Assign an agent pool to workspaces. + + Args: + agent_pool_id: Agent pool ID + options: Assignment options containing workspace IDs + + Raises: + ValueError: If parameters are invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + if not options.workspace_ids: + raise ValueError("At least one workspace ID is required") + + path = f"/api/v2/agent-pools/{agent_pool_id}/relationships/workspaces" + + # Create data payload with workspace references + workspace_data = [] + for workspace_id in options.workspace_ids: + if not valid_string_id(workspace_id): + raise ValueError(f"Invalid workspace ID: {workspace_id}") + workspace_data.append({"type": "workspaces", "id": workspace_id}) + + payload = {"data": workspace_data} + self.t.request("POST", path, json_body=payload) + + def remove_from_workspaces( + self, agent_pool_id: str, options: AgentPoolRemoveFromWorkspacesOptions + ) -> None: + """Remove an agent pool from workspaces. + + Args: + agent_pool_id: Agent pool ID + options: Removal options containing workspace IDs + + Raises: + ValueError: If parameters are invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + if not options.workspace_ids: + raise ValueError("At least one workspace ID is required") + + path = f"/api/v2/agent-pools/{agent_pool_id}/relationships/workspaces" + + # Create data payload with workspace references + workspace_data = [] + for workspace_id in options.workspace_ids: + if not valid_string_id(workspace_id): + raise ValueError(f"Invalid workspace ID: {workspace_id}") + workspace_data.append({"type": "workspaces", "id": workspace_id}) + + payload = {"data": workspace_data} + self.t.request("DELETE", path, json_body=payload) diff --git a/src/tfe/resources/agents.py b/src/tfe/resources/agents.py new file mode 100644 index 0000000..adc1ad7 --- /dev/null +++ b/src/tfe/resources/agents.py @@ -0,0 +1,346 @@ +"""Agent resource implementation for the Python TFE SDK. + +This module provides the Agents service for managing individual Terraform Enterprise/Cloud +agents within agent pools. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any, cast + +from ..models.agent import ( + Agent, + AgentListOptions, + AgentReadOptions, + AgentStatus, + AgentToken, + AgentTokenCreateOptions, + AgentTokenListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +def _safe_str(value: Any, default: str = "") -> str: + """Safely convert a value to string with optional default.""" + if value is None: + return default + return str(value) + + +def _safe_agent_status(value: Any) -> AgentStatus | None: + """Safely convert a value to an AgentStatus enum.""" + if value is None: + return None + if isinstance(value, AgentStatus): + return value + try: + # Convert string to AgentStatus + return AgentStatus(str(value)) + except (ValueError, TypeError): + return AgentStatus.UNKNOWN + + +class Agents(_Service): + """Agents service for managing individual Terraform Enterprise agents.""" + + def list( + self, agent_pool_id: str, options: AgentListOptions | None = None + ) -> Iterator[Agent]: + """List agents in an agent pool. + + Args: + agent_pool_id: Agent pool ID + options: Optional parameters for filtering and pagination + + Returns: + Iterator of Agent objects + + Raises: + ValueError: If agent_pool_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + path = f"/api/v2/agent-pools/{agent_pool_id}/agents" + params: dict[str, str | int] = {} + + if options: + if options.page_number is not None: + params["page[number]"] = options.page_number + if options.page_size is not None: + params["page[size]"] = options.page_size + if options.status: + params["filter[status]"] = options.status.value + + items_iter = self._list(path, params=params) + + for item in items_iter: + # Extract agent data from API response + attr = item.get("attributes", {}) or {} + + # Parse status + status_str = attr.get("status") + status = None + if status_str: + try: + status = AgentStatus(status_str) + except ValueError: + status = AgentStatus.UNKNOWN + + agent_data = { + "id": _safe_str(item.get("id")), + "name": _safe_str(attr.get("name")), + "status": status, + "version": _safe_str(attr.get("version")), + "last_ping_at": attr.get("last-ping-at"), + "ip_address": _safe_str(attr.get("ip-address")), + } + + yield Agent( + id=_safe_str(agent_data["id"]) or "", + name=agent_data["name"], + status=_safe_agent_status(agent_data["status"]), + version=agent_data["version"], + last_ping_at=cast(Any, agent_data["last_ping_at"]), + ip_address=agent_data["ip_address"], + ) + + def read(self, agent_id: str, options: AgentReadOptions | None = None) -> Agent: + """Get a specific agent by ID. + + Args: + agent_id: Agent ID + options: Optional parameters for including related resources + + Returns: + Agent object + + Raises: + ValueError: If agent_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_id): + raise ValueError("Agent ID is required and must be valid") + + path = f"/api/v2/agents/{agent_id}" + params: dict[str, str] = {} + + if options and options.include: + params["include"] = ",".join(options.include) + + if params: + response = self.t.request("GET", path, params=params) + else: + response = self.t.request("GET", path) + + data = response.json()["data"] + + # Extract agent data from response + attr = data.get("attributes", {}) or {} + + # Parse status + status_str = attr.get("status") + status = None + if status_str: + try: + status = AgentStatus(status_str) + except ValueError: + status = AgentStatus.UNKNOWN + + agent_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "status": status, + "version": _safe_str(attr.get("version")), + "last_ping_at": attr.get("last-ping-at"), + "ip_address": _safe_str(attr.get("ip-address")), + } + + return Agent( + id=_safe_str(agent_data["id"]) or "", + name=agent_data["name"], + status=_safe_agent_status(agent_data["status"]), + version=agent_data["version"], + last_ping_at=cast(Any, agent_data["last_ping_at"]), + ip_address=agent_data["ip_address"], + ) + + def delete(self, agent_id: str) -> None: + """Delete an agent. + + Args: + agent_id: Agent ID + + Raises: + ValueError: If agent_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_id): + raise ValueError("Agent ID is required and must be valid") + + path = f"/api/v2/agents/{agent_id}" + self.t.request("DELETE", path) + + +class AgentTokens(_Service): + """Agent Tokens service for managing authentication tokens for agents.""" + + def list( + self, agent_pool_id: str, options: AgentTokenListOptions | None = None + ) -> Iterator[AgentToken]: + """List agent tokens for an agent pool. + + Args: + agent_pool_id: Agent pool ID + options: Optional parameters for pagination + + Returns: + Iterator of AgentToken objects + + Raises: + ValueError: If agent_pool_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + path = f"/api/v2/agent-pools/{agent_pool_id}/authentication-tokens" + params: dict[str, str | int] = {} + + if options: + if options.page_number is not None: + params["page[number]"] = options.page_number + if options.page_size is not None: + params["page[size]"] = options.page_size + + items_iter = self._list(path, params=params) + + for item in items_iter: + # Extract token data from API response + attr = item.get("attributes", {}) or {} + + token_data = { + "id": _safe_str(item.get("id")), + "description": _safe_str(attr.get("description")), + "created_at": attr.get("created-at"), + "last_used_at": attr.get("last-used-at"), + # Token value is not returned in list operations for security + "token": None, + } + + yield AgentToken( + id=_safe_str(token_data["id"]) or "", + description=token_data["description"], + created_at=cast(Any, token_data["created_at"]), + last_used_at=cast(Any, token_data["last_used_at"]), + token=token_data["token"], + ) + + def create( + self, agent_pool_id: str, options: AgentTokenCreateOptions + ) -> AgentToken: + """Create a new agent token for an agent pool. + + Args: + agent_pool_id: Agent pool ID + options: Token creation options + + Returns: + Created AgentToken object (includes token value) + + Raises: + ValueError: If parameters are invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_pool_id): + raise ValueError("Agent pool ID is required and must be valid") + + if not options.description: + raise ValueError("Token description is required") + + path = f"/api/v2/agent-pools/{agent_pool_id}/authentication-tokens" + attributes = {"description": options.description} + + payload = {"data": {"type": "authentication-tokens", "attributes": attributes}} + + response = self.t.request("POST", path, json_body=payload) + data = response.json()["data"] + + # Extract token data from response + attr = data.get("attributes", {}) or {} + + token_data = { + "id": _safe_str(data.get("id")), + "description": _safe_str(attr.get("description")), + "created_at": attr.get("created-at"), + "last_used_at": attr.get("last-used-at"), + # Token value is only returned on creation + "token": _safe_str(attr.get("token")), + } + + return AgentToken( + id=_safe_str(token_data["id"]) or "", + description=token_data["description"], + created_at=cast(Any, token_data["created_at"]), + last_used_at=cast(Any, token_data["last_used_at"]), + token=token_data["token"], + ) + + def read(self, agent_token_id: str) -> AgentToken: + """Get a specific agent token by ID. + + Args: + agent_token_id: Agent token ID + + Returns: + AgentToken object (without token value for security) + + Raises: + ValueError: If agent_token_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_token_id): + raise ValueError("Agent token ID is required and must be valid") + + path = f"/api/v2/authentication-tokens/{agent_token_id}" + response = self.t.request("GET", path) + data = response.json()["data"] + + # Extract token data from response + attr = data.get("attributes", {}) or {} + + token_data = { + "id": _safe_str(data.get("id")), + "description": _safe_str(attr.get("description")), + "created_at": attr.get("created-at"), + "last_used_at": attr.get("last-used-at"), + # Token value is never returned in read operations for security + "token": None, + } + + return AgentToken( + id=_safe_str(token_data["id"]) or "", + description=token_data["description"], + created_at=cast(Any, token_data["created_at"]), + last_used_at=cast(Any, token_data["last_used_at"]), + token=token_data["token"], + ) + + def delete(self, agent_token_id: str) -> None: + """Delete an agent token. + + Args: + agent_token_id: Agent token ID + + Raises: + ValueError: If agent_token_id is invalid + TFEError: If API request fails + """ + if not valid_string_id(agent_token_id): + raise ValueError("Agent token ID is required and must be valid") + + path = f"/api/v2/authentication-tokens/{agent_token_id}" + self.t.request("DELETE", path) diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index e6a35dc..1bd02d5 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -78,7 +78,7 @@ def validate_project_update_options( def validate_project_list_options( organization: str, query: str | None = None, name: str | None = None ) -> None: - """Validate project list options following Go TFE patterns.""" + """Validate project list options.""" if not valid_organization_name(organization): raise ValueError("Organization name is required and must be valid") diff --git a/src/tfe/resources/run_task.py b/src/tfe/resources/run_task.py index 3fd2937..7783094 100644 --- a/src/tfe/resources/run_task.py +++ b/src/tfe/resources/run_task.py @@ -10,7 +10,7 @@ InvalidRunTaskURLError, RequiredNameError, ) -from ..models.agentpool import AgentPool +from ..models.agent import AgentPool from ..models.organization import Organization from ..models.run_task import ( GlobalRunTask, diff --git a/src/tfe/types.py b/src/tfe/types.py index bc115aa..2f0a4f9 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -465,10 +465,7 @@ class LockedByChoice(BaseModel): class WorkspaceListOptions(BaseModel): - """Options for listing workspaces. - - Matches the Go-TFE WorkspaceListOptions struct. - """ + """Options for listing workspaces.""" # Pagination options (from ListOptions) page_number: int | None = None diff --git a/tests/units/test_agent_pools.py b/tests/units/test_agent_pools.py new file mode 100644 index 0000000..f4e29ee --- /dev/null +++ b/tests/units/test_agent_pools.py @@ -0,0 +1,430 @@ +"""Unit tests for agent pool operations. + +These tests mock the TFE API responses and focus on: +1. Agent pool model validation and serialization +2. Agent pool CRUD operations +3. Agent token management +4. Request building and parameter handling +5. Response parsing and error handling + +Run with: + pytest tests/units/test_agent_pools.py -v +""" + +from unittest.mock import Mock + +import pytest + +from tfe.errors import AuthError, NotFound, ValidationError +from tfe.models.agent import ( + AgentPool, + AgentPoolAllowedWorkspacePolicy, + AgentPoolCreateOptions, + AgentPoolListOptions, + AgentPoolUpdateOptions, + AgentTokenCreateOptions, +) + + +class TestAgentPoolModels: + """Test agent pool model validation and serialization""" + + def test_agent_pool_model_basic(self): + """Test basic AgentPool model creation""" + agent_pool = AgentPool( + id="apool-123456789abcdef0", + name="test-pool", + created_at="2023-01-01T00:00:00Z", + organization_scoped=True, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, + agent_count=0, + ) + + assert agent_pool.id == "apool-123456789abcdef0" + assert agent_pool.name == "test-pool" + assert agent_pool.organization_scoped is True + assert ( + agent_pool.allowed_workspace_policy + == AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES + ) + assert agent_pool.agent_count == 0 + + def test_agent_pool_allowed_workspace_policy_enum(self): + """Test AgentPoolAllowedWorkspacePolicy enum values""" + assert AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES == "all-workspaces" + assert ( + AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES == "specific-workspaces" + ) + + agent_pool = AgentPool( + id="apool-123456789abcdef0", + name="test-pool", + created_at="2023-01-01T00:00:00Z", + organization_scoped=False, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES, + agent_count=3, + ) + + assert ( + agent_pool.allowed_workspace_policy + == AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES + ) + + def test_agent_pool_create_options(self): + """Test AgentPoolCreateOptions model""" + options = AgentPoolCreateOptions( + name="test-pool", + organization_scoped=True, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES, + ) + + assert options.name == "test-pool" + assert options.organization_scoped is True + assert ( + options.allowed_workspace_policy + == AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES + ) + + +class TestAgentPoolOperations: + """Test agent pool CRUD operations""" + + @pytest.fixture + def mock_transport(self): + """Mock HTTP transport.""" + transport = Mock() + return transport + + @pytest.fixture + def agent_pools_service(self, mock_transport): + """Create agent pools service with mocked transport.""" + from tfe.resources.agent_pools import AgentPools + + return AgentPools(mock_transport) + + def test_list_agent_pools(self, agent_pools_service, mock_transport): + """Test listing agent pools""" + mock_response = { + "data": [ + { + "id": "apool-123456789abcdef0", + "type": "agent-pools", + "attributes": { + "name": "test-pool-1", + "created-at": "2023-01-01T00:00:00Z", + "organization-scoped": True, + "allowed-workspace-policy": "all-workspaces", + "agent-count": 2, + }, + } + ] + } + + mock_transport.request.return_value.json.return_value = mock_response + + agent_pools = list(agent_pools_service.list("test-org")) + + assert len(agent_pools) == 1 + assert agent_pools[0].name == "test-pool-1" + assert agent_pools[0].agent_count == 2 + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert "organizations/test-org/agent-pools" in call_args[0][1] + + def test_list_agent_pools_with_options(self, agent_pools_service, mock_transport): + """Test listing agent pools with options""" + mock_response = {"data": []} + mock_transport.request.return_value.json.return_value = mock_response + + options = AgentPoolListOptions( + page_number=2, + page_size=10, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, + ) + + list(agent_pools_service.list("test-org", options)) + + # Verify API call includes query parameters + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + params = call_args[1]["params"] + assert params["page[number]"] == 2 + assert params["page[size]"] == 10 + assert params["filter[allowed_workspace_policy]"] == "all-workspaces" + + def test_create_agent_pool(self, agent_pools_service, mock_transport): + """Test creating an agent pool""" + mock_response = { + "data": { + "id": "apool-123456789abcdef0", + "type": "agent-pools", + "attributes": { + "name": "new-pool", + "created-at": "2023-01-01T00:00:00Z", + "organization-scoped": True, + "allowed-workspace-policy": "all-workspaces", + "agent-count": 0, + }, + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + options = AgentPoolCreateOptions( + name="new-pool", + organization_scoped=True, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, + ) + + agent_pool = agent_pools_service.create("test-org", options) + + assert agent_pool.id == "apool-123456789abcdef0" + assert agent_pool.name == "new-pool" + assert agent_pool.organization_scoped is True + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "POST" + assert "organizations/test-org/agent-pools" in call_args[0][1] + + def test_read_agent_pool(self, agent_pools_service, mock_transport): + """Test reading a specific agent pool""" + mock_response = { + "data": { + "id": "apool-123456789abcdef0", + "type": "agent-pools", + "attributes": { + "name": "existing-pool", + "created-at": "2023-01-01T00:00:00Z", + "organization-scoped": False, + "allowed-workspace-policy": "specific-workspaces", + "agent-count": 3, + }, + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + agent_pool = agent_pools_service.read("apool-123456789abcdef0") + + assert agent_pool.id == "apool-123456789abcdef0" + assert agent_pool.name == "existing-pool" + assert agent_pool.organization_scoped is False + assert agent_pool.agent_count == 3 + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert "agent-pools/apool-123456789abcdef0" in call_args[0][1] + + def test_update_agent_pool(self, agent_pools_service, mock_transport): + """Test updating an agent pool""" + mock_response = { + "data": { + "id": "apool-123456789abcdef0", + "type": "agent-pools", + "attributes": { + "name": "updated-pool", + "created-at": "2023-01-01T00:00:00Z", + "organization-scoped": False, + "allowed-workspace-policy": "specific-workspaces", + "agent-count": 1, + }, + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + options = AgentPoolUpdateOptions(name="updated-pool", organization_scoped=False) + + agent_pool = agent_pools_service.update("apool-123456789abcdef0", options) + + assert agent_pool.id == "apool-123456789abcdef0" + assert agent_pool.name == "updated-pool" + assert agent_pool.organization_scoped is False + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "PATCH" + assert "agent-pools/apool-123456789abcdef0" in call_args[0][1] + + def test_delete_agent_pool(self, agent_pools_service, mock_transport): + """Test deleting an agent pool""" + agent_pools_service.delete("apool-123456789abcdef0") + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "DELETE" + assert "agent-pools/apool-123456789abcdef0" in call_args[0][1] + + +class TestAgentTokenOperations: + """Test agent token operations""" + + @pytest.fixture + def mock_transport(self): + """Mock HTTP transport.""" + transport = Mock() + return transport + + @pytest.fixture + def agent_tokens_service(self, mock_transport): + """Create agent tokens service with mocked transport.""" + from tfe.resources.agents import AgentTokens + + return AgentTokens(mock_transport) + + def test_list_agent_tokens(self, agent_tokens_service, mock_transport): + """Test listing agent tokens""" + mock_response = { + "data": [ + { + "id": "at-123456789abcdef0", + "type": "agent-tokens", + "attributes": { + "description": "Token 1", + "created-at": "2023-01-01T00:00:00Z", + "last-used-at": "2023-01-02T00:00:00Z", + }, + } + ] + } + + mock_transport.request.return_value.json.return_value = mock_response + + tokens = list(agent_tokens_service.list("apool-123456789abcdef0")) + + assert len(tokens) == 1 + assert tokens[0].id == "at-123456789abcdef0" + assert tokens[0].description == "Token 1" + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert ( + "agent-pools/apool-123456789abcdef0/authentication-tokens" + in call_args[0][1] + ) + + def test_create_agent_token(self, agent_tokens_service, mock_transport): + """Test creating an agent token""" + mock_response = { + "data": { + "id": "at-123456789abcdef0", + "type": "agent-tokens", + "attributes": { + "description": "New token", + "created-at": "2023-01-01T00:00:00Z", + "last-used-at": None, + "token": "secret-token-value", + }, + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + options = AgentTokenCreateOptions(description="New token") + token = agent_tokens_service.create("apool-123456789abcdef0", options) + + assert token.id == "at-123456789abcdef0" + assert token.description == "New token" + assert token.token == "secret-token-value" + assert token.last_used_at is None + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "POST" + assert ( + "agent-pools/apool-123456789abcdef0/authentication-tokens" + in call_args[0][1] + ) + + def test_read_agent_token(self, agent_tokens_service, mock_transport): + """Test reading an agent token""" + mock_response = { + "data": { + "id": "at-123456789abcdef0", + "type": "agent-tokens", + "attributes": { + "description": "Existing token", + "created-at": "2023-01-01T00:00:00Z", + "last-used-at": "2023-01-02T00:00:00Z", + }, + } + } + + mock_transport.request.return_value.json.return_value = mock_response + + token = agent_tokens_service.read("at-123456789abcdef0") + + assert token.id == "at-123456789abcdef0" + assert token.description == "Existing token" + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert "authentication-tokens/at-123456789abcdef0" in call_args[0][1] + + def test_delete_agent_token(self, agent_tokens_service, mock_transport): + """Test deleting an agent token""" + agent_tokens_service.delete("at-123456789abcdef0") + + # Verify API call + mock_transport.request.assert_called_once() + call_args = mock_transport.request.call_args + assert call_args[0][0] == "DELETE" + assert "authentication-tokens/at-123456789abcdef0" in call_args[0][1] + + +class TestAgentPoolErrorHandling: + """Test error handling scenarios for agent pools""" + + @pytest.fixture + def mock_transport(self): + """Mock HTTP transport.""" + transport = Mock() + return transport + + @pytest.fixture + def agent_pools_service(self, mock_transport): + """Create agent pools service with mocked transport.""" + from tfe.resources.agent_pools import AgentPools + + return AgentPools(mock_transport) + + def test_not_found_error(self, agent_pools_service, mock_transport): + """Test handling of NotFound errors""" + mock_transport.request.side_effect = NotFound("Agent pool not found") + + with pytest.raises(NotFound): + agent_pools_service.read("nonexistent-pool") + + def test_validation_error(self, agent_pools_service, mock_transport): + """Test handling of ValidationError errors""" + mock_transport.request.side_effect = ValidationError("Invalid agent pool name") + + options = AgentPoolCreateOptions( + name="valid-name" + ) # Use valid name to avoid ValueError + + with pytest.raises(ValidationError): + agent_pools_service.create("test-org", options) + + def test_auth_error(self, agent_pools_service, mock_transport): + """Test handling of AuthError errors""" + mock_transport.request.side_effect = AuthError("Unauthorized") + + with pytest.raises(AuthError): + list(agent_pools_service.list("test-org")) + + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) diff --git a/tests/units/test_agents.py b/tests/units/test_agents.py new file mode 100644 index 0000000..446c98c --- /dev/null +++ b/tests/units/test_agents.py @@ -0,0 +1,173 @@ +"""Unit tests for individual agent operations. + +These tests mock the TFE API responses and focus on: +1. Agent model validation and serialization +2. Agent CRUD operations (list, read, delete) +3. Request building and parameter handling +4. Response parsing and error handling + +Run with: + pytest tests/units/test_agents.py -v +""" + +from unittest.mock import Mock + +import pytest + +from tfe.errors import AuthError, NotFound +from tfe.models.agent import ( + Agent, + AgentStatus, +) + + +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" + + +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", + }, + } + ] + } + + mock_transport.request.return_value.json.return_value = mock_response + + agents = list(agents_service.list("apool-123456789abcdef0")) + + assert len(agents) == 1 + assert agents[0].name == "test-agent-1" + assert agents[0].status == AgentStatus.IDLE + + # Verify API call + mock_transport.request.assert_called() + + 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() + + 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() + + +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_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") diff --git a/tests/units/test_run_task.py b/tests/units/test_run_task.py index 9497e36..8a8ce15 100644 --- a/tests/units/test_run_task.py +++ b/tests/units/test_run_task.py @@ -12,7 +12,7 @@ InvalidRunTaskURLError, RequiredNameError, ) -from tfe.models.agentpool import AgentPool +from tfe.models.agent import AgentPool from tfe.models.run_task import ( GlobalRunTaskOptions, RunTaskCreateOptions,