diff --git a/examples/project_integration_test_example.py b/examples/project_integration_test_example.py new file mode 100644 index 00000000..c4876f88 --- /dev/null +++ b/examples/project_integration_test_example.py @@ -0,0 +1,225 @@ +""" +Integration Test Example for python-tfe + +This file demonstrates how to create integration tests that work with real HCP Terraform API. +These tests create and delete actual resources, so use with caution! + +Setup Instructions: +1. Create a test organization in HCP Terraform (https://app.terraform.io) +2. Generate an organization or user API token with appropriate permissions +3. Set environment variables: + export TFE_TOKEN="your-api-token-here" + export TFE_ORG="your-test-organization-name" +4. Copy this file to your tests directory: + cp examples/integration_test_example.py tests/test_integration_local.py +5. Run the tests: + pytest tests/test_integration_local.py -v -s + +Important Notes: +- These tests make real API calls and create/delete actual resources +- Always use a dedicated test organization, never production +- Tests will fail if you don't have proper permissions +- Clean up is automatic, but verify resources are deleted after testing +""" + +import os +import uuid + +import pytest + +from tfe._http import HTTPTransport +from tfe.config import TFEConfig +from tfe.resources.projects import Projects + + +@pytest.fixture +def integration_client(): + """Create a real Projects client for integration testing""" + token = os.environ.get("TFE_TOKEN") + org = os.environ.get("TFE_ORG") + + if not token: + pytest.skip( + "TFE_TOKEN environment variable is required. " + "Get your token from HCP Terraform: Settings โ†’ API Tokens" + ) + + if not org: + pytest.skip( + "TFE_ORG environment variable is required. " + "Use your organization name from HCP Terraform URL" + ) + + print(f"\n๐Ÿ”ง Testing against organization: {org}") + print(f"๐Ÿ”ง Using token: {token[:10]}...") + + config = TFEConfig() + + try: + transport = HTTPTransport( + config.address, + token, + timeout=config.timeout, + verify_tls=config.verify_tls, + user_agent_suffix=None, + max_retries=3, + backoff_base=0.1, + backoff_cap=1.0, + backoff_jitter=True, + http2=False, + proxies=None, + ca_bundle=None, + ) + except Exception as e: + pytest.fail(f"Failed to create HTTP transport: {e}") + + return Projects(transport), org + + +def test_list_projects_integration(integration_client): + """Test listing projects in your HCP Terraform organization + + This is the safest test to run first - it only reads data. + """ + projects, org = integration_client + + try: + project_list = list(projects.list(org)) + print(f"โœ… Found {len(project_list)} projects in organization '{org}'") + + assert isinstance(project_list, list) + + if project_list: + project = project_list[0] + assert hasattr(project, "id"), "Project should have an ID" + assert hasattr(project, "name"), "Project should have a name" + assert hasattr(project, "organization"), ( + "Project should have an organization" + ) + print(f"๐Ÿ“‹ Example project: {project.name} (ID: {project.id})") + else: + print("๐Ÿ“‹ No projects found - this is normal for a new organization") + + except Exception as e: + pytest.fail( + f"Failed to list projects. Check your TFE_TOKEN and TFE_ORG. Error: {e}" + ) + + +def test_project_crud_integration(integration_client): + """Test complete Create, Read, Update, Delete operations + + โš ๏ธ WARNING: This test creates and deletes real resources! + Only run this in a test organization, never in production. + """ + projects, org = integration_client + + # Generate unique names to avoid conflicts + unique_id = str(uuid.uuid4())[:8] + test_name = f"integration-test-{unique_id}" + updated_name = f"integration-test-{unique_id}-updated" + project_id = None + + try: + # CREATE - Test project creation + print(f"๐Ÿ”จ Creating project: {test_name}") + created_project = projects.create(org, test_name) + + assert created_project.name == test_name, ( + f"Expected name {test_name}, got {created_project.name}" + ) + assert created_project.organization == org, ( + f"Expected org {org}, got {created_project.organization}" + ) + assert created_project.id.startswith("prj-"), ( + f"Project ID should start with 'prj-', got {created_project.id}" + ) + + project_id = created_project.id + print(f"โœ… Created project: {project_id}") + + # READ - Test reading the created project + print(f"๐Ÿ“– Reading project: {project_id}") + read_project = projects.read(project_id) + + assert read_project.id == project_id, ( + f"Expected ID {project_id}, got {read_project.id}" + ) + assert read_project.name == test_name, ( + f"Expected name {test_name}, got {read_project.name}" + ) + print(f"โœ… Successfully read project: {read_project.name}") + + # UPDATE - Test updating the project name + print(f"โœ๏ธ Updating project name to: {updated_name}") + updated_project = projects.update(project_id, updated_name) + + assert updated_project.id == project_id, ( + f"Project ID should remain {project_id}" + ) + assert updated_project.name == updated_name, ( + f"Expected updated name {updated_name}, got {updated_project.name}" + ) + print(f"โœ… Successfully updated project: {updated_project.name}") + + except Exception as e: + pytest.fail(f"CRUD operation failed: {e}") + + finally: + # DELETE - Always clean up, even if tests fail + if project_id: + try: + print(f"๐Ÿ—‘๏ธ Deleting test project: {project_id}") + projects.delete(project_id) + print("โœ… Test project deleted successfully") + except Exception as e: + print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") + print( + " You may need to manually delete this project in HCP Terraform" + ) + + +def test_error_handling_integration(integration_client): + """Test that the client handles API errors appropriately""" + projects, org = integration_client + + # Test reading a non-existent project + fake_project_id = "prj-nonexistent123456789" + + try: + projects.read(fake_project_id) + pytest.fail("Should have raised an exception for non-existent project") + except Exception as e: + print( + f"โœ… Correctly handled error for non-existent project: {type(e).__name__}" + ) + # This should raise a NotFound or similar error + assert "not found" in str(e).lower() or "404" in str(e) + + +if __name__ == "__main__": + """ + You can also run this file directly for quick testing: + + export TFE_TOKEN="your-token" + export TFE_ORG="your-org" + python examples/integration_test_example.py + """ + import sys + + token = os.environ.get("TFE_TOKEN") + org = os.environ.get("TFE_ORG") + + if not token or not org: + print("โŒ Please set TFE_TOKEN and TFE_ORG environment variables") + print(" export TFE_TOKEN='your-hcp-terraform-token'") + print(" export TFE_ORG='your-organization-name'") + sys.exit(1) + + print("๐Ÿงช Running integration tests directly...") + print( + " For full pytest features, use: pytest examples/integration_test_example.py -v -s" + ) + + # Simple direct execution + pytest.main([__file__, "-v", "-s"]) diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index 6647edc0..614a8ef3 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -19,3 +19,65 @@ def list(self, organization: str) -> Iterator[Project]: proj_id = _safe_str(item.get("id")) name = _safe_str(attr.get("name")) yield Project(id=proj_id, name=name, organization=organization) + + def create(self, organization: str, name: str) -> Project: + """Create a new project in an organization""" + path = f"/api/v2/organizations/{organization}/projects" + payload = {"data": {"type": "projects", "attributes": {"name": name}}} + + # Use json_body parameter (correct parameter name) + response = self.t.request("POST", path, json_body=payload) + data = response.json()["data"] + attr = data.get("attributes", {}) or {} + + return Project( + id=_safe_str(data.get("id")), + name=_safe_str(attr.get("name")), + organization=organization, + ) + + def read(self, project_id: str) -> Project: + """Get a specific project by ID""" + path = f"/api/v2/projects/{project_id}" + response = self.t.request("GET", path) + data = response.json()["data"] + attr = data.get("attributes", {}) or {} + + # Get organization from relationships if available + relationships = data.get("relationships", {}) + org_data = relationships.get("organization", {}).get("data", {}) + organization = _safe_str(org_data.get("id")) + + return Project( + id=_safe_str(data.get("id")), + name=_safe_str(attr.get("name")), + organization=organization, + ) + + def update(self, project_id: str, name: str) -> Project: + """Update a project's name""" + path = f"/api/v2/projects/{project_id}" + payload = { + "data": {"type": "projects", "id": project_id, "attributes": {"name": name}} + } + + # Use json_body parameter (correct parameter name) + response = self.t.request("PATCH", path, json_body=payload) + data = response.json()["data"] + attr = data.get("attributes", {}) or {} + + # Get organization from relationships if available + relationships = data.get("relationships", {}) + org_data = relationships.get("organization", {}).get("data", {}) + organization = _safe_str(org_data.get("id")) + + return Project( + id=_safe_str(data.get("id")), + name=_safe_str(attr.get("name")), + organization=organization, + ) + + def delete(self, project_id: str) -> None: + """Delete a project""" + path = f"/api/v2/projects/{project_id}" + self.t.request("DELETE", path) diff --git a/tests/units/test_project.py b/tests/units/test_project.py new file mode 100644 index 00000000..da5f222b --- /dev/null +++ b/tests/units/test_project.py @@ -0,0 +1,222 @@ +from unittest.mock import Mock + +from tfe.resources.projects import Projects, _safe_str +from tfe.types import Project + + +class TestProjects: + def setup_method(self): + """Setup method that runs before each test""" + self.mock_transport = Mock() + self.projects_service = Projects(self.mock_transport) + + def test_projects_service_init(self): + """Test that Projects service initializes correctly""" + mock_transport = Mock() + service = Projects(mock_transport) + assert service.t == mock_transport + + def test_list_projects_success(self): + """Test successful listing of projects""" + organization = "test-org" + + # Mock API response data + mock_api_response = [ + { + "id": "prj-123", + "type": "projects", + "attributes": {"name": "Test Project 1"}, + }, + { + "id": "prj-456", + "type": "projects", + "attributes": {"name": "Test Project 2"}, + }, + ] + + # Mock the _list method to return our test data + self.projects_service._list = Mock(return_value=mock_api_response) + + # Call the method under test + result = list(self.projects_service.list(organization)) + + # Assertions + assert len(result) == 2 + assert isinstance(result[0], Project) + assert isinstance(result[1], Project) + + # Check first project + assert result[0].id == "prj-123" + assert result[0].name == "Test Project 1" + assert result[0].organization == organization + + # Check second project + assert result[1].id == "prj-456" + assert result[1].name == "Test Project 2" + assert result[1].organization == organization + + # Verify the correct API path was used + expected_path = f"/api/v2/organizations/{organization}/projects" + self.projects_service._list.assert_called_once_with(expected_path) + + def test_create_project_success(self): + """Test successful project creation""" + organization = "test-org" + project_name = "New Project" + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": "prj-123", + "type": "projects", + "attributes": {"name": project_name}, + } + } + self.mock_transport.request.return_value = mock_response + + result = self.projects_service.create(organization, project_name) + + # Assertions + assert isinstance(result, Project) + assert result.id == "prj-123" + assert result.name == project_name + assert result.organization == organization + + # Verify API call + expected_path = f"/api/v2/organizations/{organization}/projects" + expected_payload = { + "data": {"type": "projects", "attributes": {"name": project_name}} + } + self.mock_transport.request.assert_called_once_with( + "POST", expected_path, json_body=expected_payload + ) + + def test_read_project_success(self): + """Test successful project read""" + project_id = "prj-123" + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": project_id, + "type": "projects", + "attributes": {"name": "Test Project"}, + "relationships": {"organization": {"data": {"id": "test-org"}}}, + } + } + self.mock_transport.request.return_value = mock_response + + result = self.projects_service.read(project_id) + + # Assertions + assert isinstance(result, Project) + assert result.id == project_id + assert result.name == "Test Project" + assert result.organization == "test-org" + + # Verify API call + expected_path = f"/api/v2/projects/{project_id}" + self.mock_transport.request.assert_called_once_with("GET", expected_path) + + def test_update_project_success(self): + """Test successful project update""" + project_id = "prj-123" + new_name = "Updated Project" + + # Mock API response + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": project_id, + "type": "projects", + "attributes": {"name": new_name}, + "relationships": {"organization": {"data": {"id": "test-org"}}}, + } + } + self.mock_transport.request.return_value = mock_response + + result = self.projects_service.update(project_id, new_name) + + # Assertions + assert isinstance(result, Project) + assert result.id == project_id + assert result.name == new_name + assert result.organization == "test-org" + + # Verify API call + expected_path = f"/api/v2/projects/{project_id}" + expected_payload = { + "data": { + "type": "projects", + "id": project_id, + "attributes": {"name": new_name}, + } + } + self.mock_transport.request.assert_called_once_with( + "PATCH", expected_path, json_body=expected_payload + ) + + def test_delete_project_success(self): + """Test successful project deletion""" + project_id = "prj-123" + + result = self.projects_service.delete(project_id) + + # Delete should return None + assert result is None + + # Verify API call + expected_path = f"/api/v2/projects/{project_id}" + self.mock_transport.request.assert_called_once_with("DELETE", expected_path) + + def test_safe_str_function(self): + """Test _safe_str utility function""" + # Test with string + assert _safe_str("test") == "test" + + # Test with None + assert _safe_str(None) == "" + + # Test with integer + assert _safe_str(123) == "123" + + # Test with custom default + assert _safe_str(None, "default") == "default" + + # Test with boolean + assert _safe_str(True) == "True" + assert _safe_str(False) == "False" + + def test_list_projects_empty_response(self): + """Test listing projects when API returns empty response""" + organization = "empty-org" + + # Mock empty API response + self.projects_service._list = Mock(return_value=[]) + + result = list(self.projects_service.list(organization)) + + assert len(result) == 0 + assert isinstance(result, list) + + def test_read_project_missing_organization(self): + """Test reading project when organization info is missing""" + project_id = "prj-123" + + # Mock API response without organization relationship + mock_response = Mock() + mock_response.json.return_value = { + "data": { + "id": project_id, + "type": "projects", + "attributes": {"name": "Test Project"}, + # No relationships field + } + } + self.mock_transport.request.return_value = mock_response + + result = self.projects_service.read(project_id) + + assert result.organization == "" # Should default to empty string