From 77fa20b98cc267977054b47ec3926da2f15a109e Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Thu, 18 Sep 2025 12:18:48 +0530 Subject: [PATCH 1/4] registry module function creation and testing --- examples/registry_module_complete_test.py | 851 ++++++++++++++++++++++ src/tfe/client.py | 2 + src/tfe/errors.py | 20 + src/tfe/resources/organizations.py | 4 +- src/tfe/resources/projects.py | 2 +- src/tfe/resources/registry_module.py | 550 ++++++++++++++ src/tfe/resources/workspaces.py | 4 +- src/tfe/types/__init__.py | 103 +++ src/tfe/types/registry_module_types.py | 343 +++++++++ src/tfe/utils.py | 8 + 10 files changed, 1882 insertions(+), 5 deletions(-) create mode 100644 examples/registry_module_complete_test.py create mode 100644 src/tfe/resources/registry_module.py create mode 100644 src/tfe/types/__init__.py create mode 100644 src/tfe/types/registry_module_types.py diff --git a/examples/registry_module_complete_test.py b/examples/registry_module_complete_test.py new file mode 100644 index 00000000..4966cbe6 --- /dev/null +++ b/examples/registry_module_complete_test.py @@ -0,0 +1,851 @@ +#!/usr/bin/env python3 +""" +Complete Registry Module Testing Suite + +This file contains individual tests for all 15 registry module functions implemented in src/tfe/resources/registry_module.py: + +PUBLIC FUNCTIONS AVAILABLE FOR TESTING: +1. list() - List registry modules in organization +2. list_commits() - List commits for a VCS-connected module +3. create() - Create a new registry module +4. create_version() - Create a new version of an existing module +5. create_with_vcs_connection() - Create module with VCS connection +6. read() - Read a specific registry module +7. read_version() - Read a specific version of a module +8. read_terraform_registry_module() - Read public Terraform Registry module +9. delete() - Delete a module by organization and name +10. delete_by_name() - Delete a module using RegistryModuleID +11. delete_provider() - Delete all modules for a provider +12. delete_version() - Delete a specific version of a module +13. update() - Update module configuration +14. upload() - Upload module content for a version +15. upload_tar_gzip() - Upload tar.gz archive to upload URL + +USAGE: +- Uncomment specific test sections to test individual functions +- Tests 1-6 have been pre-tested and are commented out (preserve this logic) +- Modify test data (module names, versions, etc.) as needed for your environment +- Ensure you have proper TFE credentials and organization access +""" + +import io +import os +import random +import sys +import tarfile +import tempfile +import time + +# Add the src directory to the path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) + +from tfe import TFEClient, TFEConfig +from tfe.types.registry_module_types import ( + AgentExecutionMode, + RegistryModuleCreateOptions, + RegistryModuleCreateVersionOptions, + RegistryModuleCreateWithVCSConnectionOptions, + RegistryModuleID, + RegistryModuleListOptions, + RegistryModuleUpdateOptions, + RegistryModuleVCSRepoOptions, + RegistryName, + TestConfig, +) + + +def main(): + """Test all registry module functions individually.""" + + print("=" * 80) + print("REGISTRY MODULE COMPLETE TESTING SUITE") + print("=" * 80) + print("Testing ALL 15 functions in src/tfe/resources/registry_module.py") + print("Comprehensive test coverage for all registry module operations") + print("=" * 80) + + # Initialize the TFE client + client = TFEClient(TFEConfig.from_env()) + organization_name = "aayush-test" # Replace with your organization + + # Variables to store created resources for dependent tests + created_module = None + created_version = None + version_object = None # Store the version object with upload URL + + # ===================================================== + # TEST 1: LIST REGISTRY MODULES [TESTED - COMMENTED] + # ===================================================== + print("\n1. Testing list() function:") + try: + options = RegistryModuleListOptions( + organization_name=organization_name, registry_name=RegistryName.PRIVATE + ) + modules = list(client.registry_modules.list(organization_name, options)) + print(f" ✓ Found {len(modules)} registry modules") + + for i, module in enumerate(modules[:3], 1): + print(f" {i}. {module.name}/{module.provider} (ID: {module.id})") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 2: CREATE REGISTRY MODULE WITH VCS CONNECTION [TESTED - COMMENTED] + # ===================================================== + print("\n2. Testing create_with_vcs_connection() function:") + created_module = None + try: + unique_suffix = f"{int(time.time())}-{random.randint(1000, 9999)}" + + vcs_options = RegistryModuleVCSRepoOptions( + identifier="aayushsingh2502/dummy-repo", # Required + **{"display-identifier": "dummy-aws"}, # Required (using alias) + **{"oauth-token-id": "ot-gAGuPJPTRrSdqjZA"}, # Optional + **{"organization-name": organization_name}, # Required when using branch + branch="main", # Optional: for branch-based modules + tags=False, # Cannot be True when branch is specified + **{"source-directory": ""}, # Optional + **{"tag-prefix": "v"}, # Optional + ) + + test_config = TestConfig( + tests_enabled=True, agent_execution_mode=AgentExecutionMode.REMOTE + ) + + vcs_create_options = RegistryModuleCreateWithVCSConnectionOptions( + **{"vcs-repo": vcs_options}, + name=f"dummy-repo-{unique_suffix}", + provider="aws", + **{"registry-name": "private"}, + **{"initial-version": "1.0.0"}, + **{"test-config": test_config}, + ) + + created_module = client.registry_modules.create_with_vcs_connection( + vcs_create_options + ) + print( + f" ✓ Created VCS module: {created_module.name}/{created_module.provider}" + ) + print(f" ID: {created_module.id}") + print(f" Status: {created_module.status}") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 3: READ REGISTRY MODULE [TESTED - COMMENTED] + # ===================================================== + if created_module: + print("\n3. Testing read() function:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + read_module = client.registry_modules.read(module_id) + print(f" ✓ Read module: {read_module.name}") + print(f" Status: {read_module.status}") + print(f" Created: {read_module.created_at}") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 4: LIST COMMITS [TESTED - COMMENTED] + # ===================================================== + if created_module: + print("\n4. Testing list_commits() function:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + commits = client.registry_modules.list_commits(module_id) + commit_list = list(commits.items) if hasattr(commits, "items") else [] + print(f" ✓ Found {len(commit_list)} commits") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 5: CREATE VERSION [TESTED - COMMENTED] + # ===================================================== + created_version = None + if created_module: + print("\n5. Testing create_version() function:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + version_options = RegistryModuleCreateVersionOptions( + version="1.0.0", commit_sha="dummy-sha-123456789abcdef" + ) + + version = client.registry_modules.create_version(module_id, version_options) + created_version = version.version + print(f" ✓ Created version: {version.version}") + print(f" Status: {version.status}") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 6: READ VERSION [TESTED - COMMENTED] + # ===================================================== + if created_module and created_version: + print("\n6. Testing read_version() function:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + read_version = client.registry_modules.read_version( + module_id, created_version + ) + print(f" ✓ Read version: {read_version.version}") + print(f" Status: {read_version.status}") + print(f" ID: {read_version.id}") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 7: READ PUBLIC TERRAFORM REGISTRY MODULE + # ===================================================== + print("\n7. Testing read_terraform_registry_module() function:") + try: + # Create a RegistryModuleID for a public module + public_module_id = RegistryModuleID( + namespace="terraform-aws-modules", # Use namespace for public modules + name="vpc", + provider="aws", + registry_name=RegistryName.PUBLIC, + ) + + # Read a specific version of the public module + version = "5.0.0" # Use a known stable version + public_module = client.registry_modules.read_terraform_registry_module( + public_module_id, version + ) + print(f" ✓ Read public module: {public_module.name}") + print(f" Version: {version}") + print(f" Downloads: {getattr(public_module, 'downloads', 'N/A')}") + print(f" Verified: {getattr(public_module, 'verified', 'N/A')}") + print(f" Source: {getattr(public_module, 'source', 'N/A')}") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 8: CREATE SIMPLE REGISTRY MODULE (Non-VCS) + # ===================================================== + print("\n8. Testing create() function (non-VCS module):") + print(" NOTE: Non-VCS modules start in PENDING status until content is uploaded") + try: + unique_suffix = f"{int(time.time())}-{random.randint(1000, 9999)}" + + create_options = RegistryModuleCreateOptions( + name=f"test-module-{unique_suffix}", + provider="aws", + registry_name=RegistryName.PRIVATE, + ) + + created_simple_module = client.registry_modules.create( + organization_name, create_options + ) + print( + f" ✓ Created simple module: {created_simple_module.name}/{created_simple_module.provider}" + ) + print(f" ID: {created_simple_module.id}") + print( + f" Status: {created_simple_module.status} (PENDING until content uploaded)" + ) + print(f" No Code: {created_simple_module.no_code}") + + # Store for later tests (will be overridden by upload test module) + created_module = created_simple_module + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 8A: LIST VERSIONS + # ===================================================== + if created_module: + print("\n8A. Testing list_versions() function:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + versions = client.registry_modules.list_versions(module_id) + versions_list = list(versions) if hasattr(versions, "__iter__") else [] + print(f" ✓ Found {len(versions_list)} versions") + + for i, version in enumerate(versions_list[:3], 1): + print(f" {i}. Version {version.version} (Status: {version.status})") + + except Exception as e: + print(f" ✗ Error: {e}") + + # ===================================================== + # TEST 8B: UPDATE MODULE + # ===================================================== + if created_module: + print("\n8B. Testing update() function:") + print(" NOTE: Update functionality may vary by TFE version") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + # First check current module status + current_module = client.registry_modules.read(module_id) + print(f" Current module no_code setting: {current_module.no_code}") + + # Try to update no_code setting + update_options = RegistryModuleUpdateOptions( + no_code=True # Set to no-code module + ) + + updated_module = client.registry_modules.update(module_id, update_options) + print(f" ✓ Updated module: {updated_module.name}") + print(f" No Code: {updated_module.no_code}") + print(f" Status: {updated_module.status}") + + except Exception as e: + print(f" ⚠ Update may not be supported: {e}") + + # ===================================================== + # TEST 9: CREATE MODULE FOR UPLOAD TESTING + # ===================================================== + print("\n9. Creating test module for upload function testing:") + + try: + # Create a module specifically for upload testing + create_options = RegistryModuleCreateOptions( + name=f"upload-test-{random.randint(100000, 999999)}", + provider="aws", + registry_name=RegistryName.PRIVATE, + ) + + created_module = client.registry_modules.create( + organization_name, create_options + ) + print(f" ✓ Created test module: {created_module.name}") + print(f" Provider: {created_module.provider}") + print(f" Status: {created_module.status}") + + except Exception as e: + print(f" ✗ Error creating module: {e}") + return + + # ===================================================== + # TEST 10: CREATE VERSION FOR UPLOAD TESTING + # ===================================================== + created_version = None + version_object = None + + if created_module: + print("\n10. Creating version for upload testing:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + version_options = RegistryModuleCreateVersionOptions(version="1.0.0") + + version = client.registry_modules.create_version(module_id, version_options) + created_version = version.version + version_object = version + print(f" ✓ Created version: {created_version}") + print(f" Status: {version.status}") + + # Check if upload URL is available + upload_url = ( + version.links.get("upload") if hasattr(version, "links") else None + ) + print(f" Upload URL available: {'Yes' if upload_url else 'No'}") + + except Exception as e: + print(f" ✗ Error creating version: {e}") + + # ===================================================== + # TEST 11: UPLOAD_TAR_GZIP FUNCTION TESTING + # ===================================================== + if created_module and created_version and version_object: + print("\n11. Testing upload_tar_gzip() function:") + print(" This will change module status from PENDING to SETUP_COMPLETE") + try: + # Create a simple module structure in memory + tar_buffer = io.BytesIO() + + with tarfile.open(fileobj=tar_buffer, mode="w:gz") as tar: + # Create main.tf content + main_tf_content = """ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = "~> 5.0" + } + } +} + +variable "name" { + description = "Name of the resource" + type = string + default = "upload-test" +} + +resource "aws_s3_bucket" "example" { + bucket = var.name +} + +output "bucket_name" { + description = "The name of the S3 bucket" + value = aws_s3_bucket.example.bucket +} +""".strip() + + # Add main.tf to archive + main_tf_info = tarfile.TarInfo(name="main.tf") + main_tf_info.size = len(main_tf_content.encode("utf-8")) + tar.addfile(main_tf_info, io.BytesIO(main_tf_content.encode("utf-8"))) + + # Create README.md content + readme_content = f""" +# {created_module.name} + +A test module created for upload function testing. + +## Usage + +```hcl +module "example" {{ + source = "app.terraform.io/{{organization_name}}/{{created_module.name}}/{{created_module.provider}}" + + name = "my-resource" +}} +``` +""".strip() + + # Add README.md to archive + readme_info = tarfile.TarInfo(name="README.md") + readme_info.size = len(readme_content.encode("utf-8")) + tar.addfile(readme_info, io.BytesIO(readme_content.encode("utf-8"))) + + tar_buffer.seek(0) + + # Get upload URL from the version object + upload_url = ( + version_object.links.get("upload") + if hasattr(version_object, "links") + else None + ) + + if upload_url: + client.registry_modules.upload_tar_gzip(upload_url, tar_buffer) + print( + " ✓ Successfully uploaded tar.gz content using upload_tar_gzip()" + ) + + # Wait for processing + print(" Waiting 5 seconds for processing...") + time.sleep(5) + + # Check module status after upload + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + updated_module = client.registry_modules.read(module_id) + print(f" Updated Module Status: {updated_module.status}") + + if updated_module.status.value != "pending": + print( + f" ✅ SUCCESS: Module status changed from PENDING to {updated_module.status}" + ) + else: + print(" ⏳ Module still processing - may take longer") + + else: + print(" ⚠ No upload URL available in version links") + + except Exception as e: + print(f" ✗ Error in upload_tar_gzip test: {e}") + + # ===================================================== + # TEST 12: UPLOAD FUNCTION TESTING + # ===================================================== + if created_module and created_version and version_object: + print("\n12. Testing upload() function:") + print(" NOTE: This function uploads from a local file path") + try: + # Create a temporary directory with module structure + with tempfile.TemporaryDirectory() as temp_dir: + # Create main.tf file + main_tf_path = os.path.join(temp_dir, "main.tf") + with open(main_tf_path, "w") as f: + f.write( + """ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = "~> 5.0" + } + } +} + +variable "test_var" { + description = "A test variable for upload() function" + type = string + default = "upload-test" +} + +resource "aws_s3_bucket" "upload_test" { + bucket = var.test_var +} + +output "bucket_name" { + description = "The name of the S3 bucket" + value = aws_s3_bucket.upload_test.bucket +} +""".strip() + ) + + # Create variables.tf file + variables_tf_path = os.path.join(temp_dir, "variables.tf") + with open(variables_tf_path, "w") as f: + f.write( + """ +variable "environment" { + description = "Environment name" + type = string + default = "dev" +} + +variable "region" { + description = "AWS region" + type = string + default = "us-west-2" +} +""".strip() + ) + + # Create outputs.tf file + outputs_tf_path = os.path.join(temp_dir, "outputs.tf") + with open(outputs_tf_path, "w") as f: + f.write( + """ +output "module_info" { + description = "Information about this module" + value = { + name = "upload-test-module" + environment = var.environment + region = var.region + } +} +""".strip() + ) + + print(f" Created temporary module files in: {temp_dir}") + print(f" Files: {os.listdir(temp_dir)}") + + # Check if upload URL is available + upload_url = ( + version_object.links.get("upload") + if hasattr(version_object, "links") + else None + ) + if upload_url: + print(" Upload URL available: Yes") + + # Try the upload function + try: + client.registry_modules.upload(version_object, temp_dir) + print(" ✓ Successfully uploaded using upload() function") + + # Wait and check status + print(" Waiting 5 seconds for processing...") + time.sleep(5) + + module_id = RegistryModuleID( + organization=organization_name, + name=created_module.name, + provider=created_module.provider, + registry_name=RegistryName.PRIVATE, + ) + + updated_module = client.registry_modules.read(module_id) + print(f" Updated Module Status: {updated_module.status}") + + except NotImplementedError as nie: + print(f" ⚠ upload() function not fully implemented: {nie}") + print(" This is expected - the function is a placeholder") + + # Fallback to upload_tar_gzip + print(" Trying fallback: upload_tar_gzip()...") + + tar_buffer = io.BytesIO() + with tarfile.open(fileobj=tar_buffer, mode="w:gz") as tar: + for file_name in os.listdir(temp_dir): + file_path = os.path.join(temp_dir, file_name) + if os.path.isfile(file_path): + with open(file_path) as file_content: + content = file_content.read() + + tarinfo = tarfile.TarInfo(name=file_name) + tarinfo.size = len(content.encode("utf-8")) + tar.addfile( + tarinfo, io.BytesIO(content.encode("utf-8")) + ) + + tar_buffer.seek(0) + client.registry_modules.upload_tar_gzip(upload_url, tar_buffer) + print( + " ✓ Successfully uploaded using upload_tar_gzip() as fallback" + ) + + except Exception as upload_error: + print(f" ✗ upload() function error: {upload_error}") + + else: + print(" ⚠ No upload URL available - cannot test upload function") + + except Exception as e: + print(f" ✗ Error in upload() test: {e}") + + # ===================================================== + # TEST 13: DELETE VERSION + # ===================================================== + # Create a test module and version for delete testing + print("\n13. Testing delete_version() function:") + print(" Creating test module and version for deletion...") + + test_module_for_deletion = None + test_version_for_deletion = None + + try: + # Create a module specifically for delete testing + delete_create_options = RegistryModuleCreateOptions( + name=f"delete-test-{random.randint(100000, 999999)}", + provider="aws", + registry_name=RegistryName.PRIVATE, + ) + + test_module_for_deletion = client.registry_modules.create( + organization_name, delete_create_options + ) + print(f" ✓ Created test module: {test_module_for_deletion.name}") + + # Create a version for deletion testing + module_id = RegistryModuleID( + organization=organization_name, + name=test_module_for_deletion.name, + provider=test_module_for_deletion.provider, + registry_name=RegistryName.PRIVATE, + ) + + version_options = RegistryModuleCreateVersionOptions(version="1.0.0") + + version = client.registry_modules.create_version(module_id, version_options) + test_version_for_deletion = version.version + print(f" ✓ Created test version: {test_version_for_deletion}") + + # Now test version deletion + print(f" Testing deletion of version {test_version_for_deletion}...") + + # Delete the version + client.registry_modules.delete_version(module_id, test_version_for_deletion) + print( + f" ✓ Successfully called delete_version() for version: {test_version_for_deletion}" + ) + + # Verify deletion by trying to read it + try: + client.registry_modules.read_version( + organization=organization_name, + registry_name=RegistryName.PRIVATE, + namespace=organization_name, + name=test_module_for_deletion.name, + provider=test_module_for_deletion.provider, + version=test_version_for_deletion, + ) + print( + " ⚠ Warning: Version still exists after deletion (may take time to process)" + ) + except Exception: + print(" ✓ Confirmed: Version no longer exists") + + except Exception as e: + print(f" ✗ Error in delete_version test: {e}") + + # ===================================================== + # TEST 14: DELETE BY NAME + # ===================================================== + if test_module_for_deletion: + print("\n14. Testing delete_by_name() function:") + try: + module_id = RegistryModuleID( + organization=organization_name, + name=test_module_for_deletion.name, + provider=test_module_for_deletion.provider, + registry_name=RegistryName.PRIVATE, + ) + + # Check module exists before deletion + try: + client.registry_modules.read(module_id) + print( + f" Module {test_module_for_deletion.name}/{test_module_for_deletion.provider} exists" + ) + + # Delete the module + client.registry_modules.delete_by_name(module_id) + print( + f" ✓ Successfully called delete_by_name() for module: {test_module_for_deletion.name}" + ) + + # Verify deletion + try: + client.registry_modules.read(module_id) + print( + " ⚠ Warning: Module still exists after deletion (may take time to process)" + ) + except Exception: + print(" ✓ Confirmed: Module no longer exists") + + except Exception as read_error: + print(f" Module not found: {read_error}") + + except Exception as e: + print(f" ✗ Error in delete_by_name test: {e}") + + # ===================================================== + # TEST 15: DELETE (Alternative delete method) + # ===================================================== + print("\n15. Testing delete() function:") + print(" NOTE: Testing with non-existent module to avoid conflicts") + try: + # This function takes organization and name directly + # We'll test with a non-existent module to avoid conflicts + test_name = "non-existent-module-for-testing" + + print(f" Testing delete with non-existent module: {test_name}") + client.registry_modules.delete(organization_name, test_name) + print( + " ✓ Delete function executed successfully (may return 404 for non-existent module)" + ) + + except Exception as e: + print(f" Expected error for non-existent module: {e}") + + # ===================================================== + # TEST 16: DELETE PROVIDER (SAFE VERSION - CREATES TEST PROVIDER) + # ===================================================== + print("\n16. Testing delete_provider() function:") + print(" Creating a test provider specifically for deletion testing...") + + try: + # Create a test module with a valid provider for deletion testing + # Use simple alphanumeric names to avoid validation issues + test_provider_name = f"testprovider{random.randint(1000, 9999)}" + + delete_provider_options = RegistryModuleCreateOptions( + name=f"testmodule{random.randint(1000, 9999)}", + provider=test_provider_name, + registry_name=RegistryName.PRIVATE, + ) + + test_provider_module = client.registry_modules.create( + organization_name, delete_provider_options + ) + print(f" ✓ Created test module with provider: {test_provider_name}") + + # Now test delete_provider function + test_provider_module_id = RegistryModuleID( + organization=organization_name, + name=test_provider_module.name, # Name doesn't matter for provider deletion + provider=test_provider_name, + registry_name=RegistryName.PRIVATE, + ) + + print(f" Testing delete_provider() for provider: {test_provider_name}") + client.registry_modules.delete_provider(test_provider_module_id) + print( + f" ✓ Successfully called delete_provider() for provider: {test_provider_name}" + ) + + # Verify deletion by trying to read the module + try: + client.registry_modules.read(test_provider_module_id) + print( + " ⚠ Warning: Module still exists after provider deletion (may take time to process)" + ) + except Exception: + print(" ✓ Confirmed: All modules for provider have been deleted") + + except Exception as e: + print(f" ✗ Error in delete_provider test: {e}") + + # ===================================================== + # TESTING SUMMARY + # ===================================================== + print("\n" + "=" * 80) + print("REGISTRY MODULE TESTING COMPLETED!") + print("=" * 80) + print("Summary of ALL 15 Functions Tested:") + print("✓ list() - List registry modules in organization") + print("✓ create_with_vcs_connection() - Create module with VCS connection") + print("✓ read() - Read module details") + print("✓ list_commits() - List VCS commits for module") + print("✓ create_version() - Create new module version") + print("✓ read_version() - Read specific version details") + print("✓ read_terraform_registry_module() - Read public registry module") + print("✓ create() - Create simple module") + print("✓ list_versions() - List all versions of a module") + print("✓ update() - Update module settings") + print("✓ upload_tar_gzip() - Upload tar.gz archive to upload URL") + print("✓ upload() - Upload from local directory path (placeholder)") + print("✓ delete_version() - Delete a specific version") + print("✓ delete_by_name() - Delete entire module by name") + print("✓ delete() - Delete module by organization and name") + print("✓ delete_provider() - Delete all modules for a provider") + if created_module: + print(f"✓ Created test module: {created_module.name}") + print("=" * 80) + print("🎉 ALL 15 REGISTRY MODULE FUNCTIONS HAVE BEEN TESTED!") + print("=" * 80) + + +if __name__ == "__main__": + main() diff --git a/src/tfe/client.py b/src/tfe/client.py index e4e27541..927eecca 100644 --- a/src/tfe/client.py +++ b/src/tfe/client.py @@ -4,6 +4,7 @@ from .config import TFEConfig from .resources.organizations import Organizations from .resources.projects import Projects +from .resources.registry_module import RegistryModules from .resources.workspaces import Workspaces @@ -27,6 +28,7 @@ def __init__(self, config: TFEConfig | None = None): self.organizations = Organizations(self._transport) self.projects = Projects(self._transport) self.workspaces = Workspaces(self._transport) + self.registry_modules = RegistryModules(self._transport) def close(self) -> None: pass diff --git a/src/tfe/errors.py b/src/tfe/errors.py index e524f321..58a5a019 100644 --- a/src/tfe/errors.py +++ b/src/tfe/errors.py @@ -57,3 +57,23 @@ class RequiredFieldMissing(TFEError): ... ERR_REQUIRED_NAME = "name is required" ERR_INVALID_ORG = "invalid organization name" ERR_REQUIRED_EMAIL = "email is required" + +# Registry Module Error Constants +ERR_REQUIRED_PROVIDER = "provider is required" +ERR_INVALID_PROVIDER = "invalid value for provider" +ERR_REQUIRED_VERSION = "version is required" +ERR_INVALID_VERSION = "invalid value for version" +ERR_REQUIRED_NAMESPACE = "namespace is required" +ERR_INVALID_REGISTRY_NAME = "invalid registry name" +ERR_UNSUPPORTED_BOTH_NAMESPACE_AND_PRIVATE_REGISTRY_NAME = ( + "namespace cannot be used with private registry" +) +ERR_REQUIRED_VCS_REPO = "VCS repo is required" +ERR_REQUIRED_BRANCH_WHEN_TESTS_ENABLED = "branch is required when tests are enabled" +ERR_BRANCH_MUST_BE_EMPTY_WHEN_TAGS_ENABLED = ( + "branch must be empty when tags are enabled" +) +ERR_AGENT_POOL_NOT_REQUIRED_FOR_REMOTE_EXECUTION = ( + "agent pool not required for remote execution" +) +ERR_INVALID_MODULE_ID = "invalid module ID" diff --git a/src/tfe/resources/organizations.py b/src/tfe/resources/organizations.py index 95ec83e3..46d5be8f 100644 --- a/src/tfe/resources/organizations.py +++ b/src/tfe/resources/organizations.py @@ -9,7 +9,7 @@ ERR_REQUIRED_EMAIL, ERR_REQUIRED_NAME, ) -from ..types import ( +from ..types import ( # type: ignore[attr-defined] Capacity, DataRetentionPolicy, DataRetentionPolicyChoice, @@ -167,7 +167,7 @@ def read_run_queue( ) data = r.json() - from ..types import Pagination, Run, RunStatus + from ..types import Pagination, Run, RunStatus # type: ignore[attr-defined] runs = [] for item in data.get("data", []): diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index 6647edc0..748319dd 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -3,7 +3,7 @@ from collections.abc import Iterator from typing import Any -from ..types import Project +from ..types import Project # type: ignore[attr-defined] from ._base import _Service diff --git a/src/tfe/resources/registry_module.py b/src/tfe/resources/registry_module.py new file mode 100644 index 00000000..a7042c9c --- /dev/null +++ b/src/tfe/resources/registry_module.py @@ -0,0 +1,550 @@ +from __future__ import annotations + +import io +from collections.abc import Iterator +from typing import Any + +from ..errors import ( + ERR_INVALID_NAME, + ERR_INVALID_ORG, + ERR_INVALID_VERSION, +) +from ..types.registry_module_types import ( + AgentExecutionMode, + Commit, + CommitList, + RegistryModule, + RegistryModuleCreateOptions, + RegistryModuleCreateVersionOptions, + RegistryModuleCreateWithVCSConnectionOptions, + RegistryModuleID, + RegistryModuleListOptions, + RegistryModulePermissions, + RegistryModuleUpdateOptions, + RegistryModuleVersion, + RegistryModuleVersionStatuses, + RegistryName, + TerraformRegistryModule, + TestConfig, + VCSRepo, +) +from ..utils import valid_string, valid_string_id, valid_version +from ._base import _Service + + +class RegistryModules(_Service): + """Registry modules service for managing Terraform registry modules.""" + + def list( + self, organization: str, options: RegistryModuleListOptions | None = None + ) -> Iterator[RegistryModule]: + """List all the registry modules within an organization.""" + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + path = f"/api/v2/organizations/{organization}/registry-modules" + params = {} + + if options: + if options.include: + params["include"] = ",".join(options.include) + if options.search: + params["q"] = options.search + if options.provider: + params["filter[provider]"] = options.provider + if options.registry_name: + params["filter[registry_name]"] = options.registry_name.value + if options.organization_name: + params["filter[organization_name]"] = options.organization_name + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + + for item in self._list(path, params=params): + yield self._parse_registry_module(item) + + def list_commits(self, module_id: RegistryModuleID) -> CommitList: + """List the commits for the registry module. + + This returns the latest 20 commits for the connected VCS repo. + Pagination is not applicable due to inconsistent support from the VCS providers. + """ + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + path = f"/api/v2/registry-modules/{module_id.organization}/{module_id.name}/{module_id.provider}/commits" + + response = self.t.request("GET", path) + data = response.json() + + commits = [] + if "data" in data: + for item in data["data"]: + commits.append(self._parse_commit(item)) + + return CommitList(items=commits) + + def create( + self, organization: str, options: RegistryModuleCreateOptions + ) -> RegistryModule: + """Create a registry module without a VCS repo.""" + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + if not self._validate_create_options(options): + raise ValueError("Invalid create options") + + body = { + "data": { + "type": "registry-modules", + "attributes": options.model_dump(exclude_none=True), + } + } + + path = f"/api/v2/organizations/{organization}/registry-modules" + response = self.t.request("POST", path, json_body=body) + data = response.json()["data"] + + return self._parse_registry_module(data) + + def create_version( + self, module_id: RegistryModuleID, options: RegistryModuleCreateVersionOptions + ) -> RegistryModuleVersion: + """Create a registry module version.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + if not self._validate_create_version_options(options): + raise ValueError("Invalid create version options") + + body = { + "data": { + "type": "registry-module-versions", + "attributes": options.model_dump(exclude_none=True), + } + } + + path = f"/api/v2/registry-modules/{module_id.organization}/{module_id.name}/{module_id.provider}/versions" + response = self.t.request("POST", path, json_body=body) + data = response.json()["data"] + + return self._parse_registry_module_version(data) + + def create_with_vcs_connection( + self, options: RegistryModuleCreateWithVCSConnectionOptions + ) -> RegistryModule: + """Create and publish a registry module with a VCS repo.""" + if not self._validate_create_with_vcs_options(options): + raise ValueError("Invalid VCS connection options") + + body = { + "data": { + "type": "registry-modules", + "attributes": options.model_dump(exclude_none=True, by_alias=True), + } + } + + # Determine the URL based on options - exactly like Go implementation + if options.vcs_repo.oauth_token_id and not options.vcs_repo.branch: + path = "/api/v2/registry-modules" + else: + if not options.vcs_repo.organization_name: + raise ValueError( + "organization_name is required in vcs_repo for VCS connection" + ) + path = f"/api/v2/organizations/{options.vcs_repo.organization_name}/registry-modules/vcs" + + # Validate agent execution mode like Go implementation + if ( + options.test_config + and options.test_config.agent_execution_mode == AgentExecutionMode.REMOTE + and options.test_config.agent_pool_id + ): + raise ValueError("Agent pool not required for remote execution") + + response = self.t.request("POST", path, json_body=body) + data = response.json()["data"] + + return self._parse_registry_module(data) + + def read(self, module_id: RegistryModuleID) -> RegistryModule: + """Read a specific registry module.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + if module_id.id: + path = f"/api/v2/registry-modules/{module_id.id}" + else: + registry_name = module_id.registry_name or RegistryName.PRIVATE + namespace = module_id.namespace or module_id.organization + + path = ( + f"/api/v2/organizations/{module_id.organization}/" + f"registry-modules/{registry_name.value}/{namespace}/" + f"{module_id.name}/{module_id.provider}" + ) + + response = self.t.request("GET", path) + data = response.json()["data"] + + return self._parse_registry_module(data) + + def read_version( + self, module_id: RegistryModuleID, version: str + ) -> RegistryModuleVersion: + """Read a registry module version.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + if not valid_string(version) or not valid_string_id(version): + raise ValueError(ERR_INVALID_VERSION) + + path = ( + f"/api/v2/organizations/{module_id.organization}/" + f"registry-modules/private/{module_id.organization}/" + f"{module_id.name}/{module_id.provider}/version" + f"?module_version={version}" + ) + + response = self.t.request("GET", path) + data = response.json()["data"] + + return self._parse_registry_module_version(data) + + def read_terraform_registry_module( + self, module_id: RegistryModuleID, version: str + ) -> TerraformRegistryModule: + """Read a registry module from the Terraform Registry.""" + if module_id.registry_name == RegistryName.PUBLIC: + path = ( + f"/api/registry/public/v1/modules/{module_id.namespace}/" + f"{module_id.name}/{module_id.provider}/{version}" + ) + else: + path = ( + f"/api/registry/v1/modules/{module_id.namespace}/" + f"{module_id.name}/{module_id.provider}/{version}" + ) + + response = self.t.request("GET", path) + data = response.json() + + return TerraformRegistryModule(**data) + + def delete(self, organization: str, name: str) -> None: + """Delete the entire registry module. + + Warning: This method is deprecated and will be removed from a future version. + Use delete_by_name instead. + """ + if not valid_string_id(organization): + raise ValueError(ERR_INVALID_ORG) + + if not valid_string(name) or not valid_string_id(name): + raise ValueError(ERR_INVALID_NAME) + + path = f"/api/v2/registry-modules/actions/delete/{organization}/{name}" + self.t.request("POST", path, json_body={}) + + def delete_by_name(self, module_id: RegistryModuleID) -> None: + """Delete the entire registry module by name.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + path = ( + f"/api/v2/registry-modules/actions/delete/" + f"{module_id.organization}/{module_id.name}" + ) + self.t.request("POST", path, json_body={}) + + def delete_provider(self, module_id: RegistryModuleID) -> None: + """Delete a specified provider for the given module along with all its versions.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + path = ( + f"/api/v2/registry-modules/actions/delete/" + f"{module_id.organization}/{module_id.name}/{module_id.provider}" + ) + self.t.request("POST", path, json_body={}) + + def delete_version(self, module_id: RegistryModuleID, version: str) -> None: + """Delete a specified version for the given provider of the module.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + if not valid_string(version) or not valid_version(version): + raise ValueError(ERR_INVALID_VERSION) + + path = ( + f"/api/v2/registry-modules/actions/delete/" + f"{module_id.organization}/{module_id.name}/" + f"{module_id.provider}/{version}" + ) + self.t.request("POST", path, json_body={}) + + def update( + self, module_id: RegistryModuleID, options: RegistryModuleUpdateOptions + ) -> RegistryModule: + """Update properties of a registry module.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + body = { + "data": { + "type": "registry-modules", + "attributes": options.model_dump(exclude_none=True), + } + } + + registry_name = module_id.registry_name or RegistryName.PRIVATE + namespace = module_id.namespace or module_id.organization + + path = ( + f"/api/v2/organizations/{module_id.organization}/" + f"registry-modules/{registry_name.value}/{namespace}/" + f"{module_id.name}/{module_id.provider}" + ) + + response = self.t.request("PATCH", path, json_body=body) + data = response.json()["data"] + + return self._parse_registry_module(data) + + def upload(self, rmv: RegistryModuleVersion, path: str) -> None: + """Upload Terraform configuration files for the provided registry module version. + + It requires a path to the configuration files on disk, which will be packaged + before being uploaded. + """ + upload_url = rmv.links.get("upload") + if not upload_url: + raise ValueError( + "provided RegistryModuleVersion does not contain an upload link" + ) + + # This would need implementation for packaging files from path + # For now, this is a placeholder + raise NotImplementedError("File packaging and upload not implemented yet") + + def upload_tar_gzip(self, upload_url: str, archive: io.IOBase) -> None: + """Upload a tar gzip archive to the specified upload URL. + + Any stream implementing io.IOBase can be passed into this method. + + Note: This method does not validate the content being uploaded and is therefore + the caller's responsibility to ensure the raw content is a valid Terraform configuration. + """ + # Use the httpx client for direct upload to external URL + response = self.t._sync.put(upload_url, content=archive.read()) + response.raise_for_status() + + # Helper methods for validation and parsing + def _validate_module_id(self, module_id: RegistryModuleID) -> bool: + """Validate registry module ID.""" + if module_id.id and valid_string_id(module_id.id): + return True + + if not valid_string_id(module_id.organization): + return False + + if not valid_string(module_id.name) or not valid_string_id(module_id.name): + return False + + if not valid_string(module_id.provider) or not valid_string_id( + module_id.provider + ): + return False + + if module_id.registry_name == RegistryName.PUBLIC: + if not valid_string(module_id.namespace): + return False + + return True + + def _validate_create_options(self, options: RegistryModuleCreateOptions) -> bool: + """Validate create options.""" + if not valid_string(options.name) or not valid_string_id(options.name): + return False + + if not valid_string(options.provider) or not valid_string_id(options.provider): + return False + + if options.registry_name == RegistryName.PUBLIC: + if not valid_string(options.namespace): + return False + elif options.registry_name == RegistryName.PRIVATE: + if valid_string(options.namespace): + return False + + return True + + def _validate_create_version_options( + self, options: RegistryModuleCreateVersionOptions + ) -> bool: + """Validate create version options.""" + if not valid_string(options.version): + return False + + if not valid_version(options.version): + return False + + return True + + def _validate_create_with_vcs_options( + self, options: RegistryModuleCreateWithVCSConnectionOptions + ) -> bool: + """Validate create with VCS connection options.""" + # Must have VCS repo + if not options.vcs_repo: + return False + + # Validate VCS repo options + if not valid_string(options.vcs_repo.identifier): + return False + + if not valid_string(options.vcs_repo.display_identifier): + return False + + # If branch is specified, organization_name is required + if valid_string(options.vcs_repo.branch) and not valid_string( + options.vcs_repo.organization_name + ): + return False + + # Cannot have both tags and branch set + if options.vcs_repo.tags and valid_string(options.vcs_repo.branch): + return False + + # Agent execution mode validation + if ( + options.test_config + and options.test_config.agent_execution_mode == AgentExecutionMode.REMOTE + and options.test_config.agent_pool_id + ): + return False + + return True + + def _parse_registry_module(self, data: dict[str, Any]) -> RegistryModule: + """Parse registry module from API response.""" + attributes = data.get("attributes", {}) + relationships = data.get("relationships", {}) + + # Parse organization relationship + organization = None + if "organization" in relationships: + org_data = relationships["organization"].get("data", {}) + if org_data: + organization = {"name": org_data.get("attributes", {}).get("name", "")} + + # Parse permissions with field name mapping + permissions = None + if "permissions" in data: + perm_data = data["permissions"] + permissions = RegistryModulePermissions( + can_delete=perm_data.get("can-delete", False), + can_resync=perm_data.get("can-resync", False), + can_retry=perm_data.get("can-retry", False), + ) + + # Parse VCS repo with field name mapping + vcs_repo = None + if "vcs-repo" in attributes: + vcs_data = attributes["vcs-repo"] + vcs_repo = VCSRepo( + branch=vcs_data.get("branch"), + display_identifier=vcs_data.get("display-identifier"), + identifier=vcs_data.get("identifier"), + ingress_submodules=vcs_data.get("ingress-submodules"), + oauth_token_id=vcs_data.get("oauth-token-id"), + repository_http_url=vcs_data.get("repository-http-url"), + service_provider=vcs_data.get("service-provider"), + webhook_url=vcs_data.get("webhook-url"), + tags=vcs_data.get("tags"), + source_directory=vcs_data.get("source-directory"), + tag_prefix=vcs_data.get("tag-prefix"), + ) + + # Parse test config with field name mapping + test_config = None + if "test-config" in attributes: + test_data = attributes["test-config"] + test_config = TestConfig( + tests_enabled=test_data.get("tests-enabled", False), + agent_execution_mode=AgentExecutionMode( + test_data.get("agent-execution-mode", "remote") + ), + ) + + # Parse version statuses with field name mapping + version_statuses = [] + if "version-statuses" in attributes: + for vs in attributes["version-statuses"]: + version_statuses.append( + RegistryModuleVersionStatuses( + version=vs.get("version", ""), + status=vs.get("status", ""), + error=vs.get("error"), + ) + ) + + return RegistryModule( + id=data.get("id", ""), + name=attributes.get("name", ""), + provider=attributes.get("provider", ""), + registry_name=RegistryName(attributes.get("registry-name", "private")), + namespace=attributes.get("namespace", ""), + no_code=attributes.get("no-code", False), + permissions=permissions, + publishing_mechanism=attributes.get("publishing-mechanism"), + status=attributes.get("status"), + test_config=test_config, + vcs_repo=vcs_repo, + version_statuses=version_statuses, + created_at=attributes.get("created-at"), + updated_at=attributes.get("updated-at"), + organization=organization, + ) + + def _parse_registry_module_version( + self, data: dict[str, Any] + ) -> RegistryModuleVersion: + """Parse registry module version from API response.""" + attributes = data.get("attributes", {}) + relationships = data.get("relationships", {}) + links = data.get("links", {}) + + # Parse registry module relationship + registry_module = None + if "registry-module" in relationships: + rm_data = relationships["registry-module"].get("data", {}) + if rm_data: + registry_module = self._parse_registry_module(rm_data) + + return RegistryModuleVersion( + id=data.get("id", ""), + source=attributes.get("source"), + status=attributes.get("status"), + version=attributes.get("version", ""), + created_at=attributes.get("created-at"), + updated_at=attributes.get("updated-at"), + registry_module=registry_module, + links=links, + ) + + def _parse_commit(self, data: dict[str, Any]) -> Commit: + """Parse commit from API response.""" + attributes = data.get("attributes", {}) + + return Commit( + id=data.get("id", ""), + sha=attributes.get("sha", ""), + date=attributes.get("date", ""), + url=attributes.get("url"), + author=attributes.get("author"), + author_avatar_url=attributes.get("author-avatar-url"), + author_html_url=attributes.get("author-html-url"), + message=attributes.get("message"), + ) diff --git a/src/tfe/resources/workspaces.py b/src/tfe/resources/workspaces.py index 9f778eb8..5aa9bd4b 100644 --- a/src/tfe/resources/workspaces.py +++ b/src/tfe/resources/workspaces.py @@ -4,7 +4,7 @@ from collections.abc import Iterator from typing import Any -from ..types import ExecutionMode, Workspace +from ..types import ExecutionMode, Workspace # type: ignore[attr-defined] from ._base import _Service @@ -16,7 +16,7 @@ def _em_safe(v: Any) -> ExecutionMode | None: # Only accept strings; map to enum if known, else None if not isinstance(v, str): return None - return ExecutionMode._value2member_map_.get(v) # type: ignore[return-value] + return ExecutionMode._value2member_map_.get(v) def _ws_from(d: dict[str, Any], org: str | None = None) -> Workspace: diff --git a/src/tfe/types/__init__.py b/src/tfe/types/__init__.py new file mode 100644 index 00000000..0a249d1d --- /dev/null +++ b/src/tfe/types/__init__.py @@ -0,0 +1,103 @@ +"""Types package for TFE client.""" + +# Import all types from the main types module by using importlib to avoid circular imports +import importlib.util +import os + +# Re-export all registry module types +from .registry_module_types import ( + AgentExecutionMode, + Commit, + CommitList, + Input, + Output, + ProviderDependency, + PublishingMechanism, + RegistryModule, + RegistryModuleCreateOptions, + RegistryModuleCreateVersionOptions, + RegistryModuleCreateWithVCSConnectionOptions, + RegistryModuleID, + RegistryModuleList, + RegistryModuleListIncludeOpt, + RegistryModuleListOptions, + RegistryModulePermissions, + RegistryModuleStatus, + RegistryModuleUpdateOptions, + RegistryModuleVCSRepoOptions, + RegistryModuleVCSRepoUpdateOptions, + RegistryModuleVersion, + RegistryModuleVersionStatus, + RegistryModuleVersionStatuses, + RegistryName, + Resource, + Root, + TestConfig, + VCSRepo, +) + +# Define what should be available when importing with * +__all__ = [ + # Registry module types + "AgentExecutionMode", + "Commit", + "CommitList", + "Input", + "Output", + "ProviderDependency", + "PublishingMechanism", + "RegistryModule", + "RegistryModuleCreateOptions", + "RegistryModuleCreateVersionOptions", + "RegistryModuleCreateWithVCSConnectionOptions", + "RegistryModuleID", + "RegistryModuleList", + "RegistryModuleListIncludeOpt", + "RegistryModuleListOptions", + "RegistryModulePermissions", + "RegistryModuleStatus", + "RegistryModuleUpdateOptions", + "RegistryModuleVCSRepoOptions", + "RegistryModuleVCSRepoUpdateOptions", + "RegistryModuleVersion", + "RegistryModuleVersionStatus", + "RegistryModuleVersionStatuses", + "RegistryName", + "Resource", + "Root", + "TestConfig", + "VCSRepo", + # Main types from types.py (will be dynamically added below) + "Capacity", + "DataRetentionPolicy", + "DataRetentionPolicyChoice", + "DataRetentionPolicyDeleteOlder", + "DataRetentionPolicyDeleteOlderSetOptions", + "DataRetentionPolicyDontDelete", + "DataRetentionPolicyDontDeleteSetOptions", + "DataRetentionPolicySetOptions", + "Entitlements", + "ExecutionMode", + "Organization", + "OrganizationCreateOptions", + "OrganizationUpdateOptions", + "Pagination", + "Project", + "ReadRunQueueOptions", + "Run", + "RunQueue", + "RunStatus", + "Workspace", +] + +# Load the main types.py file that's at the same level as this types/ directory +types_py_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "types.py") +spec = importlib.util.spec_from_file_location("main_types", types_py_path) +if spec is not None and spec.loader is not None: + main_types = importlib.util.module_from_spec(spec) + spec.loader.exec_module(main_types) + + # Re-export all main types + for name in dir(main_types): + if not name.startswith("_"): + globals()[name] = getattr(main_types, name) diff --git a/src/tfe/types/registry_module_types.py b/src/tfe/types/registry_module_types.py new file mode 100644 index 00000000..4f68c159 --- /dev/null +++ b/src/tfe/types/registry_module_types.py @@ -0,0 +1,343 @@ +from __future__ import annotations + +import importlib.util +import os +from enum import Enum +from typing import Any + +from pydantic import BaseModel, Field + +# Load the main types.py file to get Organization and other main types +# Path: from /src/tfe/types/registry_module_types.py to /src/tfe/types.py +types_py_path = os.path.join(os.path.dirname(os.path.dirname(__file__)), "types.py") +spec = importlib.util.spec_from_file_location("main_types", types_py_path) +if spec is not None and spec.loader is not None: + main_types = importlib.util.module_from_spec(spec) + spec.loader.exec_module(main_types) +else: + raise ImportError("Could not load main types module") + + +class RegistryName(str, Enum): + """Registry name enum for public/private registries.""" + + PRIVATE = "private" + PUBLIC = "public" + + +class RegistryModuleStatus(str, Enum): + """Registry module status enum.""" + + PENDING = "pending" + NO_VERSION_TAGS = "no_version_tags" + SETUP_FAILED = "setup_failed" + SETUP_COMPLETE = "setup_complete" + + +class RegistryModuleVersionStatus(str, Enum): + """Registry module version status enum.""" + + PENDING = "pending" + CLONING = "cloning" + CLONE_FAILED = "clone_failed" + REG_INGRESS_REQ_FAILED = "reg_ingress_req_failed" + REG_INGRESSING = "reg_ingressing" + REG_INGRESS_FAILED = "reg_ingress_failed" + OK = "ok" + + +class PublishingMechanism(str, Enum): + """Publishing mechanism enum.""" + + BRANCH = "branch" + TAG = "git_tag" + NON_VCS = "non_vcs" + + +class AgentExecutionMode(str, Enum): + """Agent execution mode enum.""" + + AGENT = "agent" + REMOTE = "remote" + + +class RegistryModuleListIncludeOpt(str, Enum): + """Registry module list include options.""" + + NO_CODE_MODULES = "no-code-modules" + + +# Data Models +class RegistryModuleID(BaseModel): + """Registry module identifier.""" + + id: str | None = None + organization: str | None = None + name: str | None = None + provider: str | None = None + namespace: str | None = None + registry_name: RegistryName | None = None + + +class RegistryModulePermissions(BaseModel): + """Registry module permissions.""" + + can_delete: bool + can_resync: bool + can_retry: bool + + +class VCSRepo(BaseModel): + """VCS repository configuration.""" + + branch: str | None = None + display_identifier: str | None = None + identifier: str | None = None + ingress_submodules: bool | None = None + oauth_token_id: str | None = None + repository_http_url: str | None = None + service_provider: str | None = None + webhook_url: str | None = None + tags: bool | None = None + source_directory: str | None = None + tag_prefix: str | None = None + + +class TestConfig(BaseModel): + """Test configuration for registry modules.""" + + tests_enabled: bool | None = None + agent_execution_mode: AgentExecutionMode | None = None + agent_pool_id: str | None = None + + +class RegistryModuleVersionStatuses(BaseModel): + """Registry module version status.""" + + version: str + status: RegistryModuleVersionStatus + error: str | None = None + + +class RegistryModule(BaseModel): + """Registry module model.""" + + id: str + name: str + provider: str + registry_name: RegistryName + namespace: str + no_code: bool = False + permissions: RegistryModulePermissions | None = None + publishing_mechanism: PublishingMechanism | None = None + status: RegistryModuleStatus | None = None + test_config: TestConfig | None = None + vcs_repo: VCSRepo | None = None + version_statuses: list[RegistryModuleVersionStatuses] = Field(default_factory=list) + created_at: str | None = None + updated_at: str | None = None + organization: Any | None = None # Will be Organization type from main types + + +class RegistryModuleVersion(BaseModel): + """Registry module version model.""" + + id: str + source: str | None = None + status: RegistryModuleVersionStatus | None = None + version: str + created_at: str | None = None + updated_at: str | None = None + registry_module: RegistryModule | None = None + links: dict[str, Any] = Field(default_factory=dict) + + +class Commit(BaseModel): + """Commit model.""" + + id: str + sha: str + date: str + url: str | None = None + author: str | None = None + author_avatar_url: str | None = None + author_html_url: str | None = None + message: str | None = None + + +class CommitList(BaseModel): + """Commit list model.""" + + items: list[Commit] = Field(default_factory=list) + + +class RegistryModuleList(BaseModel): + """Registry module list model.""" + + items: list[RegistryModule] = Field(default_factory=list) + + +# Terraform Registry Module Models +class Input(BaseModel): + """Terraform input variable.""" + + name: str + type: str + description: str | None = None + default: str | None = None + required: bool = False + + +class Output(BaseModel): + """Terraform output.""" + + name: str + description: str | None = None + + +class ProviderDependency(BaseModel): + """Provider dependency.""" + + name: str + namespace: str + source: str + version: str + + +class Resource(BaseModel): + """Terraform resource.""" + + name: str + type: str + + +class Root(BaseModel): + """Root module configuration.""" + + path: str | None = None + name: str + readme: str | None = None + empty: bool = False + inputs: list[Input] = Field(default_factory=list) + outputs: list[Output] = Field(default_factory=list) + provider_dependencies: list[ProviderDependency] = Field(default_factory=list) + resources: list[Resource] = Field(default_factory=list) + + +class TerraformRegistryModule(BaseModel): + """Terraform registry module from public/private registry.""" + + id: str + owner: str | None = None + namespace: str + name: str + version: str + provider: str + provider_logo_url: str | None = None + description: str | None = None + source: str | None = None + tag: str | None = None + published_at: str | None = None + downloads: int = 0 + verified: bool = False + root: Root | None = None + providers: list[str] = Field(default_factory=list) + versions: list[str] = Field(default_factory=list) + + +# Options Models +class RegistryModuleListOptions(BaseModel): + """Options for listing registry modules.""" + + include: list[RegistryModuleListIncludeOpt] = Field(default_factory=list) + search: str | None = None + provider: str | None = None + registry_name: RegistryName | None = None + organization_name: str | None = None + page_number: int | None = None + page_size: int | None = None + + +class RegistryModuleCreateOptions(BaseModel): + """Options for creating a registry module.""" + + name: str + provider: str + registry_name: RegistryName | None = RegistryName.PRIVATE + namespace: str | None = None + no_code: bool | None = None + + +class RegistryModuleCreateVersionOptions(BaseModel): + """Options for creating a registry module version.""" + + version: str + commit_sha: str | None = None + + +class RegistryModuleVCSRepoOptions(BaseModel): + """VCS repository options for registry modules.""" + + # Required fields + identifier: str = Field(description="VCS repository identifier") + display_identifier: str = Field( + alias="display-identifier", description="Display identifier" + ) + + # Optional fields + oauth_token_id: str | None = Field(alias="oauth-token-id", default=None) + github_app_installation_id: str | None = Field( + alias="github-app-installation-id", default=None + ) + organization_name: str | None = Field(alias="organization-name", default=None) + branch: str | None = Field( + default=None, description="Branch for branch-based modules" + ) + tags: bool | None = Field(default=None, description="Enable tag-based publishing") + source_directory: str | None = Field(alias="source-directory", default=None) + tag_prefix: str | None = Field(alias="tag-prefix", default=None) + + class Config: + allow_population_by_field_name = True + + +class RegistryModuleVCSRepoUpdateOptions(BaseModel): + """VCS repository update options for registry modules.""" + + branch: str | None = None + tags: bool | None = None + source_directory: str | None = None + tag_prefix: str | None = None + + +class RegistryModuleCreateWithVCSConnectionOptions(BaseModel): + """Options for creating a registry module with VCS connection.""" + + # Required: VCS repository information + vcs_repo: RegistryModuleVCSRepoOptions = Field(alias="vcs-repo") + + # Optional: Initial version for branch-based modules. Defaults to "0.0.0". + initial_version: str | None = Field(alias="initial-version", default=None) + + # Optional: Test configuration + test_config: TestConfig | None = Field(alias="test-config", default=None) + + # Additional fields that might be needed for the API + name: str | None = Field( + default=None, description="Module name (derived from repo if not provided)" + ) + provider: str | None = Field(default=None, description="Provider name") + registry_name: RegistryName | None = Field(alias="registry-name", default=None) + namespace: str | None = Field( + default=None, description="Namespace for public modules" + ) + + class Config: + allow_population_by_field_name = True + + +class RegistryModuleUpdateOptions(BaseModel): + """Options for updating a registry module.""" + + vcs_repo: RegistryModuleVCSRepoUpdateOptions | None = None + no_code: bool | None = None diff --git a/src/tfe/utils.py b/src/tfe/utils.py index 303431a4..42c8f413 100644 --- a/src/tfe/utils.py +++ b/src/tfe/utils.py @@ -5,6 +5,9 @@ from collections.abc import Callable _STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{2,}$") +_VERSION_PATTERN = re.compile( + r"^\d+\.\d+\.\d+(?:-[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)*)?(?:\+[a-zA-Z0-9]+(?:\.[a-zA-Z0-9]+)*)?$" +) def poll_until( @@ -29,3 +32,8 @@ def valid_string(v: str | None) -> bool: def valid_string_id(v: str | None) -> bool: return v is not None and _STRING_ID_PATTERN.match(str(v)) is not None + + +def valid_version(v: str | None) -> bool: + """Validate semantic version string.""" + return v is not None and _VERSION_PATTERN.match(str(v)) is not None From 9dbe9e3112be5816db07906a6e39a5b975121eff Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Thu, 18 Sep 2025 13:00:33 +0530 Subject: [PATCH 2/4] lint issue fixed --- src/tfe/resources/registry_module.py | 4 +-- src/tfe/resources/variable.py | 2 +- src/tfe/resources/workspaces.py | 6 ++-- src/tfe/types/__init__.py | 38 ++++++++++++++++++++++++-- src/tfe/types/registry_module_types.py | 7 +++-- src/tfe/utils.py | 9 ++++-- 6 files changed, 53 insertions(+), 13 deletions(-) diff --git a/src/tfe/resources/registry_module.py b/src/tfe/resources/registry_module.py index a7042c9c..0b15b7bb 100644 --- a/src/tfe/resources/registry_module.py +++ b/src/tfe/resources/registry_module.py @@ -21,12 +21,12 @@ RegistryModuleListOptions, RegistryModulePermissions, RegistryModuleUpdateOptions, + RegistryModuleVCSRepo, RegistryModuleVersion, RegistryModuleVersionStatuses, RegistryName, TerraformRegistryModule, TestConfig, - VCSRepo, ) from ..utils import valid_string, valid_string_id, valid_version from ._base import _Service @@ -453,7 +453,7 @@ def _parse_registry_module(self, data: dict[str, Any]) -> RegistryModule: vcs_repo = None if "vcs-repo" in attributes: vcs_data = attributes["vcs-repo"] - vcs_repo = VCSRepo( + vcs_repo = RegistryModuleVCSRepo( branch=vcs_data.get("branch"), display_identifier=vcs_data.get("display-identifier"), identifier=vcs_data.get("identifier"), diff --git a/src/tfe/resources/variable.py b/src/tfe/resources/variable.py index b6751f50..76778e0c 100644 --- a/src/tfe/resources/variable.py +++ b/src/tfe/resources/variable.py @@ -9,7 +9,7 @@ ERR_REQUIRED_CATEGORY, ERR_REQUIRED_KEY, ) -from ..types import ( +from ..types import ( # type: ignore[attr-defined] Variable, VariableCreateOptions, VariableListOptions, diff --git a/src/tfe/resources/workspaces.py b/src/tfe/resources/workspaces.py index 358acdc0..a8d091ee 100644 --- a/src/tfe/resources/workspaces.py +++ b/src/tfe/resources/workspaces.py @@ -16,7 +16,7 @@ WorkspaceMinimumLimitError, WorkspaceRequiredError, ) -from ..types import ( +from ..types import ( # type: ignore[attr-defined] DataRetentionPolicy, DataRetentionPolicyChoice, DataRetentionPolicyDeleteOlder, @@ -69,7 +69,7 @@ def _em_safe(v: Any) -> ExecutionMode | None: # Only accept strings; map to enum if known, else None if not isinstance(v, str): return None - return ExecutionMode._value2member_map_.get(v) # type: ignore[return-value] + return ExecutionMode._value2member_map_.get(v) def _ws_from(d: dict[str, Any], org: str | None = None) -> Workspace: @@ -1174,4 +1174,4 @@ def readme(self, workspace_id: str) -> str | None: if inc.get("type") == "workspace-readme" and inc.get("id") == readme_id: return (inc.get("attributes") or {}).get("raw-markdown") - return None \ No newline at end of file + return None diff --git a/src/tfe/types/__init__.py b/src/tfe/types/__init__.py index 0a249d1d..f176a1fc 100644 --- a/src/tfe/types/__init__.py +++ b/src/tfe/types/__init__.py @@ -24,6 +24,7 @@ RegistryModulePermissions, RegistryModuleStatus, RegistryModuleUpdateOptions, + RegistryModuleVCSRepo, RegistryModuleVCSRepoOptions, RegistryModuleVCSRepoUpdateOptions, RegistryModuleVersion, @@ -32,8 +33,8 @@ RegistryName, Resource, Root, + TerraformRegistryModule, TestConfig, - VCSRepo, ) # Define what should be available when importing with * @@ -57,6 +58,7 @@ "RegistryModulePermissions", "RegistryModuleStatus", "RegistryModuleUpdateOptions", + "RegistryModuleVCSRepo", "RegistryModuleVCSRepoOptions", "RegistryModuleVCSRepoUpdateOptions", "RegistryModuleVersion", @@ -66,7 +68,7 @@ "Resource", "Root", "TestConfig", - "VCSRepo", + "TerraformRegistryModule", # Main types from types.py (will be dynamically added below) "Capacity", "DataRetentionPolicy", @@ -76,8 +78,10 @@ "DataRetentionPolicyDontDelete", "DataRetentionPolicyDontDeleteSetOptions", "DataRetentionPolicySetOptions", + "EffectiveTagBinding", "Entitlements", "ExecutionMode", + "LockedByChoice", "Organization", "OrganizationCreateOptions", "OrganizationUpdateOptions", @@ -87,7 +91,37 @@ "Run", "RunQueue", "RunStatus", + "Tag", + "TagBinding", + "TagList", + "Variable", + "VariableCreateOptions", + "VariableListOptions", + "VariableUpdateOptions", + "VCSRepo", "Workspace", + "WorkspaceActions", + "WorkspaceAddRemoteStateConsumersOptions", + "WorkspaceAddTagBindingsOptions", + "WorkspaceAddTagsOptions", + "WorkspaceAssignSSHKeyOptions", + "WorkspaceCreateOptions", + "WorkspaceIncludeOpt", + "WorkspaceList", + "WorkspaceListOptions", + "WorkspaceListRemoteStateConsumersOptions", + "WorkspaceLockOptions", + "WorkspaceOutputs", + "WorkspacePermissions", + "WorkspaceReadOptions", + "WorkspaceRemoveRemoteStateConsumersOptions", + "WorkspaceRemoveTagsOptions", + "WorkspaceRemoveVCSConnectionOptions", + "WorkspaceSettingOverwrites", + "WorkspaceSource", + "WorkspaceTagListOptions", + "WorkspaceUpdateOptions", + "WorkspaceUpdateRemoteStateConsumersOptions", ] # Load the main types.py file that's at the same level as this types/ directory diff --git a/src/tfe/types/registry_module_types.py b/src/tfe/types/registry_module_types.py index 4f68c159..fd4a7216 100644 --- a/src/tfe/types/registry_module_types.py +++ b/src/tfe/types/registry_module_types.py @@ -87,8 +87,8 @@ class RegistryModulePermissions(BaseModel): can_retry: bool -class VCSRepo(BaseModel): - """VCS repository configuration.""" +class RegistryModuleVCSRepo(BaseModel): + """VCS repository configuration for registry modules.""" branch: str | None = None display_identifier: str | None = None @@ -101,6 +101,7 @@ class VCSRepo(BaseModel): tags: bool | None = None source_directory: str | None = None tag_prefix: str | None = None + organization_name: str | None = None class TestConfig(BaseModel): @@ -132,7 +133,7 @@ class RegistryModule(BaseModel): publishing_mechanism: PublishingMechanism | None = None status: RegistryModuleStatus | None = None test_config: TestConfig | None = None - vcs_repo: VCSRepo | None = None + vcs_repo: RegistryModuleVCSRepo | None = None version_statuses: list[RegistryModuleVersionStatuses] = Field(default_factory=list) created_at: str | None = None updated_at: str | None = None diff --git a/src/tfe/utils.py b/src/tfe/utils.py index 2a31ba06..39f3b314 100644 --- a/src/tfe/utils.py +++ b/src/tfe/utils.py @@ -15,7 +15,11 @@ UnsupportedBothTriggerPatternsAndPrefixesError, UnsupportedOperationsError, ) -from .types import VCSRepo, WorkspaceCreateOptions, WorkspaceUpdateOptions +from .types import ( # type: ignore[attr-defined] + VCSRepo, + WorkspaceCreateOptions, + WorkspaceUpdateOptions, +) _STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{2,}$") _VERSION_PATTERN = re.compile( @@ -50,7 +54,8 @@ def valid_string_id(v: str | None) -> bool: def valid_version(v: str | None) -> bool: """Validate semantic version string.""" return v is not None and _VERSION_PATTERN.match(str(v)) is not None - + + def is_valid_workspace_name(name: str | None) -> bool: """ Check if a workspace name is valid. From dbac0d96c4b385b8bf1551d89fabd633cd556820 Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Thu, 18 Sep 2025 14:13:31 +0530 Subject: [PATCH 3/4] test issues fixed --- examples/cleanup_organization_modules.py | 405 ++++++++++++++++++ examples/registry_module_complete_test.py | 5 +- src/tfe/{types => models}/__init__.py | 0 .../registry_module_types.py | 0 src/tfe/resources/_base.py | 8 +- src/tfe/resources/registry_module.py | 77 +++- 6 files changed, 486 insertions(+), 9 deletions(-) create mode 100644 examples/cleanup_organization_modules.py rename src/tfe/{types => models}/__init__.py (100%) rename src/tfe/{types => models}/registry_module_types.py (100%) diff --git a/examples/cleanup_organization_modules.py b/examples/cleanup_organization_modules.py new file mode 100644 index 00000000..ff04253c --- /dev/null +++ b/examples/cleanup_organization_modules.py @@ -0,0 +1,405 @@ +#!/usr/bin/env python3 +""" +Organization Module Cleanup Script + +This script demonstrates using the registry module list() and delete functions +to find and remove all modules from a specific organization. + +FEATURES: +- Lists all modules in the specified organization +- Shows module details (name, provider, status, versions) +- Provides options to delete individual modules or all modules +- Uses both delete_by_name() and delete() functions +- Safe mode with confirmation prompts + +USAGE: +- Set your organization name below +- Run the script to see all modules +- Choose which modules to delete +- Confirm deletion operations + +WARNING: This script can delete ALL modules in an organization! +Use with caution and test in a non-production environment first. +""" + +import os +import sys +import time + +# Add the src directory to the path +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + +from tfe import TFEClient, TFEConfig +from tfe.models.registry_module_types import ( + RegistryModule, + RegistryModuleID, + RegistryModuleListOptions, + RegistryName, +) + + +def main(): + """Main function to list and optionally delete organization modules.""" + + # ===================================================== + # CONFIGURATION - MODIFY THESE VALUES + # ===================================================== + organization_name = "aayush-test" # Replace with your organization + + print("=" * 80) + print("ORGANIZATION MODULE CLEANUP SCRIPT") + print("=" * 80) + print(f"Target Organization: {organization_name}") + print("=" * 80) + + # Initialize TFE client + try: + config = TFEConfig() + client = TFEClient(config) + print("✓ TFE Client initialized successfully") + except Exception as e: + print(f"✗ Failed to initialize TFE client: {e}") + return + + # ===================================================== + # STEP 1: LIST ALL MODULES IN ORGANIZATION + # ===================================================== + print(f"\n📋 STEP 1: Listing all modules in '{organization_name}' organization...") + + modules: list[RegistryModule] = [] + + try: + # Use the list() function to get all modules + list_options = RegistryModuleListOptions( + # You can add filters here if needed: + # provider="aws", # Filter by provider + # search="test", # Search for specific names + ) + + module_count = 0 + for module in client.registry_modules.list(organization_name, list_options): + modules.append(module) + module_count += 1 + + print(f" {module_count:2d}. Module: {module.name}") + print(f" Provider: {module.provider}") + print(f" Status: {module.status}") + print(f" Registry: {module.registry_name.value if module.registry_name else 'private'}") + print(f" No Code: {module.no_code}") + + # Show version information if available + if hasattr(module, 'version_statuses') and module.version_statuses: + print(f" Versions: {len(module.version_statuses)} version(s)") + for vs in module.version_statuses[:3]: # Show first 3 versions + if hasattr(vs, 'version') and hasattr(vs, 'status'): + # Handle objects with attributes + version_str = vs.version if vs.version else 'unknown' + status_str = vs.status if vs.status else 'unknown' + print(f" - v{version_str}: {status_str}") + elif isinstance(vs, dict): + # Handle dictionary objects + version_str = vs.get('version', 'unknown') + status_str = vs.get('status', 'unknown') + print(f" - v{version_str}: {status_str}") + else: + # Handle other types + print(f" - {vs}") + if len(module.version_statuses) > 3: + print(f" ... and {len(module.version_statuses) - 3} more") + else: + print(" Versions: No version information available") + + print(f" Created: {module.created_at}") + print() + + print(f"📊 Total modules found: {module_count}") + + except Exception as e: + print(f"✗ Error listing modules: {e}") + return + + if not modules: + print("✅ No modules found in the organization. Nothing to delete!") + return + + # ===================================================== + # STEP 2: INTERACTIVE DELETION MENU + # ===================================================== + print("\n🗑️ STEP 2: Module Deletion Options") + print("=" * 50) + + while True: + print("\nChoose an action:") + print("1. Delete specific module by number") + print("2. Delete all modules (DANGEROUS!)") + print("3. Test delete functions on non-existent module") + print("4. Show module details") + print("5. Exit without deleting") + + choice = input("\nEnter your choice (1-5): ").strip() + + if choice == "1": + delete_specific_module(client, organization_name, modules) + elif choice == "2": + delete_all_modules(client, organization_name, modules) + elif choice == "3": + test_delete_functions(client, organization_name) + elif choice == "4": + show_module_details(client, organization_name, modules) + elif choice == "5": + print("👋 Exiting without making changes.") + break + else: + print("❌ Invalid choice. Please try again.") + +def delete_specific_module(client: TFEClient, organization_name: str, modules: list[RegistryModule]): + """Delete a specific module chosen by the user.""" + + if not modules: + print("❌ No modules available to delete.") + return + + try: + print("\nAvailable modules:") + for i, module in enumerate(modules, 1): + print(f" {i:2d}. {module.name}/{module.provider}") + + choice = input(f"\nEnter module number to delete (1-{len(modules)}): ").strip() + + try: + module_index = int(choice) - 1 + if module_index < 0 or module_index >= len(modules): + print("❌ Invalid module number.") + return + except ValueError: + print("❌ Please enter a valid number.") + return + + selected_module = modules[module_index] + + print("\n⚠️ DELETION CONFIRMATION") + print(f"Module to delete: {selected_module.name}") + print(f"Provider: {selected_module.provider}") + print(f"Status: {selected_module.status}") + + confirm = input(f"\nAre you sure you want to delete '{selected_module.name}/{selected_module.provider}'? (yes/no): ").strip().lower() + + if confirm in ['yes', 'y']: + # Test both delete functions + print("\n🗑️ Deleting module using delete_by_name()...") + + module_id = RegistryModuleID( + organization=organization_name, + name=selected_module.name, + provider=selected_module.provider, + registry_name=RegistryName.PRIVATE + ) + + # Use delete_by_name function + client.registry_modules.delete_by_name(module_id) + print(f"✅ Successfully called delete_by_name() for {selected_module.name}") + + # Wait a moment and verify + print("⏳ Waiting 3 seconds for deletion to process...") + time.sleep(3) + + try: + client.registry_modules.read(module_id) + print("⚠️ Module still exists (deletion may take more time)") + except Exception: + print("✅ Confirmed: Module has been deleted") + # Remove from our local list + modules.remove(selected_module) + + else: + print("❌ Deletion cancelled.") + + except Exception as e: + print(f"✗ Error deleting module: {e}") + +def delete_all_modules(client: TFEClient, organization_name: str, modules: list[RegistryModule]): + """Delete all modules in the organization (with strong confirmation).""" + + if not modules: + print("❌ No modules to delete.") + return + + print("\n🚨 DANGER ZONE: DELETE ALL MODULES") + print("=" * 50) + print(f"This will delete ALL {len(modules)} modules in '{organization_name}':") + + for i, module in enumerate(modules, 1): + print(f" {i:2d}. {module.name}/{module.provider}") + + print("\n⚠️ This action cannot be undone!") + + # Multiple confirmations for safety + confirm1 = input(f"\nType 'DELETE ALL' to confirm deletion of {len(modules)} modules: ").strip() + if confirm1 != "DELETE ALL": + print("❌ Deletion cancelled - confirmation text did not match.") + return + + confirm2 = input(f"Type the organization name '{organization_name}' to confirm: ").strip() + if confirm2 != organization_name: + print("❌ Deletion cancelled - organization name did not match.") + return + + final_confirm = input("Type 'I understand this cannot be undone' to proceed: ").strip() + if final_confirm != "I understand this cannot be undone": + print("❌ Deletion cancelled - final confirmation failed.") + return + + print(f"\n🗑️ Proceeding with deletion of all {len(modules)} modules...") + + deleted_count = 0 + failed_count = 0 + + for i, module in enumerate(modules, 1): + try: + print(f" [{i:2d}/{len(modules)}] Deleting {module.name}/{module.provider}...") + + # Alternate between delete_by_name and delete functions + if i % 2 == 1: + # Use delete_by_name + module_id = RegistryModuleID( + organization=organization_name, + name=module.name, + provider=module.provider, + registry_name=RegistryName.PRIVATE + ) + client.registry_modules.delete_by_name(module_id) + print(" ✅ Used delete_by_name()") + else: + # Use delete + client.registry_modules.delete(organization_name, module.name) + print(" ✅ Used delete()") + + deleted_count += 1 + time.sleep(1) # Brief pause between deletions + + except Exception as e: + print(f" ✗ Failed: {e}") + failed_count += 1 + + print("\n📊 Deletion Summary:") + print(f" ✅ Successfully deleted: {deleted_count}") + print(f" ✗ Failed: {failed_count}") + print(f" 📋 Total attempted: {len(modules)}") + +def test_delete_functions(client: TFEClient, organization_name: str): + """Test delete functions on non-existent modules for demonstration.""" + + print("\n🧪 TESTING DELETE FUNCTIONS") + print("=" * 50) + print("Testing delete functions with non-existent modules (safe testing)...") + + # Test delete() function + print("\n1. Testing delete() function:") + try: + test_name = "non-existent-test-module" + print(f" Calling delete('{organization_name}', '{test_name}')") + client.registry_modules.delete(organization_name, test_name) + print(" ✅ delete() function executed successfully") + except Exception as e: + print(f" ⚠️ Expected error: {e}") + + # Test delete_by_name() function + print("\n2. Testing delete_by_name() function:") + try: + test_module_id = RegistryModuleID( + organization=organization_name, + name="non-existent-test-module", + provider="aws", + registry_name=RegistryName.PRIVATE + ) + print(" Calling delete_by_name() with non-existent module") + client.registry_modules.delete_by_name(test_module_id) + print(" ✅ delete_by_name() function executed successfully") + except Exception as e: + print(f" ⚠️ Expected error: {e}") + + print("\n✅ Delete function testing completed!") + +def show_module_details(client: TFEClient, organization_name: str, modules: list[RegistryModule]): + """Show detailed information about a specific module.""" + + if not modules: + print("❌ No modules available.") + return + + try: + print("\nAvailable modules:") + for i, module in enumerate(modules, 1): + print(f" {i:2d}. {module.name}/{module.provider}") + + choice = input(f"\nEnter module number for details (1-{len(modules)}): ").strip() + + try: + module_index = int(choice) - 1 + if module_index < 0 or module_index >= len(modules): + print("❌ Invalid module number.") + return + except ValueError: + print("❌ Please enter a valid number.") + return + + selected_module = modules[module_index] + + print("\n📋 DETAILED MODULE INFORMATION") + print("=" * 50) + print(f"Name: {selected_module.name}") + print(f"Provider: {selected_module.provider}") + print(f"Status: {selected_module.status}") + print(f"Registry: {selected_module.registry_name.value if selected_module.registry_name else 'private'}") + print(f"No Code: {selected_module.no_code}") + print(f"Created: {selected_module.created_at}") + print(f"Updated: {selected_module.updated_at}") + + if hasattr(selected_module, 'organization') and selected_module.organization: + print(f"Organization: {selected_module.organization}") + + if hasattr(selected_module, 'vcs_repo') and selected_module.vcs_repo: + print(f"VCS Repository: {selected_module.vcs_repo}") + + if hasattr(selected_module, 'version_statuses') and selected_module.version_statuses: + print(f"\nVersions ({len(selected_module.version_statuses)}):") + for vs in selected_module.version_statuses: + if hasattr(vs, 'version') and hasattr(vs, 'status'): + # Handle objects with attributes + version_str = vs.version if vs.version else 'unknown' + status_str = vs.status if vs.status else 'unknown' + print(f" - v{version_str}: {status_str}") + elif isinstance(vs, dict): + # Handle dictionary objects + version_str = vs.get('version', 'unknown') + status_str = vs.get('status', 'unknown') + print(f" - v{version_str}: {status_str}") + else: + # Handle other types + print(f" - {vs}") + + # Try to get more details using read() function + try: + print("\n🔍 Fetching additional details using read() function...") + module_id = RegistryModuleID( + organization=organization_name, + name=selected_module.name, + provider=selected_module.provider, + registry_name=RegistryName.PRIVATE + ) + + detailed_module = client.registry_modules.read(module_id) + print("✅ Successfully retrieved detailed information") + print(f"Publishing Mechanism: {detailed_module.publishing_mechanism}") + + if hasattr(detailed_module, 'permissions') and detailed_module.permissions: + print(f"Permissions: {detailed_module.permissions}") + + except Exception as e: + print(f"⚠️ Could not fetch additional details: {e}") + + except Exception as e: + print(f"✗ Error showing module details: {e}") + +if __name__ == "__main__": + main() diff --git a/examples/registry_module_complete_test.py b/examples/registry_module_complete_test.py index 4966cbe6..82ef0430 100644 --- a/examples/registry_module_complete_test.py +++ b/examples/registry_module_complete_test.py @@ -40,7 +40,8 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) from tfe import TFEClient, TFEConfig -from tfe.types.registry_module_types import ( +from tfe.errors import NotFound +from tfe.models.registry_module_types import ( AgentExecutionMode, RegistryModuleCreateOptions, RegistryModuleCreateVersionOptions, @@ -87,6 +88,8 @@ def main(): for i, module in enumerate(modules[:3], 1): print(f" {i}. {module.name}/{module.provider} (ID: {module.id})") + except NotFound: + print(" ✓ No modules found (organization may not exist or no private modules available)") except Exception as e: print(f" ✗ Error: {e}") diff --git a/src/tfe/types/__init__.py b/src/tfe/models/__init__.py similarity index 100% rename from src/tfe/types/__init__.py rename to src/tfe/models/__init__.py diff --git a/src/tfe/types/registry_module_types.py b/src/tfe/models/registry_module_types.py similarity index 100% rename from src/tfe/types/registry_module_types.py rename to src/tfe/models/registry_module_types.py diff --git a/src/tfe/resources/_base.py b/src/tfe/resources/_base.py index 9736d6c3..fdcb99c9 100644 --- a/src/tfe/resources/_base.py +++ b/src/tfe/resources/_base.py @@ -19,7 +19,13 @@ def _list( p.setdefault("page[number]", page) p.setdefault("page[size]", 100) r = self.t.request("GET", path, params=p) - data = r.json().get("data", []) + + # Handle cases where r.json() returns None or is not a dict + json_response = r.json() + if json_response is None: + json_response = {} + + data = json_response.get("data", []) yield from data if len(data) < p["page[size]"]: break diff --git a/src/tfe/resources/registry_module.py b/src/tfe/resources/registry_module.py index 0b15b7bb..2d585b5a 100644 --- a/src/tfe/resources/registry_module.py +++ b/src/tfe/resources/registry_module.py @@ -9,7 +9,7 @@ ERR_INVALID_ORG, ERR_INVALID_VERSION, ) -from ..types.registry_module_types import ( +from ..models.registry_module_types import ( AgentExecutionMode, Commit, CommitList, @@ -62,6 +62,8 @@ def list( params["page[size]"] = str(options.page_size) for item in self._list(path, params=params): + if item is None: + continue # type: ignore[unreachable] # Skip None items yield self._parse_registry_module(item) def list_commits(self, module_id: RegistryModuleID) -> CommitList: @@ -212,6 +214,63 @@ def read_version( return self._parse_registry_module_version(data) + def list_versions(self, module_id: RegistryModuleID) -> list[RegistryModuleVersion]: # type: ignore[valid-type] + """List all versions of a registry module.""" + if not self._validate_module_id(module_id): + raise ValueError("Invalid module ID") + + try: + if module_id.id: + path = f"/api/v2/registry-modules/{module_id.id}/versions" + else: + registry_name = module_id.registry_name or RegistryName.PRIVATE + namespace = module_id.namespace or module_id.organization + + path = ( + f"/api/v2/organizations/{module_id.organization}/" + f"registry-modules/{registry_name.value}/{namespace}/" + f"{module_id.name}/{module_id.provider}/versions" + ) + + response = self.t.request("GET", path) + response_data = response.json() + + # Handle the case where data might be None or empty + data = response_data.get("data", []) if response_data else [] + + versions = [] + for item in data: + if item: # Skip None items + versions.append(self._parse_registry_module_version(item)) + + return versions + + except Exception: + # Fallback: If the API endpoint doesn't exist, try to get versions from the module itself + try: + module = self.read(module_id) + versions = [] + + # Convert version_statuses to RegistryModuleVersion objects + for vs in module.version_statuses: + # Create a minimal RegistryModuleVersion from version status + version_data = { + "id": f"modver-{vs.version}", + "type": "registry-module-versions", + "attributes": { + "version": vs.version, + "status": vs.status.value, + "created-at": None, + "updated-at": None, + "error": getattr(vs, 'error', None) + } + } + versions.append(self._parse_registry_module_version(version_data)) + + return versions + except Exception: + return [] # Return empty list if all methods fail + def read_terraform_registry_module( self, module_id: RegistryModuleID, version: str ) -> TerraformRegistryModule: @@ -429,6 +488,9 @@ def _validate_create_with_vcs_options( def _parse_registry_module(self, data: dict[str, Any]) -> RegistryModule: """Parse registry module from API response.""" + if data is None: + raise ValueError("Cannot parse registry module: data is None") + attributes = data.get("attributes", {}) relationships = data.get("relationships", {}) @@ -471,12 +533,13 @@ def _parse_registry_module(self, data: dict[str, Any]) -> RegistryModule: test_config = None if "test-config" in attributes: test_data = attributes["test-config"] - test_config = TestConfig( - tests_enabled=test_data.get("tests-enabled", False), - agent_execution_mode=AgentExecutionMode( - test_data.get("agent-execution-mode", "remote") - ), - ) + if test_data is not None: + test_config = TestConfig( + tests_enabled=test_data.get("tests-enabled", False), + agent_execution_mode=AgentExecutionMode( + test_data.get("agent-execution-mode", "remote") + ), + ) # Parse version statuses with field name mapping version_statuses = [] From 063fc7efd3faf63d725d8e9726ccf4f2fe393149 Mon Sep 17 00:00:00 2001 From: aayushsingh2502 Date: Thu, 18 Sep 2025 14:21:24 +0530 Subject: [PATCH 4/4] lint issue fixed --- examples/cleanup_organization_modules.py | 405 ---------------------- examples/registry_module_complete_test.py | 4 +- src/tfe/resources/organizations.py | 4 +- src/tfe/resources/registry_module.py | 4 +- src/tfe/resources/variable.py | 2 +- src/tfe/resources/workspaces.py | 5 +- src/tfe/utils.py | 2 +- 7 files changed, 12 insertions(+), 414 deletions(-) delete mode 100644 examples/cleanup_organization_modules.py diff --git a/examples/cleanup_organization_modules.py b/examples/cleanup_organization_modules.py deleted file mode 100644 index ff04253c..00000000 --- a/examples/cleanup_organization_modules.py +++ /dev/null @@ -1,405 +0,0 @@ -#!/usr/bin/env python3 -""" -Organization Module Cleanup Script - -This script demonstrates using the registry module list() and delete functions -to find and remove all modules from a specific organization. - -FEATURES: -- Lists all modules in the specified organization -- Shows module details (name, provider, status, versions) -- Provides options to delete individual modules or all modules -- Uses both delete_by_name() and delete() functions -- Safe mode with confirmation prompts - -USAGE: -- Set your organization name below -- Run the script to see all modules -- Choose which modules to delete -- Confirm deletion operations - -WARNING: This script can delete ALL modules in an organization! -Use with caution and test in a non-production environment first. -""" - -import os -import sys -import time - -# Add the src directory to the path -sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) - -from tfe import TFEClient, TFEConfig -from tfe.models.registry_module_types import ( - RegistryModule, - RegistryModuleID, - RegistryModuleListOptions, - RegistryName, -) - - -def main(): - """Main function to list and optionally delete organization modules.""" - - # ===================================================== - # CONFIGURATION - MODIFY THESE VALUES - # ===================================================== - organization_name = "aayush-test" # Replace with your organization - - print("=" * 80) - print("ORGANIZATION MODULE CLEANUP SCRIPT") - print("=" * 80) - print(f"Target Organization: {organization_name}") - print("=" * 80) - - # Initialize TFE client - try: - config = TFEConfig() - client = TFEClient(config) - print("✓ TFE Client initialized successfully") - except Exception as e: - print(f"✗ Failed to initialize TFE client: {e}") - return - - # ===================================================== - # STEP 1: LIST ALL MODULES IN ORGANIZATION - # ===================================================== - print(f"\n📋 STEP 1: Listing all modules in '{organization_name}' organization...") - - modules: list[RegistryModule] = [] - - try: - # Use the list() function to get all modules - list_options = RegistryModuleListOptions( - # You can add filters here if needed: - # provider="aws", # Filter by provider - # search="test", # Search for specific names - ) - - module_count = 0 - for module in client.registry_modules.list(organization_name, list_options): - modules.append(module) - module_count += 1 - - print(f" {module_count:2d}. Module: {module.name}") - print(f" Provider: {module.provider}") - print(f" Status: {module.status}") - print(f" Registry: {module.registry_name.value if module.registry_name else 'private'}") - print(f" No Code: {module.no_code}") - - # Show version information if available - if hasattr(module, 'version_statuses') and module.version_statuses: - print(f" Versions: {len(module.version_statuses)} version(s)") - for vs in module.version_statuses[:3]: # Show first 3 versions - if hasattr(vs, 'version') and hasattr(vs, 'status'): - # Handle objects with attributes - version_str = vs.version if vs.version else 'unknown' - status_str = vs.status if vs.status else 'unknown' - print(f" - v{version_str}: {status_str}") - elif isinstance(vs, dict): - # Handle dictionary objects - version_str = vs.get('version', 'unknown') - status_str = vs.get('status', 'unknown') - print(f" - v{version_str}: {status_str}") - else: - # Handle other types - print(f" - {vs}") - if len(module.version_statuses) > 3: - print(f" ... and {len(module.version_statuses) - 3} more") - else: - print(" Versions: No version information available") - - print(f" Created: {module.created_at}") - print() - - print(f"📊 Total modules found: {module_count}") - - except Exception as e: - print(f"✗ Error listing modules: {e}") - return - - if not modules: - print("✅ No modules found in the organization. Nothing to delete!") - return - - # ===================================================== - # STEP 2: INTERACTIVE DELETION MENU - # ===================================================== - print("\n🗑️ STEP 2: Module Deletion Options") - print("=" * 50) - - while True: - print("\nChoose an action:") - print("1. Delete specific module by number") - print("2. Delete all modules (DANGEROUS!)") - print("3. Test delete functions on non-existent module") - print("4. Show module details") - print("5. Exit without deleting") - - choice = input("\nEnter your choice (1-5): ").strip() - - if choice == "1": - delete_specific_module(client, organization_name, modules) - elif choice == "2": - delete_all_modules(client, organization_name, modules) - elif choice == "3": - test_delete_functions(client, organization_name) - elif choice == "4": - show_module_details(client, organization_name, modules) - elif choice == "5": - print("👋 Exiting without making changes.") - break - else: - print("❌ Invalid choice. Please try again.") - -def delete_specific_module(client: TFEClient, organization_name: str, modules: list[RegistryModule]): - """Delete a specific module chosen by the user.""" - - if not modules: - print("❌ No modules available to delete.") - return - - try: - print("\nAvailable modules:") - for i, module in enumerate(modules, 1): - print(f" {i:2d}. {module.name}/{module.provider}") - - choice = input(f"\nEnter module number to delete (1-{len(modules)}): ").strip() - - try: - module_index = int(choice) - 1 - if module_index < 0 or module_index >= len(modules): - print("❌ Invalid module number.") - return - except ValueError: - print("❌ Please enter a valid number.") - return - - selected_module = modules[module_index] - - print("\n⚠️ DELETION CONFIRMATION") - print(f"Module to delete: {selected_module.name}") - print(f"Provider: {selected_module.provider}") - print(f"Status: {selected_module.status}") - - confirm = input(f"\nAre you sure you want to delete '{selected_module.name}/{selected_module.provider}'? (yes/no): ").strip().lower() - - if confirm in ['yes', 'y']: - # Test both delete functions - print("\n🗑️ Deleting module using delete_by_name()...") - - module_id = RegistryModuleID( - organization=organization_name, - name=selected_module.name, - provider=selected_module.provider, - registry_name=RegistryName.PRIVATE - ) - - # Use delete_by_name function - client.registry_modules.delete_by_name(module_id) - print(f"✅ Successfully called delete_by_name() for {selected_module.name}") - - # Wait a moment and verify - print("⏳ Waiting 3 seconds for deletion to process...") - time.sleep(3) - - try: - client.registry_modules.read(module_id) - print("⚠️ Module still exists (deletion may take more time)") - except Exception: - print("✅ Confirmed: Module has been deleted") - # Remove from our local list - modules.remove(selected_module) - - else: - print("❌ Deletion cancelled.") - - except Exception as e: - print(f"✗ Error deleting module: {e}") - -def delete_all_modules(client: TFEClient, organization_name: str, modules: list[RegistryModule]): - """Delete all modules in the organization (with strong confirmation).""" - - if not modules: - print("❌ No modules to delete.") - return - - print("\n🚨 DANGER ZONE: DELETE ALL MODULES") - print("=" * 50) - print(f"This will delete ALL {len(modules)} modules in '{organization_name}':") - - for i, module in enumerate(modules, 1): - print(f" {i:2d}. {module.name}/{module.provider}") - - print("\n⚠️ This action cannot be undone!") - - # Multiple confirmations for safety - confirm1 = input(f"\nType 'DELETE ALL' to confirm deletion of {len(modules)} modules: ").strip() - if confirm1 != "DELETE ALL": - print("❌ Deletion cancelled - confirmation text did not match.") - return - - confirm2 = input(f"Type the organization name '{organization_name}' to confirm: ").strip() - if confirm2 != organization_name: - print("❌ Deletion cancelled - organization name did not match.") - return - - final_confirm = input("Type 'I understand this cannot be undone' to proceed: ").strip() - if final_confirm != "I understand this cannot be undone": - print("❌ Deletion cancelled - final confirmation failed.") - return - - print(f"\n🗑️ Proceeding with deletion of all {len(modules)} modules...") - - deleted_count = 0 - failed_count = 0 - - for i, module in enumerate(modules, 1): - try: - print(f" [{i:2d}/{len(modules)}] Deleting {module.name}/{module.provider}...") - - # Alternate between delete_by_name and delete functions - if i % 2 == 1: - # Use delete_by_name - module_id = RegistryModuleID( - organization=organization_name, - name=module.name, - provider=module.provider, - registry_name=RegistryName.PRIVATE - ) - client.registry_modules.delete_by_name(module_id) - print(" ✅ Used delete_by_name()") - else: - # Use delete - client.registry_modules.delete(organization_name, module.name) - print(" ✅ Used delete()") - - deleted_count += 1 - time.sleep(1) # Brief pause between deletions - - except Exception as e: - print(f" ✗ Failed: {e}") - failed_count += 1 - - print("\n📊 Deletion Summary:") - print(f" ✅ Successfully deleted: {deleted_count}") - print(f" ✗ Failed: {failed_count}") - print(f" 📋 Total attempted: {len(modules)}") - -def test_delete_functions(client: TFEClient, organization_name: str): - """Test delete functions on non-existent modules for demonstration.""" - - print("\n🧪 TESTING DELETE FUNCTIONS") - print("=" * 50) - print("Testing delete functions with non-existent modules (safe testing)...") - - # Test delete() function - print("\n1. Testing delete() function:") - try: - test_name = "non-existent-test-module" - print(f" Calling delete('{organization_name}', '{test_name}')") - client.registry_modules.delete(organization_name, test_name) - print(" ✅ delete() function executed successfully") - except Exception as e: - print(f" ⚠️ Expected error: {e}") - - # Test delete_by_name() function - print("\n2. Testing delete_by_name() function:") - try: - test_module_id = RegistryModuleID( - organization=organization_name, - name="non-existent-test-module", - provider="aws", - registry_name=RegistryName.PRIVATE - ) - print(" Calling delete_by_name() with non-existent module") - client.registry_modules.delete_by_name(test_module_id) - print(" ✅ delete_by_name() function executed successfully") - except Exception as e: - print(f" ⚠️ Expected error: {e}") - - print("\n✅ Delete function testing completed!") - -def show_module_details(client: TFEClient, organization_name: str, modules: list[RegistryModule]): - """Show detailed information about a specific module.""" - - if not modules: - print("❌ No modules available.") - return - - try: - print("\nAvailable modules:") - for i, module in enumerate(modules, 1): - print(f" {i:2d}. {module.name}/{module.provider}") - - choice = input(f"\nEnter module number for details (1-{len(modules)}): ").strip() - - try: - module_index = int(choice) - 1 - if module_index < 0 or module_index >= len(modules): - print("❌ Invalid module number.") - return - except ValueError: - print("❌ Please enter a valid number.") - return - - selected_module = modules[module_index] - - print("\n📋 DETAILED MODULE INFORMATION") - print("=" * 50) - print(f"Name: {selected_module.name}") - print(f"Provider: {selected_module.provider}") - print(f"Status: {selected_module.status}") - print(f"Registry: {selected_module.registry_name.value if selected_module.registry_name else 'private'}") - print(f"No Code: {selected_module.no_code}") - print(f"Created: {selected_module.created_at}") - print(f"Updated: {selected_module.updated_at}") - - if hasattr(selected_module, 'organization') and selected_module.organization: - print(f"Organization: {selected_module.organization}") - - if hasattr(selected_module, 'vcs_repo') and selected_module.vcs_repo: - print(f"VCS Repository: {selected_module.vcs_repo}") - - if hasattr(selected_module, 'version_statuses') and selected_module.version_statuses: - print(f"\nVersions ({len(selected_module.version_statuses)}):") - for vs in selected_module.version_statuses: - if hasattr(vs, 'version') and hasattr(vs, 'status'): - # Handle objects with attributes - version_str = vs.version if vs.version else 'unknown' - status_str = vs.status if vs.status else 'unknown' - print(f" - v{version_str}: {status_str}") - elif isinstance(vs, dict): - # Handle dictionary objects - version_str = vs.get('version', 'unknown') - status_str = vs.get('status', 'unknown') - print(f" - v{version_str}: {status_str}") - else: - # Handle other types - print(f" - {vs}") - - # Try to get more details using read() function - try: - print("\n🔍 Fetching additional details using read() function...") - module_id = RegistryModuleID( - organization=organization_name, - name=selected_module.name, - provider=selected_module.provider, - registry_name=RegistryName.PRIVATE - ) - - detailed_module = client.registry_modules.read(module_id) - print("✅ Successfully retrieved detailed information") - print(f"Publishing Mechanism: {detailed_module.publishing_mechanism}") - - if hasattr(detailed_module, 'permissions') and detailed_module.permissions: - print(f"Permissions: {detailed_module.permissions}") - - except Exception as e: - print(f"⚠️ Could not fetch additional details: {e}") - - except Exception as e: - print(f"✗ Error showing module details: {e}") - -if __name__ == "__main__": - main() diff --git a/examples/registry_module_complete_test.py b/examples/registry_module_complete_test.py index 82ef0430..a6eddd95 100644 --- a/examples/registry_module_complete_test.py +++ b/examples/registry_module_complete_test.py @@ -89,7 +89,9 @@ def main(): print(f" {i}. {module.name}/{module.provider} (ID: {module.id})") except NotFound: - print(" ✓ No modules found (organization may not exist or no private modules available)") + print( + " ✓ No modules found (organization may not exist or no private modules available)" + ) except Exception as e: print(f" ✗ Error: {e}") diff --git a/src/tfe/resources/organizations.py b/src/tfe/resources/organizations.py index 46d5be8f..95ec83e3 100644 --- a/src/tfe/resources/organizations.py +++ b/src/tfe/resources/organizations.py @@ -9,7 +9,7 @@ ERR_REQUIRED_EMAIL, ERR_REQUIRED_NAME, ) -from ..types import ( # type: ignore[attr-defined] +from ..types import ( Capacity, DataRetentionPolicy, DataRetentionPolicyChoice, @@ -167,7 +167,7 @@ def read_run_queue( ) data = r.json() - from ..types import Pagination, Run, RunStatus # type: ignore[attr-defined] + from ..types import Pagination, Run, RunStatus runs = [] for item in data.get("data", []): diff --git a/src/tfe/resources/registry_module.py b/src/tfe/resources/registry_module.py index 2d585b5a..1991895b 100644 --- a/src/tfe/resources/registry_module.py +++ b/src/tfe/resources/registry_module.py @@ -262,8 +262,8 @@ def list_versions(self, module_id: RegistryModuleID) -> list[RegistryModuleVersi "status": vs.status.value, "created-at": None, "updated-at": None, - "error": getattr(vs, 'error', None) - } + "error": getattr(vs, "error", None), + }, } versions.append(self._parse_registry_module_version(version_data)) diff --git a/src/tfe/resources/variable.py b/src/tfe/resources/variable.py index 76778e0c..b6751f50 100644 --- a/src/tfe/resources/variable.py +++ b/src/tfe/resources/variable.py @@ -9,7 +9,7 @@ ERR_REQUIRED_CATEGORY, ERR_REQUIRED_KEY, ) -from ..types import ( # type: ignore[attr-defined] +from ..types import ( Variable, VariableCreateOptions, VariableListOptions, diff --git a/src/tfe/resources/workspaces.py b/src/tfe/resources/workspaces.py index a8d091ee..dbfdf5e3 100644 --- a/src/tfe/resources/workspaces.py +++ b/src/tfe/resources/workspaces.py @@ -16,7 +16,7 @@ WorkspaceMinimumLimitError, WorkspaceRequiredError, ) -from ..types import ( # type: ignore[attr-defined] +from ..types import ( DataRetentionPolicy, DataRetentionPolicyChoice, DataRetentionPolicyDeleteOlder, @@ -69,7 +69,8 @@ def _em_safe(v: Any) -> ExecutionMode | None: # Only accept strings; map to enum if known, else None if not isinstance(v, str): return None - return ExecutionMode._value2member_map_.get(v) + result = ExecutionMode._value2member_map_.get(v) + return result if isinstance(result, ExecutionMode) else None def _ws_from(d: dict[str, Any], org: str | None = None) -> Workspace: diff --git a/src/tfe/utils.py b/src/tfe/utils.py index 39f3b314..7c3a0bfb 100644 --- a/src/tfe/utils.py +++ b/src/tfe/utils.py @@ -15,7 +15,7 @@ UnsupportedBothTriggerPatternsAndPrefixesError, UnsupportedOperationsError, ) -from .types import ( # type: ignore[attr-defined] +from .types import ( VCSRepo, WorkspaceCreateOptions, WorkspaceUpdateOptions,