diff --git a/CHANGELOG.md b/CHANGELOG.md index 77af2dd7..90eaf397 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,38 @@ # Unreleased +# v0.1.2 + +## Features + +### Registry Management +* Added registry provider version resource with full CRUD operations by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) +* Added create method for registry provider versions by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) +* Added list method with pagination support for registry provider versions by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) +* Added read method for fetching specific registry provider version details by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) +* Added delete method for removing registry provider versions by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) +* Added comprehensive unit tests for registry provider versions by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) + +## Breaking Change + +### Iterator Pattern Migration for List Method +* Migrated Policy Evaluation resource to use iterator pattern for list operations and renamed attribute task_stage to policy_attachable at PolicyEvaluation Model by @isivaselvan [#68](https://github.com/hashicorp/python-tfe/pull/68) +* Migrated Policy Set Outcome resource to use iterator pattern for list operations by @isivaselvan [#68](https://github.com/hashicorp/python-tfe/pull/68) +* Migrated OAuth Token resource to use iterator pattern and removed deprecated Uid attribute by @isivaselvan [#68](https://github.com/hashicorp/python-tfe/pull/68) +* Migrated Reserved Tag Key resource to use iterator pattern, removed read method, and renamed service class by @isivaselvan [#68](https://github.com/hashicorp/python-tfe/pull/68) + +### Deprecations +* Models OAuthTokenList, PolicyEvaluationList, PolicySetOutcomeList, ReservedTagKeyList were removed from models as part of initial Iterator pattern conversion of List Method. +* page_number attribute was removed at Models of OAuthTokenListOptions, PolicyEvaluationListOptions, PolicySetOutcomeListFilter and ReservedTagKeyListOptions. +* Removed deprecated Uid attribute at OauthToken Model. + +### Enhancements +* Updated query run functions with correct api endpoints, parameters and payload options for improved performance and consistency by @aayushsingh2502 [#69](https://github.com/hashicorp/python-tfe/pull/69) +* Removed ListOptions from model and improved Cancel and Force Cancel option handling by @aayushsingh2502 [#69](https://github.com/hashicorp/python-tfe/pull/69) +* Updated function naming conventions in example files for better clarity by @aayushsingh2502 [#69](https://github.com/hashicorp/python-tfe/pull/69) + +## Bug Fixes +* Fixed the issue related to the Regex pattern on string id validation for registry resource by @isivaselvan [#66](https://github.com/hashicorp/python-tfe/pull/66) + # v0.1.1 ## Features diff --git a/examples/oauth_token.py b/examples/oauth_token.py index 16b29df9..4a31bfcc 100644 --- a/examples/oauth_token.py +++ b/examples/oauth_token.py @@ -30,7 +30,7 @@ from pytfe import TFEClient, TFEConfig from pytfe.errors import NotFound -from pytfe.models import OAuthTokenListOptions, OAuthTokenUpdateOptions +from pytfe.models import OAuthTokenUpdateOptions def main(): @@ -55,34 +55,18 @@ def main(): # ===================================================== print("\n1. Testing list() function:") try: - # Test basic list without options - token_list = client.oauth_tokens.list(organization_name) - print(f"Found {len(token_list.items)} OAuth tokens") - - # Show token details - for i, token in enumerate(token_list.items[:3], 1): # Show first 3 - print(f"{i}. Token ID: {token.id}") - print(f"UID: {token.uid}") + for token in client.oauth_tokens.list(organization_name): + print(f"Token ID: {token.id}") print(f"Service Provider User: {token.service_provider_user}") print(f"Has SSH Key: {token.has_ssh_key}") print(f"Created: {token.created_at}") if token.oauth_client: print(f"OAuth Client: {token.oauth_client.id}") - # Store first token for subsequent tests - if token_list.items: - test_token_id = token_list.items[0].id - print(f"\n Using token {test_token_id} for subsequent tests") - - # Test list with options - print("\nTesting list() with pagination options:") - options = OAuthTokenListOptions(page_size=10, page_number=1) - token_list_with_options = client.oauth_tokens.list(organization_name, options) - print(f"Found {len(token_list_with_options.items)} tokens with options") - if token_list_with_options.current_page: - print(f"Current page: {token_list_with_options.current_page}") - if token_list_with_options.total_count: - print(f"Total count: {token_list_with_options.total_count}") + # Store first token for subsequent tests + if token and not test_token_id: + test_token_id = token.id + print(f"\n Using token {test_token_id} for subsequent tests \n") except NotFound: print( @@ -99,7 +83,6 @@ def main(): try: token = client.oauth_tokens.read(test_token_id) print(f"Read OAuth token: {token.id}") - print(f"UID: {token.uid}") print(f"Service Provider User: {token.service_provider_user}") print(f"Has SSH Key: {token.has_ssh_key}") print(f"Created: {token.created_at}") diff --git a/examples/policy_evaluation.py b/examples/policy_evaluation.py index fecf0c59..d7cb2fd8 100644 --- a/examples/policy_evaluation.py +++ b/examples/policy_evaluation.py @@ -26,7 +26,6 @@ def main(): required=True, help="Task stage ID to list policy evaluations for", ) - parser.add_argument("--page", type=int, default=1) parser.add_argument("--page-size", type=int, default=20) args = parser.parse_args() @@ -41,58 +40,55 @@ def main(): _print_header(f"Listing policy evaluations for task stage: {args.task_stage_id}") options = PolicyEvaluationListOptions( - page_number=args.page, page_size=args.page_size, ) try: - pe_list = client.policy_evaluations.list(args.task_stage_id, options) - - print(f"Total policy evaluations: {pe_list.total_count}") - print(f"Page {pe_list.current_page} of {pe_list.total_pages}") - print() - - if not pe_list.items: + pe_count = 0 + for pe in client.policy_evaluations.list(args.task_stage_id, options): + pe_count += 1 + print(f"- ID: {pe.id}") + print(f"Status: {pe.status}") + print(f"Policy Kind: {pe.policy_kind}") + + if pe.result_count: + print(" Result Count:") + if pe.result_count.passed is not None: + print(f"- Passed: {pe.result_count.passed}") + if pe.result_count.advisory_failed is not None: + print(f"- Advisory Failed: {pe.result_count.advisory_failed}") + if pe.result_count.mandatory_failed is not None: + print(f"- Mandatory Failed: {pe.result_count.mandatory_failed}") + if pe.result_count.errored is not None: + print(f"- Errored: {pe.result_count.errored}") + + if pe.status_timestamp: + print(" Status Timestamps:") + if pe.status_timestamp.passed_at: + print(f"- Passed At: {pe.status_timestamp.passed_at}") + if pe.status_timestamp.failed_at: + print(f"- Failed At: {pe.status_timestamp.failed_at}") + if pe.status_timestamp.running_at: + print(f"- Running At: {pe.status_timestamp.running_at}") + if pe.status_timestamp.canceled_at: + print(f"- Canceled At: {pe.status_timestamp.canceled_at}") + if pe.status_timestamp.errored_at: + print(f"- Errored At: {pe.status_timestamp.errored_at}") + + if pe.policy_attachable: + print(f"Task Stage: {pe.task_stage.id} ({pe.task_stage.type})") + + if pe.created_at: + print(f"Created At: {pe.created_at}") + if pe.updated_at: + print(f"Updated At: {pe.updated_at}") + + print() + + if pe_count == 0: print("No policy evaluations found for this task stage.") else: - for pe in pe_list.items: - print(f"- ID: {pe.id}") - print(f"Status: {pe.status}") - print(f"Policy Kind: {pe.policy_kind}") - - if pe.result_count: - print(" Result Count:") - if pe.result_count.passed is not None: - print(f"- Passed: {pe.result_count.passed}") - if pe.result_count.advisory_failed is not None: - print(f"- Advisory Failed: {pe.result_count.advisory_failed}") - if pe.result_count.mandatory_failed is not None: - print(f"- Mandatory Failed: {pe.result_count.mandatory_failed}") - if pe.result_count.errored is not None: - print(f"- Errored: {pe.result_count.errored}") - - if pe.status_timestamp: - print(" Status Timestamps:") - if pe.status_timestamp.passed_at: - print(f"- Passed At: {pe.status_timestamp.passed_at}") - if pe.status_timestamp.failed_at: - print(f"- Failed At: {pe.status_timestamp.failed_at}") - if pe.status_timestamp.running_at: - print(f"- Running At: {pe.status_timestamp.running_at}") - if pe.status_timestamp.canceled_at: - print(f"- Canceled At: {pe.status_timestamp.canceled_at}") - if pe.status_timestamp.errored_at: - print(f"- Errored At: {pe.status_timestamp.errored_at}") - - if pe.task_stage: - print(f"Task Stage: {pe.task_stage.id} ({pe.task_stage.type})") - - if pe.created_at: - print(f"Created At: {pe.created_at}") - if pe.updated_at: - print(f"Updated At: {pe.updated_at}") - - print() + print(f"\nTotal: {pe_count} policy evaluations") except Exception as e: print(f"Error listing policy evaluations: {e}") diff --git a/examples/query_run.py b/examples/query_run.py index 610caa77..66ee8042 100644 --- a/examples/query_run.py +++ b/examples/query_run.py @@ -1,413 +1,290 @@ #!/usr/bin/env python3 """ -Query Run Management Example +Query Run Individual Function Tests -This example demonstrates all available query run operations in the Python TFE SDK, -including create, read, list, logs, results, cancel, and force cancel operations. +This file provides individual test functions for each query run operation. +You can run specific functions to test individual parts of the API. + +Functions available: +- run_list() - List query runs in a workspace +- run_create() - Create a new query run +- run_read() - Read a specific query run +- run_logs() - Retrieve logs for a query run +- run_cancel() - Cancel a query run +- run_force_cancel() - Force cancel a query run Usage: - python examples/query_run.py - -Requirements: - - TFE_TOKEN environment variable set - - TFE_ADDRESS # Get logs - logs = client.query_runs.logs(query_run_id) - print(f" ✓ Retrieved execution logs ({len(logs.logs)} characters)")ironment variable set (optional, defaults to Terraform Cloud) - - An existing organization in your Terraform Cloud/Enterprise instance - -Query Run Operations Demonstrated: - 1. List query runs with various filters - 2. Create new query runs with different types - 3. Read query run details - 4. Read query run with additional options - 5. Retrieve query run logs - 6. Retrieve query run results - 7. Cancel running query runs - 8. Force cancel stuck query runs + python query_run.py + +Note: Query Runs require Terraform ~>1.14 which includes the 'terraform query' command. + These tests may fail with error status since the feature is not fully available yet. """ import os import time -from datetime import datetime from pytfe import TFEClient, TFEConfig from pytfe.models import ( - QueryRunCancelOptions, QueryRunCreateOptions, - QueryRunForceCancelOptions, QueryRunListOptions, - QueryRunReadOptions, - QueryRunStatus, - QueryRunType, + QueryRunSource, ) -def test_list_query_runs(client, organization_name): - """Test listing query runs with various options.""" - print("=== Testing Query Run List Operations ===") +def get_client_and_workspace(): + """Initialize client and get workspace ID.""" + client = TFEClient(TFEConfig.from_env()) + organization = os.getenv("TFE_ORG", "aayush-test") + workspace_name = "query-test" # Default workspace for testing - # 1. List all query runs - print("\n1. Listing All Query Runs:") - try: - query_runs = client.query_runs.list(organization_name) - print(f" ✓ Found {len(query_runs.items)} query runs") - if query_runs.items: - print(f" ✓ Latest query run: {query_runs.items[0].id}") - print(f" ✓ Status: {query_runs.items[0].status}") - print(f" ✓ Query type: {query_runs.items[0].query_type}") - except Exception as e: - print(f" ✗ Error: {e}") + # Get workspace + workspace = client.workspaces.read(workspace_name, organization=organization) + return client, workspace - # 2. List with pagination - print("\n2. Listing Query Runs with Pagination:") - try: - options = QueryRunListOptions(page_number=1, page_size=5) - query_runs = client.query_runs.list(organization_name, options) - print(f" ✓ Page 1 has {len(query_runs.items)} query runs") - print(f" ✓ Total pages: {query_runs.total_pages}") - print(f" ✓ Total count: {query_runs.total_count}") - except Exception as e: - print(f" ✗ Error: {e}") - # 3. List with filters - print("\n3. Listing Query Runs with Filters:") - try: - options = QueryRunListOptions( - query_type=QueryRunType.FILTER, - status=QueryRunStatus.COMPLETED, - page_size=10, - ) - query_runs = client.query_runs.list(organization_name, options) - print(f" ✓ Found {len(query_runs.items)} completed filter query runs") - for qr in query_runs.items[:3]: # Show first 3 - print(f" - {qr.id}: {qr.query[:50]}...") - except Exception as e: - print(f" ✗ Error: {e}") +def run_list(): + """Test 1: List query runs in a workspace.""" + print("=== Test 1: List Query Runs ===") - return query_runs.items[0] if query_runs.items else None + client, workspace = get_client_and_workspace() + try: + # Simple list + query_runs = list(client.query_runs.list(workspace.id)) + print(f"Found {len(query_runs)} query runs in workspace '{workspace.name}'") -def test_create_query_runs(client, organization_name): - """Test creating different types of query runs.""" - print("\n=== Testing Query Run Creation ===") + for i, qr in enumerate(query_runs[:5], 1): + print(f" {i}. {qr.id}") + print(f" Status: {qr.status}") + print(f" Created: {qr.created_at}") + print() - created_query_runs = [] + # List with options + options = QueryRunListOptions(page_size=5) + limited_runs = list(client.query_runs.list(workspace.id, options)) + print(f"Retrieved {len(limited_runs)} query runs (page_size=5)") - # 1. Create a filter query run - print("\n1. Creating Filter Query Run:") - try: - options = QueryRunCreateOptions( - query="SELECT id, status, created_at FROM runs WHERE status = 'completed' ORDER BY created_at DESC", - query_type=QueryRunType.FILTER, - organization_name=organization_name, - timeout_seconds=300, - max_results=100, - ) - query_run = client.query_runs.create(organization_name, options) - created_query_runs.append(query_run) - print(f" ✓ Created filter query run: {query_run.id}") - print(f" ✓ Status: {query_run.status}") - print(f" ✓ Query: {query_run.query}") - except Exception as e: - print(f" ✗ Error: {e}") + return query_runs - # 2. Create a search query run - print("\n2. Creating Search Query Run:") - try: - options = QueryRunCreateOptions( - query="SEARCH workspaces WHERE name CONTAINS 'production'", - query_type=QueryRunType.SEARCH, - organization_name=organization_name, - timeout_seconds=180, - max_results=50, - ) - query_run = client.query_runs.create(organization_name, options) - created_query_runs.append(query_run) - print(f" ✓ Created search query run: {query_run.id}") - print(f" ✓ Status: {query_run.status}") - print(f" ✓ Query type: {query_run.query_type}") except Exception as e: - print(f" ✗ Error: {e}") + print(f"Error: {e}") + return [] + + +def run_create(): + """Test 2: Create a new query run.""" + print("\n=== Test 2: Create Query Run ===") + + client, workspace = get_client_and_workspace() - # 3. Create an analytics query run - print("\n3. Creating Analytics Query Run:") try: + # Get the latest configuration version + config_versions = list(client.configuration_versions.list(workspace.id)) + if not config_versions: + print("ERROR: No configuration versions found in workspace") + return None + + config_version = config_versions[0] + print(f"Using configuration version: {config_version.id}") + + # Create query run options = QueryRunCreateOptions( - query="ANALYZE run_durations GROUP BY workspace_id ORDER BY avg_duration DESC", - query_type=QueryRunType.ANALYTICS, - organization_name=organization_name, - timeout_seconds=600, - max_results=200, - filters={"time_range": "last_30_days", "include_failed": False}, + source=QueryRunSource.API, + workspace_id=workspace.id, + configuration_version_id=config_version.id, ) - query_run = client.query_runs.create(organization_name, options) - created_query_runs.append(query_run) - print(f" ✓ Created analytics query run: {query_run.id}") - print(f" ✓ Status: {query_run.status}") - print(f" ✓ Timeout: {query_run.timeout_seconds}s") - print(f" ✓ Max results: {query_run.max_results}") + + query_run = client.query_runs.create(options) + print(f"Created query run: {query_run.id}") + print(f" Status: {query_run.status}") + print(f" Source: {query_run.source}") + print(f" Created: {query_run.created_at}") + + return query_run + except Exception as e: - print(f" ✗ Error: {e}") + print(f"Error: {e}") + return None - return created_query_runs +def run_read(query_run_id=None): + """Test 3: Read a specific query run.""" + print("\n=== Test 3: Read Query Run ===") -def test_read_query_run(client, query_run_id): - """Test reading query run details.""" - print(f"\n=== Testing Query Run Read Operations for {query_run_id} ===") + client, workspace = get_client_and_workspace() - # 1. Basic read - print("\n1. Reading Query Run Details:") try: + # If no query_run_id provided, get the first one from the list + if not query_run_id: + query_runs = list(client.query_runs.list(workspace.id)) + if not query_runs: + print("ERROR: No query runs found to read") + return None + query_run_id = query_runs[0].id + print(f"Using first query run from list: {query_run_id}") + + # Read the query run query_run = client.query_runs.read(query_run_id) - print(f" ✓ Query Run ID: {query_run.id}") - print(f" ✓ Status: {query_run.status}") - print(f" ✓ Query Type: {query_run.query_type}") - print(f" ✓ Created: {query_run.created_at}") - print(f" ✓ Updated: {query_run.updated_at}") - if query_run.results_count: - print(f" ✓ Results Count: {query_run.results_count}") - if query_run.error_message: - print(f" ✗ Error: {query_run.error_message}") - except Exception as e: - print(f" ✗ Error: {e}") - return None + print(f"Read query run: {query_run.id}") + print(f" Status: {query_run.status}") + print(f" Source: {query_run.source}") + print(f" Created: {query_run.created_at}") + + if query_run.status_timestamps: + print(" Status Timestamps:") + if query_run.status_timestamps.queued_at: + print(f" Queued: {query_run.status_timestamps.queued_at}") + if query_run.status_timestamps.running_at: + print(f" Running: {query_run.status_timestamps.running_at}") + if query_run.status_timestamps.finished_at: + print(f" Finished: {query_run.status_timestamps.finished_at}") + if query_run.status_timestamps.errored_at: + print(f" Errored: {query_run.status_timestamps.errored_at}") + + return query_run - # 2. Read with options - print("\n2. Reading Query Run with Options:") - try: - options = QueryRunReadOptions(include_results=True, include_logs=True) - query_run = client.query_runs.read_with_options(query_run_id, options) - print(" ✓ Read query run with additional data") - print(f" ✓ Status: {query_run.status}") - if query_run.logs_url: - print(f" ✓ Logs URL available: {query_run.logs_url[:50]}...") - if query_run.results_url: - print(f" ✓ Results URL available: {query_run.results_url[:50]}...") except Exception as e: - print(f" ✗ Error: {e}") + print(f"Error: {e}") + return None - return query_run +def run_logs(query_run_id=None): + """Test 4: Retrieve logs for a query run.""" + print("\n=== Test 4: Get Query Run Logs ===") -def test_query_run_logs(client, query_run_id): - """Test retrieving query run logs.""" - print(f"\n=== Testing Query Run Logs for {query_run_id} ===") + client, workspace = get_client_and_workspace() try: + # If no query_run_id provided, get the first one from the list + if not query_run_id: + query_runs = list(client.query_runs.list(workspace.id)) + if not query_runs: + print("ERROR: No query runs found to get logs") + return None + query_run_id = query_runs[0].id + print(f"Using first query run from list: {query_run_id}") + + # Get logs logs = client.query_runs.logs(query_run_id) - print(f" ✓ Retrieved logs for query run: {logs.query_run_id}") - print(f" ✓ Log level: {logs.log_level}") - if logs.timestamp: - print(f" ✓ Log timestamp: {logs.timestamp}") - - # Show first few lines of logs - log_lines = logs.logs.split("\n")[:5] - print(" ✓ Log preview:") - for line in log_lines: - if line.strip(): - print(f" {line}") - except Exception as e: - print(f" ✗ Error retrieving logs: {e}") + log_content = logs.read().decode("utf-8") + print(f"Retrieved logs for query run: {query_run_id}") + print(f" Log size: {len(log_content)} bytes") + print("\n--- Log Preview (first 500 chars) ---") + print(log_content[:500]) + if len(log_content) > 500: + print(f"\n... ({len(log_content) - 500} more characters)") + print("--- End of Log Preview ---") -def test_query_run_results(client, query_run_id): - """Test retrieving query run results.""" - print(f"\n=== Testing Query Run Results for {query_run_id} ===") + return log_content - try: - results = client.query_runs.results(query_run_id) - print(f" ✓ Retrieved results for query run: {results.query_run_id}") - print(f" ✓ Total results: {results.total_count}") - print(f" ✓ Truncated: {results.truncated}") - - # Show first few results - if results.results: - print(" ✓ Sample results:") - for i, result in enumerate(results.results[:3]): - print(f" {i + 1}. {result}") - else: - print(" ℹ No results available") except Exception as e: - print(f" ✗ Error retrieving results: {e}") + print(f"Error: {e}") + print(" Note: Logs may not be available if the query run hasn't started yet") + return None -def test_query_run_cancellation(client, query_run_id): - """Test canceling query runs.""" - print(f"\n=== Testing Query Run Cancellation for {query_run_id} ===") +def run_cancel(query_run_id=None): + """Test 5: Cancel a query run.""" + print("\n=== Test 5: Cancel Query Run ===") - # First check if the query run is in a cancelable state - try: - query_run = client.query_runs.read(query_run_id) - if query_run.status not in [QueryRunStatus.PENDING, QueryRunStatus.RUNNING]: - print( - f" ℹ Query run is {query_run.status}, creating new one for cancellation test" - ) - - # Create a new query run for cancellation test - options = QueryRunCreateOptions( - query="SELECT * FROM runs LIMIT 10000", # Large query to ensure it runs long enough - query_type=QueryRunType.FILTER, - organization_name=query_run.organization_name, - timeout_seconds=300, - ) - query_run = client.query_runs.create(query_run.organization_name, options) - query_run_id = query_run.id - print(f" ✓ Created new query run for cancellation: {query_run_id}") - except Exception as e: - print(f" ✗ Error checking query run status: {e}") - return + client, workspace = get_client_and_workspace() - # 1. Test regular cancel - print("\n1. Testing Regular Cancellation:") - try: - cancel_options = QueryRunCancelOptions( - reason="User requested cancellation for testing" - ) - canceled_query_run = client.query_runs.cancel(query_run_id, cancel_options) - print(f" ✓ Canceled query run: {canceled_query_run.id}") - print(f" ✓ New status: {canceled_query_run.status}") - except Exception as e: - print(f" ✗ Error canceling query run: {e}") - - # If regular cancel fails, try force cancel - print("\n2. Testing Force Cancellation:") - try: - force_cancel_options = QueryRunForceCancelOptions( - reason="Force cancel after regular cancel failed" - ) - force_canceled_query_run = client.query_runs.force_cancel( - query_run_id, force_cancel_options - ) - print(f" ✓ Force canceled query run: {force_canceled_query_run.id}") - print(f" ✓ New status: {force_canceled_query_run.status}") - except Exception as e: - print(f" ✗ Error force canceling query run: {e}") - - -def test_query_run_workflow(client, organization_name): - """Test a complete query run workflow.""" - print("\n=== Testing Complete Query Run Workflow ===") - - # 1. Create a query run - print("\n1. Creating Query Run:") - try: - options = QueryRunCreateOptions( - query="SELECT id, name, status FROM workspaces ORDER BY created_at DESC LIMIT 10", - query_type=QueryRunType.FILTER, - organization_name=organization_name, - timeout_seconds=120, - max_results=50, - ) - query_run = client.query_runs.create(organization_name, options) - print(f" ✓ Created: {query_run.id}") - query_run_id = query_run.id - except Exception as e: - print(f" ✗ Error creating query run: {e}") - return - - # 2. Monitor execution - print("\n2. Monitoring Execution:") - max_attempts = 30 - attempt = 0 - - while attempt < max_attempts: - try: - query_run = client.query_runs.read(query_run_id) - print(f" Attempt {attempt + 1}: Status = {query_run.status}") - - if query_run.status in [ - QueryRunStatus.COMPLETED, - QueryRunStatus.ERRORED, - QueryRunStatus.CANCELED, - ]: - break - - time.sleep(2) # Wait 2 seconds before checking again - attempt += 1 - except Exception as e: - print(f" ✗ Error monitoring query run: {e}") - break - - # 3. Get final results - print("\n3. Getting Final Results:") try: - if query_run.status == QueryRunStatus.COMPLETED: - results = client.query_runs.results(query_run_id) - print(" ✓ Query completed successfully") - print(f" ✓ Total results: {results.total_count}") - print(f" ✓ Truncated: {results.truncated}") - - # Get logs - logs = client.query_runs.logs(query_run_id) - print(f" ✓ Retrieved execution logs ({len(logs.logs)} characters)") - else: - print(f" ✗ Query run finished with status: {query_run.status}") - if query_run.error_message: - print(f" ✗ Error message: {query_run.error_message}") - except Exception as e: - print(f" ✗ Error getting final results: {e}") + # If no query_run_id provided, create a new one + if not query_run_id: + print("Creating a new query run to cancel...") + new_run = run_create() + if not new_run: + print("ERROR: Could not create query run to cancel") + return False + query_run_id = new_run.id + time.sleep(1) # Give it a moment to start + + # Cancel the query run + client.query_runs.cancel(query_run_id) + print(f"Cancel requested for query run: {query_run_id}") + + # Verify cancellation + time.sleep(2) + query_run = client.query_runs.read(query_run_id) + print(f" Status after cancel: {query_run.status}") - return query_run_id + return True + except Exception as e: + print(f"Error: {e}") + print(" Note: Query run may not be in a cancelable state") + return False -def main(): - """Main function to demonstrate query run operations.""" - # Get configuration from environment - token = os.environ.get("TFE_TOKEN") - org = os.environ.get("TFE_ORG") - address = os.environ.get("TFE_ADDRESS", "https://app.terraform.io") - if not token: - print("Error: TFE_TOKEN environment variable is required") - return 1 +def run_force_cancel(query_run_id=None): + """Test 6: Force cancel a query run.""" + print("\n=== Test 6: Force Cancel Query Run ===") - if not org: - print("Error: TFE_ORG environment variable is required") - return 1 + client, workspace = get_client_and_workspace() - # Initialize client - print("=== Terraform Enterprise Query Run SDK Example ===") - print(f"Address: {address}") - print(f"Organization: {org}") - print(f"Timestamp: {datetime.now()}") + try: + # If no query_run_id provided, create a new one + if not query_run_id: + print("Creating a new query run to force cancel...") + new_run = run_create() + if not new_run: + print("ERROR: Could not create query run to force cancel") + return False + query_run_id = new_run.id + time.sleep(1) # Give it a moment to start + + # Force cancel the query run + client.query_runs.force_cancel(query_run_id) + print(f"Force cancel requested for query run: {query_run_id}") + + # Verify force cancellation + time.sleep(2) + query_run = client.query_runs.read(query_run_id) + print(f" Status after force cancel: {query_run.status}") - config = TFEConfig(address=address, token=token) - client = TFEClient(config) + return True - try: - # 1. List existing query runs - existing_query_run = test_list_query_runs(client, org) + except Exception as e: + print(f"Error: {e}") + print(" Note: Query run may not be in a force-cancelable state") + return False - # 2. Create new query runs - created_query_runs = test_create_query_runs(client, org) - # 3. Test read operations - if existing_query_run: - test_read_query_run(client, existing_query_run.id) +def main(): + """Run all tests in sequence.""" + print("=" * 80) + print("QUERY RUN FUNCTION TESTS") + print("=" * 80) + print("Testing all Query Run API operations") + print() + print("NOTE: Query Runs require Terraform 1.10+ with 'terraform query' command.") + print(" Most query runs will error since this feature is not yet available.") + print("=" * 80) - # Only test logs and results if query run is completed - if existing_query_run.status == QueryRunStatus.COMPLETED: - test_query_run_logs(client, existing_query_run.id) - test_query_run_results(client, existing_query_run.id) + # Test 1: List query runs + query_runs = run_list() - # 4. Test cancellation (with a new query run if needed) - if created_query_runs: - test_query_run_cancellation(client, created_query_runs[0].id) + # Test 2: Create a query run + new_query_run = run_create() - # 5. Test complete workflow - test_query_run_workflow(client, org) + # Test 3: Read a query run + if query_runs: + run_read(query_runs[0].id) + elif new_query_run: + run_read(new_query_run.id) - print("\n" + "=" * 80) - print("Query Run operations completed successfully!") - print("=" * 80) + # Test 4: Get logs (use first query run from list) + if query_runs: + run_logs(query_runs[0].id) - except Exception as e: - print(f"\nUnexpected error: {e}") - return 1 + # Test 5: Cancel a query run (creates new one) + run_cancel() - return 0 + # Test 6: Force cancel a query run (creates new one) + run_force_cancel() if __name__ == "__main__": - exit(main()) + main() diff --git a/examples/registry_provider_version.py b/examples/registry_provider_version.py new file mode 100644 index 00000000..4da1c83a --- /dev/null +++ b/examples/registry_provider_version.py @@ -0,0 +1,244 @@ +from __future__ import annotations + +import argparse +import os + +from pytfe import TFEClient, TFEConfig +from pytfe.models import ( + RegistryProviderID, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, + RegistryProviderVersionListOptions, +) + + +def _print_header(title: str): + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def main(): + parser = argparse.ArgumentParser( + description="Registry Provider Versions demo for python-tfe SDK" + ) + parser.add_argument( + "--address", default=os.getenv("TFE_ADDRESS", "https://app.terraform.io") + ) + parser.add_argument("--token", default=os.getenv("TFE_TOKEN", "")) + parser.add_argument("--organization", required=True, help="Organization name") + parser.add_argument( + "--registry-name", + default="private", + help="Registry name (default: private)", + ) + parser.add_argument("--namespace", required=True, help="Provider namespace") + parser.add_argument("--name", required=True, help="Provider name") + parser.add_argument( + "--page-size", + type=int, + default=100, + help="Page size for fetching versions", + ) + parser.add_argument("--create", action="store_true", help="Create a test version") + parser.add_argument("--read", action="store_true", help="Read a specific version") + parser.add_argument( + "--delete", action="store_true", help="Delete a specific version" + ) + parser.add_argument("--version", help="Version number (e.g., 1.0.0)") + parser.add_argument("--key-id", help="GPG key ID for version signing") + parser.add_argument( + "--protocols", + nargs="+", + help="Supported protocols (e.g., 5.0 6.0)", + ) + args = parser.parse_args() + + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) + + # 1) List all versions for the registry provider + _print_header( + f"Listing versions for {args.registry_name}/{args.namespace}/{args.name}" + ) + provider_id = RegistryProviderID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + ) + + options = RegistryProviderVersionListOptions( + page_size=args.page_size, + ) + + version_count = 0 + for version in client.registry_provider_versions.list( + provider_id=provider_id, + options=options, + ): + version_count += 1 + print(f"- Version {version.version} (ID: {version.id})") + print(f" Created: {version.created_at}") + print(f" Updated: {version.updated_at}") + print(f" Key ID: {version.key_id}") + print(f" Protocols: {', '.join(version.protocols)}") + print(f" Shasums Uploaded: {version.shasums_uploaded}") + print(f" Shasums Signature Uploaded: {version.shasums_sig_uploaded}") + if version.permissions: + print(" Permissions:") + print(f" Can Delete: {version.permissions.can_delete}") + print(f" Can Upload Asset: {version.permissions.can_upload_asset}") + print() + + if version_count == 0: + print("No versions found.") + else: + print(f"Total: {version_count} versions") + + # 2) Create a new version (if --create flag is provided) + if args.create: + if not args.version: + print("Error: --version is required for create operation") + return + if not args.key_id: + print("Error: --key-id is required for create operation") + return + if not args.protocols: + print("Error: --protocols is required for create operation") + return + + _print_header(f"Creating new version: {args.version}") + + create_options = RegistryProviderVersionCreateOptions( + version=args.version, + key_id=args.key_id, + protocols=args.protocols, + ) + + new_version = client.registry_provider_versions.create( + provider_id=provider_id, + options=create_options, + ) + + print(f"Created version: {new_version.id}") + print(f" Version: {new_version.version}") + print(f" Created: {new_version.created_at}") + print(f" Key ID: {new_version.key_id}") + print(f" Protocols: {', '.join(new_version.protocols)}") + print(f" Shasums Uploaded: {new_version.shasums_uploaded}") + print(f" Shasums Signature Uploaded: {new_version.shasums_sig_uploaded}") + + # Show upload URLs if available in links + if new_version.links: + print("\n Upload URLs:") + if "shasums-upload" in new_version.links: + print(f" Shasums: {new_version.links['shasums-upload']}") + if "shasums-sig-upload" in new_version.links: + print( + f" Shasums Signature: {new_version.links['shasums-sig-upload']}" + ) + + # 3) Read a specific version (if --read flag is provided) + if args.read: + if not args.version: + print("Error: --version is required for read operation") + return + + _print_header(f"Reading version: {args.version}") + + version_id = RegistryProviderVersionID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + ) + + version = client.registry_provider_versions.read(version_id) + + print(f"Version ID: {version.id}") + print(f" Version: {version.version}") + print(f" Created: {version.created_at}") + print(f" Updated: {version.updated_at}") + print(f" Key ID: {version.key_id}") + print(f" Protocols: {', '.join(version.protocols)}") + print(f" Shasums Uploaded: {version.shasums_uploaded}") + print(f" Shasums Signature Uploaded: {version.shasums_sig_uploaded}") + + if version.permissions: + print(" Permissions:") + print(f" Can Delete: {version.permissions.can_delete}") + print(f" Can Upload Asset: {version.permissions.can_upload_asset}") + + # Show links if available + if version.links: + print(" Links:") + for key, value in version.links.items(): + print(f" {key}: {value}") + + # 4) Delete a version (if --delete flag is provided) + if args.delete: + if not args.version: + print("Error: --version is required for delete operation") + return + + _print_header(f"Deleting version: {args.version}") + + version_id = RegistryProviderVersionID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + version=args.version, + ) + + # First read the version to show what's being deleted + try: + version_to_delete = client.registry_provider_versions.read(version_id) + print("Version to delete:") + print(f" ID: {version_to_delete.id}") + print(f" Version: {version_to_delete.version}") + print(f" Protocols: {', '.join(version_to_delete.protocols)}") + print(f" Key ID: {version_to_delete.key_id}") + except Exception as e: + print(f"Error reading version: {e}") + return + + # Delete the version + client.registry_provider_versions.delete(version_id) + print(f"\n Successfully deleted version: {args.version}") + + # List remaining versions + _print_header("Listing versions after deletion") + provider_id = RegistryProviderID( + organization_name=args.organization, + registry_name=args.registry_name, + namespace=args.namespace, + name=args.name, + ) + + options = RegistryProviderVersionListOptions( + page_size=args.page_size, + ) + print("Remaining versions:") + remaining_count = 0 + for version in client.registry_provider_versions.list( + provider_id=provider_id, + options=options, + ): + remaining_count += 1 + print( + f"- Version {version.version}: " + f" protocols={', '.join(version.protocols)}, " + f" shasums_uploaded={version.shasums_uploaded}" + ) + + if remaining_count == 0: + print("No versions remaining.") + else: + print(f"\nTotal: {remaining_count} versions") + + +if __name__ == "__main__": + main() diff --git a/examples/reserved_tag_key.py b/examples/reserved_tag_key.py index b0056c04..8e62b1a8 100644 --- a/examples/reserved_tag_key.py +++ b/examples/reserved_tag_key.py @@ -53,9 +53,7 @@ def main(): try: # 1. List existing reserved tag keys print("\n1. Listing reserved tag keys...") - reserved_tag_keys = client.reserved_tag_key.list(TFE_ORG) - print(f"Found {len(reserved_tag_keys.items)} reserved tag keys:") - for rtk in reserved_tag_keys.items: + for rtk in client.reserved_tag_key.list(TFE_ORG): print( f" - ID: {rtk.id}, Key: {rtk.key}, Disable Overrides: {rtk.disable_overrides}" ) @@ -87,16 +85,16 @@ def main(): # 5. Verify deletion by listing again print("\n5. Verifying deletion...") - reserved_tag_keys_after = client.reserved_tag_key.list(TFE_ORG) - print(f"Reserved tag keys after deletion: {len(reserved_tag_keys_after.items)}") + reserved_tag_keys_after = list(client.reserved_tag_key.list(TFE_ORG)) + print(f"Reserved tag keys after deletion: {len(reserved_tag_keys_after)}") # 6. Demonstrate pagination with options print("\n6. Demonstrating pagination options...") - list_options = ReservedTagKeyListOptions(page_size=5, page_number=1) - paginated_rtks = client.reserved_tag_key.list(TFE_ORG, list_options) - print(f"Page 1 with page size 5: {len(paginated_rtks.items)} keys") - print(f"Total pages: {paginated_rtks.total_pages}") - print(f"Total count: {paginated_rtks.total_count}") + list_options = ReservedTagKeyListOptions(page_size=5) + for rtk in client.reserved_tag_key.list(TFE_ORG, list_options): + print( + f" - ID: {rtk.id}, Key: {rtk.key}, Disable Overrides: {rtk.disable_overrides}" + ) print("\n Reserved Tag Keys API example completed successfully!") diff --git a/pyproject.toml b/pyproject.toml index 5d42334f..ad9117ef 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pytfe" -version = "0.1.1" +version = "0.1.2" description = "Official Python SDK for HashiCorp Terraform Cloud / Terraform Enterprise (TFE) API v2" readme = "README.md" license = { text = "MPL-2.0" } diff --git a/src/pytfe/client.py b/src/pytfe/client.py index f50cdfec..d1c83373 100644 --- a/src/pytfe/client.py +++ b/src/pytfe/client.py @@ -16,14 +16,15 @@ from .resources.policy_check import PolicyChecks from .resources.policy_evaluation import PolicyEvaluations from .resources.policy_set import PolicySets -from .resources.policy_set_outcome import PolicySets as PolicySetOutcomes +from .resources.policy_set_outcome import PolicySetOutcomes from .resources.policy_set_parameter import PolicySetParameters from .resources.policy_set_version import PolicySetVersions from .resources.projects import Projects from .resources.query_run import QueryRuns from .resources.registry_module import RegistryModules from .resources.registry_provider import RegistryProviders -from .resources.reserved_tag_key import ReservedTagKey +from .resources.registry_provider_version import RegistryProviderVersions +from .resources.reserved_tag_key import ReservedTagKeys from .resources.run import Runs from .resources.run_event import RunEvents from .resources.run_task import RunTasks @@ -76,6 +77,7 @@ def __init__(self, config: TFEConfig | None = None): self.workspace_resources = WorkspaceResourcesService(self._transport) self.registry_modules = RegistryModules(self._transport) self.registry_providers = RegistryProviders(self._transport) + self.registry_provider_versions = RegistryProviderVersions(self._transport) # State and execution resources self.state_versions = StateVersions(self._transport) @@ -97,7 +99,7 @@ def __init__(self, config: TFEConfig | None = None): self.ssh_keys = SSHKeys(self._transport) # Reserved Tag Key - self.reserved_tag_key = ReservedTagKey(self._transport) + self.reserved_tag_key = ReservedTagKeys(self._transport) def close(self) -> None: try: diff --git a/src/pytfe/errors.py b/src/pytfe/errors.py index 61853d10..168d37b4 100644 --- a/src/pytfe/errors.py +++ b/src/pytfe/errors.py @@ -497,3 +497,33 @@ class RequiredKeyError(RequiredFieldMissing): def __init__(self, message: str = "key is required"): super().__init__(message) + + +# Policy Set Outcome errors +class InvalidPolicySetOutcomeIDError(InvalidValues): + """Raised when an invalid policy set outcome ID is provided.""" + + def __init__(self, message: str = "invalid value for policy set outcome ID"): + super().__init__(message) + + +# Registry Provider Version errors +class RequiredPrivateRegistryError(RequiredFieldMissing): + """Raised when a required private registry field is missing.""" + + def __init__(self, message: str = "only private registry is allowed"): + super().__init__(message) + + +class InvalidVersionError(InvalidValues): + """Raised when an invalid version is provided.""" + + def __init__(self, message: str = "invalid value for version"): + super().__init__(message) + + +class InvalidKeyIDError(InvalidValues): + """Raised when an invalid key ID is provided.""" + + def __init__(self, message: str = "invalid value for key-id"): + super().__init__(message) diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index c70dd050..8524e6b1 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -71,7 +71,6 @@ ) from .oauth_token import ( OAuthToken, - OAuthTokenList, OAuthTokenListOptions, OAuthTokenUpdateOptions, ) @@ -118,7 +117,6 @@ from .policy_evaluation import ( PolicyAttachable, PolicyEvaluation, - PolicyEvaluationList, PolicyEvaluationListOptions, PolicyEvaluationStatus, PolicyEvaluationStatusTimestamps, @@ -162,16 +160,15 @@ # ── Query Runs ──────────────────────────────────────────────────────────────── from .query_run import ( QueryRun, - QueryRunCancelOptions, + QueryRunActions, QueryRunCreateOptions, - QueryRunForceCancelOptions, - QueryRunList, + QueryRunIncludeOpt, QueryRunListOptions, - QueryRunLogs, QueryRunReadOptions, - QueryRunResults, + QueryRunSource, QueryRunStatus, - QueryRunType, + QueryRunStatusTimestamps, + QueryRunVariable, ) # ── Registry Modules / Providers ────────────────────────────────────────────── @@ -217,12 +214,18 @@ RegistryProviderPermissions, RegistryProviderReadOptions, ) +from .registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, + RegistryProviderVersionListOptions, + RegistryProviderVersionPermissions, +) # ── Reserved Tag Keys ───────────────────────────────────────────────────────── from .reserved_tag_key import ( ReservedTagKey, ReservedTagKeyCreateOptions, - ReservedTagKeyList, ReservedTagKeyListOptions, ReservedTagKeyUpdateOptions, ) @@ -374,7 +377,6 @@ "ServiceProviderType", # OAuth token "OAuthToken", - "OAuthTokenList", "OAuthTokenListOptions", "OAuthTokenUpdateOptions", # SSH keys @@ -386,7 +388,6 @@ # Reserved tag keys "ReservedTagKey", "ReservedTagKeyCreateOptions", - "ReservedTagKeyList", "ReservedTagKeyListOptions", "ReservedTagKeyUpdateOptions", # Agent & pools @@ -455,18 +456,23 @@ "RegistryProviderListOptions", "RegistryProviderPermissions", "RegistryProviderReadOptions", + # Registry provider versions + "RegistryProviderVersion", + "RegistryProviderVersionCreateOptions", + "RegistryProviderVersionID", + "RegistryProviderVersionListOptions", + "RegistryProviderVersionPermissions", # Query runs "QueryRun", - "QueryRunCancelOptions", + "QueryRunActions", "QueryRunCreateOptions", - "QueryRunForceCancelOptions", - "QueryRunList", + "QueryRunIncludeOpt", "QueryRunListOptions", - "QueryRunLogs", "QueryRunReadOptions", - "QueryRunResults", + "QueryRunSource", "QueryRunStatus", - "QueryRunType", + "QueryRunStatusTimestamps", + "QueryRunVariable", # Core (from old types.py, now split) "Entitlements", "ExecutionMode", @@ -596,7 +602,6 @@ # Policy Evaluation "PolicyAttachable", "PolicyEvaluation", - "PolicyEvaluationList", "PolicyEvaluationListOptions", "PolicyEvaluationStatus", "PolicyEvaluationStatusTimestamps", diff --git a/src/pytfe/models/oauth_token.py b/src/pytfe/models/oauth_token.py index c6b004b3..a70c20e8 100644 --- a/src/pytfe/models/oauth_token.py +++ b/src/pytfe/models/oauth_token.py @@ -15,7 +15,6 @@ class OAuthToken(BaseModel): model_config = ConfigDict(extra="forbid") id: str = Field(..., description="OAuth token ID") - uid: str = Field(..., description="OAuth token UID") created_at: datetime = Field(..., description="Creation timestamp") has_ssh_key: bool = Field(..., description="Whether the token has an SSH key") service_provider_user: str = Field(..., description="Service provider user") @@ -26,26 +25,12 @@ class OAuthToken(BaseModel): ) -class OAuthTokenList(BaseModel): - """List of OAuth tokens with pagination information.""" - - model_config = ConfigDict(extra="forbid") - - items: list[OAuthToken] = Field(default_factory=list, description="OAuth tokens") - current_page: int | None = Field(None, description="Current page number") - prev_page: int | None = Field(None, description="Previous page number") - next_page: int | None = Field(None, description="Next page number") - total_pages: int | None = Field(None, description="Total number of pages") - total_count: int | None = Field(None, description="Total count of items") - - class OAuthTokenListOptions(BaseModel): """Options for listing OAuth tokens.""" - model_config = ConfigDict(extra="forbid") + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - page_number: int | None = Field(None, description="Page number") - page_size: int | None = Field(None, description="Page size") + page_size: int | None = Field(None, alias="page[size]", description="Page size") class OAuthTokenUpdateOptions(BaseModel): @@ -63,7 +48,6 @@ class OAuthTokenUpdateOptions(BaseModel): from .oauth_client import OAuthClient # noqa: F401 OAuthToken.model_rebuild() - OAuthTokenList.model_rebuild() except ImportError: # If OAuthClient is not available, create a dummy class pass diff --git a/src/pytfe/models/policy_evaluation.py b/src/pytfe/models/policy_evaluation.py index 86175e9c..49ad257c 100644 --- a/src/pytfe/models/policy_evaluation.py +++ b/src/pytfe/models/policy_evaluation.py @@ -37,7 +37,7 @@ class PolicyEvaluation(BaseModel): updated_at: datetime | None = Field(None, alias="updated-at") # The task stage the policy evaluation belongs to - task_stage: PolicyAttachable | None = Field(None, alias="policy-attachable") + policy_attachable: PolicyAttachable | None = Field(None, alias="policy-attachable") class PolicyEvaluationStatusTimestamps(BaseModel): @@ -72,23 +72,9 @@ class PolicyResultCount(BaseModel): errored: int | None = Field(None, alias="errored") -class PolicyEvaluationList(BaseModel): - """PolicyEvaluationList represents a list of policy evaluations""" - - model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - - items: list[PolicyEvaluation] | None = Field(default_factory=list) - current_page: int | None = None - next_page: str | None = None - prev_page: str | None = None - total_count: int | None = None - total_pages: int | None = None - - class PolicyEvaluationListOptions(BaseModel): """PolicyEvaluationListOptions represents the options for listing policy evaluations""" model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/models/policy_set_outcome.py b/src/pytfe/models/policy_set_outcome.py index 40595462..ffb97230 100644 --- a/src/pytfe/models/policy_set_outcome.py +++ b/src/pytfe/models/policy_set_outcome.py @@ -34,19 +34,6 @@ class Outcome(BaseModel): description: str | None = Field(None, alias="description") -class PolicySetOutcomeList(BaseModel): - """PolicySetOutcomeList represents a list of policy set outcomes""" - - model_config = ConfigDict(populate_by_name=True, validate_by_name=True) - - items: list[PolicySetOutcome] | None = Field(default_factory=list) - current_page: int | None = None - next_page: str | None = None - prev_page: str | None = None - total_count: int | None = None - total_pages: int | None = None - - class PolicySetOutcomeListFilter(BaseModel): """PolicySetOutcomeListFilter represents the filters that are supported while listing a policy set outcome""" @@ -62,5 +49,4 @@ class PolicySetOutcomeListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True, validate_by_name=True) filter: dict[str, PolicySetOutcomeListFilter] | None = None - page_number: int | None = Field(None, alias="page[number]") page_size: int | None = Field(None, alias="page[size]") diff --git a/src/pytfe/models/query_run.py b/src/pytfe/models/query_run.py index 3670830c..cdcfe57d 100644 --- a/src/pytfe/models/query_run.py +++ b/src/pytfe/models/query_run.py @@ -2,7 +2,6 @@ from datetime import datetime from enum import Enum -from typing import Any from pydantic import BaseModel, ConfigDict, Field @@ -11,204 +10,166 @@ class QueryRunStatus(str, Enum): """QueryRunStatus represents the status of a query run operation.""" PENDING = "pending" + QUEUED = "queued" RUNNING = "running" - COMPLETED = "completed" + FINISHED = "finished" ERRORED = "errored" CANCELED = "canceled" -class QueryRunType(str, Enum): - """QueryRunType represents different types of query runs.""" +class QueryRunSource(str, Enum): + """QueryRunSource represents the source of a query run.""" - FILTER = "filter" - SEARCH = "search" - ANALYTICS = "analytics" + API = "tfe-api" -class QueryRun(BaseModel): - """Represents a query run in Terraform Enterprise.""" +class QueryRunActions(BaseModel): + """Actions available on a query run.""" model_config = ConfigDict(populate_by_name=True) - id: str = Field(..., description="The unique identifier for this query run") - type: str = Field(default="query-runs", description="The type of this resource") - query: str = Field(..., description="The query string used for this run") - query_type: QueryRunType = Field( - ..., alias="query-type", description="The type of query being executed" - ) - status: QueryRunStatus = Field( - ..., description="The current status of the query run" + is_cancelable: bool = Field( + ..., alias="is-cancelable", description="Whether the query run can be canceled" ) - results_count: int | None = Field( - None, alias="results-count", description="The number of results returned" + is_force_cancelable: bool = Field( + ..., + alias="is-force-cancelable", + description="Whether the query run can be force canceled", ) - created_at: datetime = Field( - ..., alias="created-at", description="The time this query run was created" + + +class QueryRunStatusTimestamps(BaseModel): + """Timestamps for each status of a query run.""" + + model_config = ConfigDict(populate_by_name=True) + + pending_at: datetime | None = Field( + None, alias="pending-at", description="When the query run was created" ) - updated_at: datetime = Field( - ..., alias="updated-at", description="The time this query run was last updated" + queued_at: datetime | None = Field( + None, alias="queued-at", description="When the query run was queued" ) - started_at: datetime | None = Field( - None, alias="started-at", description="The time this query run was started" + running_at: datetime | None = Field( + None, alias="running-at", description="When the query run started running" ) finished_at: datetime | None = Field( - None, alias="finished-at", description="The time this query run was finished" - ) - error_message: str | None = Field( - None, alias="error-message", description="Error message if the query run failed" - ) - logs_url: str | None = Field( - None, alias="logs-url", description="URL to retrieve the query run logs" - ) - results_url: str | None = Field( - None, alias="results-url", description="URL to retrieve the query run results" - ) - workspace_id: str | None = Field( None, - alias="workspace-id", - description="The workspace ID if query is workspace-scoped", + alias="finished-at", + description="When the query run finished successfully", ) - organization_name: str | None = Field( - None, alias="organization-name", description="The organization name" + errored_at: datetime | None = Field( + None, alias="errored-at", description="When the query run encountered an error" ) - timeout_seconds: int | None = Field( - None, alias="timeout-seconds", description="Query timeout in seconds" - ) - max_results: int | None = Field( - None, alias="max-results", description="Maximum number of results to return" + canceled_at: datetime | None = Field( + None, alias="canceled-at", description="When the query run was canceled" ) -class QueryRunCreateOptions(BaseModel): - """Options for creating a new query run.""" +class QueryRunVariable(BaseModel): + """A variable for a query run.""" + + key: str = Field(..., description="Variable key") + value: str = Field(..., description="Variable value") + + +class QueryRun(BaseModel): + """Represents a query run in Terraform Enterprise.""" model_config = ConfigDict(populate_by_name=True) - query: str = Field(..., description="The query string to execute") - query_type: QueryRunType = Field( - ..., alias="query-type", description="The type of query being executed" + id: str = Field(..., description="The unique identifier for this query run") + type: str = Field(default="queries", description="The type of this resource") + actions: QueryRunActions | None = Field( + None, description="Actions available on this query run" ) - workspace_id: str | None = Field( - None, - alias="workspace-id", - description="The workspace ID if query is workspace-scoped", + canceled_at: datetime | None = Field( + None, alias="canceled-at", description="When the query run was canceled" ) - organization_name: str | None = Field( - None, alias="organization-name", description="The organization name" + created_at: datetime = Field( + ..., alias="created-at", description="The time this query run was created" ) - timeout_seconds: int | None = Field( - None, - alias="timeout-seconds", - description="Query timeout in seconds", - gt=0, - le=3600, + updated_at: datetime | None = Field( + None, alias="updated-at", description="The time this query run was last updated" ) - max_results: int | None = Field( + source: QueryRunSource | str = Field(..., description="The source of the query run") + status: QueryRunStatus = Field( + ..., description="The current status of the query run" + ) + status_timestamps: QueryRunStatusTimestamps | None = Field( None, - alias="max-results", - description="Maximum number of results to return", - gt=0, - le=10000, + alias="status-timestamps", + description="Timestamps for each status of the query run", ) - filters: dict[str, Any] | None = Field( - None, description="Additional filters to apply to the query" + variables: list[QueryRunVariable] | None = Field( + None, description="Run-specific variable values" ) - - -class QueryRunListOptions(BaseModel): - """Options for listing query runs.""" - - model_config = ConfigDict(populate_by_name=True) - - page_number: int | None = Field( - None, alias="page[number]", description="Page number to retrieve", ge=1 + log_read_url: str | None = Field( + None, alias="log-read-url", description="URL to retrieve the query run logs" ) - page_size: int | None = Field( - None, alias="page[size]", description="Number of items per page", ge=1, le=100 + # Relationships + workspace_id: str | None = Field( + None, description="The workspace ID associated with this query run" ) - query_type: QueryRunType | None = Field( - None, alias="filter[query-type]", description="Filter by query type" + configuration_version_id: str | None = Field( + None, description="The configuration version ID used for this query run" ) - status: QueryRunStatus | None = Field( - None, alias="filter[status]", description="Filter by status" + created_by_id: str | None = Field( + None, description="The user ID who created this query run" ) - workspace_id: str | None = Field( - None, alias="filter[workspace-id]", description="Filter by workspace ID" - ) - organization_name: str | None = Field( - None, - alias="filter[organization-name]", - description="Filter by organization name", + canceled_by_id: str | None = Field( + None, description="The user ID who canceled this query run" ) -class QueryRunReadOptions(BaseModel): - """Options for reading a query run with additional data.""" +class QueryRunCreateOptions(BaseModel): + """Options for creating a new query run.""" model_config = ConfigDict(populate_by_name=True) - include_results: bool | None = Field( - None, alias="include[results]", description="Include query results in response" + source: QueryRunSource | str = Field(..., description="The source of the query run") + variables: list[QueryRunVariable] | None = Field( + None, description="Run-specific variable values" ) - include_logs: bool | None = Field( - None, alias="include[logs]", description="Include query logs in response" + workspace_id: str = Field( + ..., + alias="workspace-id", + description="The workspace ID to run the query against", ) - - -class QueryRunCancelOptions(BaseModel): - """Options for canceling a query run.""" - - model_config = ConfigDict(populate_by_name=True) - - reason: str | None = Field(None, description="Reason for canceling the query run") - - -class QueryRunForceCancelOptions(BaseModel): - """Options for force canceling a query run.""" - - model_config = ConfigDict(populate_by_name=True) - - reason: str | None = Field( - None, description="Reason for force canceling the query run" + configuration_version_id: str | None = Field( + None, + alias="configuration-version-id", + description="The configuration version ID to use for the query", ) -class QueryRunList(BaseModel): - """Represents a paginated list of query runs.""" +class QueryRunIncludeOpt(str, Enum): + """Options for including related resources in query run requests.""" - model_config = ConfigDict(populate_by_name=True) - - items: list[QueryRun] = Field( - default_factory=list, description="List of query runs" + CREATED_BY = "created_by" + CONFIGURATION_VERSION = "configuration_version" + CONFIGURATION_VERSION_INGRESS_ATTRIBUTES = ( + "configuration_version.ingress_attributes" ) - current_page: int | None = Field(None, description="Current page number") - total_pages: int | None = Field(None, description="Total number of pages") - prev_page: str | None = Field(None, description="URL of the previous page") - next_page: str | None = Field(None, description="URL of the next page") - total_count: int | None = Field(None, description="Total number of items") -class QueryRunResults(BaseModel): - """Represents the results of a query run.""" +class QueryRunListOptions(BaseModel): + """Options for listing query runs.""" model_config = ConfigDict(populate_by_name=True) - query_run_id: str = Field(..., description="The ID of the query run") - results: list[dict[str, Any]] = Field( - default_factory=list, description="The query results" + page_size: int | None = Field( + None, alias="page[size]", description="Number of items per page", ge=1, le=100 ) - total_count: int = Field(..., description="Total number of results") - truncated: bool = Field( - False, description="Whether the results were truncated due to limits" + include: list[QueryRunIncludeOpt] | None = Field( + None, description="List of related resources to include" ) -class QueryRunLogs(BaseModel): - """Represents the logs of a query run.""" +class QueryRunReadOptions(BaseModel): + """Options for reading a query run with additional data.""" model_config = ConfigDict(populate_by_name=True) - query_run_id: str = Field(..., description="The ID of the query run") - logs: str = Field(..., description="The query run logs") - log_level: str | None = Field(None, description="The log level") - timestamp: datetime | None = Field(None, description="When the logs were generated") + include: list[QueryRunIncludeOpt] | None = Field( + None, description="List of related resources to include" + ) diff --git a/src/pytfe/models/registry_provider_version.py b/src/pytfe/models/registry_provider_version.py new file mode 100644 index 00000000..6c043d37 --- /dev/null +++ b/src/pytfe/models/registry_provider_version.py @@ -0,0 +1,174 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +from ..errors import ( + InvalidKeyIDError, + InvalidVersionError, + RequiredPrivateRegistryError, +) +from ..utils import valid_string_id +from .registry_provider import ( + RegistryName, + RegistryProviderID, +) + + +class RegistryProviderVersionPermissions(BaseModel): + """Registry provider version permissions.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + can_delete: bool = Field(alias="can-delete") + can_upload_asset: bool = Field(alias="can-upload-asset") + + +class RegistryProviderVersion(BaseModel): + """Registry provider version model.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + id: str + version: str + created_at: datetime = Field(alias="created-at") + updated_at: datetime = Field(alias="updated-at") + key_id: str = Field(alias="key-id") + protocols: list[str] + permissions: RegistryProviderVersionPermissions + shasums_uploaded: bool = Field(alias="shasums-uploaded") + shasums_sig_uploaded: bool = Field(alias="shasums-sig-uploaded") + + # Relations + registry_provider: dict[str, Any] | None = Field( + alias="registry-provider", default=None + ) + registry_provider_platforms: list[dict[str, Any]] | None = Field( + alias="platforms", default=None + ) + + # Links + links: dict[str, Any] | None = None + + def shasums_upload_url(self) -> str: + """ShasumsUploadURL returns the upload URL to upload shasums if one is available""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums upload link" + ) + upload_url = str(self.links.get("shasums-upload")) + if not upload_url: + raise ValueError( + "The registry provider version does not contain a shasums upload link" + ) + + if upload_url == "": + raise ValueError( + "The registry provider version shasums upload URL is empty" + ) + + return upload_url + + def shasums_sig_upload_url(self) -> str: + """ShasumsSigUploadURL returns the URL to upload a shasums sig""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums sig upload link" + ) + upload_url = str(self.links.get("shasums-sig-upload")) + if not upload_url: + raise ValueError( + "The registry provider version does not contain a shasums sig upload link" + ) + + if upload_url == "": + raise ValueError( + "The registry provider version shasums sig upload URL is empty" + ) + + return upload_url + + def shasums_download_url(self) -> str: + """ShasumsDownloadURL returns the URL to download the shasums for the registry version""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums download link" + ) + download_url = str(self.links.get("shasums-download")) + if not download_url: + raise ValueError( + "The registry provider version does not contain a shasums download link" + ) + + if download_url == "": + raise ValueError( + "The registry provider version shasums download URL is empty" + ) + + return download_url + + def shasums_sig_download_url(self) -> str: + """ShasumsSigDownloadURL returns the URL to download the shasums sig for the registry version""" + if self.links is None: + raise ValueError( + "The registry provider version does not contain a shasums sig download link" + ) + download_url = str(self.links.get("shasums-sig-download")) + if not download_url: + raise ValueError( + "The registry provider version does not contain a shasums sig download link" + ) + + if download_url == "": + raise ValueError( + "The registry provider version shasums sig download URL is empty" + ) + + return download_url + + +class RegistryProviderVersionID(RegistryProviderID): + """Registry provider version identifier. + + This extends RegistryProviderID with a version field to uniquely + identify a specific version of a provider. + """ + + version: str + + @model_validator(mode="after") + def valid(self) -> RegistryProviderVersionID: + if not valid_string_id(self.version): + raise InvalidVersionError() + if self.registry_name != RegistryName.PRIVATE: + raise RequiredPrivateRegistryError() + return self + + +class RegistryProviderVersionCreateOptions(BaseModel): + """Options for creating a registry provider version.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + version: str + key_id: str = Field(alias="key-id") + protocols: list[str] + + # validation method for version and key_id + @model_validator(mode="after") + def valid(self) -> RegistryProviderVersionCreateOptions: + if not valid_string_id(self.version): + raise InvalidVersionError() + if not valid_string_id(self.key_id): + raise InvalidKeyIDError() + return self + + +class RegistryProviderVersionListOptions(BaseModel): + """Options for listing registry provider versions.""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + page_size: int | None = Field(alias="page[size]", default=None) diff --git a/src/pytfe/models/reserved_tag_key.py b/src/pytfe/models/reserved_tag_key.py index eb125eae..c332742c 100644 --- a/src/pytfe/models/reserved_tag_key.py +++ b/src/pytfe/models/reserved_tag_key.py @@ -65,24 +65,6 @@ class ReservedTagKeyListOptions(BaseModel): model_config = ConfigDict(populate_by_name=True) - page_number: int | None = Field( - None, alias="page[number]", description="Page number to retrieve", ge=1 - ) page_size: int | None = Field( None, alias="page[size]", description="Number of items per page", ge=1, le=100 ) - - -class ReservedTagKeyList(BaseModel): - """Represents a paginated list of reserved tag keys.""" - - model_config = ConfigDict(populate_by_name=True) - - items: list[ReservedTagKey] = Field( - default_factory=list, description="List of reserved tag keys" - ) - current_page: int | None = Field(None, description="Current page number") - total_pages: int | None = Field(None, description="Total number of pages") - prev_page: str | None = Field(None, description="URL of the previous page") - next_page: str | None = Field(None, description="URL of the next page") - total_count: int | None = Field(None, description="Total number of items") diff --git a/src/pytfe/resources/oauth_token.py b/src/pytfe/resources/oauth_token.py index fb25074a..337fa02c 100644 --- a/src/pytfe/resources/oauth_token.py +++ b/src/pytfe/resources/oauth_token.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Iterator from datetime import datetime from typing import Any from urllib.parse import quote @@ -7,11 +8,10 @@ from ..errors import ERR_INVALID_OAUTH_TOKEN_ID, ERR_INVALID_ORG from ..models.oauth_token import ( OAuthToken, - OAuthTokenList, OAuthTokenListOptions, OAuthTokenUpdateOptions, ) -from ..utils import encode_query, valid_string_id +from ..utils import valid_string_id from ._base import _Service @@ -20,7 +20,7 @@ class OAuthTokens(_Service): def list( self, organization: str, options: OAuthTokenListOptions | None = None - ) -> OAuthTokenList: + ) -> Iterator[OAuthToken]: """List all the OAuth tokens for a given organization.""" if not valid_string_id(organization): raise ValueError(ERR_INVALID_ORG) @@ -29,37 +29,11 @@ def list( params = {} if options: - if options.page_number: - params["page[number]"] = str(options.page_number) if options.page_size: params["page[size]"] = str(options.page_size) - query_string = encode_query(params) - full_path = f"{path}{query_string}" - - response = self.t.request("GET", full_path) - data = response.json() - - tokens = [] - if "data" in data: - for item in data["data"]: - tokens.append(self._parse_oauth_token(item)) - - # Parse pagination metadata - pagination = {} - if "meta" in data: - meta = data["meta"] - if "pagination" in meta: - page_info = meta["pagination"] - pagination = { - "current_page": page_info.get("current-page"), - "prev_page": page_info.get("prev-page"), - "next_page": page_info.get("next-page"), - "total_pages": page_info.get("total-pages"), - "total_count": page_info.get("total-count"), - } - - return OAuthTokenList(items=tokens, **pagination) + for item in self._list(path, params=params): + yield self._parse_oauth_token(item) def read(self, oauth_token_id: str) -> OAuthToken: """Read an OAuth token by its ID.""" @@ -128,7 +102,6 @@ def _parse_oauth_token(self, data: dict[str, Any]) -> OAuthToken: return OAuthToken( id=data.get("id", ""), - uid=attributes.get("uid", ""), created_at=created_at, has_ssh_key=attributes.get("has-ssh-key", False), service_provider_user=attributes.get("service-provider-user", ""), diff --git a/src/pytfe/resources/policy_evaluation.py b/src/pytfe/resources/policy_evaluation.py index bc301937..2f911f60 100644 --- a/src/pytfe/resources/policy_evaluation.py +++ b/src/pytfe/resources/policy_evaluation.py @@ -1,11 +1,12 @@ from __future__ import annotations +from collections.abc import Iterator + from ..errors import ( InvalidTaskStageIDError, ) from ..models.policy_evaluation import ( PolicyEvaluation, - PolicyEvaluationList, PolicyEvaluationListOptions, ) from ..utils import valid_string_id @@ -20,34 +21,21 @@ class PolicyEvaluations(_Service): def list( self, task_stage_id: str, options: PolicyEvaluationListOptions | None = None - ) -> PolicyEvaluationList: + ) -> Iterator[PolicyEvaluation]: """ **Note: This method is still in BETA and subject to change.** - List all policy evaluations in the task stage. Only available for OPA policies. + List all policy evaluations in the task stage. Only available for OPA policies. """ if not valid_string_id(task_stage_id): raise InvalidTaskStageIDError() params = options.model_dump(by_alias=True) if options else {} path = f"api/v2/task-stages/{task_stage_id}/policy-evaluations" - r = self.t.request("GET", path, params=params) - jd = r.json() - items = [] - meta = jd.get("meta", {}) - pagination = meta.get("pagination", {}) - for item in jd.get("data", []): + for item in self._list(path, params=params): attrs = item.get("attributes", {}) attrs["id"] = item.get("id") - attrs["task-stage"] = ( + attrs["policy-attachable"] = ( item.get("relationships", {}) .get("policy-attachable", {}) .get("data", {}) ) - items.append(PolicyEvaluation.model_validate(attrs)) - return PolicyEvaluationList( - items=items, - current_page=pagination.get("current-page"), - next_page=pagination.get("next-page"), - prev_page=pagination.get("prev-page"), - total_count=pagination.get("total-count"), - total_pages=pagination.get("total-pages"), - ) + yield PolicyEvaluation.model_validate(attrs) diff --git a/src/pytfe/resources/policy_set_outcome.py b/src/pytfe/resources/policy_set_outcome.py index 56f7f342..42389d36 100644 --- a/src/pytfe/resources/policy_set_outcome.py +++ b/src/pytfe/resources/policy_set_outcome.py @@ -1,18 +1,21 @@ from __future__ import annotations +from collections.abc import Iterator +from typing import Any + from ..errors import ( InvalidPolicyEvaluationIDError, + InvalidPolicySetOutcomeIDError, ) from ..models.policy_set_outcome import ( PolicySetOutcome, - PolicySetOutcomeList, PolicySetOutcomeListOptions, ) from ..utils import valid_string_id from ._base import _Service -class PolicySets(_Service): +class PolicySetOutcomes(_Service): """ PolicySetOutcomes describes all the policy set outcome related methods that the Terraform Enterprise API supports. TFE API docs: https://developer.hashicorp.com/terraform/cloud-docs/api-docs/policy-checks @@ -22,7 +25,7 @@ def list( self, policy_evaluation_id: str, options: PolicySetOutcomeListOptions | None = None, - ) -> PolicySetOutcomeList: + ) -> Iterator[PolicySetOutcome]: """ **Note: This method is still in BETA and subject to change.** List all policy set outcomes in the policy evaluation. Only available for OPA policies. @@ -35,28 +38,8 @@ def list( if additional_query_params: params.update(additional_query_params) path = f"api/v2/policy-evaluations/{policy_evaluation_id}/policy-set-outcomes" - r = self.t.request("GET", path, params=params) - jd = r.json() - items = [] - meta = jd.get("meta", {}) - pagination = meta.get("pagination", {}) - for item in jd.get("data", []): - attrs = item.get("attributes", {}) - attrs["id"] = item.get("id") - attrs["policy-evaluation"] = ( - item.get("relationships", {}) - .get("policy-evaluation", {}) - .get("data", {}) - ) - items.append(PolicySetOutcome.model_validate(attrs)) - return PolicySetOutcomeList( - items=items, - current_page=pagination.get("current-page"), - next_page=pagination.get("next-page"), - prev_page=pagination.get("prev-page"), - total_count=pagination.get("total-count"), - total_pages=pagination.get("total-pages"), - ) + for item in self._list(path, params=params): + yield self._policy_set_outcome_from(item) def build_query_string( self, options: PolicySetOutcomeListOptions | None @@ -77,14 +60,17 @@ def read(self, policy_set_outcome_id: str) -> PolicySetOutcome: **Note: This method is still in BETA and subject to change.** Read a single policy set outcome by ID. Only available for OPA policies.""" if not valid_string_id(policy_set_outcome_id): - raise InvalidPolicyEvaluationIDError() + raise InvalidPolicySetOutcomeIDError() path = f"api/v2/policy-set-outcomes/{policy_set_outcome_id}" r = self.t.request("GET", path) - jd = r.json() - item = jd.get("data", {}) - attrs = item.get("attributes", {}) - attrs["id"] = item.get("id") + data = r.json().get("data", {}) + return PolicySetOutcome.model_validate(data) + + def _policy_set_outcome_from(self, d: dict[str, Any]) -> PolicySetOutcome: + """Convert API response dict to PolicySetParameter model.""" + attrs = d.get("attributes", {}) + attrs["id"] = d.get("id") attrs["policy-evaluation"] = ( - item.get("relationships", {}).get("policy-evaluation", {}).get("data", {}) + d.get("relationships", {}).get("policy-evaluation", {}).get("data", {}) ) return PolicySetOutcome.model_validate(attrs) diff --git a/src/pytfe/resources/query_run.py b/src/pytfe/resources/query_run.py index 1540c703..a552e644 100644 --- a/src/pytfe/resources/query_run.py +++ b/src/pytfe/resources/query_run.py @@ -1,21 +1,18 @@ from __future__ import annotations +import io +from collections.abc import Iterator from typing import Any from ..errors import ( - InvalidOrgError, InvalidQueryRunIDError, + InvalidWorkspaceIDError, ) from ..models.query_run import ( QueryRun, - QueryRunCancelOptions, QueryRunCreateOptions, - QueryRunForceCancelOptions, - QueryRunList, QueryRunListOptions, - QueryRunLogs, QueryRunReadOptions, - QueryRunResults, ) from ..utils import valid_string_id from ._base import _Service @@ -25,57 +22,69 @@ class QueryRuns(_Service): """Query Runs API for Terraform Enterprise.""" def list( - self, organization: str, options: QueryRunListOptions | None = None - ) -> QueryRunList: - """List query runs for the given organization.""" - if not valid_string_id(organization): - raise InvalidOrgError() - - params = ( - options.model_dump(by_alias=True, exclude_none=True) if options else None - ) + self, workspace_id: str, options: QueryRunListOptions | None = None + ) -> Iterator[QueryRun]: + """Iterate through all query runs for the given workspace. + + This method automatically handles pagination and yields QueryRun objects one at a time. + + Args: + workspace_id: The ID of the workspace + options: Optional list options (page_size, include, etc.) + + Yields: + QueryRun objects one at a time + + Example: + for query_run in client.query_runs.list(workspace_id): + print(f"Query Run: {query_run.id} - Status: {query_run.status}") + """ + if not valid_string_id(workspace_id): + raise InvalidWorkspaceIDError() + + params: dict[str, Any] = {} + if options: + params = options.model_dump(by_alias=True, exclude_none=True) + # Convert include list to comma-separated string + if "include" in params and params["include"] and options.include: + params["include"] = ",".join([i.value for i in options.include]) + + path = f"/api/v2/workspaces/{workspace_id}/queries" + for item in self._list(path, params=params): + attrs = item.get("attributes", {}) + attrs["id"] = item.get("id") + yield QueryRun.model_validate(attrs) + + def create(self, options: QueryRunCreateOptions) -> QueryRun: + """Create a new query run.""" + attrs = options.model_dump(by_alias=True, exclude_none=True) - r = self.t.request( - "GET", - f"/api/v2/organizations/{organization}/query-runs", - params=params, - ) + # Build relationships + relationships: dict[str, Any] = {} - jd = r.json() - items = [] - meta = jd.get("meta", {}) - pagination = meta.get("pagination", {}) - - for d in jd.get("data", []): - attrs = d.get("attributes", {}) - attrs["id"] = d.get("id") - items.append(QueryRun.model_validate(attrs)) - - return QueryRunList( - items=items, - current_page=pagination.get("current-page"), - total_pages=pagination.get("total-pages"), - prev_page=pagination.get("prev-page"), - next_page=pagination.get("next-page"), - total_count=pagination.get("total-count"), - ) + if workspace_id := attrs.pop("workspace-id", None): + relationships["workspace"] = { + "data": {"type": "workspaces", "id": workspace_id} + } - def create(self, organization: str, options: QueryRunCreateOptions) -> QueryRun: - """Create a new query run for the given organization.""" - if not valid_string_id(organization): - raise InvalidOrgError() + if config_version_id := attrs.pop("configuration-version-id", None): + relationships["configuration-version"] = { + "data": {"type": "configuration-versions", "id": config_version_id} + } - attrs = options.model_dump(by_alias=True, exclude_none=True) body: dict[str, Any] = { "data": { + "type": "queries", "attributes": attrs, - "type": "query-runs", } } + if relationships: + body["data"]["relationships"] = relationships + r = self.t.request( "POST", - f"/api/v2/organizations/{organization}/query-runs", + "/api/v2/queries", json_body=body, ) @@ -91,7 +100,7 @@ def read(self, query_run_id: str) -> QueryRun: if not valid_string_id(query_run_id): raise InvalidQueryRunIDError() - r = self.t.request("GET", f"/api/v2/query-runs/{query_run_id}") + r = self.t.request("GET", f"/api/v2/queries/{query_run_id}") jd = r.json() data = jd.get("data", {}) @@ -108,8 +117,11 @@ def read_with_options( raise InvalidQueryRunIDError() params = options.model_dump(by_alias=True, exclude_none=True) + # Convert include list to comma-separated string + if "include" in params and params["include"] and options.include: + params["include"] = ",".join([i.value for i in options.include]) - r = self.t.request("GET", f"/api/v2/query-runs/{query_run_id}", params=params) + r = self.t.request("GET", f"/api/v2/queries/{query_run_id}", params=params) jd = r.json() data = jd.get("data", {}) @@ -118,99 +130,48 @@ def read_with_options( return QueryRun.model_validate(attrs) - def logs(self, query_run_id: str) -> QueryRunLogs: - """Retrieve the logs for a query run.""" - if not valid_string_id(query_run_id): - raise InvalidQueryRunIDError() + def logs(self, query_run_id: str) -> io.IOBase: + """Retrieve the logs for a query run. - r = self.t.request("GET", f"/api/v2/query-runs/{query_run_id}/logs") - - # Handle both JSON and plain text responses - content_type = r.headers.get("content-type", "").lower() - - if "application/json" in content_type: - jd = r.json() - return QueryRunLogs.model_validate(jd.get("data", {})) - else: - # Plain text logs - return QueryRunLogs( - query_run_id=query_run_id, - logs=r.text, - log_level="info", - timestamp=None, - ) - - def results(self, query_run_id: str) -> QueryRunResults: - """Retrieve the results for a query run.""" + Returns an IO stream that can be read to get the log content. + """ if not valid_string_id(query_run_id): raise InvalidQueryRunIDError() - r = self.t.request("GET", f"/api/v2/query-runs/{query_run_id}/results") + # First get the query run to retrieve the log read URL + query_run = self.read(query_run_id) - jd = r.json() - data = jd.get("data", {}) + if not query_run.log_read_url: + raise ValueError(f"Query run {query_run_id} does not have a log URL") - return QueryRunResults( - query_run_id=query_run_id, - results=data.get("results", []), - total_count=data.get("total_count", 0), - truncated=data.get("truncated", False), - ) + # Fetch the logs from the URL (absolute URLs are handled by _build_url) + r = self.t.request("GET", query_run.log_read_url) - def cancel( - self, query_run_id: str, options: QueryRunCancelOptions | None = None - ) -> QueryRun: - """Cancel a query run.""" - if not valid_string_id(query_run_id): - raise InvalidQueryRunIDError() + # Return the content as a BytesIO stream + return io.BytesIO(r.content) - attrs = options.model_dump(by_alias=True, exclude_none=True) if options else {} + def cancel(self, query_run_id: str) -> None: + """Cancel a query run. - body: dict[str, Any] = { - "data": { - "attributes": attrs, - "type": "query-runs", - } - } + Returns 202 on success with empty body. + """ + if not valid_string_id(query_run_id): + raise InvalidQueryRunIDError() - r = self.t.request( + self.t.request( "POST", - f"/api/v2/query-runs/{query_run_id}/actions/cancel", - json_body=body, + f"/api/v2/queries/{query_run_id}/actions/cancel", ) - jd = r.json() - data = jd.get("data", {}) - attrs = data.get("attributes", {}) - attrs["id"] = data.get("id") + def force_cancel(self, query_run_id: str) -> None: + """Force cancel a query run. - return QueryRun.model_validate(attrs) - - def force_cancel( - self, query_run_id: str, options: QueryRunForceCancelOptions | None = None - ) -> QueryRun: - """Force cancel a query run.""" + Returns 202 on success with empty body. + """ if not valid_string_id(query_run_id): raise InvalidQueryRunIDError() - attrs = options.model_dump(by_alias=True, exclude_none=True) if options else {} - - body: dict[str, Any] = { - "data": { - "attributes": attrs, - "type": "query-runs", - } - } - - r = self.t.request( + self.t.request( "POST", - f"/api/v2/query-runs/{query_run_id}/actions/force-cancel", - json_body=body, + f"/api/v2/queries/{query_run_id}/actions/force-cancel", ) - - jd = r.json() - data = jd.get("data", {}) - attrs = data.get("attributes", {}) - attrs["id"] = data.get("id") - - return QueryRun.model_validate(attrs) diff --git a/src/pytfe/resources/registry_provider_version.py b/src/pytfe/resources/registry_provider_version.py new file mode 100644 index 00000000..f2d4fb34 --- /dev/null +++ b/src/pytfe/resources/registry_provider_version.py @@ -0,0 +1,124 @@ +from __future__ import annotations + +from collections.abc import Iterator +from typing import Any + +from ..errors import ( + RequiredPrivateRegistryError, +) +from ..models.registry_provider import ( + RegistryName, + RegistryProviderID, +) +from ..models.registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, + RegistryProviderVersionListOptions, +) +from ..utils import valid_string_id +from ._base import _Service + + +class RegistryProviderVersions(_Service): + """Registry providers service for managing Terraform registry providers.""" + + def create( + self, + provider_id: RegistryProviderID, + options: RegistryProviderVersionCreateOptions, + ) -> RegistryProviderVersion: + """Create a registry provider version""" + if not self._validate_provider_id(provider_id): + raise ValueError("Invalid provider ID") + + if provider_id.registry_name != RegistryName.PRIVATE: + raise RequiredPrivateRegistryError() + path = f"/api/v2/organizations/{provider_id.organization_name}/registry-providers/{provider_id.registry_name.value}/{provider_id.namespace}/{provider_id.name}/versions" + attributes = options.model_dump(by_alias=True, exclude_none=True) + payload = { + "data": { + "type": "registry-provider-versions", + "attributes": attributes, + } + } + r = self.t.request( + "POST", + path=path, + json_body=payload, + ) + data = r.json().get("data", {}) + return self._registry_provider_version_from(data) + + def _validate_provider_id(self, provider_id: RegistryProviderID) -> bool: + """Validate a registry provider ID.""" + if not valid_string_id(provider_id.organization_name): + return False + if not valid_string_id(provider_id.name): + return False + if not valid_string_id(provider_id.namespace): + return False + if provider_id.registry_name not in [RegistryName.PRIVATE, RegistryName.PUBLIC]: + return False + return True + + def _registry_provider_version_from( + self, data: dict[str, Any] + ) -> RegistryProviderVersion: + """Parse a registry provider version from API response data.""" + + attrs = data.get("attributes", {}) + relationships = data.get("relationships", {}) + attrs["id"] = data.get("id") + + # Parse relationships + if "registry-provider" in relationships: + attrs["registry_provider"] = relationships["registry-provider"].get( + "data", {} + ) + + if "platforms" in relationships: + attrs["registry_provider_platforms"] = relationships["platforms"].get( + "data", [] + ) + + return RegistryProviderVersion.model_validate(attrs) + + def list( + self, + provider_id: RegistryProviderID, + options: RegistryProviderVersionListOptions | None = None, + ) -> Iterator[RegistryProviderVersion]: + """List registry provider versions""" + if not self._validate_provider_id(provider_id): + raise ValueError("Invalid provider ID") + + path = f"/api/v2/organizations/{provider_id.organization_name}/registry-providers/{provider_id.registry_name.value}/{provider_id.namespace}/{provider_id.name}/versions" + params = options.model_dump(by_alias=True) if options else {} + for item in self._list(path=path, params=params): + yield self._registry_provider_version_from(item) + + def read(self, version_id: RegistryProviderVersionID) -> RegistryProviderVersion: + """Read a specific registry provider version""" + if not self._validate_provider_id(version_id): + raise ValueError("Invalid provider ID") + + path = f"/api/v2/organizations/{version_id.organization_name}/registry-providers/{version_id.registry_name.value}/{version_id.namespace}/{version_id.name}/versions/{version_id.version}" + r = self.t.request( + "GET", + path=path, + ) + data = r.json().get("data", {}) + return self._registry_provider_version_from(data) + + def delete(self, version_id: RegistryProviderVersionID) -> None: + """Delete a specific registry provider version""" + if not self._validate_provider_id(version_id): + raise ValueError("Invalid provider ID") + + path = f"/api/v2/organizations/{version_id.organization_name}/registry-providers/{version_id.registry_name.value}/{version_id.namespace}/{version_id.name}/versions/{version_id.version}" + self.t.request( + "DELETE", + path=path, + ) + return None diff --git a/src/pytfe/resources/reserved_tag_key.py b/src/pytfe/resources/reserved_tag_key.py index aeff161c..8eed7fa8 100644 --- a/src/pytfe/resources/reserved_tag_key.py +++ b/src/pytfe/resources/reserved_tag_key.py @@ -1,5 +1,6 @@ from __future__ import annotations +from collections.abc import Iterator from typing import Any from ..errors import ( @@ -7,11 +8,8 @@ ValidationError, ) from ..models.reserved_tag_key import ( - ReservedTagKey as ReservedTagKeyModel, -) -from ..models.reserved_tag_key import ( + ReservedTagKey, ReservedTagKeyCreateOptions, - ReservedTagKeyList, ReservedTagKeyListOptions, ReservedTagKeyUpdateOptions, ) @@ -19,12 +17,12 @@ from ._base import _Service -class ReservedTagKey(_Service): +class ReservedTagKeys(_Service): """Reserved Tag Key API for Terraform Enterprise.""" def list( self, organization: str, options: ReservedTagKeyListOptions | None = None - ) -> ReservedTagKeyList: + ) -> Iterator[ReservedTagKey]: """List reserved tag keys for the given organization.""" if not valid_string_id(organization): raise InvalidOrgError() @@ -32,33 +30,13 @@ def list( params = ( options.model_dump(by_alias=True, exclude_none=True) if options else None ) - - r = self.t.request( - "GET", - f"/api/v2/organizations/{organization}/reserved-tag-keys", - params=params, - ) - - jd = r.json() - items = [] - meta = jd.get("meta", {}) - pagination = meta.get("pagination", {}) - - for d in jd.get("data", []): - items.append(self._parse_reserved_tag_key(d)) - - return ReservedTagKeyList( - items=items, - current_page=pagination.get("current-page"), - total_pages=pagination.get("total-pages"), - prev_page=pagination.get("prev-page"), - next_page=pagination.get("next-page"), - total_count=pagination.get("total-count"), - ) + path = f"/api/v2/organizations/{organization}/reserved-tag-keys" + for item in self._list(path, params=params): + yield self._parse_reserved_tag_key(item) def create( self, organization: str, options: ReservedTagKeyCreateOptions - ) -> ReservedTagKeyModel: + ) -> ReservedTagKey: """Create a new reserved tag key for the given organization.""" if not valid_string_id(organization): raise InvalidOrgError() @@ -82,20 +60,9 @@ def create( return self._parse_reserved_tag_key(data) - def read(self, reserved_tag_key_id: str) -> ReservedTagKeyModel: - """Read a reserved tag key by its ID.""" - if not valid_string_id(reserved_tag_key_id): - raise ValidationError("Invalid reserved tag key ID") - - # Note: Based on the API docs, there's no explicit GET endpoint for individual reserved tag keys - # This method would need to be implemented if such an endpoint exists - raise NotImplementedError( - "Individual reserved tag key read is not supported by the API" - ) - def update( self, reserved_tag_key_id: str, options: ReservedTagKeyUpdateOptions - ) -> ReservedTagKeyModel: + ) -> ReservedTagKey: """Update a reserved tag key.""" if not valid_string_id(reserved_tag_key_id): raise ValidationError("Invalid reserved tag key ID") @@ -125,10 +92,10 @@ def delete(self, reserved_tag_key_id: str) -> None: raise ValidationError("Invalid reserved tag key ID") self.t.request("DELETE", f"/api/v2/reserved-tag-keys/{reserved_tag_key_id}") - # DELETE returns 204 No Content on success + return None - def _parse_reserved_tag_key(self, data: dict[str, Any]) -> ReservedTagKeyModel: + def _parse_reserved_tag_key(self, data: dict[str, Any]) -> ReservedTagKey: """Parse reserved tag key data from API response.""" attrs = data.get("attributes", {}) attrs["id"] = data.get("id") - return ReservedTagKeyModel.model_validate(attrs) + return ReservedTagKey.model_validate(attrs) diff --git a/src/pytfe/utils.py b/src/pytfe/utils.py index d6e9b385..02c43cc1 100644 --- a/src/pytfe/utils.py +++ b/src/pytfe/utils.py @@ -37,7 +37,7 @@ WorkspaceUpdateOptions, ) -_STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{2,}$") +_STRING_ID_PATTERN = re.compile(r"^[^/\s]+$") _WS_ID_RE = re.compile(r"^ws-[A-Za-z0-9]+$") _VERSION_PATTERN = re.compile( r"^\d+\.\d+\.\d+(?:-[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)*)?(?:\+[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)*)?$" diff --git a/tests/units/test_apply.py b/tests/units/test_apply.py index 458c87bd..62f7509d 100644 --- a/tests/units/test_apply.py +++ b/tests/units/test_apply.py @@ -25,7 +25,7 @@ def test_read_apply_validation_errors(self): self.applies.read("") with self.assertRaises(InvalidApplyIDError): - self.applies.read("a") + self.applies.read("! / nope") # Contains spaces and slashes def test_read_apply_success(self): """Test successful apply read.""" diff --git a/tests/units/test_oauth_token.py b/tests/units/test_oauth_token.py index 1af07084..b60ee9bb 100644 --- a/tests/units/test_oauth_token.py +++ b/tests/units/test_oauth_token.py @@ -5,7 +5,7 @@ """ from datetime import datetime -from unittest.mock import Mock +from unittest.mock import Mock, patch import pytest @@ -35,7 +35,6 @@ def test_parse_oauth_token_minimal(self, oauth_tokens_service): data = { "id": "ot-test123", "attributes": { - "uid": "uid-test123", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": False, "service-provider-user": "testuser", @@ -46,7 +45,6 @@ def test_parse_oauth_token_minimal(self, oauth_tokens_service): result = oauth_tokens_service._parse_oauth_token(data) assert result.id == "ot-test123" - assert result.uid == "uid-test123" assert isinstance(result.created_at, datetime) assert result.has_ssh_key is False assert result.service_provider_user == "testuser" @@ -57,7 +55,6 @@ def test_parse_oauth_token_with_oauth_client(self, oauth_tokens_service): data = { "id": "ot-test123", "attributes": { - "uid": "uid-test123", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": True, "service-provider-user": "testuser", @@ -84,7 +81,6 @@ def test_parse_oauth_token_empty_relationships(self, oauth_tokens_service): data = { "id": "ot-test123", "attributes": { - "uid": "uid-test123", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": False, "service-provider-user": "testuser", @@ -119,7 +115,6 @@ def test_list_oauth_tokens_basic(self, oauth_tokens_service, mock_transport): { "id": "ot-test1", "attributes": { - "uid": "uid-test1", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": False, "service-provider-user": "testuser1", @@ -129,7 +124,6 @@ def test_list_oauth_tokens_basic(self, oauth_tokens_service, mock_transport): { "id": "ot-test2", "attributes": { - "uid": "uid-test2", "created-at": "2023-01-02T00:00:00Z", "has-ssh-key": True, "service-provider-user": "testuser2", @@ -149,38 +143,33 @@ def test_list_oauth_tokens_basic(self, oauth_tokens_service, mock_transport): } mock_transport.request.return_value = mock_response - result = oauth_tokens_service.list("test-org") + result = list(oauth_tokens_service.list("test-org")) - mock_transport.request.assert_called_once_with( - "GET", "/api/v2/organizations/test-org/oauth-tokens" - ) - assert len(result.items) == 2 - assert result.items[0].id == "ot-test1" - assert result.items[1].id == "ot-test2" - assert result.current_page == 1 - assert result.total_count == 2 + assert mock_transport.request.call_count == 1 + assert len(result) == 2 + assert result[0].id == "ot-test1" + assert result[1].id == "ot-test2" def test_list_oauth_tokens_with_options(self, oauth_tokens_service, mock_transport): """Test listing OAuth tokens with pagination options.""" - mock_response = Mock() - mock_response.json.return_value = { - "data": [], - "meta": {"pagination": {"current-page": 2}}, - } - mock_transport.request.return_value = mock_response + options = OAuthTokenListOptions(page_size=50) - options = OAuthTokenListOptions(page_number=2, page_size=50) - oauth_tokens_service.list("test-org", options) + with patch.object(oauth_tokens_service, "_list") as mock_list: + mock_list.return_value = [] - mock_transport.request.assert_called_once_with( - "GET", - "/api/v2/organizations/test-org/oauth-tokens?page[number]=2&page[size]=50", - ) + list(oauth_tokens_service.list("test-org", options)) + + expected_params = { + "page[size]": "50", + } + mock_list.assert_called_once_with( + "/api/v2/organizations/test-org/oauth-tokens", params=expected_params + ) def test_list_oauth_tokens_invalid_org(self, oauth_tokens_service): """Test listing OAuth tokens with invalid organization ID.""" with pytest.raises(ValueError, match=ERR_INVALID_ORG): - oauth_tokens_service.list("") + list(oauth_tokens_service.list("")) def test_read_oauth_token_success(self, oauth_tokens_service, mock_transport): """Test reading an OAuth token successfully.""" @@ -189,7 +178,6 @@ def test_read_oauth_token_success(self, oauth_tokens_service, mock_transport): "data": { "id": "ot-test123", "attributes": { - "uid": "uid-test123", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": False, "service-provider-user": "testuser", @@ -205,7 +193,6 @@ def test_read_oauth_token_success(self, oauth_tokens_service, mock_transport): "GET", "/api/v2/oauth-tokens/ot-test123" ) assert result.id == "ot-test123" - assert result.uid == "uid-test123" def test_read_oauth_token_invalid_id(self, oauth_tokens_service): """Test reading an OAuth token with invalid ID.""" @@ -219,7 +206,6 @@ def test_update_oauth_token_success(self, oauth_tokens_service, mock_transport): "data": { "id": "ot-test123", "attributes": { - "uid": "uid-test123", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": True, "service-provider-user": "testuser", @@ -253,7 +239,6 @@ def test_update_oauth_token_no_ssh_key(self, oauth_tokens_service, mock_transpor "data": { "id": "ot-test123", "attributes": { - "uid": "uid-test123", "created-at": "2023-01-01T00:00:00Z", "has-ssh-key": False, "service-provider-user": "testuser", @@ -308,9 +293,8 @@ def oauth_tokens_service(self): def test_oauth_token_list_options(self, oauth_tokens_service): """Test OAuth token list options creation.""" - options = OAuthTokenListOptions(page_number=1, page_size=25) + options = OAuthTokenListOptions(page_size=25) - assert options.page_number == 1 assert options.page_size == 25 def test_oauth_token_update_options(self, oauth_tokens_service): diff --git a/tests/units/test_policy_evaluation.py b/tests/units/test_policy_evaluation.py new file mode 100644 index 00000000..820496b8 --- /dev/null +++ b/tests/units/test_policy_evaluation.py @@ -0,0 +1,211 @@ +"""Unit tests for the policy evaluation module.""" + +from unittest.mock import Mock + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import InvalidTaskStageIDError +from pytfe.models.policy_evaluation import ( + PolicyEvaluation, + PolicyEvaluationListOptions, + PolicyEvaluationStatus, +) +from pytfe.resources.policy_evaluation import PolicyEvaluations + + +class TestPolicyEvaluations: + """Test the PolicyEvaluations service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def policy_evaluations_service(self, mock_transport): + """Create a PolicyEvaluations service with mocked transport.""" + return PolicyEvaluations(mock_transport) + + def test_list_validations(self, policy_evaluations_service): + """Test list method with invalid task stage ID.""" + + # Test empty task stage ID + with pytest.raises(InvalidTaskStageIDError): + list(policy_evaluations_service.list("")) + + # Test None task stage ID + with pytest.raises(InvalidTaskStageIDError): + list(policy_evaluations_service.list(None)) + + def test_list_success_with_options( + self, policy_evaluations_service, mock_transport + ): + """Test successful iteration with custom pagination options.""" + + mock_response_data = { + "data": [ + { + "id": "poleval-456", + "type": "policy-evaluations", + "attributes": { + "status": "failed", + "policy-kind": "opa", + "status-timestamp": { + "passed-at": None, + "failed-at": "2023-01-02T12:00:00Z", + "running-at": "2023-01-02T11:59:00Z", + "canceled-at": None, + "errored-at": None, + }, + "result-count": { + "advisory-failed": 2, + "mandatory-failed": 1, + "passed": 3, + "errored": 0, + }, + "created-at": "2023-01-02T11:58:00Z", + "updated-at": "2023-01-02T12:00:00Z", + }, + "relationships": { + "policy-attachable": { + "data": {"id": "ts-456", "type": "task-stages"} + } + }, + } + ] + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = PolicyEvaluationListOptions(page_size=5) + result = list(policy_evaluations_service.list("ts-456", options=options)) + + # Verify the request was made with correct parameters + assert mock_transport.request.call_count == 1 + call_args = mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert call_args[0][1] == "api/v2/task-stages/ts-456/policy-evaluations" + + # Verify custom options were passed and merged with _list defaults + params = call_args[1]["params"] + assert params["page[size]"] == 5 # Custom value from options + + # Verify the result + assert len(result) == 1 + assert isinstance(result[0], PolicyEvaluation) + assert result[0].id == "poleval-456" + assert result[0].status == PolicyEvaluationStatus.POLICYEVALUATIONFAILED + assert result[0].result_count.advisory_failed == 2 + assert result[0].result_count.mandatory_failed == 1 + + def test_list_empty_result(self, policy_evaluations_service, mock_transport): + """Test iteration with no results.""" + + mock_response_data = {"data": []} + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + result = list(policy_evaluations_service.list("ts-empty")) + + # Verify the request was made + assert mock_transport.request.call_count == 1 + + # Verify iterator yields no items + assert len(result) == 0 + assert result == [] + + def test_list_with_different_statuses( + self, policy_evaluations_service, mock_transport + ): + """Test list operation returns evaluations with different statuses.""" + + mock_response_data = { + "data": [ + { + "id": "poleval-pending", + "type": "policy-evaluations", + "attributes": { + "status": "pending", + "policy-kind": "opa", + "status-timestamp": {}, + "result-count": { + "advisory-failed": 0, + "mandatory-failed": 0, + "passed": 0, + "errored": 0, + }, + "created-at": "2023-01-01T11:58:00Z", + "updated-at": "2023-01-01T11:58:00Z", + }, + "relationships": { + "policy-attachable": { + "data": {"id": "ts-multi", "type": "task-stages"} + } + }, + }, + { + "id": "poleval-running", + "type": "policy-evaluations", + "attributes": { + "status": "running", + "policy-kind": "opa", + "status-timestamp": {"running-at": "2023-01-01T11:59:00Z"}, + "result-count": { + "advisory-failed": 0, + "mandatory-failed": 0, + "passed": 0, + "errored": 0, + }, + "created-at": "2023-01-01T11:58:00Z", + "updated-at": "2023-01-01T11:59:00Z", + }, + "relationships": { + "policy-attachable": { + "data": {"id": "ts-multi", "type": "task-stages"} + } + }, + }, + { + "id": "poleval-errored", + "type": "policy-evaluations", + "attributes": { + "status": "errored", + "policy-kind": "opa", + "status-timestamp": {"errored-at": "2023-01-01T12:00:00Z"}, + "result-count": { + "advisory-failed": 0, + "mandatory-failed": 0, + "passed": 0, + "errored": 1, + }, + "created-at": "2023-01-01T11:58:00Z", + "updated-at": "2023-01-01T12:00:00Z", + }, + "relationships": { + "policy-attachable": { + "data": {"id": "ts-multi", "type": "task-stages"} + } + }, + }, + ] + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + result = list(policy_evaluations_service.list("ts-multi")) + + # Verify the iterator yields all items with correct statuses + assert len(result) == 3 + assert result[0].status == PolicyEvaluationStatus.POLICYEVALUATIONPENDING + assert result[1].status == PolicyEvaluationStatus.POLICYEVALUATIONRUNNING + assert result[2].status == PolicyEvaluationStatus.POLICYEVALUATIONERRORED + + # Verify all are PolicyEvaluation instances + assert all(isinstance(item, PolicyEvaluation) for item in result) diff --git a/tests/units/test_project.py b/tests/units/test_project.py index 262787af..801a29f8 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -345,7 +345,9 @@ def test_list_tag_bindings_invalid_project_id(self): with pytest.raises( ValueError, match="Project ID is required and must be valid" ): - self.projects_service.list_tag_bindings("x") # Too short + self.projects_service.list_tag_bindings( + "! / nope" + ) # Contains spaces and slashes def test_list_effective_tag_bindings_success(self): """Test successful listing of effective tag bindings""" @@ -541,5 +543,5 @@ def test_delete_tag_bindings_invalid_project_id(self): ValueError, match="Project ID is required and must be valid" ): self.projects_service.delete_tag_bindings( - "ab" - ) # Too short (needs at least 3 chars) + "bad/id" + ) # Contains forward slash diff --git a/tests/units/test_query_run.py b/tests/units/test_query_run.py index 8808090c..3409d13f 100644 --- a/tests/units/test_query_run.py +++ b/tests/units/test_query_run.py @@ -1,564 +1,551 @@ -from datetime import datetime -from unittest.mock import MagicMock, Mock +""" +Comprehensive unit tests for query run operations in the Python TFE SDK. + +This test suite covers all query run methods including: +1. list() - List query runs for a workspace with pagination +2. create() - Create new query runs +3. read() - Read query run details +4. read_with_options() - Read with include options +5. logs() - Retrieve query run logs +6. cancel() - Cancel a query run +7. force_cancel() - Force cancel a query run + +Usage: + pytest tests/units/test_query_run.py -v +""" + +from unittest.mock import Mock, patch import pytest -from pytfe import TFEClient, TFEConfig -from pytfe.errors import InvalidOrgError, InvalidQueryRunIDError -from pytfe.models.query_run import ( +from pytfe.errors import InvalidQueryRunIDError, InvalidWorkspaceIDError +from pytfe.models import ( QueryRun, - QueryRunCancelOptions, QueryRunCreateOptions, - QueryRunForceCancelOptions, - QueryRunList, + QueryRunIncludeOpt, QueryRunListOptions, - QueryRunLogs, QueryRunReadOptions, - QueryRunResults, + QueryRunSource, QueryRunStatus, - QueryRunType, + QueryRunStatusTimestamps, + QueryRunVariable, ) +from pytfe.resources.query_run import QueryRuns + +# ============================================================================ +# Fixtures +# ============================================================================ + + +@pytest.fixture +def mock_transport(): + """Create a mock HTTPTransport.""" + return Mock() + + +@pytest.fixture +def query_runs_service(mock_transport): + """Create a QueryRuns service with mocked transport.""" + return QueryRuns(mock_transport) + + +@pytest.fixture +def sample_query_run_data(): + """Sample query run data from API.""" + return { + "id": "qr-123abc456def", + "type": "queries", + "attributes": { + "source": "tfe-api", + "status": "finished", + "created-at": "2024-01-15T10:00:00Z", + "updated-at": "2024-01-15T10:05:00Z", + "canceled-at": None, + "log-read-url": "https://app.terraform.io/api/v2/queries/qr-123abc456def/logs", + "status-timestamps": { + "queued-at": "2024-01-15T10:00:00Z", + "running-at": "2024-01-15T10:01:00Z", + "finished-at": "2024-01-15T10:05:00Z", + }, + "variables": [ + {"key": "environment", "value": "production"}, + {"key": "region", "value": "us-east-1"}, + ], + "actions": { + "is-cancelable": True, + "is-force-cancelable": False, + }, + }, + "relationships": { + "workspace": {"data": {"id": "ws-abc123", "type": "workspaces"}}, + "configuration-version": { + "data": {"id": "cv-def456", "type": "configuration-versions"} + }, + "created-by": {"data": {"id": "user-123", "type": "users"}}, + }, + } + + +@pytest.fixture +def sample_query_run_list_response(sample_query_run_data): + """Sample query run list response.""" + return { + "data": [ + sample_query_run_data, + { + "id": "qr-789ghi012jkl", + "type": "queries", + "attributes": { + "source": "tfe-api", + "status": "running", + "created-at": "2024-01-15T11:00:00Z", + "updated-at": "2024-01-15T11:02:00Z", + "canceled-at": None, + "log-read-url": None, + "status-timestamps": { + "queued-at": "2024-01-15T11:00:00Z", + "running-at": "2024-01-15T11:01:00Z", + }, + "variables": [], + "actions": { + "is-cancelable": True, + "is-force-cancelable": False, + }, + }, + }, + ], + "meta": { + "pagination": { + "current-page": 1, + "page-size": 20, + "total-pages": 1, + "total-count": 2, + } + }, + "links": {"next": None}, + } -class TestQueryRunModels: - """Test query run models and validation.""" +# ============================================================================ +# List Operations Tests +# ============================================================================ - def test_query_run_model_basic(self): - """Test basic QueryRun model creation.""" - query_run = QueryRun( - id="qr-test123", - query="SELECT * FROM runs WHERE status = 'completed'", - query_type=QueryRunType.FILTER, - status=QueryRunStatus.COMPLETED, - created_at=datetime.now(), - updated_at=datetime.now(), - ) - assert query_run.id == "qr-test123" - assert query_run.query == "SELECT * FROM runs WHERE status = 'completed'" - assert query_run.query_type == QueryRunType.FILTER - assert query_run.status == QueryRunStatus.COMPLETED - - def test_query_run_status_enum(self): - """Test QueryRunStatus enum values.""" - assert QueryRunStatus.PENDING == "pending" - assert QueryRunStatus.RUNNING == "running" - assert QueryRunStatus.COMPLETED == "completed" - assert QueryRunStatus.ERRORED == "errored" - assert QueryRunStatus.CANCELED == "canceled" - - def test_query_run_type_enum(self): - """Test QueryRunType enum values.""" - assert QueryRunType.FILTER == "filter" - assert QueryRunType.SEARCH == "search" - assert QueryRunType.ANALYTICS == "analytics" - - def test_query_run_create_options(self): - """Test QueryRunCreateOptions model.""" - options = QueryRunCreateOptions( - query="SELECT * FROM workspaces", - query_type=QueryRunType.SEARCH, - organization_name="test-org", - timeout_seconds=300, - max_results=1000, - ) - assert options.query == "SELECT * FROM workspaces" - assert options.query_type == QueryRunType.SEARCH - assert options.organization_name == "test-org" - assert options.timeout_seconds == 300 - assert options.max_results == 1000 - - def test_query_run_list_options(self): - """Test QueryRunListOptions model.""" - options = QueryRunListOptions( - page_number=2, - page_size=50, - query_type=QueryRunType.FILTER, - status=QueryRunStatus.COMPLETED, - organization_name="test-org", - ) - assert options.page_number == 2 - assert options.page_size == 50 - assert options.query_type == QueryRunType.FILTER - assert options.status == QueryRunStatus.COMPLETED - assert options.organization_name == "test-org" - - -class TestQueryRunOperations: - """Test query run operations.""" - - @pytest.fixture - def client(self): - """Create a test client.""" - config = TFEConfig(address="https://test.terraform.io", token="test-token") - return TFEClient(config) - - @pytest.fixture - def mock_response(self): - """Create a mock response.""" - mock = Mock() - mock.json.return_value = { - "data": [ - { - "id": "qr-test123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs", - "query-type": "filter", - "status": "completed", - "results-count": 42, - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:05:00Z", - "started-at": "2023-01-01T00:01:00Z", - "finished-at": "2023-01-01T00:05:00Z", - "organization-name": "test-org", - }, - } - ], - "meta": { - "pagination": { - "current-page": 1, - "total-pages": 1, - "prev-page": None, - "next-page": None, - "total-count": 1, - } - }, - } - return mock - def test_list_query_runs(self, client, mock_response): - """Test listing query runs.""" - client._transport.request = MagicMock(return_value=mock_response) +class TestQueryRunsList: + """Test suite for query run list operations.""" - result = client.query_runs.list("test-org") + def test_list_basic( + self, query_runs_service, mock_transport, sample_query_run_list_response + ): + """Test basic query run listing.""" + mock_response = Mock() + mock_response.json.return_value = sample_query_run_list_response + mock_transport.request.return_value = mock_response - assert isinstance(result, QueryRunList) - assert len(result.items) == 1 - assert result.items[0].id == "qr-test123" - assert result.items[0].query == "SELECT * FROM runs" - assert result.current_page == 1 - assert result.total_count == 1 + workspace_id = "ws-abc123" + query_runs = list(query_runs_service.list(workspace_id)) - client._transport.request.assert_called_once_with( - "GET", "/api/v2/organizations/test-org/query-runs", params=None + # Verify the request + mock_transport.request.assert_called_with( + "GET", + f"/api/v2/workspaces/{workspace_id}/queries", + params={"page[number]": 1, "page[size]": 100}, ) - def test_list_query_runs_with_options(self, client, mock_response): - """Test listing query runs with options.""" - client._transport.request = MagicMock(return_value=mock_response) + # Verify the results + assert len(query_runs) == 2 + + # Check first query run + qr1 = query_runs[0] + assert qr1.id == "qr-123abc456def" + assert qr1.status == QueryRunStatus.FINISHED + assert qr1.source == QueryRunSource.API + assert qr1.log_read_url is not None + assert len(qr1.variables) == 2 + assert qr1.variables[0].key == "environment" + assert qr1.variables[0].value == "production" + + # Check second query run + qr2 = query_runs[1] + assert qr2.id == "qr-789ghi012jkl" + assert qr2.status == QueryRunStatus.RUNNING + assert qr2.log_read_url is None + assert len(qr2.variables) == 0 + + def test_list_with_options( + self, query_runs_service, mock_transport, sample_query_run_list_response + ): + """Test list with options.""" + mock_response = Mock() + mock_response.json.return_value = sample_query_run_list_response + mock_transport.request.return_value = mock_response + workspace_id = "ws-abc123" options = QueryRunListOptions( - page_number=2, - page_size=25, - query_type=QueryRunType.FILTER, - status=QueryRunStatus.COMPLETED, + page_number=1, + page_size=10, + include=[ + QueryRunIncludeOpt.CREATED_BY, + QueryRunIncludeOpt.CONFIGURATION_VERSION, + ], ) - result = client.query_runs.list("test-org", options) - assert isinstance(result, QueryRunList) - client._transport.request.assert_called_once_with( - "GET", - "/api/v2/organizations/test-org/query-runs", - params={ - "page[number]": 2, - "page[size]": 25, - "filter[query-type]": "filter", - "filter[status]": "completed", - }, - ) + query_runs = list(query_runs_service.list(workspace_id, options)) + + # Verify the request includes options + call_args = mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert call_args[0][1] == f"/api/v2/workspaces/{workspace_id}/queries" + params = call_args[1]["params"] + assert params["page[number]"] == 1 + assert params["page[size]"] == 10 + assert params["include"] == "created_by,configuration_version" + + assert len(query_runs) == 2 + + def test_list_invalid_workspace_id(self, query_runs_service): + """Test list with invalid workspace ID.""" + with pytest.raises(InvalidWorkspaceIDError): + list(query_runs_service.list("")) + + with pytest.raises(InvalidWorkspaceIDError): + list(query_runs_service.list(None)) - def test_create_query_run(self, client): - """Test creating a query run.""" + +# ============================================================================ +# Create Operations Tests +# ============================================================================ + + +class TestQueryRunsCreate: + """Test suite for query run create operations.""" + + def test_create_basic( + self, query_runs_service, mock_transport, sample_query_run_data + ): + """Test basic query run creation.""" mock_response = Mock() - mock_response.json.return_value = { - "data": { - "id": "qr-new123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM workspaces", - "query-type": "search", - "status": "pending", - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:00:00Z", - "organization-name": "test-org", - }, - } - } - client._transport.request = MagicMock(return_value=mock_response) + mock_response.json.return_value = {"data": sample_query_run_data} + mock_transport.request.return_value = mock_response options = QueryRunCreateOptions( - query="SELECT * FROM workspaces", - query_type=QueryRunType.SEARCH, - organization_name="test-org", - timeout_seconds=300, + source=QueryRunSource.API, + workspace_id="ws-abc123", + configuration_version_id="cv-def456", ) - result = client.query_runs.create("test-org", options) - assert isinstance(result, QueryRun) - assert result.id == "qr-new123" - assert result.query == "SELECT * FROM workspaces" - assert result.status == QueryRunStatus.PENDING + result = query_runs_service.create(options) - client._transport.request.assert_called_once_with( - "POST", - "/api/v2/organizations/test-org/query-runs", - json_body={ - "data": { - "attributes": { - "query": "SELECT * FROM workspaces", - "query-type": "search", - "organization-name": "test-org", - "timeout-seconds": 300, - }, - "type": "query-runs", - } - }, + # Verify the request + call_args = mock_transport.request.call_args + assert call_args[0][0] == "POST" + assert call_args[0][1] == "/api/v2/queries" + + json_body = call_args[1]["json_body"] + assert json_body["data"]["type"] == "queries" + assert json_body["data"]["attributes"]["source"] == "tfe-api" + assert ( + json_body["data"]["relationships"]["workspace"]["data"]["id"] == "ws-abc123" + ) + assert ( + json_body["data"]["relationships"]["configuration-version"]["data"]["id"] + == "cv-def456" ) - def test_read_query_run(self, client): - """Test reading a query run.""" + # Verify the result + assert isinstance(result, QueryRun) + assert result.id == "qr-123abc456def" + assert result.status == QueryRunStatus.FINISHED + assert result.source == QueryRunSource.API + + def test_create_with_variables( + self, query_runs_service, mock_transport, sample_query_run_data + ): + """Test query run creation with variables.""" mock_response = Mock() - mock_response.json.return_value = { - "data": { - "id": "qr-test123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs", - "query-type": "filter", - "status": "completed", - "results-count": 42, - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:05:00Z", - }, - } - } - client._transport.request = MagicMock(return_value=mock_response) + mock_response.json.return_value = {"data": sample_query_run_data} + mock_transport.request.return_value = mock_response - result = client.query_runs.read("qr-test123") - - assert isinstance(result, QueryRun) - assert result.id == "qr-test123" - assert result.status == QueryRunStatus.COMPLETED - assert result.results_count == 42 + variables = [ + QueryRunVariable(key="environment", value="production"), + QueryRunVariable(key="region", value="us-east-1"), + ] - client._transport.request.assert_called_once_with( - "GET", "/api/v2/query-runs/qr-test123" + options = QueryRunCreateOptions( + source=QueryRunSource.API, + workspace_id="ws-abc123", + configuration_version_id="cv-def456", + variables=variables, ) - def test_read_query_run_with_options(self, client): - """Test reading a query run with options.""" - mock_response = Mock() - mock_response.json.return_value = { - "data": { - "id": "qr-test123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs", - "query-type": "filter", - "status": "completed", - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:05:00Z", - }, - } - } - client._transport.request = MagicMock(return_value=mock_response) + result = query_runs_service.create(options) - options = QueryRunReadOptions(include_results=True, include_logs=True) - result = client.query_runs.read_with_options("qr-test123", options) + # Verify variables in request + call_args = mock_transport.request.call_args + json_body = call_args[1]["json_body"] + assert "variables" in json_body["data"]["attributes"] + assert len(json_body["data"]["attributes"]["variables"]) == 2 - assert isinstance(result, QueryRun) - assert result.id == "qr-test123" + # Verify result + assert result.id == "qr-123abc456def" + assert len(result.variables) == 2 - client._transport.request.assert_called_once_with( - "GET", - "/api/v2/query-runs/qr-test123", - params={"include[results]": True, "include[logs]": True}, - ) - def test_query_run_logs(self, client): - """Test retrieving query run logs.""" +# ============================================================================ +# Read Operations Tests +# ============================================================================ + + +class TestQueryRunsRead: + """Test suite for query run read operations.""" + + def test_read_success( + self, query_runs_service, mock_transport, sample_query_run_data + ): + """Test successful query run read.""" mock_response = Mock() - mock_response.headers = {"content-type": "text/plain"} - mock_response.text = ( - "Starting query execution...\nQuery completed successfully." + mock_response.json.return_value = {"data": sample_query_run_data} + mock_transport.request.return_value = mock_response + + result = query_runs_service.read("qr-123abc456def") + + # Verify the request + mock_transport.request.assert_called_once_with( + "GET", "/api/v2/queries/qr-123abc456def" ) - client._transport.request = MagicMock(return_value=mock_response) - result = client.query_runs.logs("qr-test123") + # Verify the result + assert isinstance(result, QueryRun) + assert result.id == "qr-123abc456def" + assert result.status == QueryRunStatus.FINISHED + assert result.source == QueryRunSource.API + assert result.log_read_url is not None - assert isinstance(result, QueryRunLogs) - assert result.query_run_id == "qr-test123" - assert "Starting query execution" in result.logs - assert result.log_level == "info" + def test_read_invalid_id(self, query_runs_service): + """Test read with invalid query run ID.""" + with pytest.raises(InvalidQueryRunIDError): + query_runs_service.read("") - client._transport.request.assert_called_once_with( - "GET", "/api/v2/query-runs/qr-test123/logs" - ) + with pytest.raises(InvalidQueryRunIDError): + query_runs_service.read(None) - def test_query_run_results(self, client): - """Test retrieving query run results.""" + def test_read_with_options_success( + self, query_runs_service, mock_transport, sample_query_run_data + ): + """Test read with options.""" mock_response = Mock() - mock_response.json.return_value = { - "data": { - "results": [ - {"id": "run-1", "status": "completed"}, - {"id": "run-2", "status": "pending"}, - ], - "total_count": 2, - "truncated": False, - } - } - client._transport.request = MagicMock(return_value=mock_response) + mock_response.json.return_value = {"data": sample_query_run_data} + mock_transport.request.return_value = mock_response + + options = QueryRunReadOptions( + include=[ + QueryRunIncludeOpt.CREATED_BY, + QueryRunIncludeOpt.CONFIGURATION_VERSION, + ] + ) - result = client.query_runs.results("qr-test123") + result = query_runs_service.read_with_options("qr-123abc456def", options) - assert isinstance(result, QueryRunResults) - assert result.query_run_id == "qr-test123" - assert len(result.results) == 2 - assert result.total_count == 2 - assert not result.truncated + # Verify the request includes options + call_args = mock_transport.request.call_args + assert call_args[0][0] == "GET" + assert call_args[0][1] == "/api/v2/queries/qr-123abc456def" + params = call_args[1]["params"] + assert params["include"] == "created_by,configuration_version" - client._transport.request.assert_called_once_with( - "GET", "/api/v2/query-runs/qr-test123/results" - ) + # Verify the result + assert result.id == "qr-123abc456def" - def test_cancel_query_run(self, client): - """Test canceling a query run.""" - mock_response = Mock() - mock_response.json.return_value = { - "data": { - "id": "qr-test123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs", - "query-type": "filter", - "status": "canceled", - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:02:00Z", - }, - } - } - client._transport.request = MagicMock(return_value=mock_response) - options = QueryRunCancelOptions(reason="User requested cancellation") - result = client.query_runs.cancel("qr-test123", options) +# ============================================================================ +# Logs Operations Tests +# ============================================================================ - assert isinstance(result, QueryRun) - assert result.id == "qr-test123" - assert result.status == QueryRunStatus.CANCELED - client._transport.request.assert_called_once_with( - "POST", - "/api/v2/query-runs/qr-test123/actions/cancel", - json_body={ - "data": { - "attributes": {"reason": "User requested cancellation"}, - "type": "query-runs", - } - }, +class TestQueryRunsLogs: + """Test suite for query run logs operations.""" + + def test_logs_success(self, query_runs_service, mock_transport): + """Test successful logs retrieval.""" + # Mock the read method to return a query run with log URL + mock_query_run = Mock() + mock_query_run.log_read_url = ( + "https://app.terraform.io/api/v2/queries/qr-123/logs" ) - def test_force_cancel_query_run(self, client): - """Test force canceling a query run.""" - mock_response = Mock() - mock_response.json.return_value = { - "data": { - "id": "qr-test123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs", - "query-type": "filter", - "status": "canceled", - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:02:00Z", - }, - } - } - client._transport.request = MagicMock(return_value=mock_response) + # Mock the logs content + mock_logs_response = Mock() + mock_logs_response.content = b"Query run logs content\nLine 2\nLine 3" - options = QueryRunForceCancelOptions(reason="Force cancel due to timeout") - result = client.query_runs.force_cancel("qr-test123", options) + with patch.object(query_runs_service, "read", return_value=mock_query_run): + mock_transport.request.return_value = mock_logs_response - assert isinstance(result, QueryRun) - assert result.id == "qr-test123" - assert result.status == QueryRunStatus.CANCELED + result = query_runs_service.logs("qr-123abc456def") - client._transport.request.assert_called_once_with( - "POST", - "/api/v2/query-runs/qr-test123/actions/force-cancel", - json_body={ - "data": { - "attributes": {"reason": "Force cancel due to timeout"}, - "type": "query-runs", - } - }, - ) + # Verify read was called + query_runs_service.read.assert_called_once_with("qr-123abc456def") + # Verify logs request was made + mock_transport.request.assert_called_once_with( + "GET", "https://app.terraform.io/api/v2/queries/qr-123/logs" + ) -class TestQueryRunErrorHandling: - """Test query run error handling.""" + # Verify the result is an IO stream + assert result.read() == b"Query run logs content\nLine 2\nLine 3" - @pytest.fixture - def client(self): - """Create a test client.""" - config = TFEConfig(address="https://test.terraform.io", token="test-token") - return TFEClient(config) + def test_logs_no_url_error(self, query_runs_service): + """Test logs method when query run has no log URL.""" + mock_query_run = Mock() + mock_query_run.log_read_url = None - def test_invalid_organization_error(self, client): - """Test invalid organization error.""" - with pytest.raises(InvalidOrgError): - client.query_runs.list("") + with patch.object(query_runs_service, "read", return_value=mock_query_run): + with pytest.raises(ValueError) as exc: + query_runs_service.logs("qr-123abc456def") - with pytest.raises(InvalidOrgError): - client.query_runs.list(None) + assert "does not have a log URL" in str(exc.value) - def test_invalid_query_run_id_error(self, client): - """Test invalid query run ID error.""" + def test_logs_invalid_id(self, query_runs_service): + """Test logs with invalid query run ID.""" with pytest.raises(InvalidQueryRunIDError): - client.query_runs.read("") + query_runs_service.logs("") - with pytest.raises(InvalidQueryRunIDError): - client.query_runs.read(None) - with pytest.raises(InvalidQueryRunIDError): - client.query_runs.logs("") +# ============================================================================ +# Cancel Operations Tests +# ============================================================================ - with pytest.raises(InvalidQueryRunIDError): - client.query_runs.results("") - with pytest.raises(InvalidQueryRunIDError): - client.query_runs.cancel("") +class TestQueryRunsCancel: + """Test suite for query run cancel operations.""" + + def test_cancel_success(self, query_runs_service, mock_transport): + """Test successful query run cancellation.""" + mock_response = Mock() + mock_transport.request.return_value = mock_response + + query_runs_service.cancel("qr-123abc456def") + + # Verify the request + mock_transport.request.assert_called_once_with( + "POST", + "/api/v2/queries/qr-123abc456def/actions/cancel", + ) + def test_cancel_invalid_id(self, query_runs_service): + """Test cancel with invalid query run ID.""" with pytest.raises(InvalidQueryRunIDError): - client.query_runs.force_cancel("") + query_runs_service.cancel("") - def test_create_query_run_validation_errors(self, client): - """Test create query run validation errors.""" - with pytest.raises(InvalidOrgError): - options = QueryRunCreateOptions( - query="SELECT * FROM runs", query_type=QueryRunType.FILTER - ) - client.query_runs.create("", options) +# ============================================================================ +# Force Cancel Operations Tests +# ============================================================================ -class TestQueryRunIntegration: - """Test query run integration scenarios.""" - @pytest.fixture - def client(self): - """Create a test client with mocked transport.""" - from unittest.mock import MagicMock, patch +class TestQueryRunsForceCancel: + """Test suite for query run force cancel operations.""" - # Mock the HTTPTransport to prevent any network calls during initialization - with patch("pytfe.client.HTTPTransport") as mock_transport_class: - mock_transport_instance = MagicMock() - mock_transport_class.return_value = mock_transport_instance + def test_force_cancel_success(self, query_runs_service, mock_transport): + """Test successful force cancellation.""" + mock_response = Mock() + mock_transport.request.return_value = mock_response - config = TFEConfig(address="https://test.terraform.io", token="test-token") - client = TFEClient(config) - return client + query_runs_service.force_cancel("qr-123abc456def") - def test_full_query_run_workflow(self, client): - """Test a complete query run workflow simulation.""" - # Use the already mocked transport from the fixture - mock_transport = client._transport + # Verify the request + mock_transport.request.assert_called_once_with( + "POST", + "/api/v2/queries/qr-123abc456def/actions/force-cancel", + ) - # 1. Create query run - create_response = Mock() - create_response.json.return_value = { - "data": { - "id": "qr-workflow123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs WHERE status = 'completed'", - "query-type": "filter", - "status": "pending", - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:00:00Z", - "organization-name": "test-org", - }, - } - } - - # 2. Read query run (running state) - read_response = Mock() - read_response.json.return_value = { - "data": { - "id": "qr-workflow123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs WHERE status = 'completed'", - "query-type": "filter", - "status": "running", - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:01:00Z", - "started-at": "2023-01-01T00:01:00Z", - }, - } - } - - # 3. Read query run (completed state) - completed_response = Mock() - completed_response.json.return_value = { - "data": { - "id": "qr-workflow123", - "type": "query-runs", - "attributes": { - "query": "SELECT * FROM runs WHERE status = 'completed'", - "query-type": "filter", - "status": "completed", - "results-count": 15, - "created-at": "2023-01-01T00:00:00Z", - "updated-at": "2023-01-01T00:05:00Z", - "started-at": "2023-01-01T00:01:00Z", - "finished-at": "2023-01-01T00:05:00Z", - }, - } - } - - # 4. Get results - results_response = Mock() - results_response.json.return_value = { - "data": { - "results": [ - {"id": f"run-{i}", "status": "completed"} for i in range(15) - ], - "total_count": 15, - "truncated": False, - } - } + def test_force_cancel_invalid_id(self, query_runs_service): + """Test force cancel with invalid query run ID.""" + with pytest.raises(InvalidQueryRunIDError): + query_runs_service.force_cancel("") - mock_transport.request.side_effect = [ - create_response, - read_response, - completed_response, - results_response, + +# ============================================================================ +# Unit Tests - Model Validation +# ============================================================================ + + +class TestQueryRunCreateOptions: + """Unit tests for QueryRunCreateOptions model.""" + + def test_create_with_required_fields(self): + """Test creating options with required fields only.""" + options = QueryRunCreateOptions( + source=QueryRunSource.API, + workspace_id="ws-123", + ) + + assert options.source == QueryRunSource.API + assert options.workspace_id == "ws-123" + assert options.configuration_version_id is None + assert options.variables is None + + def test_create_with_all_fields(self): + """Test creating options with all fields.""" + variables = [ + QueryRunVariable(key="var1", value="value1"), + QueryRunVariable(key="var2", value="value2"), ] - # Execute workflow options = QueryRunCreateOptions( - query="SELECT * FROM runs WHERE status = 'completed'", - query_type=QueryRunType.FILTER, - organization_name="test-org", + source=QueryRunSource.API, + workspace_id="ws-123", + configuration_version_id="cv-456", + variables=variables, ) - # 1. Create - query_run = client.query_runs.create("test-org", options) - assert query_run.status == QueryRunStatus.PENDING + assert options.source == QueryRunSource.API + assert options.workspace_id == "ws-123" + assert options.configuration_version_id == "cv-456" + assert len(options.variables) == 2 + assert options.variables[0].key == "var1" + + +class TestQueryRunModel: + """Unit tests for QueryRun model.""" + + def test_status_enum_values(self): + """Test all status enum values.""" + assert QueryRunStatus.PENDING.value == "pending" + assert QueryRunStatus.QUEUED.value == "queued" + assert QueryRunStatus.RUNNING.value == "running" + assert QueryRunStatus.FINISHED.value == "finished" + assert QueryRunStatus.ERRORED.value == "errored" + assert QueryRunStatus.CANCELED.value == "canceled" + + def test_source_enum_value(self): + """Test source enum value.""" + assert QueryRunSource.API.value == "tfe-api" + + +# ============================================================================ +# Test Utilities +# ============================================================================ + + +def test_query_run_variable(): + """Test QueryRunVariable model.""" + var = QueryRunVariable(key="test_key", value="test_value") - # 2. Check status (running) - query_run = client.query_runs.read(query_run.id) - assert query_run.status == QueryRunStatus.RUNNING + assert var.key == "test_key" + assert var.value == "test_value" - # 3. Check status (completed) - query_run = client.query_runs.read(query_run.id) - assert query_run.status == QueryRunStatus.COMPLETED - assert query_run.results_count == 15 - # 4. Get results - results = client.query_runs.results(query_run.id) - assert len(results.results) == 15 - assert not results.truncated +def test_query_run_status_timestamps(): + """Test QueryRunStatusTimestamps model.""" + timestamps = QueryRunStatusTimestamps( + queued_at="2024-01-15T10:00:00Z", + running_at="2024-01-15T10:05:00Z", + errored_at="2024-01-15T10:10:00Z", + ) - # Verify all calls were made - assert mock_transport.request.call_count == 4 + # Timestamps are datetime objects + assert timestamps.queued_at is not None + assert timestamps.running_at is not None + assert timestamps.errored_at is not None + assert timestamps.finished_at is None + assert timestamps.canceled_at is None diff --git a/tests/units/test_registry_provider_version.py b/tests/units/test_registry_provider_version.py new file mode 100644 index 00000000..46b76bdc --- /dev/null +++ b/tests/units/test_registry_provider_version.py @@ -0,0 +1,402 @@ +"""Unit tests for the registry_provider_version module.""" + +from unittest.mock import Mock, patch + +import pytest + +from pytfe._http import HTTPTransport +from pytfe.errors import ( + InvalidKeyIDError, + InvalidVersionError, + RequiredPrivateRegistryError, +) +from pytfe.models.registry_provider import ( + RegistryName, + RegistryProviderID, +) +from pytfe.models.registry_provider_version import ( + RegistryProviderVersion, + RegistryProviderVersionCreateOptions, + RegistryProviderVersionID, +) +from pytfe.resources.registry_provider_version import RegistryProviderVersions + + +class TestRegistryProviderVersions: + """Test the RegistryProviderVersions service class.""" + + @pytest.fixture + def mock_transport(self): + """Create a mock HTTPTransport.""" + return Mock(spec=HTTPTransport) + + @pytest.fixture + def versions_service(self, mock_transport): + """Create a RegistryProviderVersions service with mocked transport.""" + return RegistryProviderVersions(mock_transport) + + @pytest.fixture + def valid_provider_id(self): + """Create a valid provider ID.""" + return RegistryProviderID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + ) + + @pytest.fixture + def valid_version_id(self): + """Create a valid version ID.""" + return RegistryProviderVersionID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + ) + + def test_validate_provider_id_success(self, versions_service, valid_provider_id): + """Test _validate_provider_id with valid provider ID.""" + result = versions_service._validate_provider_id(valid_provider_id) + assert result is True + + def test_validate_provider_id_invalid_organization( + self, versions_service, valid_provider_id + ): + """Test _validate_provider_id with invalid organization name.""" + valid_provider_id.organization_name = "" + result = versions_service._validate_provider_id(valid_provider_id) + assert result is False + + def test_create_version_validations(self, versions_service): + """Test create method validations.""" + # Test with invalid provider ID + invalid_provider_id = RegistryProviderID( + organization_name="", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + ) + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + with pytest.raises(ValueError, match="Invalid provider ID"): + versions_service.create(invalid_provider_id, options) + + def test_create_version_requires_private_registry( + self, versions_service, mock_transport + ): + """Test create method requires private registry.""" + public_provider_id = RegistryProviderID( + organization_name="test-org", + registry_name=RegistryName.PUBLIC, + namespace="hashicorp", + name="aws", + ) + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + with pytest.raises(RequiredPrivateRegistryError): + versions_service.create(public_provider_id, options) + + def test_create_version_success( + self, versions_service, valid_provider_id, mock_transport + ): + """Test successful create operation.""" + mock_response_data = { + "data": { + "id": "provver-123", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0"], + "shasums-uploaded": False, + "shasums-sig-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + "relationships": { + "registry-provider": { + "data": {"id": "prov-123", "type": "registry-providers"} + } + }, + "links": { + "shasums-upload": "https://example.com/upload", + "shasums-sig-upload": "https://example.com/sig-upload", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + result = versions_service.create(valid_provider_id, options) + + mock_transport.request.assert_called_once_with( + "POST", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions", + json_body={ + "data": { + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "key-id": "test-key-id", + "protocols": ["5.0"], + }, + } + }, + ) + + assert isinstance(result, RegistryProviderVersion) + assert result.id == "provver-123" + assert result.version == "1.0.0" + assert result.key_id == "test-key-id" + assert result.protocols == ["5.0"] + assert result.permissions.can_delete is True + + def test_list_versions_success_without_options( + self, versions_service, valid_provider_id, mock_transport + ): + """Test successful list operation without options.""" + mock_response_data = { + "data": [ + { + "id": "provver-123", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0"], + "shasums-uploaded": False, + "shasums-sig-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + }, + { + "id": "provver-456", + "type": "registry-provider-versions", + "attributes": { + "version": "1.1.0", + "created-at": "2023-02-01T12:00:00Z", + "updated-at": "2023-02-01T12:00:00Z", + "key-id": "test-key-id-2", + "protocols": ["5.0", "6.0"], + "shasums-uploaded": True, + "shasums-sig-uploaded": True, + "permissions": { + "can-delete": True, + "can-upload-asset": False, + }, + }, + }, + ], + "meta": { + "pagination": { + "current-page": 1, + "total-pages": 1, + "prev-page": None, + "next-page": None, + "total-count": 2, + } + }, + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + with patch.object( + versions_service, "_list", return_value=mock_response_data["data"] + ): + result = list(versions_service.list(valid_provider_id)) + + assert len(result) == 2 + assert result[0].id == "provver-123" + assert result[0].version == "1.0.0" + assert result[0].shasums_uploaded is False + assert result[1].id == "provver-456" + assert result[1].version == "1.1.0" + assert result[1].shasums_uploaded is True + + def test_read_version_validations(self, versions_service): + """Test read method with invalid version ID.""" + invalid_version_id = RegistryProviderVersionID( + organization_name="", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + ) + + with pytest.raises(ValueError, match="Invalid provider ID"): + versions_service.read(invalid_version_id) + + def test_read_version_success( + self, versions_service, valid_version_id, mock_transport + ): + """Test successful read operation.""" + mock_response_data = { + "data": { + "id": "provver-789", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0", "6.0"], + "shasums-uploaded": True, + "shasums-sig-uploaded": True, + "permissions": { + "can-delete": True, + "can-upload-asset": False, + }, + }, + "relationships": { + "registry-provider": { + "data": {"id": "prov-123", "type": "registry-providers"} + }, + "platforms": { + "data": [ + {"id": "plat-123", "type": "registry-provider-platforms"} + ] + }, + }, + "links": { + "shasums-download": "https://example.com/download", + "shasums-sig-download": "https://example.com/sig-download", + }, + } + } + + mock_response = Mock() + mock_response.json.return_value = mock_response_data + mock_transport.request.return_value = mock_response + + result = versions_service.read(valid_version_id) + + mock_transport.request.assert_called_once_with( + "GET", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions/1.0.0", + ) + + assert isinstance(result, RegistryProviderVersion) + assert result.id == "provver-789" + assert result.version == "1.0.0" + assert result.key_id == "test-key-id" + assert result.protocols == ["5.0", "6.0"] + assert result.shasums_uploaded is True + assert result.shasums_sig_uploaded is True + + def test_delete_version_success( + self, versions_service, valid_version_id, mock_transport + ): + """Test successful delete operation.""" + result = versions_service.delete(valid_version_id) + + mock_transport.request.assert_called_once_with( + "DELETE", + path="/api/v2/organizations/test-org/registry-providers/private/test-namespace/test-provider/versions/1.0.0", + ) + + assert result is None + + def test_registry_provider_version_from_success(self, versions_service): + """Test _registry_provider_version_from with valid data.""" + data = { + "id": "provver-123", + "type": "registry-provider-versions", + "attributes": { + "version": "1.0.0", + "created-at": "2023-01-01T12:00:00Z", + "updated-at": "2023-01-01T12:00:00Z", + "key-id": "test-key-id", + "protocols": ["5.0"], + "shasums-uploaded": False, + "shasums-sig-uploaded": False, + "permissions": { + "can-delete": True, + "can-upload-asset": True, + }, + }, + "relationships": { + "registry-provider": { + "data": {"id": "prov-123", "type": "registry-providers"} + }, + "platforms": { + "data": [ + {"id": "plat-123", "type": "registry-provider-platforms"}, + {"id": "plat-456", "type": "registry-provider-platforms"}, + ] + }, + }, + } + + result = versions_service._registry_provider_version_from(data) + + assert isinstance(result, RegistryProviderVersion) + assert result.id == "provver-123" + assert result.version == "1.0.0" + assert result.key_id == "test-key-id" + assert result.registry_provider == { + "id": "prov-123", + "type": "registry-providers", + } + assert result.registry_provider_platforms is not None + assert len(result.registry_provider_platforms) == 2 + + def test_create_options_validation_invalid_version(self): + """Test RegistryProviderVersionCreateOptions with invalid version.""" + with pytest.raises(InvalidVersionError): + RegistryProviderVersionCreateOptions( + version="", **{"key-id": "test-key-id"}, protocols=["5.0"] + ) + + def test_create_options_validation_invalid_key_id(self): + """Test RegistryProviderVersionCreateOptions with invalid key_id.""" + with pytest.raises(InvalidKeyIDError): + RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": ""}, protocols=["5.0"] + ) + + def test_create_options_validation_success(self): + """Test RegistryProviderVersionCreateOptions with valid data.""" + options = RegistryProviderVersionCreateOptions( + version="1.0.0", **{"key-id": "test-key-id"}, protocols=["5.0", "6.0"] + ) + assert options.version == "1.0.0" + assert options.key_id == "test-key-id" + assert options.protocols == ["5.0", "6.0"] + + def test_version_id_validation_success(self): + """Test RegistryProviderVersionID with valid data.""" + version_id = RegistryProviderVersionID( + organization_name="test-org", + registry_name=RegistryName.PRIVATE, + namespace="test-namespace", + name="test-provider", + version="1.0.0", + ) + assert version_id.organization_name == "test-org" + assert version_id.registry_name == RegistryName.PRIVATE + assert version_id.namespace == "test-namespace" + assert version_id.name == "test-provider" + assert version_id.version == "1.0.0" diff --git a/tests/units/test_reserved_tag_key.py b/tests/units/test_reserved_tag_key.py index 490a93a9..d7ca66b7 100644 --- a/tests/units/test_reserved_tag_key.py +++ b/tests/units/test_reserved_tag_key.py @@ -14,7 +14,7 @@ ReservedTagKeyListOptions, ReservedTagKeyUpdateOptions, ) -from pytfe.resources.reserved_tag_key import ReservedTagKey +from pytfe.resources.reserved_tag_key import ReservedTagKeys class TestReservedTagKeyParsing: @@ -24,7 +24,7 @@ class TestReservedTagKeyParsing: def reserved_tag_key_service(self): """Create a ReservedTagKey service for testing parsing.""" mock_transport = Mock(spec=HTTPTransport) - return ReservedTagKey(mock_transport) + return ReservedTagKeys(mock_transport) def test_parse_reserved_tag_key_minimal(self, reserved_tag_key_service): """Test _parse_reserved_tag_key with minimal data.""" @@ -68,12 +68,12 @@ class TestReservedTagKey: def reserved_tag_key_service(self): """Create a ReservedTagKey service for testing.""" mock_transport = Mock(spec=HTTPTransport) - return ReservedTagKey(mock_transport) + return ReservedTagKeys(mock_transport) def test_list_reserved_tag_keys_invalid_org(self, reserved_tag_key_service): """Test listing reserved tag keys with invalid organization.""" with pytest.raises(InvalidOrgError): - reserved_tag_key_service.list("") + list(reserved_tag_key_service.list("")) def test_create_reserved_tag_key_invalid_org(self, reserved_tag_key_service): """Test creating reserved tag key with invalid organization.""" @@ -83,11 +83,6 @@ def test_create_reserved_tag_key_invalid_org(self, reserved_tag_key_service): with pytest.raises(InvalidOrgError): reserved_tag_key_service.create("", options) - def test_read_reserved_tag_key_not_implemented(self, reserved_tag_key_service): - """Test reading reserved tag key raises NotImplementedError.""" - with pytest.raises(NotImplementedError): - reserved_tag_key_service.read("rtk-123") - def test_update_reserved_tag_key_invalid_id(self, reserved_tag_key_service): """Test updating reserved tag key with invalid ID.""" options = ReservedTagKeyUpdateOptions(key="updated-key") @@ -115,6 +110,5 @@ def test_reserved_tag_key_update_options_model(self): def test_reserved_tag_key_list_options_model(self): """Test ReservedTagKeyListOptions model validation.""" - options = ReservedTagKeyListOptions(page_number=2, page_size=50) - assert options.page_number == 2 + options = ReservedTagKeyListOptions(page_size=50) assert options.page_size == 50