fix: better support for non-aws providers

We only have an import ID generator for the AWS provider, but for other
providers we can still create the import blocks and do our best at
linking to the provider documentation for that specific resource with
instructions on how to construct the import ID.

Author: stekern

src/tfblocks/main.py

@@ -17,12 +17,7 @@ def get_aws_resource_import_id_generators() -> Dict[str, type]:
     }
 
 
-def file_exists(file_path: str) -> bool:
-    """Check if a file exists."""
-    return os.path.exists(file_path)
-
-
-def extract_resource_addresses_from_content(content: str) -> List[str]:
+def extract_addresses_from_content(content: str) -> List[str]:
     """Extract resource and module addresses from Terraform content."""
     addresses = []
 
@@ -41,20 +36,19 @@ def extract_resource_addresses_from_content(content: str) -> List[str]:
     return addresses
 
 
-def extract_resource_addresses_from_file(file_path: str) -> List[str]:
+def extract_addresses_from_file(file_path: str) -> List[str]:
     """Extract resource and module addresses from a Terraform file."""
     addresses = []
 
-    # First check if file exists
-    if not file_exists(file_path):
+    if not os.path.exists(file_path):
         print(f"Error: File {file_path} does not exist", file=sys.stderr)
         sys.exit(1)
 
     try:
         with open(file_path, "r") as f:
             content = f.read()
 
-        addresses = extract_resource_addresses_from_content(content)
+        addresses = extract_addresses_from_content(content)
 
     except Exception as e:
         print(f"Warning: Could not process file {file_path}: {str(e)}", file=sys.stderr)
@@ -158,12 +152,12 @@ def matches_address_list(addr: str, addr_list: List[str]) -> bool:
 def filter_resources(
     state: Dict[str, Any], addresses: List[str] = [], files: List[str] = []
 ) -> List[Dict[str, Any]]:
-    """Extract matching AWS resources from Terraform state."""
+    """Extract matching resources from Terraform state."""
     # Extract addresses from files if provided
     file_addresses = []
     if files:
         for file_path in files:
-            extracted = extract_resource_addresses_from_file(file_path)
+            extracted = extract_addresses_from_file(file_path)
             file_addresses.extend(extracted)
 
         if not file_addresses:
@@ -185,10 +179,8 @@ def filter_resources(
 
         # Process resources in current module
         for resource in module.get("resources", []):
-            if (
-                resource.get("type", "").startswith("aws_")
-                and resource.get("mode") == "managed"
-                and is_resource_match(resource["address"], addresses, file_addresses)
+            if resource.get("mode") == "managed" and is_resource_match(
+                resource["address"], addresses, file_addresses
             ):
                 resources.append(resource)
 
@@ -202,20 +194,48 @@ def filter_resources(
 
 
 def generate_import_block(
-    resource: Dict[str, Any], schema_classes: Dict[str, type]
-) -> str:
+    resource: Dict[str, Any],
+    schema_classes: Dict[str, type],
+    supported_providers_only: bool = False,
+) -> str | None:
     """Generate Terraform import block for a resource."""
-    matching_class = schema_classes.get(resource["type"])
-    documentation = f"https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/{resource['type'].replace('aws_', '')}#import"
-    import_id = f'"" # TODO: {documentation}'
-
-    if matching_class:
-        try:
-            instance = matching_class(resource["address"], resource["values"])
-            if instance.import_id is not None:
-                import_id = f'"{instance.import_id}"'
-        except Exception:
-            pass
+    provider_name = resource["type"].split("_")[0] if "_" in resource["type"] else ""
+
+    if provider_name == "aws":
+        # For AWS resources, we have import ID generators
+        matching_class = schema_classes.get(resource["type"])
+        documentation = f"https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/{resource['type'].replace('aws_', '')}#import"
+        import_id = f'"" # TODO: {documentation}'
+
+        if matching_class:
+            try:
+                instance = matching_class(resource["address"], resource["values"])
+                if instance.import_id is not None:
+                    import_id = f'"{instance.import_id}"'
+            except Exception:
+                pass
+    elif not supported_providers_only:
+        # For providers we don't have import ID generators for
+        # we create a block with a link to the resource documentation
+        resource_type = resource["type"]
+
+        # Use provider_name field when available to get the right documentation URL
+        if resource.get("provider_name", "").startswith("registry.terraform.io/"):
+            parts = resource["provider_name"].split("/")
+            if len(parts) >= 3:
+                org = parts[1]
+                provider = parts[2]
+                # Best-effort attempt at linking to the right place
+                docs_url = f"https://registry.terraform.io/providers/{org}/{provider}/latest/docs/resources/{resource_type.removeprefix(f'{provider_name}_')}#import"
+                import_id = f'"" # TODO: {docs_url}'
+            else:
+                import_id = '"" # TODO'
+        else:
+            # Fallback to generic message if provider is not on the Terraform registry
+            import_id = '"" # TODO'
+    else:
+        # For unsupported providers when restricting to supported providers only
+        return None  # Skip this resource
 
     return f"""import {{
   to = {resource["address"]}
@@ -239,7 +259,10 @@ def generate_removed_block(resource_addr: str, destroy: bool = False) -> str:
 
 
 def generate_blocks_for_command(
-    resources: List[Dict[str, Any]], command: str, destroy: bool = False
+    resources: List[Dict[str, Any]],
+    command: str,
+    destroy: bool = False,
+    supported_providers_only: bool = False,
 ) -> List[str]:
     """Generate Terraform code blocks based on command."""
     blocks = []
@@ -261,7 +284,14 @@ def generate_blocks_for_command(
     elif command == "import":
         # For import blocks, we need the full resource data
         schema_classes = get_aws_resource_import_id_generators()
-        blocks = [generate_import_block(r, schema_classes) for r in resources]
+        blocks = [
+            block
+            for block in [
+                generate_import_block(r, schema_classes, supported_providers_only)
+                for r in resources
+            ]
+            if block is not None
+        ]
     else:
         raise ValueError(f"Invalid command '{command}'")
     return blocks
@@ -295,6 +325,11 @@ def add_filter_args(cmd_parser: argparse.ArgumentParser):
 
     import_parser = subparsers.add_parser("import", help="Generate import blocks")
     add_filter_args(import_parser)
+    import_parser.add_argument(
+        "--supported-providers-only",
+        action="store_true",
+        help="Only generate import IDs for supported providers (currently only AWS)",
+    )
 
     remove_parser = subparsers.add_parser("remove", help="Generate removed blocks")
     add_filter_args(remove_parser)
@@ -330,7 +365,10 @@ def main():
         return
 
     blocks = generate_blocks_for_command(
-        resources, args.command, getattr(args, "destroy", False)
+        resources,
+        args.command,
+        getattr(args, "destroy", False),
+        getattr(args, "supported_providers_only", False),
     )
 
     if args.no_color:

tests/test_main.py

@@ -1,6 +1,5 @@
-import tempfile
 import unittest
-from unittest.mock import mock_open, patch
+from unittest.mock import patch
 
 from tfblocks import main
 
@@ -42,7 +41,7 @@ def test_module_match(self):
                 "module.my_module[0].aws_s3_bucket.test", ["module.my_module"], []
             )
         )
-        
+
     def test_resource_type_name_match(self):
         """Test matching resources by type and name across different module paths"""
         # Resource in a module should match the same resource type and name from a file
@@ -51,14 +50,14 @@ def test_resource_type_name_match(self):
                 "module.my_module.aws_s3_bucket.test", [], ["aws_s3_bucket.test"]
             )
         )
-        
+
         # Different resource name should not match
         self.assertFalse(
             main.is_resource_match(
                 "module.my_module.aws_s3_bucket.test", [], ["aws_s3_bucket.other"]
             )
         )
-        
+
         # Different resource type should not match
         self.assertFalse(
             main.is_resource_match(
@@ -78,18 +77,24 @@ def test_wildcard_matching(self):
         self.assertFalse(
             main.is_resource_match("aws_lambda_function.test", ["aws_s3_bucket.*"], [])
         )
-        
+
         # Match resources using wildcards in module paths
         self.assertTrue(
-            main.is_resource_match("module.my_module.aws_s3_bucket.test", ["*.aws_s3_bucket.test"], [])
+            main.is_resource_match(
+                "module.my_module.aws_s3_bucket.test", ["*.aws_s3_bucket.test"], []
+            )
         )
-        
+
         # Match more complex patterns
         self.assertTrue(
-            main.is_resource_match("module.my_module.aws_s3_bucket.test", ["module.*.aws_s3_bucket.*"], [])
+            main.is_resource_match(
+                "module.my_module.aws_s3_bucket.test", ["module.*.aws_s3_bucket.*"], []
+            )
         )
         self.assertFalse(
-            main.is_resource_match("other.my_module.aws_s3_bucket.test", ["module.*.aws_s3_bucket.*"], [])
+            main.is_resource_match(
+                "other.my_module.aws_s3_bucket.test", ["module.*.aws_s3_bucket.*"], []
+            )
         )
 
     def test_intersection_filter(self):
@@ -107,7 +112,7 @@ def test_intersection_filter(self):
 
 
 class TestFileProcessing(unittest.TestCase):
-    def test_extract_resource_addresses_from_content(self):
+    def test_extract_addresses_from_content(self):
         """Test extracting resource addresses from content"""
         terraform_content = """
         resource "aws_s3_bucket" "bucket" {
@@ -123,28 +128,12 @@ def test_extract_resource_addresses_from_content(self):
         }
         """
 
-        addresses = main.extract_resource_addresses_from_content(terraform_content)
+        addresses = main.extract_addresses_from_content(terraform_content)
 
         self.assertEqual(len(addresses), 3)
         self.assertIn("aws_s3_bucket.bucket", addresses)
         self.assertIn("aws_dynamodb_table.table", addresses)
         self.assertIn("module.vpc", addresses)
-        
-    def test_file_exists(self):
-        """Test file existence check"""
-        # Test with tempfile to avoid dependencies on filesystem
-        with tempfile.NamedTemporaryFile() as temp_file:
-            self.assertTrue(main.file_exists(temp_file.name))
-            
-        # This path should not exist
-        self.assertFalse(main.file_exists("/path/that/does/not/exist/file.tf"))
-            
-    def test_extract_resource_addresses_nonexistent_file(self):
-        """Test behavior with nonexistent file"""
-        with patch("tfblocks.main.file_exists", return_value=False), \
-             patch("sys.exit") as mock_exit:
-            main.extract_resource_addresses_from_file("nonexistent.tf")
-            mock_exit.assert_called_once_with(1)
 
     def test_filter_resources_basic(self):
         """Test basic resource filtering without files"""
@@ -201,7 +190,7 @@ def test_filter_resources_basic(self):
         resources = main.filter_resources(test_state, ["module.test"])
         self.assertEqual(len(resources), 1)
         self.assertEqual(resources[0]["address"], "module.test.aws_s3_bucket.nested")
-        
+
     def test_filter_resources_with_file_filters(self):
         """Test resource filtering with file filters"""
         # Create a test state
@@ -221,13 +210,13 @@ def test_filter_resources_with_file_filters(self):
         }
 
         # Mock the file handling part without mocking implementation details
-        with patch("tfblocks.main.extract_resource_addresses_from_file") as mock_extract:
+        with patch("tfblocks.main.extract_addresses_from_file") as mock_extract:
             # Case 1: File contains matching resource
             mock_extract.return_value = ["aws_s3_bucket.test"]
             resources = main.filter_resources(test_state, [], ["matching_file.tf"])
             self.assertEqual(len(resources), 1)
             self.assertEqual(resources[0]["address"], "aws_s3_bucket.test")
-            
+
             # Case 2: File contains non-matching resource
             mock_extract.return_value = ["aws_lambda_function.not_in_state"]
             resources = main.filter_resources(test_state, [], ["non_matching_file.tf"])
@@ -250,6 +239,29 @@ def test_generate_import_block(self):
         self.assertIn("to = aws_s3_bucket.test", block)
         self.assertIn("id =", block)
 
+    def test_generate_import_block_for_unsupported_provider(self):
+        """Test generating an import block for an unsupported provider resource"""
+        resource = {
+            "address": "google_storage_bucket.test",
+            "type": "google_storage_bucket",
+            "values": {"name": "test-bucket"},
+        }
+
+        schema_classes = {}
+
+        # By default, we should generate blocks for all providers
+        block = main.generate_import_block(resource, schema_classes)
+        self.assertIn("to = google_storage_bucket.test", block)
+        self.assertIn("TODO", block)
+        self.assertIn("google", block)
+        self.assertIn("storage_bucket", block)
+
+        # When supported_providers_only=True, we should not generate a block
+        block = main.generate_import_block(
+            resource, schema_classes, supported_providers_only=True
+        )
+        self.assertIsNone(block)
+
     def test_generate_removed_block(self):
         """Test generating a removed block for a resource"""
         # Test with destroy=False (default)
@@ -303,6 +315,36 @@ def test_generate_blocks_for_command(self):
         # Should have 2 blocks, not 3, due to deduplication
         self.assertEqual(len(removed_blocks), 2)
 
+    def test_generate_blocks_with_mixed_providers(self):
+        """Test that supported_providers_only flag filters non-AWS resources"""
+        resources = [
+            {
+                "address": "aws_s3_bucket.test",
+                "type": "aws_s3_bucket",
+                "values": {"bucket": "test-bucket"},
+            },
+            {
+                "address": "google_storage_bucket.test",
+                "type": "google_storage_bucket",
+                "values": {"name": "test-bucket"},
+            },
+            {
+                "address": "azurerm_storage_account.test",
+                "type": "azurerm_storage_account",
+                "values": {"name": "teststorage"},
+            },
+        ]
+
+        # Test with all providers (default behavior)
+        import_blocks = main.generate_blocks_for_command(resources, "import")
+        self.assertEqual(len(import_blocks), 3)  # All resources should be included
+
+        # Test with supported_providers_only=True
+        import_blocks = main.generate_blocks_for_command(
+            resources, "import", supported_providers_only=True
+        )
+        self.assertEqual(len(import_blocks), 1)  # Only AWS resource should be included
+
 
 if __name__ == "__main__":
     unittest.main()