Merge pull request #107 from hashicorp/bugfix/config-version-upload

fix(config-version): Updated upload method and ConfigurationSource Enum in the configuration version

Author: isivaselvan

examples/configuration_version.py

@@ -277,29 +277,8 @@ def main():
                     print(f"       - {filename} ({size} bytes)")
 
                 try:
-                    # Create tar.gz archive manually since go-slug isn't available
-                    print("Creating tar.gz archive manually...")
-
-                    import tarfile
-
-                    # Create tar.gz archive in memory
-                    archive_buffer = io.BytesIO()
-                    with tarfile.open(fileobj=archive_buffer, mode="w:gz") as tar:
-                        # Add all files from the temp directory
-                        for filename in files:
-                            filepath = os.path.join(temp_dir, filename)
-                            tar.add(filepath, arcname=filename)
-
-                    archive_buffer.seek(0)
-                    archive_bytes = archive_buffer.getvalue()
-                    print(f"Created archive: {len(archive_bytes)} bytes")
-
-                    # Use the SDK's upload_tar_gzip method instead of direct HTTP calls
-                    print("Uploading archive using SDK method...")
-                    archive_buffer.seek(0)  # Reset buffer position
-                    client.configuration_versions.upload_tar_gzip(
-                        new_cv.upload_url, archive_buffer
-                    )
+                    print("Uploading Terraform configuration...")
+                    client.configuration_versions.upload(new_cv.upload_url, temp_dir)
                     print("Terraform configuration uploaded successfully!")
 
                     # Wait and check status
@@ -408,8 +387,6 @@ def main():
     # =====================================================
     # TEST 4: UPLOAD CONFIGURATION VERSION
     # =====================================================
-    # Test 4: Upload function (requires go-slug)
-    # =====================================================
     print("\n4. Testing upload() function:")
     try:
         # Create a fresh configuration version specifically for upload testing
@@ -441,33 +418,19 @@ def main():
                 print(f"\n Uploading configuration to CV: {fresh_cv.id}")
                 print(f"Upload URL: {upload_url[:60]}...")
 
-                try:
-                    client.configuration_versions.upload(upload_url, temp_dir)
-                    print("Configuration uploaded successfully!")
-
-                    # Check status after upload
-                    print("\n Checking status after upload:")
-                    time.sleep(3)  # Give TFE time to process
-                    updated_cv = client.configuration_versions.read(fresh_cv.id)
-                    print(f"Status after upload: {updated_cv.status}")
+                client.configuration_versions.upload(upload_url, temp_dir)
+                print("Configuration uploaded successfully!")
 
-                    if updated_cv.status.value != "pending":
-                        print("Status changed (upload processed)")
-                    else:
-                        print("Status still pending (may need more time)")
+                # Check status after upload
+                print("\n Checking status after upload:")
+                time.sleep(3)  # Give TFE time to process
+                updated_cv = client.configuration_versions.read(fresh_cv.id)
+                print(f"Status after upload: {updated_cv.status}")
 
-                except ImportError as e:
-                    if "go-slug" in str(e):
-                        print("go-slug package not available")
-                        print("Install with: pip install go-slug")
-                        print(
-                            "Upload function exists but requires go-slug for packaging"
-                        )
-                        print(
-                            "Function correctly raises ImportError when go-slug unavailable"
-                        )
-                    else:
-                        raise
+                if updated_cv.status.value != "pending":
+                    print("Status changed (upload processed)")
+                else:
+                    print("Status still pending (may need more time)")
 
     except Exception as e:
         print(f"Error: {e}")
@@ -868,7 +831,7 @@ def main():
         "TEST 2:  create() - Create new configuration versions with different options"
     )
     print("TEST 3:  read() - Read configuration version details and validate fields")
-    print("TEST 4:  upload() - Upload Terraform configurations (requires go-slug)")
+    print("TEST 4:  upload() - Upload Terraform configurations (stdlib tarfile)")
     print("TEST 5:  download() - Download configuration version archives")
     print("TEST 6:  archive() - Archive configuration versions")
     print("TEST 7:  read_with_options() - Read with include options")

src/pytfe/models/configuration_version.py

@@ -25,6 +25,7 @@ class ConfigurationSource(str, Enum):
     GITLAB = "gitlab"
     ADO = "ado"
     TERRAFORM = "terraform"
