diff --git a/examples/configuration_version.py b/examples/configuration_version.py index 87fa6d1c..3d22d677 100644 --- a/examples/configuration_version.py +++ b/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") diff --git a/src/pytfe/models/configuration_version.py b/src/pytfe/models/configuration_version.py index 6bc613a8..d32223f4 100644 --- a/src/pytfe/models/configuration_version.py +++ b/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): diff --git a/src/pytfe/utils.py b/src/pytfe/utils.py index 02c43cc1..646b0994 100644 --- a/src/pytfe/utils.py +++ b/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 diff --git a/tests/units/test_configuration_version.py b/tests/units/test_configuration_version.py index f4f0fa20..58ca328f 100644 --- a/tests/units/test_configuration_version.py +++ b/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: