From 39f4247f35a7d1b5a66736eca1a32cf5cfb83055 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Wed, 24 Sep 2025 12:48:16 +0530 Subject: [PATCH 1/5] PythonTFE agent and agent_pool --- examples/agent.py | 529 ++++++++++++++++++++++++ examples/agent_pool.py | 664 +++++++++++++++++++++++++++++++ src/tfe/client.py | 9 + src/tfe/models/__init__.py | 35 ++ src/tfe/models/agent.py | 172 ++++++++ src/tfe/models/agent_pool.py | 29 ++ src/tfe/models/run_task.py | 2 +- src/tfe/resources/agent_pools.py | 427 ++++++++++++++++++++ src/tfe/resources/agents.py | 349 ++++++++++++++++ src/tfe/resources/run_task.py | 2 +- tests/units/test_agent_pools.py | 430 ++++++++++++++++++++ tests/units/test_agents.py | 173 ++++++++ tests/units/test_run_task.py | 2 +- 13 files changed, 2820 insertions(+), 3 deletions(-) create mode 100644 examples/agent.py create mode 100644 examples/agent_pool.py create mode 100644 src/tfe/models/agent.py create mode 100644 src/tfe/models/agent_pool.py create mode 100644 src/tfe/resources/agent_pools.py create mode 100644 src/tfe/resources/agents.py create mode 100644 tests/units/test_agent_pools.py create mode 100644 tests/units/test_agents.py diff --git a/examples/agent.py b/examples/agent.py new file mode 100644 index 00000000..8ae420a0 --- /dev/null +++ b/examples/agent.py @@ -0,0 +1,529 @@ +"""Comprehensive example for Individual Agent operations with the TFE Python SDK. + +This example demonstrates: +1. Listing agents within agent pools +2. Reading individual agent details +3. Deleting agents +4. Agent status monitoring +5. Error handling and best practices + +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.py +""" + +import os + +import httpx +import pytest + +from tfe.client import TFEClient +from tfe.config import TFEConfig +from tfe.models.agent import ( + AgentListOptions, + AgentPoolCreateOptions, + AgentReadOptions, + AgentStatus, +) + + +def get_token_display(client: TFEClient) -> str: + """Get a safe display version of the token for logging.""" + try: + token = client._transport.token + if token and len(token) > 10: + return f"{token[:10]}..." + return "Not set" + except Exception: + return "Error reading token" + + +@pytest.fixture(scope="session") +def integration_client(): + """Create TFE 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") + if not org: + pytest.skip("TFE_ORG environment variable is required") + + config = TFEConfig(token=token) + client = TFEClient(config) + + return client, org + + +def test_agent_authentication_and_prerequisites(integration_client): + """Test authentication and verify agent pool prerequisites.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing agent operations for organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Test 1: Basic authentication + print("1๏ธโƒฃ Testing basic API authentication...") + headers = client._transport.headers.copy() + response = httpx.get( + f"{client._transport.base}/api/v2/organizations", + headers=headers, + timeout=30, + ) + + if response.status_code == 200: + print("โœ… Organizations API accessible") + else: + print(f"โŒ Authentication failed (status: {response.status_code})") + pytest.fail("Authentication failed - cannot proceed with agent tests") + + # Test 2: Check if any agent pools exist + print("2๏ธโƒฃ Checking for existing agent pools...") + agent_pools = list(client.agent_pools.list(org)) + print(f"โœ… Found {len(agent_pools)} agent pools in organization") + + if len(agent_pools) == 0: + print("โš ๏ธ No agent pools found - creating one for agent testing...") + # Create a test agent pool + create_options = AgentPoolCreateOptions( + name="test-agent-pool-for-agents", organization_scoped=True + ) + test_pool = client.agent_pools.create(org, create_options) + print(f"โœ… Created test agent pool: {test_pool.id}") + print("โœ… Prerequisites verified - agent pool available for testing") + else: + print(f"โœ… Using existing agent pool: {agent_pools[0].id}") + print("โœ… Prerequisites verified - agent pools exist for testing") + + except Exception as e: + print(f"โŒ Prerequisites check failed: {e}") + pytest.fail(f"Prerequisites not met: {e}") + + +def test_list_agents_integration(integration_client): + """Test LIST operation - Get all agents in an agent pool.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Get an agent pool to test with + agent_pools = list(client.agent_pools.list(org)) + if not agent_pools: + print("โš ๏ธ No agent pools found - creating one...") + create_options = AgentPoolCreateOptions( + name="test-agents-list", organization_scoped=True + ) + test_pool = client.agent_pools.create(org, create_options) + agent_pool_id = test_pool.id + cleanup_pool = True + else: + agent_pool_id = agent_pools[0].id + cleanup_pool = False + + print(f"๐Ÿ“‹ Testing LIST agents in pool: {agent_pool_id}") + + # Test basic list + agents = list(client.agents.list(agent_pool_id)) + print(f"โœ… Found {len(agents)} agents in agent pool") + + # Test list with options + print("๐Ÿ“‹ Testing LIST with filtering options...") + list_options = AgentListOptions(page_size=10, status=AgentStatus.IDLE) + idle_agents = list(client.agents.list(agent_pool_id, list_options)) + print(f"โœ… Found {len(idle_agents)} idle agents") + + # Test different status filters + for status in [AgentStatus.BUSY, AgentStatus.UNKNOWN]: + status_options = AgentListOptions(status=status) + status_agents = list(client.agents.list(agent_pool_id, status_options)) + print(f"โœ… Found {len(status_agents)} {status.value} agents") + + if len(agents) == 0: + print( + "โ„น๏ธ No agents found - this is normal if no agent binaries are running" + ) + print("โ„น๏ธ To see agents, run the tfc-agent binary connected to this pool") + else: + print(f"๐ŸŽ‰ Successfully listed {len(agents)} agents") + for agent in agents[:3]: # Show first 3 agents + print( + f" - Agent: {agent.name} (ID: {agent.id}, Status: {agent.status})" + ) + + # Cleanup if we created a pool + if cleanup_pool: + print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + + except Exception as e: + print(f"โŒ List agents operation failed: {e}") + pytest.fail(f"List agents failed: {e}") + + +def test_read_agent_integration(integration_client): + """Test READ operation - Get specific agent details.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Get an agent pool and list agents + agent_pools = list(client.agent_pools.list(org)) + if not agent_pools: + print("โš ๏ธ No agent pools found - creating one...") + create_options = AgentPoolCreateOptions( + name="test-agent-read", organization_scoped=True + ) + test_pool = client.agent_pools.create(org, create_options) + agent_pool_id = test_pool.id + cleanup_pool = True + else: + agent_pool_id = agent_pools[0].id + cleanup_pool = False + + # List agents to get one to read + agents = list(client.agents.list(agent_pool_id)) + + if not agents: + print("โ„น๏ธ No agents found in pool - cannot test read operation") + print("โ„น๏ธ To test this, run tfc-agent connected to an agent pool") + print("โœ… Read test skipped (no agents available)") + else: + # Test reading the first agent + test_agent = agents[0] + print(f"๐Ÿ“– Testing READ operation for agent: {test_agent.id}") + + # Read without options + agent = client.agents.read(test_agent.id) + print(f"โœ… READ successful: {agent.name}") + print(f"โœ… Agent status: {agent.status}") + print(f"โœ… Agent version: {agent.version}") + print(f"โœ… Last ping: {agent.last_ping_at}") + print(f"โœ… IP address: {agent.ip_address}") + + # Read with options + read_options = AgentReadOptions(include=["agent-pool"]) + agent_detailed = client.agents.read(test_agent.id, read_options) + print(f"โœ… READ with options successful: {agent_detailed.name}") + + # Cleanup if we created a pool + if cleanup_pool: + print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + + except Exception as e: + print(f"โŒ Read agent operation failed: {e}") + pytest.fail(f"Read agent failed: {e}") + + +def test_delete_agent_integration(integration_client): + """Test DELETE operation - Remove an agent.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Get an agent pool and list agents + agent_pools = list(client.agent_pools.list(org)) + if not agent_pools: + print("โš ๏ธ No agent pools found - creating one...") + create_options = AgentPoolCreateOptions( + name="test-agent-delete", organization_scoped=True + ) + test_pool = client.agent_pools.create(org, create_options) + agent_pool_id = test_pool.id + cleanup_pool = True + else: + agent_pool_id = agent_pools[0].id + cleanup_pool = False + + # List agents to get one to delete + agents = list(client.agents.list(agent_pool_id)) + + if not agents: + print("โ„น๏ธ No agents found in pool - cannot test delete operation") + print("โ„น๏ธ To test this, run tfc-agent connected to an agent pool") + print("โœ… Delete test skipped (no agents available)") + else: + # Test deleting an agent (only if there are multiple or it's a test agent) + if len(agents) > 1: + test_agent = agents[-1] # Delete the last one + print(f"๐Ÿ—‘๏ธ Testing DELETE operation for agent: {test_agent.id}") + print(f"๐Ÿ—‘๏ธ Agent name: {test_agent.name}") + + # Confirm agent exists + try: + agent_before = client.agents.read(test_agent.id) + print(f"โœ… Agent confirmed to exist: {agent_before.name}") + except Exception: + print("โŒ Agent doesn't exist - cannot test delete") + return + + # Delete the agent + print(f"๐Ÿ—‘๏ธ Deleting agent: {test_agent.id}") + client.agents.delete(test_agent.id) + print("โœ… DELETE operation completed") + + # Verify deletion + try: + client.agents.read(test_agent.id) + print("โŒ Agent still exists after deletion") + except Exception: + print("โœ… Agent successfully deleted - confirmed by error on read") + + else: + print("โš ๏ธ Only one agent found - skipping delete to avoid disruption") + print("โœ… Delete test skipped (preserving single agent)") + + # Cleanup if we created a pool + if cleanup_pool: + print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + + except Exception as e: + print(f"โŒ Delete agent operation failed: {e}") + pytest.fail(f"Delete agent failed: {e}") + + +def test_agent_status_monitoring_integration(integration_client): + """Test agent status monitoring and filtering.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Get agent pools and their agents + agent_pools = list(client.agent_pools.list(org)) + + if not agent_pools: + print("โš ๏ธ No agent pools found - creating one...") + create_options = AgentPoolCreateOptions( + name="test-agent-monitoring", organization_scoped=True + ) + test_pool = client.agent_pools.create(org, create_options) + agent_pool_id = test_pool.id + cleanup_pool = True + else: + agent_pool_id = agent_pools[0].id + cleanup_pool = False + + print(f"๐Ÿ“Š Testing agent status monitoring for pool: {agent_pool_id}") + + # Get all agents + all_agents = list(client.agents.list(agent_pool_id)) + print(f"๐Ÿ“Š Total agents in pool: {len(all_agents)}") + + if len(all_agents) == 0: + print("โ„น๏ธ No agents found - status monitoring test requires running agents") + print("โœ… Status monitoring test skipped (no agents available)") + else: + # Count agents by status + status_counts = {} + for agent in all_agents: + status = agent.status or AgentStatus.UNKNOWN + status_counts[status] = status_counts.get(status, 0) + 1 + + print("๐Ÿ“Š Agent status summary:") + for status, count in status_counts.items(): + print(f" - {status.value}: {count} agents") + + # Test filtering by each status + for status in AgentStatus: + filter_options = AgentListOptions(status=status) + filtered_agents = list( + client.agents.list(agent_pool_id, filter_options) + ) + expected_count = status_counts.get(status, 0) + print( + f"โœ… Status filter '{status.value}': found {len(filtered_agents)} agents (expected {expected_count})" + ) + + # Show detailed info for first few agents + print("๐Ÿ“Š Detailed agent information:") + for i, agent in enumerate(all_agents[:3]): + print(f" Agent {i + 1}:") + print(f" - ID: {agent.id}") + print(f" - Name: {agent.name}") + print(f" - Status: {agent.status}") + print(f" - Version: {agent.version}") + print(f" - Last ping: {agent.last_ping_at}") + print(f" - IP: {agent.ip_address}") + + # Cleanup if we created a pool + if cleanup_pool: + print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + + except Exception as e: + print(f"โŒ Agent status monitoring failed: {e}") + pytest.fail(f"Agent status monitoring failed: {e}") + + +def test_agent_error_handling_integration(integration_client): + """Test error handling for agent operations.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + print("๐Ÿšซ Testing agent error handling scenarios") + + # Test 1: Read non-existent agent + print("๐Ÿšซ Testing read non-existent agent") + fake_agent_id = "agent-nonexistent123" + try: + client.agents.read(fake_agent_id) + print("โŒ Expected error for non-existent agent, but got success") + except Exception as e: + error_type = type(e).__name__ + print(f"โœ… Correctly handled error for non-existent agent: {error_type}") + + # Test 2: Delete non-existent agent + print("๐Ÿšซ Testing delete non-existent agent") + try: + client.agents.delete(fake_agent_id) + print("โŒ Expected error for deleting non-existent agent, but got success") + except Exception as e: + error_type = type(e).__name__ + print( + f"โœ… Correctly handled delete error for non-existent agent: {error_type}" + ) + + # Test 3: List agents for non-existent pool + print("๐Ÿšซ Testing list agents for non-existent pool") + fake_pool_id = "apool-nonexistent123" + try: + list(client.agents.list(fake_pool_id)) + print("โŒ Expected error for non-existent pool, but got success") + except Exception as e: + error_type = type(e).__name__ + print(f"โœ… Correctly handled error for non-existent pool: {error_type}") + + # Test 4: Invalid agent pool ID format + print("๐Ÿšซ Testing invalid agent pool ID format") + try: + list(client.agents.list("invalid-id")) + print("โŒ Expected error for invalid pool ID, but got success") + except Exception as e: + error_type = type(e).__name__ + print(f"โœ… Correctly handled error for invalid pool ID: {error_type}") + + print("โœ… All agent error handling scenarios tested successfully") + + except Exception as e: + print(f"โŒ Agent error handling test failed: {e}") + pytest.fail(f"Agent error handling failed: {e}") + + +def test_comprehensive_agent_workflow(integration_client): + """Test complete agent management workflow.""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + print("๐Ÿ”„ Starting comprehensive agent workflow") + + # Step 1: Setup - ensure we have an agent pool + agent_pools = list(client.agent_pools.list(org)) + if not agent_pools: + print("1๏ธโƒฃ SETUP: Creating agent pool for workflow...") + create_options = AgentPoolCreateOptions( + name="comprehensive-agent-workflow", organization_scoped=True + ) + test_pool = client.agent_pools.create(org, create_options) + agent_pool_id = test_pool.id + cleanup_pool = True + print(f"โœ… SETUP: Created agent pool {agent_pool_id}") + else: + agent_pool_id = agent_pools[0].id + cleanup_pool = False + print(f"โœ… SETUP: Using existing agent pool {agent_pool_id}") + + # Step 2: List all agents + print("2๏ธโƒฃ LIST: Getting all agents in pool...") + all_agents = list(client.agents.list(agent_pool_id)) + print(f"โœ… LIST: Found {len(all_agents)} agents") + + if len(all_agents) == 0: + print("โ„น๏ธ No agents found - workflow limited without running agents") + print("โ„น๏ธ To see full workflow, run tfc-agent connected to this pool") + else: + # Step 3: Read detailed agent info + print("3๏ธโƒฃ READ: Getting detailed info for first agent...") + first_agent = all_agents[0] + agent_details = client.agents.read(first_agent.id) + print( + f"โœ… READ: Agent {agent_details.name} (Status: {agent_details.status})" + ) + + # Step 4: Monitor status changes (simulated) + print("4๏ธโƒฃ MONITOR: Checking agent status...") + for status in [AgentStatus.IDLE, AgentStatus.BUSY, AgentStatus.UNKNOWN]: + status_agents = list( + client.agents.list(agent_pool_id, AgentListOptions(status=status)) + ) + print( + f"โœ… MONITOR: {len(status_agents)} agents with status '{status.value}'" + ) + + # Step 5: Agent health check + print("5๏ธโƒฃ HEALTH: Performing agent health check...") + healthy_agents = [] + for agent in all_agents: + if agent.status == AgentStatus.IDLE or agent.status == AgentStatus.BUSY: + healthy_agents.append(agent) + print( + f"โœ… HEALTH: {len(healthy_agents)}/{len(all_agents)} agents are healthy" + ) + + # Step 6: Cleanup + print("6๏ธโƒฃ CLEANUP: Workflow completed") + if cleanup_pool: + print(f"๐Ÿ—‘๏ธ Cleaning up workflow agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + + print("๐ŸŽ‰ Comprehensive agent workflow completed successfully!") + + except Exception as e: + print(f"โŒ Comprehensive agent workflow failed: {e}") + pytest.fail(f"Comprehensive agent workflow failed: {e}") + + +if __name__ == "__main__": + # Check environment variables + if not os.environ.get("TFE_TOKEN"): + print("โŒ TFE_TOKEN environment variable is required") + print("๐Ÿ’ก Set it with: export TFE_TOKEN='your-token-here'") + exit(1) + + if not os.environ.get("TFE_ORG"): + print("โŒ TFE_ORG environment variable is required") + print("๐Ÿ’ก Set it with: export TFE_ORG='your-organization-name'") + exit(1) + + print("๐Ÿงช Running individual agent integration tests directly...") + print(" For full pytest features, use: pytest examples/agent.py -v -s") + + # Simple direct execution + pytest.main([__file__, "-v", "-s"]) diff --git a/examples/agent_pool.py b/examples/agent_pool.py new file mode 100644 index 00000000..0ff524de --- /dev/null +++ b/examples/agent_pool.py @@ -0,0 +1,664 @@ +"""Comprehensive example for Agent Pool operations with the TFE Python SDK. + +This example demonstrates: +1. Agent Pool CRUD operations (Create, Read, Update, Delete) +2. Agent token creation and management +3. Workspace assignments to agent pools +4. Error handling and best practices +5. Authentication diagnostics + +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.py +""" + +import os +import uuid + +import pytest + +from tfe import TFEClient, TFEConfig +from tfe.errors import NotFound +from tfe.models.agent import ( + AgentPoolAllowedWorkspacePolicy, + AgentPoolCreateOptions, + AgentPoolListOptions, + AgentPoolReadOptions, + AgentPoolUpdateOptions, + AgentTokenCreateOptions, +) + + +def get_token_display(client) -> str: + """Get a safe display version of the token from the client.""" + auth_header = client._transport.headers.get("Authorization", "Bearer [not-set]") + token_display = ( + auth_header.replace("Bearer ", "")[:10] + if "Bearer " in auth_header + else "[not-set]" + ) + return token_display + + +@pytest.fixture +def integration_client(): + """Create a real TFE client for integration testing""" + 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 or not org: + pytest.skip("TFE_TOKEN and TFE_ORG environment variables required") + + config = TFEConfig(token=token, address=address) + client = TFEClient(config=config) + + return client, org + + +def test_authentication_and_organization_access(integration_client): + """Test basic authentication and organization access before running agent tests""" + client, org = integration_client + + print(f"๐Ÿ”ง Testing authentication for organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + print(f"๐Ÿ”ง TFE Address: {client._transport.base}") + + try: + # Test 1: Try to access organizations endpoint (basic auth test) + print("1๏ธโƒฃ Testing basic API authentication...") + import httpx + + headers = client._transport.headers.copy() + try: + response = httpx.get( + f"{client._transport.base}/api/v2/organizations", + headers=headers, + timeout=30, + ) + print(f"Response status: {response.status_code}") + if response.status_code == 200: + print("โœ… Organizations API accessible") + elif response.status_code == 401: + print("โŒ 401 Unauthorized - Token is invalid, expired, or malformed") + print("๐Ÿ’ก Solution: Generate a new API token from HCP Terraform") + elif response.status_code == 403: + print("โŒ 403 Forbidden - Token doesn't have required permissions") + else: + print(f"โŒ Unexpected status: {response.status_code}") + print(f"Response: {response.text[:500]}...") + except Exception as e: + print(f"โŒ Request failed: {e}") + return + + # Only continue if auth worked + if response.status_code != 200: + print("๐Ÿ›‘ Stopping diagnostics - basic authentication failed") + print("\n๐Ÿ”ง SOLUTIONS:") + print("1. Generate a new API token from HCP Terraform:") + print(" - Go to https://app.terraform.io/app/settings/tokens") + print(" - Click 'Create an API token'") + print(" - Copy the token and set: export TFE_TOKEN='your-new-token'") + print("2. Verify your organization name:") + print(f" - Current: {org}") + print(" - Should match your HCP Terraform organization exactly") + print("3. Check token permissions:") + print(" - Ensure token has organization-level permissions") + print(" - Team tokens may have limited access") + return + + # Test 2: Try to access the specific organization + print(f"2๏ธโƒฃ Testing access to organization '{org}'...") + try: + response = httpx.get( + f"{client._transport.base}/api/v2/organizations/{org}", + headers=headers, + timeout=30, + ) + if response.status_code == 200: + org_data = response.json().get("data", {}) + org_name = org_data.get("attributes", {}).get("name", "unknown") + print(f"โœ… Organization '{org}' accessible (name: {org_name})") + else: + print(f"โŒ Organization access failed (status: {response.status_code})") + if response.status_code == 404: + print(f"๐Ÿ’ก Organization '{org}' not found - check the name") + return + except Exception as e: + print(f"โŒ Organization test failed: {e}") + return + + # Test 3: Check organization entitlements for agents + print("3๏ธโƒฃ Testing organization entitlements...") + try: + response = httpx.get( + f"{client._transport.base}/api/v2/organizations/{org}/entitlement-set", + headers=headers, + timeout=30, + ) + if response.status_code == 200: + entitlements = response.json().get("data", {}).get("attributes", {}) + agents_enabled = entitlements.get("agents", False) + print(f"โœ… Entitlements accessible - Agents enabled: {agents_enabled}") + if not agents_enabled: + print("โš ๏ธ WARNING: Agents are not enabled for this organization!") + print("โš ๏ธ Agent functionality requires a paid HCP Terraform plan") + print("โš ๏ธ Contact your organization admin to enable agents") + else: + print(f"โŒ Entitlements check failed (status: {response.status_code})") + except Exception as e: + print(f"โŒ Entitlements test failed: {e}") + + # Test 4: Test basic agent pools endpoint access + print("4๏ธโƒฃ Testing agent pools endpoint access...") + try: + response = httpx.get( + f"{client._transport.base}/api/v2/organizations/{org}/agent-pools", + headers=headers, + timeout=30, + ) + print(f"Agent pools endpoint status: {response.status_code}") + if response.status_code == 200: + pools_data = response.json().get("data", []) + print( + f"โœ… Agent pools endpoint accessible - Found {len(pools_data)} pools" + ) + elif response.status_code == 401: + print("โŒ Unauthorized - Token may be invalid or expired") + elif response.status_code == 403: + print( + "โŒ Forbidden - Token doesn't have sufficient permissions or agents not enabled" + ) + elif response.status_code == 404: + print( + "โŒ Not Found - Organization may not exist or agents not available" + ) + else: + print(f"โŒ Unexpected status: {response.status_code}") + print(f"Response: {response.text[:200]}...") + except Exception as e: + print(f"โŒ Agent pools test failed: {e}") + + except Exception as e: + print(f"โŒ Authentication test failed: {e}") + raise + + +def test_list_agent_pools_integration(integration_client): + """Test LIST operation - Get all agent pools in organization""" + client, org = integration_client + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Test basic list + print("๐Ÿ“‹ Testing LIST operation: basic list") + agent_pools = list(client.agent_pools.list(org)) + print(f"โœ… Found {len(agent_pools)} agent pools in organization '{org}'") + + if agent_pools: + example_pool = agent_pools[0] + print(f"๐Ÿ“‹ Example agent pool: {example_pool.name} (ID: {example_pool.id})") + print( + f"๐Ÿ“‹ Created: {example_pool.created_at}, Agent count: {example_pool.agent_count}" + ) + + # Test list with options + print("๐Ÿ“‹ Testing LIST operation: with options") + options = AgentPoolListOptions( + page_size=10, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, + ) + pools_with_options = list(client.agent_pools.list(org, options)) + print(f"โœ… List with options returned {len(pools_with_options)} agent pools") + + except Exception as e: + print(f"โŒ List operation failed: {e}") + raise + + +def test_create_agent_pool_integration(integration_client): + """Test CREATE operation - Add new agent pools""" + client, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"test-pool-{unique_id}" + agent_pool_id = None + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + print(f"๐Ÿ”จ Testing CREATE operation: {test_name}") + + # Create agent pool with organization scoped policy + options = AgentPoolCreateOptions( + name=test_name, + organization_scoped=True, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, + ) + + agent_pool = client.agent_pools.create(org, options) + agent_pool_id = agent_pool.id + + print(f"โœ… CREATE successful: {agent_pool.id}") + print( + f"โœ… Agent pool details: {agent_pool.name} - Organization scoped: {agent_pool.organization_scoped}" + ) + + except Exception as e: + print(f"โŒ Create operation failed: {e}") + raise + + finally: + # Cleanup + if agent_pool_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up created agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โš ๏ธ Cleanup failed: {e}") + + +def test_read_agent_pool_integration(integration_client): + """Test READ operation - Get specific agent pool details""" + client, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"read-pool-{unique_id}" + agent_pool_id = None + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Create agent pool for read test + print(f"๐Ÿ”จ Creating agent pool for READ test: {test_name}") + create_options = AgentPoolCreateOptions( + name=test_name, + organization_scoped=False, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES, + ) + created_pool = client.agent_pools.create(org, create_options) + agent_pool_id = created_pool.id + + # Test read operation + print(f"๐Ÿ“– Testing READ operation: {agent_pool_id}") + read_options = AgentPoolReadOptions(include=["allowed-workspaces"]) + agent_pool = client.agent_pools.read(agent_pool_id, read_options) + + print(f"โœ… READ successful: {agent_pool.name}") + print(f"โœ… Agent pool created: {agent_pool.created_at}") + print(f"โœ… Workspace policy: {agent_pool.allowed_workspace_policy}") + print("โœ… READ operation completed successfully") + + except Exception as e: + print(f"โŒ Read operation failed: {e}") + raise + + finally: + if agent_pool_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up read test agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โš ๏ธ Cleanup failed: {e}") + + +def test_update_agent_pool_integration(integration_client): + """Test UPDATE operation - Modify existing agent pools""" + client, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + original_name = f"update-pool-{unique_id}" + updated_name = f"updated-pool-{unique_id}" + agent_pool_id = None + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Create agent pool for update test + print(f"๐Ÿ”จ Creating agent pool for UPDATE test: {original_name}") + create_options = AgentPoolCreateOptions( + name=original_name, + organization_scoped=True, + ) + created_pool = client.agent_pools.create(org, create_options) + agent_pool_id = created_pool.id + + # Test update name only + print("โœ๏ธ Testing UPDATE operation: name only") + update_options = AgentPoolUpdateOptions(name=updated_name) + updated_pool = client.agent_pools.update(agent_pool_id, update_options) + print(f"โœ… UPDATE name successful: {updated_pool.name}") + + # Test update organization scoped policy + print("โœ๏ธ Testing UPDATE operation: organization scoped") + update_options = AgentPoolUpdateOptions(organization_scoped=False) + updated_pool = client.agent_pools.update(agent_pool_id, update_options) + print( + f"โœ… UPDATE policy successful: organization_scoped={updated_pool.organization_scoped}" + ) + + # Test update workspace policy + print("โœ๏ธ Testing UPDATE operation: workspace policy") + update_options = AgentPoolUpdateOptions( + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES + ) + updated_pool = client.agent_pools.update(agent_pool_id, update_options) + print( + f"โœ… UPDATE workspace policy successful: {updated_pool.allowed_workspace_policy}" + ) + + except Exception as e: + print(f"โŒ Update operation failed: {e}") + raise + + finally: + if agent_pool_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up update test agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โš ๏ธ Cleanup failed: {e}") + + +def test_delete_agent_pool_integration(integration_client): + """Test DELETE operation - Remove agent pools""" + client, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"delete-pool-{unique_id}" + agent_pool_id = None + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Create agent pool for delete test + print(f"๐Ÿ”จ Creating agent pool for DELETE test: {test_name}") + create_options = AgentPoolCreateOptions(name=test_name) + created_pool = client.agent_pools.create(org, create_options) + agent_pool_id = created_pool.id + print(f"โœ… Agent pool created for deletion: {agent_pool_id}") + + # Verify agent pool exists before deletion + print("๐Ÿ“– Verifying agent pool exists before deletion") + agent_pool = client.agent_pools.read(agent_pool_id) + print(f"โœ… Agent pool confirmed to exist: {agent_pool.name}") + + # Test delete operation + print(f"๐Ÿ—‘๏ธ Testing DELETE operation: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… DELETE operation completed") + + # Verify agent pool is deleted + print("๐Ÿ“– Verifying agent pool is deleted") + try: + client.agent_pools.read(agent_pool_id) + print("โŒ Agent pool still exists after deletion") + except NotFound: + print("โœ… Agent pool successfully deleted - confirmed by 404 error") + agent_pool_id = None # Don't try to clean up again + + except Exception as e: + print(f"โŒ Delete operation failed: {e}") + raise + + +def test_agent_token_management_integration(integration_client): + """Test agent token creation and management""" + client, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + pool_name = f"token-pool-{unique_id}" + agent_pool_id = None + agent_token_id = None + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + # Create agent pool for token testing + print(f"๐Ÿ”จ Creating agent pool for token test: {pool_name}") + create_options = AgentPoolCreateOptions(name=pool_name) + agent_pool = client.agent_pools.create(org, create_options) + agent_pool_id = agent_pool.id + print(f"โœ… Agent pool created: {agent_pool_id}") + + # Test creating agent token + print("๐Ÿ”‘ Testing agent token creation") + token_options = AgentTokenCreateOptions( + description=f"Test token for {pool_name}" + ) + agent_token = client.agent_tokens.create(agent_pool_id, token_options) + agent_token_id = agent_token.id + + print(f"โœ… Agent token created: {agent_token.id}") + print(f"โœ… Token description: {agent_token.description}") + print(f"โœ… Token value available: {'Yes' if agent_token.token else 'No'}") + + # Test listing agent tokens + print("๐Ÿ“‹ Testing agent token list") + tokens = list(client.agent_tokens.list(agent_pool_id)) + print(f"โœ… Found {len(tokens)} tokens for agent pool") + + # Test reading agent token + print("๐Ÿ“– Testing agent token read") + read_token = client.agent_tokens.read(agent_token_id) + print(f"โœ… Read token: {read_token.description}") + print( + f"โœ… Token value in read: {'Yes' if read_token.token else 'No (security)'}" + ) + + # Test deleting agent token + print("๐Ÿ—‘๏ธ Testing agent token deletion") + client.agent_tokens.delete(agent_token_id) + print("โœ… Agent token deleted successfully") + agent_token_id = None + + except Exception as e: + print(f"โŒ Agent token operation failed: {e}") + raise + + finally: + # Cleanup + if agent_token_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up agent token: {agent_token_id}") + client.agent_tokens.delete(agent_token_id) + except Exception as e: + print(f"โš ๏ธ Token cleanup failed: {e}") + + if agent_pool_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up agent pool: {agent_pool_id}") + client.agent_pools.delete(agent_pool_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โš ๏ธ Pool cleanup failed: {e}") + + +def test_agent_pool_error_handling_integration(integration_client): + """Test error handling scenarios""" + client, org = integration_client + + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + print("๐Ÿšซ Testing error handling scenarios") + + # Test reading a non-existent agent pool + print("๐Ÿšซ Testing read non-existent agent pool") + fake_pool_id = "apool-nonexistent123456789" + try: + client.agent_pools.read(fake_pool_id) + print("โŒ Should have raised NotFound") + except NotFound: + print("โœ… Correctly handled error for non-existent agent pool: NotFound") + except Exception as e: + print( + f"โœ… Correctly handled error for non-existent agent pool: {type(e).__name__}" + ) + + # Test updating a non-existent agent pool + print("๐Ÿšซ Testing update non-existent agent pool") + try: + update_options = AgentPoolUpdateOptions(name="nonexistent") + client.agent_pools.update(fake_pool_id, update_options) + print("โŒ Should have raised NotFound") + except NotFound: + print("โœ… Correctly handled update error for non-existent agent pool: NotFound") + except Exception as e: + print( + f"โœ… Correctly handled update error for non-existent agent pool: {type(e).__name__}" + ) + + # Test deleting a non-existent agent pool + print("๐Ÿšซ Testing delete non-existent agent pool") + try: + client.agent_pools.delete(fake_pool_id) + print("โŒ Should have raised NotFound") + except NotFound: + print("โœ… Correctly handled delete error for non-existent agent pool: NotFound") + except Exception as e: + print( + f"โœ… Correctly handled delete error for non-existent agent pool: {type(e).__name__}" + ) + + print("โœ… All error handling scenarios tested successfully") + + +def test_comprehensive_agent_pool_workflow(integration_client): + """Test complete agent pool workflow""" + client, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"comprehensive-pool-{unique_id}" + updated_name = f"comprehensive-updated-{unique_id}" + agent_pool_id = None + agent_token_id = None + + try: + print(f"๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + + print(f"๐Ÿ”„ Starting comprehensive agent pool workflow: {test_name}") + + # 1. Create agent pool + print("1๏ธโƒฃ CREATE: Creating agent pool") + create_options = AgentPoolCreateOptions( + name=test_name, + organization_scoped=True, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, + ) + agent_pool = client.agent_pools.create(org, create_options) + agent_pool_id = agent_pool.id + print(f"โœ… CREATE: {agent_pool_id}") + + # 2. Read agent pool + print("2๏ธโƒฃ READ: Reading created agent pool") + read_pool = client.agent_pools.read(agent_pool_id) + print(f"โœ… READ: {read_pool.name}") + + # 3. Update agent pool + print("3๏ธโƒฃ UPDATE: Updating agent pool") + update_options = AgentPoolUpdateOptions( + name=updated_name, + organization_scoped=False, + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES, + ) + updated_pool = client.agent_pools.update(agent_pool_id, update_options) + print(f"โœ… UPDATE: {updated_pool.name}") + + # 4. Create agent token + print("4๏ธโƒฃ TOKEN: Creating agent token") + token_options = AgentTokenCreateOptions(description=f"Token for {updated_name}") + token = client.agent_tokens.create(agent_pool_id, token_options) + agent_token_id = token.id + print(f"โœ… TOKEN: Created with description '{token.description}'") + + # 5. List agent pools + print("5๏ธโƒฃ LIST: Verifying agent pool appears in list") + pools = list(client.agent_pools.list(org)) + pool_ids = [pool.id for pool in pools] + if agent_pool_id in pool_ids: + print("โœ… LIST: Found updated agent pool in list") + else: + print("โš ๏ธ LIST: Agent pool not found in list") + + # 6. Clean up token + print("6๏ธโƒฃ TOKEN_DELETE: Deleting agent token") + client.agent_tokens.delete(agent_token_id) + print("โœ… TOKEN_DELETE: Token deleted") + agent_token_id = None + + # 7. Delete agent pool + print("7๏ธโƒฃ DELETE: Deleting agent pool") + client.agent_pools.delete(agent_pool_id) + print("โœ… DELETE: Agent pool deleted") + + # 8. Verify deletion + print("8๏ธโƒฃ VERIFY: Confirming deletion") + try: + client.agent_pools.read(agent_pool_id) + print("โŒ VERIFY: Agent pool still exists") + except NotFound: + print("โœ… VERIFY: Deletion confirmed") + agent_pool_id = None + + print("๐ŸŽ‰ Comprehensive agent pool workflow completed successfully!") + + except Exception as e: + print(f"โŒ Comprehensive workflow failed: {e}") + raise + + finally: + # Emergency cleanup + if agent_token_id: + try: + client.agent_tokens.delete(agent_token_id) + except Exception: + pass + + if agent_pool_id: + try: + client.agent_pools.delete(agent_pool_id) + except Exception: + pass + + +if __name__ == "__main__": + """ + You can also run this file directly for quick testing: + + export TFE_TOKEN="your-token" + export TFE_ORG="your-org" + python examples/agent_example.py + """ + + token = os.environ.get("TFE_TOKEN") + org = os.environ.get("TFE_ORG") + + 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'") + exit(1) + + print("๐Ÿงช Running agent pool integration tests directly...") + print(" For full pytest features, use: pytest examples/agent_pool.py -v -s") + + # Simple direct execution + pytest.main([__file__, "-v", "-s"]) diff --git a/src/tfe/client.py b/src/tfe/client.py index 38fc5e57..c6904984 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.organizations import Organizations from .resources.projects import Projects from .resources.registry_module import RegistryModules @@ -32,6 +34,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.organizations = Organizations(self._transport) self.projects = Projects(self._transport) self.variables = Variables(self._transport) @@ -41,6 +49,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 15fe5920..e26c4d35 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 registry module types from .registry_module_types import ( AgentExecutionMode, @@ -51,6 +70,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", # Registry module types "AgentExecutionMode", "Commit", diff --git a/src/tfe/models/agent.py b/src/tfe/models/agent.py new file mode 100644 index 00000000..f156c83b --- /dev/null +++ b/src/tfe/models/agent.py @@ -0,0 +1,172 @@ +"""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. + +Based on the Go TFE implementation: +https://github.com/hashicorp/go-tfe/blob/main/agent.go +https://github.com/hashicorp/go-tfe/blob/main/agent_pool.go +""" + +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/agent_pool.py b/src/tfe/models/agent_pool.py new file mode 100644 index 00000000..2378c25f --- /dev/null +++ b/src/tfe/models/agent_pool.py @@ -0,0 +1,29 @@ +"""Legacy Agent Pool model - DEPRECATED. + +This file is kept for backward compatibility. +Please use src/tfe/models/agent.py for new agent and agent pool models. +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel + +# Re-export from the new agent module + + +class AgentPool(BaseModel): + """Legacy Agent Pool model - use agent.AgentPool instead.""" + + id: str + + def __init_subclass__(cls, **kwargs: Any) -> None: + import warnings + + warnings.warn( + "AgentPool from agentpool.py is deprecated. Use agent.AgentPool instead.", + DeprecationWarning, + stacklevel=2, + ) + super().__init_subclass__(**kwargs) diff --git a/src/tfe/models/run_task.py b/src/tfe/models/run_task.py index afb83b33..7cda9448 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_pool import AgentPool from .organization import Organization from .workspace_run_task import WorkspaceRunTask diff --git a/src/tfe/resources/agent_pools.py b/src/tfe/resources/agent_pools.py new file mode 100644 index 00000000..907ac0ec --- /dev/null +++ b/src/tfe/resources/agent_pools.py @@ -0,0 +1,427 @@ +"""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. + +Based on the Go TFE implementation: +https://github.com/hashicorp/go-tfe/blob/main/agent_pool.go +""" + +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 00000000..b304f997 --- /dev/null +++ b/src/tfe/resources/agents.py @@ -0,0 +1,349 @@ +"""Agent resource implementation for the Python TFE SDK. + +This module provides the Agents service for managing individual Terraform Enterprise/Cloud +agents within agent pools. + +Based on the Go TFE implementation: +https://github.com/hashicorp/go-tfe/blob/main/agent.go +""" + +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/run_task.py b/src/tfe/resources/run_task.py index 3fd2937b..5ef26e1c 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_pool import AgentPool from ..models.organization import Organization from ..models.run_task import ( GlobalRunTask, diff --git a/tests/units/test_agent_pools.py b/tests/units/test_agent_pools.py new file mode 100644 index 00000000..f4e29ee3 --- /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 00000000..446c98c4 --- /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 9497e36a..6c84d6e0 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_pool import AgentPool from tfe.models.run_task import ( GlobalRunTaskOptions, RunTaskCreateOptions, From b654fd6b5bce95ff05869e25b882efdefb67ad51 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Wed, 24 Sep 2025 12:48:24 +0530 Subject: [PATCH 2/5] PythonTFE agent and agent_pool --- src/tfe/models/agentpool.py | 7 ------- 1 file changed, 7 deletions(-) delete mode 100644 src/tfe/models/agentpool.py diff --git a/src/tfe/models/agentpool.py b/src/tfe/models/agentpool.py deleted file mode 100644 index 2c1780c7..00000000 --- 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 From e02c0ad60e39e6a93ff35214361f92948a75eb82 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Wed, 24 Sep 2025 12:56:11 +0530 Subject: [PATCH 3/5] PythonTFE agent and agent_pool --- src/tfe/resources/agent_pools.py | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/src/tfe/resources/agent_pools.py b/src/tfe/resources/agent_pools.py index 907ac0ec..edb0d1fd 100644 --- a/src/tfe/resources/agent_pools.py +++ b/src/tfe/resources/agent_pools.py @@ -91,7 +91,7 @@ def _safe_bool(value: Any) -> bool | None: if isinstance(value, bool): return value if isinstance(value, str): - return value.lower() in ('true', '1', 'yes', 'on') + return value.lower() in ("true", "1", "yes", "on") return bool(value) @@ -173,7 +173,9 @@ def list( 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"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), agent_count=_safe_int(agent_pool_data["agent_count"]), ) @@ -225,7 +227,9 @@ def create(self, organization: str, options: AgentPoolCreateOptions) -> AgentPoo 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"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), agent_count=_safe_int(agent_pool_data["agent_count"]), ) @@ -283,7 +287,9 @@ def read( 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"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), agent_count=_safe_int(agent_pool_data["agent_count"]), ) @@ -344,7 +350,9 @@ def update(self, agent_pool_id: str, options: AgentPoolUpdateOptions) -> AgentPo 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"]), + allowed_workspace_policy=_safe_workspace_policy( + agent_pool_data["allowed_workspace_policy"] + ), agent_count=_safe_int(agent_pool_data["agent_count"]), ) From 532d0c10cbe7bcf39e266bd9aa0daeffec3debb8 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Fri, 26 Sep 2025 08:13:00 +0530 Subject: [PATCH 4/5] PythonTFE Agent and Agent_pool --- examples/agent.py | 574 ++++---------------------- examples/agent_pool.py | 676 ++++--------------------------- src/tfe/models/agent.py | 4 - src/tfe/models/run_task.py | 2 +- src/tfe/project.py | 83 ++++ src/tfe/resources/agent_pools.py | 3 - src/tfe/resources/agents.py | 3 - src/tfe/resources/projects.py | 2 +- src/tfe/resources/run_task.py | 2 +- src/tfe/types.py | 5 +- tests/units/test_run_task.py | 2 +- 11 files changed, 250 insertions(+), 1106 deletions(-) create mode 100644 src/tfe/project.py diff --git a/examples/agent.py b/examples/agent.py index 8ae420a0..01daf88c 100644 --- a/examples/agent.py +++ b/examples/agent.py @@ -1,11 +1,10 @@ -"""Comprehensive example for Individual Agent operations with the TFE Python SDK. +"""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. Deleting agents -4. Agent status monitoring -5. Error handling and best practices +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. @@ -18,512 +17,111 @@ Usage: export TFE_TOKEN="your-token-here" export TFE_ORG="your-organization" - python examples/agent.py + python examples/agent_simple.py """ import os -import httpx -import pytest - from tfe.client import TFEClient from tfe.config import TFEConfig -from tfe.models.agent import ( - AgentListOptions, - AgentPoolCreateOptions, - AgentReadOptions, - AgentStatus, -) - - -def get_token_display(client: TFEClient) -> str: - """Get a safe display version of the token for logging.""" - try: - token = client._transport.token - if token and len(token) > 10: - return f"{token[:10]}..." - return "Not set" - except Exception: - return "Error reading token" +from tfe.errors import NotFound +from tfe.models.agent import AgentListOptions -@pytest.fixture(scope="session") -def integration_client(): - """Create TFE client for integration testing.""" +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: - pytest.skip("TFE_TOKEN environment variable is required") - if not org: - pytest.skip("TFE_ORG environment variable is required") - - config = TFEConfig(token=token) - client = TFEClient(config) - - return client, org - - -def test_agent_authentication_and_prerequisites(integration_client): - """Test authentication and verify agent pool prerequisites.""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing agent operations for organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Test 1: Basic authentication - print("1๏ธโƒฃ Testing basic API authentication...") - headers = client._transport.headers.copy() - response = httpx.get( - f"{client._transport.base}/api/v2/organizations", - headers=headers, - timeout=30, - ) - - if response.status_code == 200: - print("โœ… Organizations API accessible") - else: - print(f"โŒ Authentication failed (status: {response.status_code})") - pytest.fail("Authentication failed - cannot proceed with agent tests") - - # Test 2: Check if any agent pools exist - print("2๏ธโƒฃ Checking for existing agent pools...") - agent_pools = list(client.agent_pools.list(org)) - print(f"โœ… Found {len(agent_pools)} agent pools in organization") - - if len(agent_pools) == 0: - print("โš ๏ธ No agent pools found - creating one for agent testing...") - # Create a test agent pool - create_options = AgentPoolCreateOptions( - name="test-agent-pool-for-agents", organization_scoped=True - ) - test_pool = client.agent_pools.create(org, create_options) - print(f"โœ… Created test agent pool: {test_pool.id}") - print("โœ… Prerequisites verified - agent pool available for testing") - else: - print(f"โœ… Using existing agent pool: {agent_pools[0].id}") - print("โœ… Prerequisites verified - agent pools exist for testing") - - except Exception as e: - print(f"โŒ Prerequisites check failed: {e}") - pytest.fail(f"Prerequisites not met: {e}") - - -def test_list_agents_integration(integration_client): - """Test LIST operation - Get all agents in an agent pool.""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Get an agent pool to test with - agent_pools = list(client.agent_pools.list(org)) - if not agent_pools: - print("โš ๏ธ No agent pools found - creating one...") - create_options = AgentPoolCreateOptions( - name="test-agents-list", organization_scoped=True - ) - test_pool = client.agent_pools.create(org, create_options) - agent_pool_id = test_pool.id - cleanup_pool = True - else: - agent_pool_id = agent_pools[0].id - cleanup_pool = False - - print(f"๐Ÿ“‹ Testing LIST agents in pool: {agent_pool_id}") - - # Test basic list - agents = list(client.agents.list(agent_pool_id)) - print(f"โœ… Found {len(agents)} agents in agent pool") - - # Test list with options - print("๐Ÿ“‹ Testing LIST with filtering options...") - list_options = AgentListOptions(page_size=10, status=AgentStatus.IDLE) - idle_agents = list(client.agents.list(agent_pool_id, list_options)) - print(f"โœ… Found {len(idle_agents)} idle agents") - - # Test different status filters - for status in [AgentStatus.BUSY, AgentStatus.UNKNOWN]: - status_options = AgentListOptions(status=status) - status_agents = list(client.agents.list(agent_pool_id, status_options)) - print(f"โœ… Found {len(status_agents)} {status.value} agents") - - if len(agents) == 0: - print( - "โ„น๏ธ No agents found - this is normal if no agent binaries are running" - ) - print("โ„น๏ธ To see agents, run the tfc-agent binary connected to this pool") - else: - print(f"๐ŸŽ‰ Successfully listed {len(agents)} agents") - for agent in agents[:3]: # Show first 3 agents - print( - f" - Agent: {agent.name} (ID: {agent.id}, Status: {agent.status})" - ) - - # Cleanup if we created a pool - if cleanup_pool: - print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - - except Exception as e: - print(f"โŒ List agents operation failed: {e}") - pytest.fail(f"List agents failed: {e}") - - -def test_read_agent_integration(integration_client): - """Test READ operation - Get specific agent details.""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Get an agent pool and list agents - agent_pools = list(client.agent_pools.list(org)) - if not agent_pools: - print("โš ๏ธ No agent pools found - creating one...") - create_options = AgentPoolCreateOptions( - name="test-agent-read", organization_scoped=True - ) - test_pool = client.agent_pools.create(org, create_options) - agent_pool_id = test_pool.id - cleanup_pool = True - else: - agent_pool_id = agent_pools[0].id - cleanup_pool = False - - # List agents to get one to read - agents = list(client.agents.list(agent_pool_id)) - - if not agents: - print("โ„น๏ธ No agents found in pool - cannot test read operation") - print("โ„น๏ธ To test this, run tfc-agent connected to an agent pool") - print("โœ… Read test skipped (no agents available)") - else: - # Test reading the first agent - test_agent = agents[0] - print(f"๐Ÿ“– Testing READ operation for agent: {test_agent.id}") - - # Read without options - agent = client.agents.read(test_agent.id) - print(f"โœ… READ successful: {agent.name}") - print(f"โœ… Agent status: {agent.status}") - print(f"โœ… Agent version: {agent.version}") - print(f"โœ… Last ping: {agent.last_ping_at}") - print(f"โœ… IP address: {agent.ip_address}") - - # Read with options - read_options = AgentReadOptions(include=["agent-pool"]) - agent_detailed = client.agents.read(test_agent.id, read_options) - print(f"โœ… READ with options successful: {agent_detailed.name}") - - # Cleanup if we created a pool - if cleanup_pool: - print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") + print("โŒ TFE_TOKEN environment variable is required") + return 1 - except Exception as e: - print(f"โŒ Read agent operation failed: {e}") - pytest.fail(f"Read agent failed: {e}") + 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) -def test_delete_agent_integration(integration_client): - """Test DELETE operation - Remove an agent.""" - client, org = integration_client + print(f"๐Ÿ”— Connected to: {address}") + print(f"๐Ÿข Organization: {org}") try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Get an agent pool and list agents - agent_pools = list(client.agent_pools.list(org)) - if not agent_pools: - print("โš ๏ธ No agent pools found - creating one...") - create_options = AgentPoolCreateOptions( - name="test-agent-delete", organization_scoped=True - ) - test_pool = client.agent_pools.create(org, create_options) - agent_pool_id = test_pool.id - cleanup_pool = True - else: - agent_pool_id = agent_pools[0].id - cleanup_pool = False - - # List agents to get one to delete - agents = list(client.agents.list(agent_pool_id)) - - if not agents: - print("โ„น๏ธ No agents found in pool - cannot test delete operation") - print("โ„น๏ธ To test this, run tfc-agent connected to an agent pool") - print("โœ… Delete test skipped (no agents available)") - else: - # Test deleting an agent (only if there are multiple or it's a test agent) - if len(agents) > 1: - test_agent = agents[-1] # Delete the last one - print(f"๐Ÿ—‘๏ธ Testing DELETE operation for agent: {test_agent.id}") - print(f"๐Ÿ—‘๏ธ Agent name: {test_agent.name}") - - # Confirm agent exists - try: - agent_before = client.agents.read(test_agent.id) - print(f"โœ… Agent confirmed to exist: {agent_before.name}") - except Exception: - print("โŒ Agent doesn't exist - cannot test delete") - return - - # Delete the agent - print(f"๐Ÿ—‘๏ธ Deleting agent: {test_agent.id}") - client.agents.delete(test_agent.id) - print("โœ… DELETE operation completed") - - # Verify deletion - try: - client.agents.read(test_agent.id) - print("โŒ Agent still exists after deletion") - except Exception: - print("โœ… Agent successfully deleted - confirmed by error on read") - + # 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("โš ๏ธ Only one agent found - skipping delete to avoid disruption") - print("โœ… Delete test skipped (preserving single agent)") - - # Cleanup if we created a pool - if cleanup_pool: - print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - - except Exception as e: - print(f"โŒ Delete agent operation failed: {e}") - pytest.fail(f"Delete agent failed: {e}") - - -def test_agent_status_monitoring_integration(integration_client): - """Test agent status monitoring and filtering.""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Get agent pools and their agents - agent_pools = list(client.agent_pools.list(org)) - - if not agent_pools: - print("โš ๏ธ No agent pools found - creating one...") - create_options = AgentPoolCreateOptions( - name="test-agent-monitoring", organization_scoped=True - ) - test_pool = client.agent_pools.create(org, create_options) - agent_pool_id = test_pool.id - cleanup_pool = True + 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: - agent_pool_id = agent_pools[0].id - cleanup_pool = False - - print(f"๐Ÿ“Š Testing agent status monitoring for pool: {agent_pool_id}") - - # Get all agents - all_agents = list(client.agents.list(agent_pool_id)) - print(f"๐Ÿ“Š Total agents in pool: {len(all_agents)}") - - if len(all_agents) == 0: - print("โ„น๏ธ No agents found - status monitoring test requires running agents") - print("โœ… Status monitoring test skipped (no agents available)") - else: - # Count agents by status - status_counts = {} - for agent in all_agents: - status = agent.status or AgentStatus.UNKNOWN - status_counts[status] = status_counts.get(status, 0) + 1 - - print("๐Ÿ“Š Agent status summary:") - for status, count in status_counts.items(): - print(f" - {status.value}: {count} agents") - - # Test filtering by each status - for status in AgentStatus: - filter_options = AgentListOptions(status=status) - filtered_agents = list( - client.agents.list(agent_pool_id, filter_options) - ) - expected_count = status_counts.get(status, 0) - print( - f"โœ… Status filter '{status.value}': found {len(filtered_agents)} agents (expected {expected_count})" - ) + print(f"\n๐Ÿ“Š Total agents found across all pools: {total_agents}") - # Show detailed info for first few agents - print("๐Ÿ“Š Detailed agent information:") - for i, agent in enumerate(all_agents[:3]): - print(f" Agent {i + 1}:") - print(f" - ID: {agent.id}") - print(f" - Name: {agent.name}") - print(f" - Status: {agent.status}") - print(f" - Version: {agent.version}") - print(f" - Last ping: {agent.last_ping_at}") - print(f" - IP: {agent.ip_address}") - - # Cleanup if we created a pool - if cleanup_pool: - print(f"๐Ÿ—‘๏ธ Cleaning up test agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") + 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"โŒ Agent status monitoring failed: {e}") - pytest.fail(f"Agent status monitoring failed: {e}") - - -def test_agent_error_handling_integration(integration_client): - """Test error handling for agent operations.""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - print("๐Ÿšซ Testing agent error handling scenarios") - - # Test 1: Read non-existent agent - print("๐Ÿšซ Testing read non-existent agent") - fake_agent_id = "agent-nonexistent123" - try: - client.agents.read(fake_agent_id) - print("โŒ Expected error for non-existent agent, but got success") - except Exception as e: - error_type = type(e).__name__ - print(f"โœ… Correctly handled error for non-existent agent: {error_type}") - - # Test 2: Delete non-existent agent - print("๐Ÿšซ Testing delete non-existent agent") - try: - client.agents.delete(fake_agent_id) - print("โŒ Expected error for deleting non-existent agent, but got success") - except Exception as e: - error_type = type(e).__name__ - print( - f"โœ… Correctly handled delete error for non-existent agent: {error_type}" - ) - - # Test 3: List agents for non-existent pool - print("๐Ÿšซ Testing list agents for non-existent pool") - fake_pool_id = "apool-nonexistent123" - try: - list(client.agents.list(fake_pool_id)) - print("โŒ Expected error for non-existent pool, but got success") - except Exception as e: - error_type = type(e).__name__ - print(f"โœ… Correctly handled error for non-existent pool: {error_type}") - - # Test 4: Invalid agent pool ID format - print("๐Ÿšซ Testing invalid agent pool ID format") - try: - list(client.agents.list("invalid-id")) - print("โŒ Expected error for invalid pool ID, but got success") - except Exception as e: - error_type = type(e).__name__ - print(f"โœ… Correctly handled error for invalid pool ID: {error_type}") - - print("โœ… All agent error handling scenarios tested successfully") - - except Exception as e: - print(f"โŒ Agent error handling test failed: {e}") - pytest.fail(f"Agent error handling failed: {e}") - - -def test_comprehensive_agent_workflow(integration_client): - """Test complete agent management workflow.""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - print("๐Ÿ”„ Starting comprehensive agent workflow") - - # Step 1: Setup - ensure we have an agent pool - agent_pools = list(client.agent_pools.list(org)) - if not agent_pools: - print("1๏ธโƒฃ SETUP: Creating agent pool for workflow...") - create_options = AgentPoolCreateOptions( - name="comprehensive-agent-workflow", organization_scoped=True - ) - test_pool = client.agent_pools.create(org, create_options) - agent_pool_id = test_pool.id - cleanup_pool = True - print(f"โœ… SETUP: Created agent pool {agent_pool_id}") - else: - agent_pool_id = agent_pools[0].id - cleanup_pool = False - print(f"โœ… SETUP: Using existing agent pool {agent_pool_id}") - - # Step 2: List all agents - print("2๏ธโƒฃ LIST: Getting all agents in pool...") - all_agents = list(client.agents.list(agent_pool_id)) - print(f"โœ… LIST: Found {len(all_agents)} agents") - - if len(all_agents) == 0: - print("โ„น๏ธ No agents found - workflow limited without running agents") - print("โ„น๏ธ To see full workflow, run tfc-agent connected to this pool") - else: - # Step 3: Read detailed agent info - print("3๏ธโƒฃ READ: Getting detailed info for first agent...") - first_agent = all_agents[0] - agent_details = client.agents.read(first_agent.id) - print( - f"โœ… READ: Agent {agent_details.name} (Status: {agent_details.status})" - ) - - # Step 4: Monitor status changes (simulated) - print("4๏ธโƒฃ MONITOR: Checking agent status...") - for status in [AgentStatus.IDLE, AgentStatus.BUSY, AgentStatus.UNKNOWN]: - status_agents = list( - client.agents.list(agent_pool_id, AgentListOptions(status=status)) - ) - print( - f"โœ… MONITOR: {len(status_agents)} agents with status '{status.value}'" - ) - - # Step 5: Agent health check - print("5๏ธโƒฃ HEALTH: Performing agent health check...") - healthy_agents = [] - for agent in all_agents: - if agent.status == AgentStatus.IDLE or agent.status == AgentStatus.BUSY: - healthy_agents.append(agent) - print( - f"โœ… HEALTH: {len(healthy_agents)}/{len(all_agents)} agents are healthy" - ) - - # Step 6: Cleanup - print("6๏ธโƒฃ CLEANUP: Workflow completed") - if cleanup_pool: - print(f"๐Ÿ—‘๏ธ Cleaning up workflow agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - - print("๐ŸŽ‰ Comprehensive agent workflow completed successfully!") - - except Exception as e: - print(f"โŒ Comprehensive agent workflow failed: {e}") - pytest.fail(f"Comprehensive agent workflow failed: {e}") + print(f"โŒ Error: {e}") + return 1 if __name__ == "__main__": - # Check environment variables - if not os.environ.get("TFE_TOKEN"): - print("โŒ TFE_TOKEN environment variable is required") - print("๐Ÿ’ก Set it with: export TFE_TOKEN='your-token-here'") - exit(1) - - if not os.environ.get("TFE_ORG"): - print("โŒ TFE_ORG environment variable is required") - print("๐Ÿ’ก Set it with: export TFE_ORG='your-organization-name'") - exit(1) - - print("๐Ÿงช Running individual agent integration tests directly...") - print(" For full pytest features, use: pytest examples/agent.py -v -s") - - # Simple direct execution - pytest.main([__file__, "-v", "-s"]) + exit(main()) diff --git a/examples/agent_pool.py b/examples/agent_pool.py index 0ff524de..d9a37dfe 100644 --- a/examples/agent_pool.py +++ b/examples/agent_pool.py @@ -1,11 +1,10 @@ -"""Comprehensive example for Agent Pool operations with the TFE Python SDK. +"""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. Workspace assignments to agent pools -4. Error handling and best practices -5. Authentication diagnostics +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 @@ -15,650 +14,127 @@ Usage: export TFE_TOKEN="your-token-here" export TFE_ORG="your-organization" - python examples/agent_pool.py + python examples/agent_pool_simple.py """ import os import uuid -import pytest - from tfe import TFEClient, TFEConfig from tfe.errors import NotFound from tfe.models.agent import ( AgentPoolAllowedWorkspacePolicy, AgentPoolCreateOptions, AgentPoolListOptions, - AgentPoolReadOptions, AgentPoolUpdateOptions, AgentTokenCreateOptions, ) -def get_token_display(client) -> str: - """Get a safe display version of the token from the client.""" - auth_header = client._transport.headers.get("Authorization", "Bearer [not-set]") - token_display = ( - auth_header.replace("Bearer ", "")[:10] - if "Bearer " in auth_header - else "[not-set]" - ) - return token_display - - -@pytest.fixture -def integration_client(): - """Create a real TFE client for integration testing""" +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 or not org: - pytest.skip("TFE_TOKEN and TFE_ORG environment variables required") + 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) - return client, org - - -def test_authentication_and_organization_access(integration_client): - """Test basic authentication and organization access before running agent tests""" - client, org = integration_client - - print(f"๐Ÿ”ง Testing authentication for organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - print(f"๐Ÿ”ง TFE Address: {client._transport.base}") - - try: - # Test 1: Try to access organizations endpoint (basic auth test) - print("1๏ธโƒฃ Testing basic API authentication...") - import httpx - - headers = client._transport.headers.copy() - try: - response = httpx.get( - f"{client._transport.base}/api/v2/organizations", - headers=headers, - timeout=30, - ) - print(f"Response status: {response.status_code}") - if response.status_code == 200: - print("โœ… Organizations API accessible") - elif response.status_code == 401: - print("โŒ 401 Unauthorized - Token is invalid, expired, or malformed") - print("๐Ÿ’ก Solution: Generate a new API token from HCP Terraform") - elif response.status_code == 403: - print("โŒ 403 Forbidden - Token doesn't have required permissions") - else: - print(f"โŒ Unexpected status: {response.status_code}") - print(f"Response: {response.text[:500]}...") - except Exception as e: - print(f"โŒ Request failed: {e}") - return - - # Only continue if auth worked - if response.status_code != 200: - print("๐Ÿ›‘ Stopping diagnostics - basic authentication failed") - print("\n๐Ÿ”ง SOLUTIONS:") - print("1. Generate a new API token from HCP Terraform:") - print(" - Go to https://app.terraform.io/app/settings/tokens") - print(" - Click 'Create an API token'") - print(" - Copy the token and set: export TFE_TOKEN='your-new-token'") - print("2. Verify your organization name:") - print(f" - Current: {org}") - print(" - Should match your HCP Terraform organization exactly") - print("3. Check token permissions:") - print(" - Ensure token has organization-level permissions") - print(" - Team tokens may have limited access") - return - - # Test 2: Try to access the specific organization - print(f"2๏ธโƒฃ Testing access to organization '{org}'...") - try: - response = httpx.get( - f"{client._transport.base}/api/v2/organizations/{org}", - headers=headers, - timeout=30, - ) - if response.status_code == 200: - org_data = response.json().get("data", {}) - org_name = org_data.get("attributes", {}).get("name", "unknown") - print(f"โœ… Organization '{org}' accessible (name: {org_name})") - else: - print(f"โŒ Organization access failed (status: {response.status_code})") - if response.status_code == 404: - print(f"๐Ÿ’ก Organization '{org}' not found - check the name") - return - except Exception as e: - print(f"โŒ Organization test failed: {e}") - return - - # Test 3: Check organization entitlements for agents - print("3๏ธโƒฃ Testing organization entitlements...") - try: - response = httpx.get( - f"{client._transport.base}/api/v2/organizations/{org}/entitlement-set", - headers=headers, - timeout=30, - ) - if response.status_code == 200: - entitlements = response.json().get("data", {}).get("attributes", {}) - agents_enabled = entitlements.get("agents", False) - print(f"โœ… Entitlements accessible - Agents enabled: {agents_enabled}") - if not agents_enabled: - print("โš ๏ธ WARNING: Agents are not enabled for this organization!") - print("โš ๏ธ Agent functionality requires a paid HCP Terraform plan") - print("โš ๏ธ Contact your organization admin to enable agents") - else: - print(f"โŒ Entitlements check failed (status: {response.status_code})") - except Exception as e: - print(f"โŒ Entitlements test failed: {e}") - - # Test 4: Test basic agent pools endpoint access - print("4๏ธโƒฃ Testing agent pools endpoint access...") - try: - response = httpx.get( - f"{client._transport.base}/api/v2/organizations/{org}/agent-pools", - headers=headers, - timeout=30, - ) - print(f"Agent pools endpoint status: {response.status_code}") - if response.status_code == 200: - pools_data = response.json().get("data", []) - print( - f"โœ… Agent pools endpoint accessible - Found {len(pools_data)} pools" - ) - elif response.status_code == 401: - print("โŒ Unauthorized - Token may be invalid or expired") - elif response.status_code == 403: - print( - "โŒ Forbidden - Token doesn't have sufficient permissions or agents not enabled" - ) - elif response.status_code == 404: - print( - "โŒ Not Found - Organization may not exist or agents not available" - ) - else: - print(f"โŒ Unexpected status: {response.status_code}") - print(f"Response: {response.text[:200]}...") - except Exception as e: - print(f"โŒ Agent pools test failed: {e}") - - except Exception as e: - print(f"โŒ Authentication test failed: {e}") - raise - - -def test_list_agent_pools_integration(integration_client): - """Test LIST operation - Get all agent pools in organization""" - client, org = integration_client - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Test basic list - print("๐Ÿ“‹ Testing LIST operation: basic list") - agent_pools = list(client.agent_pools.list(org)) - print(f"โœ… Found {len(agent_pools)} agent pools in organization '{org}'") - - if agent_pools: - example_pool = agent_pools[0] - print(f"๐Ÿ“‹ Example agent pool: {example_pool.name} (ID: {example_pool.id})") - print( - f"๐Ÿ“‹ Created: {example_pool.created_at}, Agent count: {example_pool.agent_count}" - ) - - # Test list with options - print("๐Ÿ“‹ Testing LIST operation: with options") - options = AgentPoolListOptions( - page_size=10, - allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, - ) - pools_with_options = list(client.agent_pools.list(org, options)) - print(f"โœ… List with options returned {len(pools_with_options)} agent pools") - - except Exception as e: - print(f"โŒ List operation failed: {e}") - raise - - -def test_create_agent_pool_integration(integration_client): - """Test CREATE operation - Add new agent pools""" - client, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"test-pool-{unique_id}" - agent_pool_id = None - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - print(f"๐Ÿ”จ Testing CREATE operation: {test_name}") - - # Create agent pool with organization scoped policy - options = AgentPoolCreateOptions( - name=test_name, - organization_scoped=True, - allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, - ) - - agent_pool = client.agent_pools.create(org, options) - agent_pool_id = agent_pool.id - - print(f"โœ… CREATE successful: {agent_pool.id}") - print( - f"โœ… Agent pool details: {agent_pool.name} - Organization scoped: {agent_pool.organization_scoped}" - ) - - except Exception as e: - print(f"โŒ Create operation failed: {e}") - raise - - finally: - # Cleanup - if agent_pool_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up created agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โš ๏ธ Cleanup failed: {e}") - - -def test_read_agent_pool_integration(integration_client): - """Test READ operation - Get specific agent pool details""" - client, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"read-pool-{unique_id}" - agent_pool_id = None + print(f"๐Ÿ”— Connected to: {address}") + print(f"๐Ÿข Organization: {org}") try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") + # 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) - # Create agent pool for read test - print(f"๐Ÿ”จ Creating agent pool for READ test: {test_name}") - create_options = AgentPoolCreateOptions( - name=test_name, - organization_scoped=False, - allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES, - ) - created_pool = client.agent_pools.create(org, create_options) - agent_pool_id = created_pool.id + # 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})") - # Test read operation - print(f"๐Ÿ“– Testing READ operation: {agent_pool_id}") - read_options = AgentPoolReadOptions(include=["allowed-workspaces"]) - agent_pool = client.agent_pools.read(agent_pool_id, read_options) + # Example 2: Create a new agent pool + print("\n๐Ÿ†• Creating a new agent pool...") + unique_name = f"sdk-example-pool-{uuid.uuid4().hex[:8]}" - print(f"โœ… READ successful: {agent_pool.name}") - print(f"โœ… Agent pool created: {agent_pool.created_at}") - print(f"โœ… Workspace policy: {agent_pool.allowed_workspace_policy}") - print("โœ… READ operation completed successfully") - - except Exception as e: - print(f"โŒ Read operation failed: {e}") - raise - - finally: - if agent_pool_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up read test agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โš ๏ธ Cleanup failed: {e}") - - -def test_update_agent_pool_integration(integration_client): - """Test UPDATE operation - Modify existing agent pools""" - client, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - original_name = f"update-pool-{unique_id}" - updated_name = f"updated-pool-{unique_id}" - agent_pool_id = None - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Create agent pool for update test - print(f"๐Ÿ”จ Creating agent pool for UPDATE test: {original_name}") create_options = AgentPoolCreateOptions( - name=original_name, - organization_scoped=True, + name=unique_name, + organization_scoped=True, # Optional parameter + allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, # Optional ) - created_pool = client.agent_pools.create(org, create_options) - agent_pool_id = created_pool.id - # Test update name only - print("โœ๏ธ Testing UPDATE operation: name only") - update_options = AgentPoolUpdateOptions(name=updated_name) - updated_pool = client.agent_pools.update(agent_pool_id, update_options) - print(f"โœ… UPDATE name successful: {updated_pool.name}") + new_pool = client.agent_pools.create(org, create_options) + print(f"โœ… Created agent pool: {new_pool.name} (ID: {new_pool.id})") - # Test update organization scoped policy - print("โœ๏ธ Testing UPDATE operation: organization scoped") - update_options = AgentPoolUpdateOptions(organization_scoped=False) - updated_pool = client.agent_pools.update(agent_pool_id, update_options) - print( - f"โœ… UPDATE policy successful: organization_scoped={updated_pool.organization_scoped}" - ) + # 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}") - # Test update workspace policy - print("โœ๏ธ Testing UPDATE operation: workspace policy") + # Example 4: Update the agent pool + print("\nโœ๏ธ Updating agent pool...") update_options = AgentPoolUpdateOptions( - allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES + name=f"{unique_name}-updated", + organization_scoped=False, # Making this optional parameter different ) - updated_pool = client.agent_pools.update(agent_pool_id, update_options) - print( - f"โœ… UPDATE workspace policy successful: {updated_pool.allowed_workspace_policy}" - ) - - except Exception as e: - print(f"โŒ Update operation failed: {e}") - raise - - finally: - if agent_pool_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up update test agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โš ๏ธ Cleanup failed: {e}") + updated_pool = client.agent_pools.update(new_pool.id, update_options) + print(f"โœ… Updated agent pool name to: {updated_pool.name}") -def test_delete_agent_pool_integration(integration_client): - """Test DELETE operation - Remove agent pools""" - client, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"delete-pool-{unique_id}" - agent_pool_id = None - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Create agent pool for delete test - print(f"๐Ÿ”จ Creating agent pool for DELETE test: {test_name}") - create_options = AgentPoolCreateOptions(name=test_name) - created_pool = client.agent_pools.create(org, create_options) - agent_pool_id = created_pool.id - print(f"โœ… Agent pool created for deletion: {agent_pool_id}") - - # Verify agent pool exists before deletion - print("๐Ÿ“– Verifying agent pool exists before deletion") - agent_pool = client.agent_pools.read(agent_pool_id) - print(f"โœ… Agent pool confirmed to exist: {agent_pool.name}") - - # Test delete operation - print(f"๐Ÿ—‘๏ธ Testing DELETE operation: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… DELETE operation completed") - - # Verify agent pool is deleted - print("๐Ÿ“– Verifying agent pool is deleted") - try: - client.agent_pools.read(agent_pool_id) - print("โŒ Agent pool still exists after deletion") - except NotFound: - print("โœ… Agent pool successfully deleted - confirmed by 404 error") - agent_pool_id = None # Don't try to clean up again - - except Exception as e: - print(f"โŒ Delete operation failed: {e}") - raise - - -def test_agent_token_management_integration(integration_client): - """Test agent token creation and management""" - client, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - pool_name = f"token-pool-{unique_id}" - agent_pool_id = None - agent_token_id = None - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - # Create agent pool for token testing - print(f"๐Ÿ”จ Creating agent pool for token test: {pool_name}") - create_options = AgentPoolCreateOptions(name=pool_name) - agent_pool = client.agent_pools.create(org, create_options) - agent_pool_id = agent_pool.id - print(f"โœ… Agent pool created: {agent_pool_id}") - - # Test creating agent token - print("๐Ÿ”‘ Testing agent token creation") + # Example 5: Create an agent token + print("\n๐Ÿ”‘ Creating agent token...") token_options = AgentTokenCreateOptions( - description=f"Test token for {pool_name}" + description="SDK example token" # Optional description ) - agent_token = client.agent_tokens.create(agent_pool_id, token_options) - agent_token_id = agent_token.id - - print(f"โœ… Agent token created: {agent_token.id}") - print(f"โœ… Token description: {agent_token.description}") - print(f"โœ… Token value available: {'Yes' if agent_token.token else 'No'}") - # Test listing agent tokens - print("๐Ÿ“‹ Testing agent token list") - tokens = list(client.agent_tokens.list(agent_pool_id)) - print(f"โœ… Found {len(tokens)} tokens for agent pool") + 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]}...") - # Test reading agent token - print("๐Ÿ“– Testing agent token read") - read_token = client.agent_tokens.read(agent_token_id) - print(f"โœ… Read token: {read_token.description}") - print( - f"โœ… Token value in read: {'Yes' if read_token.token else 'No (security)'}" - ) + # Example 6: List agent tokens + print("\n๐Ÿ“ Listing agent tokens...") + tokens = client.agent_tokens.list(new_pool.id) - # Test deleting agent token - print("๐Ÿ—‘๏ธ Testing agent token deletion") - client.agent_tokens.delete(agent_token_id) - print("โœ… Agent token deleted successfully") - agent_token_id = None + # 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})") - except Exception as e: - print(f"โŒ Agent token operation failed: {e}") - raise + # Example 7: Clean up - delete the token and pool + print("\n๐Ÿงน Cleaning up...") + client.agent_tokens.delete(agent_token.id) + print("โœ… Deleted agent token") - finally: - # Cleanup - if agent_token_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up agent token: {agent_token_id}") - client.agent_tokens.delete(agent_token_id) - except Exception as e: - print(f"โš ๏ธ Token cleanup failed: {e}") + client.agent_pools.delete(new_pool.id) + print("โœ… Deleted agent pool") - if agent_pool_id: - try: - print(f"๐Ÿ—‘๏ธ Cleaning up agent pool: {agent_pool_id}") - client.agent_pools.delete(agent_pool_id) - print("โœ… Cleanup successful") - except Exception as e: - print(f"โš ๏ธ Pool cleanup failed: {e}") + print("\n๐ŸŽ‰ Agent pool operations completed successfully!") + return 0 - -def test_agent_pool_error_handling_integration(integration_client): - """Test error handling scenarios""" - client, org = integration_client - - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - print("๐Ÿšซ Testing error handling scenarios") - - # Test reading a non-existent agent pool - print("๐Ÿšซ Testing read non-existent agent pool") - fake_pool_id = "apool-nonexistent123456789" - try: - client.agent_pools.read(fake_pool_id) - print("โŒ Should have raised NotFound") - except NotFound: - print("โœ… Correctly handled error for non-existent agent pool: NotFound") + except NotFound as e: + print(f"โŒ Resource not found: {e}") + return 1 except Exception as e: - print( - f"โœ… Correctly handled error for non-existent agent pool: {type(e).__name__}" - ) - - # Test updating a non-existent agent pool - print("๐Ÿšซ Testing update non-existent agent pool") - try: - update_options = AgentPoolUpdateOptions(name="nonexistent") - client.agent_pools.update(fake_pool_id, update_options) - print("โŒ Should have raised NotFound") - except NotFound: - print("โœ… Correctly handled update error for non-existent agent pool: NotFound") - except Exception as e: - print( - f"โœ… Correctly handled update error for non-existent agent pool: {type(e).__name__}" - ) - - # Test deleting a non-existent agent pool - print("๐Ÿšซ Testing delete non-existent agent pool") - try: - client.agent_pools.delete(fake_pool_id) - print("โŒ Should have raised NotFound") - except NotFound: - print("โœ… Correctly handled delete error for non-existent agent pool: NotFound") - except Exception as e: - print( - f"โœ… Correctly handled delete error for non-existent agent pool: {type(e).__name__}" - ) - - print("โœ… All error handling scenarios tested successfully") - - -def test_comprehensive_agent_pool_workflow(integration_client): - """Test complete agent pool workflow""" - client, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"comprehensive-pool-{unique_id}" - updated_name = f"comprehensive-updated-{unique_id}" - agent_pool_id = None - agent_token_id = None - - try: - print(f"๐Ÿ”ง Testing against organization: {org}") - print(f"๐Ÿ”ง Using token: {get_token_display(client)}...") - - print(f"๐Ÿ”„ Starting comprehensive agent pool workflow: {test_name}") - - # 1. Create agent pool - print("1๏ธโƒฃ CREATE: Creating agent pool") - create_options = AgentPoolCreateOptions( - name=test_name, - organization_scoped=True, - allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.ALL_WORKSPACES, - ) - agent_pool = client.agent_pools.create(org, create_options) - agent_pool_id = agent_pool.id - print(f"โœ… CREATE: {agent_pool_id}") - - # 2. Read agent pool - print("2๏ธโƒฃ READ: Reading created agent pool") - read_pool = client.agent_pools.read(agent_pool_id) - print(f"โœ… READ: {read_pool.name}") - - # 3. Update agent pool - print("3๏ธโƒฃ UPDATE: Updating agent pool") - update_options = AgentPoolUpdateOptions( - name=updated_name, - organization_scoped=False, - allowed_workspace_policy=AgentPoolAllowedWorkspacePolicy.SPECIFIC_WORKSPACES, - ) - updated_pool = client.agent_pools.update(agent_pool_id, update_options) - print(f"โœ… UPDATE: {updated_pool.name}") - - # 4. Create agent token - print("4๏ธโƒฃ TOKEN: Creating agent token") - token_options = AgentTokenCreateOptions(description=f"Token for {updated_name}") - token = client.agent_tokens.create(agent_pool_id, token_options) - agent_token_id = token.id - print(f"โœ… TOKEN: Created with description '{token.description}'") - - # 5. List agent pools - print("5๏ธโƒฃ LIST: Verifying agent pool appears in list") - pools = list(client.agent_pools.list(org)) - pool_ids = [pool.id for pool in pools] - if agent_pool_id in pool_ids: - print("โœ… LIST: Found updated agent pool in list") - else: - print("โš ๏ธ LIST: Agent pool not found in list") - - # 6. Clean up token - print("6๏ธโƒฃ TOKEN_DELETE: Deleting agent token") - client.agent_tokens.delete(agent_token_id) - print("โœ… TOKEN_DELETE: Token deleted") - agent_token_id = None - - # 7. Delete agent pool - print("7๏ธโƒฃ DELETE: Deleting agent pool") - client.agent_pools.delete(agent_pool_id) - print("โœ… DELETE: Agent pool deleted") - - # 8. Verify deletion - print("8๏ธโƒฃ VERIFY: Confirming deletion") - try: - client.agent_pools.read(agent_pool_id) - print("โŒ VERIFY: Agent pool still exists") - except NotFound: - print("โœ… VERIFY: Deletion confirmed") - agent_pool_id = None - - print("๐ŸŽ‰ Comprehensive agent pool workflow completed successfully!") - - except Exception as e: - print(f"โŒ Comprehensive workflow failed: {e}") - raise - - finally: - # Emergency cleanup - if agent_token_id: - try: - client.agent_tokens.delete(agent_token_id) - except Exception: - pass - - if agent_pool_id: - try: - client.agent_pools.delete(agent_pool_id) - except Exception: - pass + print(f"โŒ Error: {e}") + return 1 if __name__ == "__main__": - """ - You can also run this file directly for quick testing: - - export TFE_TOKEN="your-token" - export TFE_ORG="your-org" - python examples/agent_example.py - """ - - token = os.environ.get("TFE_TOKEN") - org = os.environ.get("TFE_ORG") - - 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'") - exit(1) - - print("๐Ÿงช Running agent pool integration tests directly...") - print(" For full pytest features, use: pytest examples/agent_pool.py -v -s") - - # Simple direct execution - pytest.main([__file__, "-v", "-s"]) + exit(main()) diff --git a/src/tfe/models/agent.py b/src/tfe/models/agent.py index f156c83b..48c30554 100644 --- a/src/tfe/models/agent.py +++ b/src/tfe/models/agent.py @@ -2,10 +2,6 @@ This module contains Pydantic models for Terraform Enterprise/Cloud agents and agent pools, including all necessary option classes for CRUD operations. - -Based on the Go TFE implementation: -https://github.com/hashicorp/go-tfe/blob/main/agent.go -https://github.com/hashicorp/go-tfe/blob/main/agent_pool.go """ from __future__ import annotations diff --git a/src/tfe/models/run_task.py b/src/tfe/models/run_task.py index 7cda9448..2fde05ab 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 .agent_pool 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 00000000..9fcfa719 --- /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 index edb0d1fd..e0ff7761 100644 --- a/src/tfe/resources/agent_pools.py +++ b/src/tfe/resources/agent_pools.py @@ -2,9 +2,6 @@ This module provides the AgentPools service for managing Terraform Enterprise/Cloud agent pools, including CRUD operations and workspace assignments. - -Based on the Go TFE implementation: -https://github.com/hashicorp/go-tfe/blob/main/agent_pool.go """ from __future__ import annotations diff --git a/src/tfe/resources/agents.py b/src/tfe/resources/agents.py index b304f997..adc1ad79 100644 --- a/src/tfe/resources/agents.py +++ b/src/tfe/resources/agents.py @@ -2,9 +2,6 @@ This module provides the Agents service for managing individual Terraform Enterprise/Cloud agents within agent pools. - -Based on the Go TFE implementation: -https://github.com/hashicorp/go-tfe/blob/main/agent.go """ from __future__ import annotations diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index e6a35dc4..1bd02d57 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 5ef26e1c..7783094e 100644 --- a/src/tfe/resources/run_task.py +++ b/src/tfe/resources/run_task.py @@ -10,7 +10,7 @@ InvalidRunTaskURLError, RequiredNameError, ) -from ..models.agent_pool 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 bc115aaa..2f0a4f9f 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_run_task.py b/tests/units/test_run_task.py index 6c84d6e0..8a8ce15b 100644 --- a/tests/units/test_run_task.py +++ b/tests/units/test_run_task.py @@ -12,7 +12,7 @@ InvalidRunTaskURLError, RequiredNameError, ) -from tfe.models.agent_pool import AgentPool +from tfe.models.agent import AgentPool from tfe.models.run_task import ( GlobalRunTaskOptions, RunTaskCreateOptions, From 61c1c8321ec1fa2b291d0ed4138f0e9067671efb Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Fri, 26 Sep 2025 08:13:07 +0530 Subject: [PATCH 5/5] PythonTFE Agent and Agent_pool --- src/tfe/models/agent_pool.py | 29 ----------------------------- 1 file changed, 29 deletions(-) delete mode 100644 src/tfe/models/agent_pool.py diff --git a/src/tfe/models/agent_pool.py b/src/tfe/models/agent_pool.py deleted file mode 100644 index 2378c25f..00000000 --- a/src/tfe/models/agent_pool.py +++ /dev/null @@ -1,29 +0,0 @@ -"""Legacy Agent Pool model - DEPRECATED. - -This file is kept for backward compatibility. -Please use src/tfe/models/agent.py for new agent and agent pool models. -""" - -from __future__ import annotations - -from typing import Any - -from pydantic import BaseModel - -# Re-export from the new agent module - - -class AgentPool(BaseModel): - """Legacy Agent Pool model - use agent.AgentPool instead.""" - - id: str - - def __init_subclass__(cls, **kwargs: Any) -> None: - import warnings - - warnings.warn( - "AgentPool from agentpool.py is deprecated. Use agent.AgentPool instead.", - DeprecationWarning, - stacklevel=2, - ) - super().__init_subclass__(**kwargs)