+    TERRAFORM_CLOUD = "terraform+cloud"
 
 
 class ConfigVerIncludeOpt(str, Enum):

src/pytfe/utils.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import io
+import os
 import re
+import tarfile
 import time
 from collections.abc import Callable, Mapping
 from typing import TYPE_CHECKING, Any
@@ -15,11 +17,6 @@
 
 from urllib.parse import urlparse
 
-try:
-    import slug  # type: ignore[import-not-found]
-except ImportError:
-    slug = None
-
 from .errors import (
     InvalidNameError,
     RequiredAgentModeError,
@@ -366,26 +363,25 @@ def pack_contents(path: str) -> io.BytesIO:
         BytesIO buffer containing the tar.gz archive
 
     Raises:
-        ImportError: If go-slug is not available
         ValueError: If path is invalid
     """
-    if slug is None:
-        raise ImportError(
-            "go-slug package is required for packing configuration files. "
-            "Install it with: pip install go-slug"
+    if not path or not os.path.isdir(path):
+        raise ValueError(
+            f"Failed to pack directory {path}: path must be an existing directory"
         )
 
     body = io.BytesIO()
 
-    # Use go-slug to pack the configuration directory
-    # This handles .terraformignore and other Terraform-specific behaviors
-    packer = slug.Packer()
-    _, err = packer.pack(path, body)
-
-    if err:
-        raise ValueError(f"Failed to pack directory {path}: {err}")
+    with tarfile.open(fileobj=body, mode="w:gz") as tar:
+        for root, _, files in os.walk(path):
+            rel_root = os.path.relpath(root, path)
+            for filename in files:
+                full_path = os.path.join(root, filename)
+                arcname = (
+                    filename if rel_root == "." else os.path.join(rel_root, filename)
+                )
+                tar.add(full_path, arcname=arcname, recursive=False)
 
-    # Reset buffer position to beginning for reading
     body.seek(0)
     return body
 

tests/units/test_configuration_version.py

@@ -17,7 +17,7 @@
 """
 
 import io
-from unittest.mock import Mock, patch
+from unittest.mock import Mock
 
 import pytest
 
@@ -343,35 +343,33 @@ def test_read_with_options_no_ingress(
 class TestConfigurationVersionsUpload:
     """Test configuration versions upload functionality."""
 
-    def test_upload_missing_slug(self, configuration_versions_service):
-        """Test upload when go-slug is not available."""
+    def test_upload_packs_with_tar(self, configuration_versions_service, tmp_path):
+        """Test upload works by packing a directory to tar.gz with stdlib."""
         upload_url = "https://example.com/upload"
-        directory_path = "/tmp/test"
+        directory_path = tmp_path
+        (directory_path / "main.tf").write_text('resource "null_resource" "test" {}')
 
-        with patch("src.pytfe.utils.slug", None):
-            with pytest.raises(ImportError, match="go-slug package is required"):
-                configuration_versions_service.upload(upload_url, directory_path)
+        mock_response = Mock()
+        mock_response.status_code = 200
+        configuration_versions_service.t._sync.put.return_value = mock_response
 
-    @patch("src.pytfe.utils.slug")
-    def test_upload_success(self, mock_slug, configuration_versions_service):
-        """Test successful upload."""
-        # Mock slug.pack
-        mock_packer = Mock()
-        mock_packer.pack.return_value = (None, None)  # (size, error)
-        mock_slug.Packer.return_value = mock_packer
+        configuration_versions_service.upload(upload_url, str(directory_path))
 
+        configuration_versions_service.t._sync.put.assert_called_once()
+
+    def test_upload_success(self, configuration_versions_service, tmp_path):
+        """Test successful upload."""
         upload_url = "https://example.com/upload"
-        directory_path = "/tmp/test"
+        directory_path = tmp_path
+        (directory_path / "main.tf").write_text('resource "null_resource" "test" {}')
 
         # Mock transport's underlying httpx client instead of direct httpx
         mock_response = Mock()
         mock_response.status_code = 200
         configuration_versions_service.t._sync.put.return_value = mock_response
 
-        configuration_versions_service.upload(upload_url, directory_path)
-
-        # Verify slug.pack was called
-        mock_packer.pack.assert_called_once()
+        configuration_versions_service.upload(upload_url, str(directory_path))
+        configuration_versions_service.t._sync.put.assert_called_once()
 
 
 class TestConfigurationVersionsUploadTarGzip: