From 323e34f6193361d26c9f96c3ecd4fc72862e3a20 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 12:53:36 +0530 Subject: [PATCH 01/10] Hi this is first version of project.py please take a look --- src/tfe/resources/projects.py | 74 ++++++++++ tests/units/test_project.py | 249 ++++++++++++++++++++++++++++++++++ 2 files changed, 323 insertions(+) create mode 100644 tests/units/test_project.py diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index 6647edc0..0237cd30 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -19,3 +19,77 @@ 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 + } + } + } + + response = self.t.request("POST", path, json=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 + } + } + } + + response = self.t.request("PATCH", path, json=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) \ No newline at end of file diff --git a/tests/units/test_project.py b/tests/units/test_project.py new file mode 100644 index 00000000..e32bef13 --- /dev/null +++ b/tests/units/test_project.py @@ -0,0 +1,249 @@ +from unittest.mock import Mock +import pytest +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=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=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 \ No newline at end of file From 50ba6554ee081cb20d38dfe6545746a28fdc95fa Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 17:20:40 +0530 Subject: [PATCH 02/10] Project.py CRUD operations with unit tests and examples --- examples/project_integration_test_example.py | 97 ++++++++++++++++++++ src/tfe/resources/projects.py | 7 +- 2 files changed, 101 insertions(+), 3 deletions(-) create mode 100644 examples/project_integration_test_example.py diff --git a/examples/project_integration_test_example.py b/examples/project_integration_test_example.py new file mode 100644 index 00000000..c4892934 --- /dev/null +++ b/examples/project_integration_test_example.py @@ -0,0 +1,97 @@ +""" +Example integration test template for local development. + +This file shows how to create integration tests for your local development. +Copy this file to your tests/ directory and modify for your needs. + +Requirements: +1. Set environment variables: TFE_TOKEN and TFE_ORG +2. Ensure your token has appropriate permissions +3. Be careful - this makes real API calls to HCP Terraform! + +Usage: + cp examples/integration_test_example.py tests/test_local_integration.py + export TFE_TOKEN="your-token" + export TFE_ORG="your-org" + pytest tests/test_local_integration.py -v -s +""" + +import os +import pytest +from tfe._http import HTTPTransport +from tfe.resources.projects import Projects +from tfe.config import TFEConfig + + +@pytest.fixture +def real_projects_client(): + """Create a real Projects client for local integration testing""" + token = os.environ.get("TFE_TOKEN") + org = os.environ.get("TFE_ORG") + + if not token or not org: + pytest.skip("TFE_TOKEN and TFE_ORG environment variables required") + + config = TFEConfig() + 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, + ) + + return Projects(transport), org + + +def test_list_projects_integration(real_projects_client): + """Example integration test for listing projects""" + projects, org = real_projects_client + + project_list = list(projects.list(org)) + print(f"Found {len(project_list)} projects in organization '{org}'") + + assert isinstance(project_list, list) + # Add your assertions here + + +def test_project_crud_integration(real_projects_client): + """Example integration test for full CRUD operations + + WARNING: This creates and deletes real resources in HCP Terraform! + """ + projects, org = real_projects_client + + test_name = "example-test-project" + project_id = None + + try: + # CREATE + created_project = projects.create(org, test_name) + project_id = created_project.id + assert created_project.name == test_name + + # READ + read_project = projects.read(project_id) + assert read_project.id == project_id + + # UPDATE + updated_name = f"{test_name}-updated" + updated_project = projects.update(project_id, updated_name) + assert updated_project.name == updated_name + + finally: + # CLEANUP - Always delete test resources + if project_id: + try: + projects.delete(project_id) + print(f"โœ… Cleaned up project: {project_id}") + except Exception as e: + print(f"โŒ Cleanup failed: {e}") \ No newline at end of file diff --git a/src/tfe/resources/projects.py b/src/tfe/resources/projects.py index 0237cd30..aefe4e05 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -19,7 +19,6 @@ 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""" @@ -33,7 +32,8 @@ def create(self, organization: str, name: str) -> Project: } } - response = self.t.request("POST", path, json=payload) + # 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 {} @@ -74,7 +74,8 @@ def update(self, project_id: str, name: str) -> Project: } } - response = self.t.request("PATCH", path, json=payload) + # 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 {} From 4a8c584a8890980f2f7c549fb06381ca639dece1 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 17:23:08 +0530 Subject: [PATCH 03/10] Project.py CRUD operations with unit tests and examples --- examples/integration_test_example.py | 201 +++++++++++++++++++++++++++ 1 file changed, 201 insertions(+) create mode 100644 examples/integration_test_example.py diff --git a/examples/integration_test_example.py b/examples/integration_test_example.py new file mode 100644 index 00000000..071c049a --- /dev/null +++ b/examples/integration_test_example.py @@ -0,0 +1,201 @@ +""" +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.resources.projects import Projects +from tfe.config import TFEConfig + + +@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"]) \ No newline at end of file From 6f24efe1dc34cec2a99bfd309f1d25ae969e7401 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 17:27:42 +0530 Subject: [PATCH 04/10] Project CRUD operations with unit tests and examples --- examples/integration_test_example.py | 201 --------------------------- 1 file changed, 201 deletions(-) delete mode 100644 examples/integration_test_example.py diff --git a/examples/integration_test_example.py b/examples/integration_test_example.py deleted file mode 100644 index 071c049a..00000000 --- a/examples/integration_test_example.py +++ /dev/null @@ -1,201 +0,0 @@ -""" -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.resources.projects import Projects -from tfe.config import TFEConfig - - -@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"]) \ No newline at end of file From df24ccd054df22076d382133ae036952e9602149 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 18:29:14 +0530 Subject: [PATCH 05/10] CRUD operations for project.py along with unit tesst and example for integration testing --- examples/project_integration_test_example.py | 234 ++++++++++++++----- src/tfe/resources/projects.py | 16 +- tests/units/test_project.py | 70 +++--- 3 files changed, 213 insertions(+), 107 deletions(-) diff --git a/examples/project_integration_test_example.py b/examples/project_integration_test_example.py index c4892934..61aac6d9 100644 --- a/examples/project_integration_test_example.py +++ b/examples/project_integration_test_example.py @@ -1,97 +1,203 @@ """ -Example integration test template for local development. +Integration Test Example for python-tfe -This file shows how to create integration tests for your local development. -Copy this file to your tests/ directory and modify for your needs. +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! -Requirements: -1. Set environment variables: TFE_TOKEN and TFE_ORG -2. Ensure your token has appropriate permissions -3. Be careful - this makes real API calls to HCP Terraform! +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 -Usage: - cp examples/integration_test_example.py tests/test_local_integration.py - export TFE_TOKEN="your-token" - export TFE_ORG="your-org" - pytest tests/test_local_integration.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.resources.projects import Projects from tfe.config import TFEConfig +from tfe.resources.projects import Projects @pytest.fixture -def real_projects_client(): - """Create a real Projects client for local integration testing""" +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 or not org: - pytest.skip("TFE_TOKEN and TFE_ORG environment variables required") - + + 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() - 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, - ) - + + 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(real_projects_client): - """Example integration test for listing projects""" - projects, org = real_projects_client - - project_list = list(projects.list(org)) - print(f"Found {len(project_list)} projects in organization '{org}'") - - assert isinstance(project_list, list) - # Add your assertions here +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(real_projects_client): - """Example integration test for full CRUD operations - - WARNING: This creates and deletes real resources in HCP Terraform! + +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 = real_projects_client - - test_name = "example-test-project" + 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 + # 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 - assert created_project.name == test_name - - # READ + 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 - - # UPDATE - updated_name = f"{test_name}-updated" + + 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.name == 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: - # CLEANUP - Always delete test resources + # DELETE - Always clean up, even if tests fail if project_id: try: + print(f"๐Ÿ—‘๏ธ Deleting test project: {project_id}") projects.delete(project_id) - print(f"โœ… Cleaned up project: {project_id}") + print("โœ… Test project deleted successfully") except Exception as e: - print(f"โŒ Cleanup failed: {e}") \ No newline at end of file + 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 aefe4e05..bd4c2906 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -31,12 +31,12 @@ def create(self, organization: str, name: str) -> Project: } } } - + # 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")), @@ -49,12 +49,12 @@ def read(self, project_id: str) -> Project: 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")), @@ -73,17 +73,17 @@ def update(self, project_id: str, name: str) -> Project: } } } - + # 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")), @@ -93,4 +93,4 @@ def update(self, project_id: str, name: str) -> Project: def delete(self, project_id: str) -> None: """Delete a project""" path = f"/api/v2/projects/{project_id}" - self.t.request("DELETE", path) \ No newline at end of file + self.t.request("DELETE", path) diff --git a/tests/units/test_project.py b/tests/units/test_project.py index e32bef13..92de407c 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -1,5 +1,5 @@ from unittest.mock import Mock -import pytest + from tfe.resources.projects import Projects, _safe_str from tfe.types import Project @@ -19,7 +19,7 @@ def test_projects_service_init(self): def test_list_projects_success(self): """Test successful listing of projects""" organization = "test-org" - + # Mock API response data mock_api_response = [ { @@ -30,35 +30,35 @@ def test_list_projects_success(self): } }, { - "id": "prj-456", + "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) @@ -67,7 +67,7 @@ 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 = { @@ -80,15 +80,15 @@ def test_create_project_success(self): } } 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 = { @@ -104,7 +104,7 @@ def test_create_project_success(self): 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 = { @@ -124,15 +124,15 @@ def test_read_project_success(self): } } 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) @@ -141,7 +141,7 @@ 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 = { @@ -161,15 +161,15 @@ def test_update_project_success(self): } } 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 = { @@ -186,12 +186,12 @@ def test_update_project_success(self): 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) @@ -200,16 +200,16 @@ 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" @@ -217,19 +217,19 @@ def test_safe_str_function(self): 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 = { @@ -243,7 +243,7 @@ def test_read_project_missing_organization(self): } } self.mock_transport.request.return_value = mock_response - + result = self.projects_service.read(project_id) - - assert result.organization == "" # Should default to empty string \ No newline at end of file + + assert result.organization == "" # Should default to empty string From 6df4ddc972d82b877e617e7f250f7dad279213cf Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 18:51:29 +0530 Subject: [PATCH 06/10] CRUD operations for project for python TFE --- tests/units/test_project.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/units/test_project.py b/tests/units/test_project.py index 92de407c..10eb79ab 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -99,7 +99,7 @@ def test_create_project_success(self): } } } - self.mock_transport.request.assert_called_once_with("POST", expected_path, json=expected_payload) + 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""" @@ -181,7 +181,7 @@ def test_update_project_success(self): } } } - self.mock_transport.request.assert_called_once_with("PATCH", expected_path, json=expected_payload) + 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""" From 4465f0cd1006bf232be4ec3ad1cadd72350ff3d2 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Thu, 11 Sep 2025 18:56:03 +0530 Subject: [PATCH 07/10] CRUD operation for Project of Python TFE --- examples/project_integration_test_example.py | 48 ++++++++++----- src/tfe/resources/projects.py | 23 ++------ tests/units/test_project.py | 61 ++++++-------------- 3 files changed, 57 insertions(+), 75 deletions(-) diff --git a/examples/project_integration_test_example.py b/examples/project_integration_test_example.py index 61aac6d9..c4876f88 100644 --- a/examples/project_integration_test_example.py +++ b/examples/project_integration_test_example.py @@ -91,9 +91,11 @@ def test_list_projects_integration(integration_client): 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" + 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") @@ -123,9 +125,15 @@ def test_project_crud_integration(integration_client): 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}" + 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}") @@ -134,16 +142,24 @@ def test_project_crud_integration(integration_client): 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}" + 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}" + 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: @@ -158,7 +174,9 @@ def test_project_crud_integration(integration_client): 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") + print( + " You may need to manually delete this project in HCP Terraform" + ) def test_error_handling_integration(integration_client): @@ -172,7 +190,9 @@ def test_error_handling_integration(integration_client): 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__}") + 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) @@ -197,7 +217,9 @@ def test_error_handling_integration(integration_client): sys.exit(1) print("๐Ÿงช Running integration tests directly...") - print(" For full pytest features, use: pytest examples/integration_test_example.py -v -s") + 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 bd4c2906..614a8ef3 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -23,14 +23,7 @@ def list(self, organization: str) -> Iterator[Project]: 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 - } - } - } + payload = {"data": {"type": "projects", "attributes": {"name": name}}} # Use json_body parameter (correct parameter name) response = self.t.request("POST", path, json_body=payload) @@ -40,7 +33,7 @@ def create(self, organization: str, name: str) -> Project: return Project( id=_safe_str(data.get("id")), name=_safe_str(attr.get("name")), - organization=organization + organization=organization, ) def read(self, project_id: str) -> Project: @@ -58,20 +51,14 @@ def read(self, project_id: str) -> Project: return Project( id=_safe_str(data.get("id")), name=_safe_str(attr.get("name")), - organization=organization + 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 - } - } + "data": {"type": "projects", "id": project_id, "attributes": {"name": name}} } # Use json_body parameter (correct parameter name) @@ -87,7 +74,7 @@ def update(self, project_id: str, name: str) -> Project: return Project( id=_safe_str(data.get("id")), name=_safe_str(attr.get("name")), - organization=organization + organization=organization, ) def delete(self, project_id: str) -> None: diff --git a/tests/units/test_project.py b/tests/units/test_project.py index 10eb79ab..da5f222b 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -25,17 +25,13 @@ def test_list_projects_success(self): { "id": "prj-123", "type": "projects", - "attributes": { - "name": "Test Project 1" - } + "attributes": {"name": "Test Project 1"}, }, { "id": "prj-456", "type": "projects", - "attributes": { - "name": "Test Project 2" - } - } + "attributes": {"name": "Test Project 2"}, + }, ] # Mock the _list method to return our test data @@ -74,9 +70,7 @@ def test_create_project_success(self): "data": { "id": "prj-123", "type": "projects", - "attributes": { - "name": project_name - } + "attributes": {"name": project_name}, } } self.mock_transport.request.return_value = mock_response @@ -92,14 +86,11 @@ def test_create_project_success(self): # Verify API call expected_path = f"/api/v2/organizations/{organization}/projects" expected_payload = { - "data": { - "type": "projects", - "attributes": { - "name": project_name - } - } + "data": {"type": "projects", "attributes": {"name": project_name}} } - self.mock_transport.request.assert_called_once_with("POST", expected_path, json_body=expected_payload) + 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""" @@ -111,16 +102,8 @@ def test_read_project_success(self): "data": { "id": project_id, "type": "projects", - "attributes": { - "name": "Test Project" - }, - "relationships": { - "organization": { - "data": { - "id": "test-org" - } - } - } + "attributes": {"name": "Test Project"}, + "relationships": {"organization": {"data": {"id": "test-org"}}}, } } self.mock_transport.request.return_value = mock_response @@ -148,16 +131,8 @@ def test_update_project_success(self): "data": { "id": project_id, "type": "projects", - "attributes": { - "name": new_name - }, - "relationships": { - "organization": { - "data": { - "id": "test-org" - } - } - } + "attributes": {"name": new_name}, + "relationships": {"organization": {"data": {"id": "test-org"}}}, } } self.mock_transport.request.return_value = mock_response @@ -176,12 +151,12 @@ def test_update_project_success(self): "data": { "type": "projects", "id": project_id, - "attributes": { - "name": new_name - } + "attributes": {"name": new_name}, } } - self.mock_transport.request.assert_called_once_with("PATCH", expected_path, json_body=expected_payload) + 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""" @@ -236,9 +211,7 @@ def test_read_project_missing_organization(self): "data": { "id": project_id, "type": "projects", - "attributes": { - "name": "Test Project" - } + "attributes": {"name": "Test Project"}, # No relationships field } } From 5419e1c08873611fe5c4fce425fccee1668bbd39 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Tue, 16 Sep 2025 15:49:16 +0530 Subject: [PATCH 08/10] CRUD operation for Project of Python TFE --- examples/project.py | 595 ++++++++++++++++++++++++++++++++++ src/tfe/resources/projects.py | 251 +++++++++++--- src/tfe/types.py | 39 +++ src/tfe/utils.py | 15 + tests/units/test_project.py | 8 +- 5 files changed, 868 insertions(+), 40 deletions(-) create mode 100644 examples/project.py diff --git a/examples/project.py b/examples/project.py new file mode 100644 index 00000000..afc385c3 --- /dev/null +++ b/examples/project.py @@ -0,0 +1,595 @@ +""" +Comprehensive Integration Test for python-tfe Projects CRUD Operations + +This file tests all CRUD operations from src/tfe/resources/projects.py: +- List: Get all projects in an organization +- Create: Add new projects with validation +- Read: Get specific project details +- Update: Modify existing projects +- Delete: Remove projects + +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. Run the tests: + pytest examples/project.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 +from tfe.types import ProjectCreateOptions, ProjectListOptions, ProjectUpdateOptions + + +@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 LIST operation - Get all projects in organization + + This is the safest test to run first - it only reads data. + Tests: projects.list(organization, options) + """ + projects, org = integration_client + + try: + # Test basic list without options + print("๐Ÿ“‹ Testing LIST operation: basic list") + 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" + ) + assert hasattr(project, "description"), "Project should have a description" + assert hasattr(project, "created_at"), "Project should have created_at" + assert hasattr(project, "updated_at"), "Project should have updated_at" + print(f"๐Ÿ“‹ Example project: {project.name} (ID: {project.id})") + print(f"๐Ÿ“‹ Created: {project.created_at}, Updated: {project.updated_at}") + else: + print("๐Ÿ“‹ No projects found - this is normal for a new organization") + + # Test list with options + print("๐Ÿ“‹ Testing LIST operation: with options") + list_options = ProjectListOptions(page_size=5) + project_list_with_options = list(projects.list(org, list_options)) + print( + f"โœ… List with options returned {len(project_list_with_options)} projects" + ) + + except Exception as e: + pytest.fail( + f"LIST operation failed. Check your TFE_TOKEN and TFE_ORG. Error: {e}" + ) + + +def test_create_project_integration(integration_client): + """Test CREATE operation - Add new projects + + Tests: projects.create(organization, options) + Validates: ProjectCreateOptions with name and description + """ + projects, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"create-test-{unique_id}" + test_description = f"Integration test project created at {unique_id}" + project_id = None + + try: + # Test CREATE operation + print(f"๐Ÿ”จ Testing CREATE operation: {test_name}") + create_options = ProjectCreateOptions( + name=test_name, description=test_description + ) + created_project = projects.create(org, create_options) + + # Validate created project + assert created_project.name == test_name, ( + f"Expected name {test_name}, got {created_project.name}" + ) + assert created_project.description == test_description, ( + f"Expected description {test_description}, got {created_project.description}" + ) + 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}" + ) + assert created_project.workspace_count == 0, ( + "New project should have 0 workspaces" + ) + + project_id = created_project.id + print(f"โœ… CREATE successful: {project_id}") + print( + f"โœ… Project details: {created_project.name} - {created_project.description}" + ) + + except Exception as e: + pytest.fail(f"CREATE operation failed: {e}") + + finally: + # Clean up created project + if project_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up created project: {project_id}") + projects.delete(project_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") + + +def test_read_project_integration(integration_client): + """Test READ operation - Get specific project details + + Tests: projects.read(project_id, include) + Creates a project, reads it, then cleans up + """ + projects, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"read-test-{unique_id}" + project_id = None + + try: + # Create a project to read + print(f"๏ฟฝ Creating project for READ test: {test_name}") + create_options = ProjectCreateOptions( + name=test_name, description="Project for read test" + ) + created_project = projects.create(org, create_options) + project_id = created_project.id + + # Test READ operation + print(f"๐Ÿ“– Testing READ operation: {project_id}") + read_project = projects.read(project_id) + + # Validate read project + 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}" + ) + assert read_project.organization == org, ( + f"Expected org {org}, got {read_project.organization}" + ) + assert hasattr(read_project, "created_at"), "Project should have created_at" + assert hasattr(read_project, "updated_at"), "Project should have updated_at" + + print(f"โœ… READ successful: {read_project.name}") + print(f"โœ… Project created: {read_project.created_at}") + + # Note: Projects API doesn't support include parameters in the current API version + print("โœ… READ operation completed successfully") + + except Exception as e: + pytest.fail(f"READ operation failed: {e}") + + finally: + # Clean up created project + if project_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up read test project: {project_id}") + projects.delete(project_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") + + +def test_update_project_integration(integration_client): + """Test UPDATE operation - Modify existing projects + + Tests: projects.update(project_id, options) + Validates: ProjectUpdateOptions with name and description changes + """ + projects, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + original_name = f"update-test-{unique_id}" + updated_name = f"updated-test-{unique_id}" + original_description = "Original description for update test" + updated_description = "Updated description for update test" + project_id = None + + try: + # Create a project to update + print(f"๐Ÿ”จ Creating project for UPDATE test: {original_name}") + create_options = ProjectCreateOptions( + name=original_name, description=original_description + ) + created_project = projects.create(org, create_options) + project_id = created_project.id + + # Test UPDATE operation - name only + print("โœ๏ธ Testing UPDATE operation: name only") + update_options = ProjectUpdateOptions(name=updated_name) + updated_project = projects.update(project_id, update_options) + + 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}" + ) + assert updated_project.description == original_description, ( + "Description should remain unchanged" + ) + print(f"โœ… UPDATE name successful: {updated_project.name}") + + # Test UPDATE operation - description only + print("โœ๏ธ Testing UPDATE operation: description only") + update_options = ProjectUpdateOptions(description=updated_description) + updated_project = projects.update(project_id, update_options) + + assert updated_project.name == updated_name, "Name should remain unchanged" + assert updated_project.description == updated_description, ( + f"Expected updated description {updated_description}, got {updated_project.description}" + ) + print("โœ… UPDATE description successful") + + # Test UPDATE operation - both name and description + final_name = f"final-{unique_id}" + final_description = "Final description for update test" + print("โœ๏ธ Testing UPDATE operation: both name and description") + update_options = ProjectUpdateOptions( + name=final_name, description=final_description + ) + updated_project = projects.update(project_id, update_options) + + assert updated_project.name == final_name, ( + f"Expected final name {final_name}, got {updated_project.name}" + ) + assert updated_project.description == final_description, ( + f"Expected final description {final_description}, got {updated_project.description}" + ) + print(f"โœ… UPDATE both fields successful: {updated_project.name}") + + except Exception as e: + pytest.fail(f"UPDATE operation failed: {e}") + + finally: + # Clean up created project + if project_id: + try: + print(f"๐Ÿ—‘๏ธ Cleaning up update test project: {project_id}") + projects.delete(project_id) + print("โœ… Cleanup successful") + except Exception as e: + print(f"โŒ Warning: Failed to clean up project {project_id}: {e}") + + +def test_delete_project_integration(integration_client): + """Test DELETE operation - Remove projects + + Tests: projects.delete(project_id) + Creates a project, deletes it, verifies it's gone + """ + projects, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"delete-test-{unique_id}" + project_id = None + + try: + # Create a project to delete + print(f"๐Ÿ”จ Creating project for DELETE test: {test_name}") + create_options = ProjectCreateOptions( + name=test_name, description="Project for delete test" + ) + created_project = projects.create(org, create_options) + project_id = created_project.id + print(f"โœ… Project created for deletion: {project_id}") + + # Verify project exists + print("๐Ÿ“– Verifying project exists before deletion") + read_project = projects.read(project_id) + assert read_project.id == project_id + print(f"โœ… Project confirmed to exist: {read_project.name}") + + # Test DELETE operation + print(f"๐Ÿ—‘๏ธ Testing DELETE operation: {project_id}") + projects.delete(project_id) + print("โœ… DELETE operation completed") + + # Verify project is deleted + print("๐Ÿ“– Verifying project is deleted") + try: + projects.read(project_id) + pytest.fail("Project should not exist after deletion") + except Exception as e: + if "404" in str(e) or "not found" in str(e).lower(): + print("โœ… Project successfully deleted - confirmed by 404 error") + else: + raise e + + # Clear project_id since it's been deleted + project_id = None + + except Exception as e: + pytest.fail(f"DELETE operation failed: {e}") + + finally: + # Additional cleanup attempt (should be unnecessary) + if project_id: + try: + print(f"๐Ÿ—‘๏ธ Additional cleanup attempt: {project_id}") + projects.delete(project_id) + except Exception: + pass # Project might already be deleted + + +def test_comprehensive_crud_integration(integration_client): + """Test all CRUD operations in sequence + + โš ๏ธ WARNING: This test creates and deletes real resources! + Tests complete workflow: CREATE โ†’ READ โ†’ UPDATE โ†’ LIST โ†’ DELETE + """ + projects, org = integration_client + + unique_id = str(uuid.uuid4())[:8] + test_name = f"comprehensive-{unique_id}" + updated_name = f"comprehensive-updated-{unique_id}" + test_description = f"Comprehensive CRUD test {unique_id}" + updated_description = f"Updated comprehensive CRUD test {unique_id}" + project_id = None + + try: + print(f"๐Ÿ”„ Starting comprehensive CRUD test: {test_name}") + + # 1. CREATE + print("1๏ธโƒฃ CREATE: Creating project") + create_options = ProjectCreateOptions( + name=test_name, description=test_description + ) + created_project = projects.create(org, create_options) + project_id = created_project.id + + assert created_project.name == test_name + assert created_project.description == test_description + print(f"โœ… CREATE: {project_id}") + + # 2. READ + print("2๏ธโƒฃ READ: Reading created project") + read_project = projects.read(project_id) + + assert read_project.id == project_id + assert read_project.name == test_name + assert read_project.description == test_description + print(f"โœ… READ: {read_project.name}") + + # 3. UPDATE + print("3๏ธโƒฃ UPDATE: Updating project") + update_options = ProjectUpdateOptions( + name=updated_name, description=updated_description + ) + updated_project = projects.update(project_id, update_options) + + assert updated_project.id == project_id + assert updated_project.name == updated_name + assert updated_project.description == updated_description + print(f"โœ… UPDATE: {updated_project.name}") + + # 4. LIST (verify updated project appears) + print("4๏ธโƒฃ LIST: Verifying project appears in list") + project_list = list(projects.list(org)) + found_project = None + for p in project_list: + if p.id == project_id: + found_project = p + break + + assert found_project is not None, ( + f"Updated project {project_id} should appear in list" + ) + assert found_project.name == updated_name + print("โœ… LIST: Found updated project in list") + + # 5. DELETE + print("5๏ธโƒฃ DELETE: Deleting project") + projects.delete(project_id) + print("โœ… DELETE: Project deleted") + + # 6. Verify deletion + print("6๏ธโƒฃ VERIFY: Confirming deletion") + try: + projects.read(project_id) + pytest.fail("Project should not exist after deletion") + except Exception as e: + if "404" in str(e) or "not found" in str(e).lower(): + print("โœ… VERIFY: Deletion confirmed") + else: + raise e + + project_id = None # Clear since deleted + print("๐ŸŽ‰ Comprehensive CRUD test completed successfully!") + + except Exception as e: + pytest.fail(f"Comprehensive CRUD test failed: {e}") + + finally: + if project_id: + try: + print(f"๐Ÿ—‘๏ธ Final cleanup: {project_id}") + projects.delete(project_id) + except Exception: + pass + + +def test_validation_integration(integration_client): + """Test validation functions work with real API + + Tests all validation scenarios with actual API calls + """ + projects, org = integration_client + + print("๐Ÿ” Testing validation with real API calls") + + try: + # Test valid project creation + unique_id = str(uuid.uuid4())[:8] + valid_name = f"validation-test-{unique_id}" + + print(f"โœ… Testing valid project creation: {valid_name}") + create_options = ProjectCreateOptions( + name=valid_name, description="Valid project" + ) + created_project = projects.create(org, create_options) + + assert created_project.name == valid_name + project_id = created_project.id + print(f"โœ… Valid project created successfully: {project_id}") + + # Test valid project update + updated_name = f"validation-updated-{unique_id}" + print(f"โœ… Testing valid project update: {updated_name}") + update_options = ProjectUpdateOptions(name=updated_name) + updated_project = projects.update(project_id, update_options) + + assert updated_project.name == updated_name + print("โœ… Valid project updated successfully") + + # Clean up + projects.delete(project_id) + print("โœ… Validation test cleanup completed") + + except Exception as e: + pytest.fail(f"Validation integration test failed: {e}") + + +def test_error_handling_integration(integration_client): + """Test error handling with real API calls + + Tests various error scenarios to ensure proper error handling + """ + projects, org = integration_client + + print("๐Ÿšซ Testing error handling scenarios") + + # Test reading a non-existent project + print("๐Ÿšซ Testing read 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__}" + ) + assert "404" in str(e) or "not found" in str(e).lower() + + # Test updating a non-existent project + print("๐Ÿšซ Testing update non-existent project") + try: + update_options = ProjectUpdateOptions(name="should-fail") + projects.update(fake_project_id, update_options) + pytest.fail("Should have raised an exception for non-existent project") + except Exception as e: + print( + f"โœ… Correctly handled update error for non-existent project: {type(e).__name__}" + ) + assert "404" in str(e) or "not found" in str(e).lower() + + # Test deleting a non-existent project + print("๐Ÿšซ Testing delete non-existent project") + try: + projects.delete(fake_project_id) + pytest.fail("Should have raised an exception for non-existent project") + except Exception as e: + print( + f"โœ… Correctly handled delete error for non-existent project: {type(e).__name__}" + ) + assert "404" in str(e) or "not found" in str(e).lower() + + print("โœ… All error handling scenarios tested successfully") + + +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 614a8ef3..6c75cb35 100644 --- a/src/tfe/resources/projects.py +++ b/src/tfe/resources/projects.py @@ -1,83 +1,260 @@ from __future__ import annotations +import re from collections.abc import Iterator from typing import Any -from ..types import Project +from ..types import ( + Project, + ProjectCreateOptions, + ProjectListOptions, + ProjectUpdateOptions, +) +from ..utils import valid_string, valid_string_id from ._base import _Service -def _safe_str(v: Any, default: str = "") -> str: - return v if isinstance(v, str) else (str(v) if v is not None else default) +# Project validation functions +def valid_project_name(name: str) -> bool: + """Validate project name format""" + if not valid_string(name): + return False + # Project names can contain letters, numbers, spaces, hyphens, underscores, and periods + # Must be between 1 and 90 characters + if len(name) > 90: + return False + # Allow most printable characters except some special ones + # Based on Terraform Cloud API documentation + pattern = re.compile(r"^[a-zA-Z0-9\s._-]+$") + return bool(pattern.match(name)) + + +def valid_organization_name(org_name: str) -> bool: + """Validate organization name format""" + if not valid_string(org_name): + return False + # Organization names must be valid identifiers + return valid_string_id(org_name) + + +def validate_project_create_options( + organization: str, name: str, description: str | None = None +) -> None: + """Validate project creation parameters""" + if not valid_organization_name(organization): + raise ValueError("Organization name is required and must be valid") + + if not valid_string(name): + raise ValueError("Project name is required") + + if not valid_project_name(name): + raise ValueError("Project name contains invalid characters or is too long") + + if description is not None and not valid_string(description): + raise ValueError("Description must be a valid string") + + +def validate_project_update_options( + project_id: str, name: str | None = None, description: str | None = None +) -> None: + """Validate project update parameters""" + if not valid_string_id(project_id): + raise ValueError("Project ID is required") + + if name is not None: + if not valid_string(name): + raise ValueError("Project name cannot be empty") + if not valid_project_name(name): + raise ValueError("Project name contains invalid characters or is too long") + + if description is not None and not valid_string(description): + raise ValueError("Description must be a valid string") + + +def validate_project_list_options( + organization: str, query: str | None = None, name: str | None = None +) -> None: + """Validate project list options following Go TFE patterns.""" + if not valid_organization_name(organization): + raise ValueError("Organization name is required and must be valid") + + if query and not valid_string(query): + raise ValueError("Query must be a valid string") + + if name and not valid_project_name(name): + raise ValueError("Project name must be valid") + + +def _safe_str(value: Any, default: str = "") -> str: + """Safely convert a value to string with optional default.""" + if value is None: + return default + return str(value) class Projects(_Service): - def list(self, organization: str) -> Iterator[Project]: + """Projects service for managing Terraform Enterprise projects""" + + def list( + self, organization: str, options: ProjectListOptions | None = None + ) -> Iterator[Project]: + """List projects in an organization""" + # Validate inputs following Go patterns + validate_project_list_options(organization) + path = f"/api/v2/organizations/{organization}/projects" - for item in self._list(path): + params = {} + + if options: + if options.include: + params["include"] = ",".join(options.include) + if options.query: + params["q"] = options.query + if options.name: + params["filter[names]"] = options.name + if options.page_number: + params["page[number]"] = str(options.page_number) + if options.page_size: + params["page[size]"] = str(options.page_size) + + if params: + items_iter = self._list(path, params=params) + else: + items_iter = self._list(path) + + for item in items_iter: + # Extract project data following Go patterns attr = item.get("attributes", {}) or {} - proj_id = _safe_str(item.get("id")) - name = _safe_str(attr.get("name")) - yield Project(id=proj_id, name=name, organization=organization) + project_data = { + "id": _safe_str(item.get("id")), + "name": _safe_str(attr.get("name")), + "description": _safe_str(attr.get("description")), + "organization": organization, + "created_at": _safe_str(attr.get("created-at")), + "updated_at": _safe_str(attr.get("updated-at")), + "workspace_count": attr.get("workspace-count", 0), + "default_execution_mode": _safe_str( + attr.get("default-execution-mode"), "remote" + ), + } + yield Project(**project_data) - def create(self, organization: str, name: str) -> Project: + def create(self, organization: str, options: ProjectCreateOptions) -> Project: """Create a new project in an organization""" + # Validate inputs following Go patterns + validate_project_create_options(organization, options.name, options.description) + path = f"/api/v2/organizations/{organization}/projects" - payload = {"data": {"type": "projects", "attributes": {"name": name}}} + attributes = {"name": options.name} + if options.description: + attributes["description"] = options.description + + payload = {"data": {"type": "projects", "attributes": attributes}} - # 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, - ) + # Extract project data following Go patterns + attr = data.get("attributes", {}) or {} + project_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "description": _safe_str(attr.get("description")), + "organization": organization, + "created_at": _safe_str(attr.get("created-at")), + "updated_at": _safe_str(attr.get("updated-at")), + "workspace_count": attr.get("workspace-count", 0), + "default_execution_mode": _safe_str( + attr.get("default-execution-mode"), "remote" + ), + } + return Project(**project_data) - def read(self, project_id: str) -> Project: + def read(self, project_id: str, include: list[str] | None = None) -> Project: """Get a specific project by ID""" + # Validate inputs following Go patterns + if not valid_string_id(project_id): + raise ValueError("Project ID is required and must be valid") + path = f"/api/v2/projects/{project_id}" - response = self.t.request("GET", path) + params: dict[str, str] = {} + if include: + params["include"] = ",".join(include) + + if params: + response = self.t.request("GET", path, params=params) + else: + response = self.t.request("GET", path) + data = response.json()["data"] - attr = data.get("attributes", {}) or {} - # Get organization from relationships if available + # Extract organization from relationships 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, - ) + # Extract project data following Go patterns + attr = data.get("attributes", {}) or {} + project_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "description": _safe_str(attr.get("description")), + "organization": organization, + "created_at": _safe_str(attr.get("created-at")), + "updated_at": _safe_str(attr.get("updated-at")), + "workspace_count": attr.get("workspace-count", 0), + "default_execution_mode": _safe_str( + attr.get("default-execution-mode"), "remote" + ), + } + return Project(**project_data) + + def update(self, project_id: str, options: ProjectUpdateOptions) -> Project: + """Update a project's name and/or description""" + # Validate inputs following Go patterns + validate_project_update_options(project_id, options.name, options.description) - def update(self, project_id: str, name: str) -> Project: - """Update a project's name""" path = f"/api/v2/projects/{project_id}" + attributes = {} + + if options.name is not None: + attributes["name"] = options.name + if options.description is not None: + attributes["description"] = options.description + payload = { - "data": {"type": "projects", "id": project_id, "attributes": {"name": name}} + "data": {"type": "projects", "id": project_id, "attributes": attributes} } - # 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 + # Extract organization from relationships 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, - ) + # Extract project data following Go patterns + attr = data.get("attributes", {}) or {} + project_data = { + "id": _safe_str(data.get("id")), + "name": _safe_str(attr.get("name")), + "description": _safe_str(attr.get("description")), + "organization": organization, + "created_at": _safe_str(attr.get("created-at")), + "updated_at": _safe_str(attr.get("updated-at")), + "workspace_count": attr.get("workspace-count", 0), + "default_execution_mode": _safe_str( + attr.get("default-execution-mode"), "remote" + ), + } + return Project(**project_data) def delete(self, project_id: str) -> None: """Delete a project""" + # Validate inputs following Go patterns + if not valid_string_id(project_id): + raise ValueError("Project ID is required and must be valid") + path = f"/api/v2/projects/{project_id}" self.t.request("DELETE", path) diff --git a/src/tfe/types.py b/src/tfe/types.py index 478614eb..6586f30c 100644 --- a/src/tfe/types.py +++ b/src/tfe/types.py @@ -26,9 +26,48 @@ class Organization(BaseModel): class Project(BaseModel): + """Project represents a Terraform Enterprise project""" + id: str name: str + description: str = "" organization: str + created_at: str = "" + updated_at: str = "" + workspace_count: int = 0 + default_execution_mode: str = "remote" + + +class ProjectListOptions(BaseModel): + """Options for listing projects""" + + # Optional: String used to filter results by complete project name + name: str | None = None + # Optional: Query string to search projects by names + query: str | None = None + # Optional: Include related resources + include: list[str] | None = None + # Pagination options + page_number: int | None = None + page_size: int | None = None + + +class ProjectCreateOptions(BaseModel): + """Options for creating a project""" + + # Required: A name to identify the project + name: str + # Optional: A description for the project + description: str | None = None + + +class ProjectUpdateOptions(BaseModel): + """Options for updating a project""" + + # Optional: A name to identify the project + name: str | None = None + # Optional: A description for the project + description: str | None = None class Workspace(BaseModel): diff --git a/src/tfe/utils.py b/src/tfe/utils.py index 968bb4d0..874bc92a 100644 --- a/src/tfe/utils.py +++ b/src/tfe/utils.py @@ -1,8 +1,14 @@ +"""Basic utility functions for the TFE client.""" + from __future__ import annotations +import re import time from collections.abc import Callable +# Pattern for validating string IDs +_STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9\-_]+$") + def poll_until( fn: Callable[[], bool], @@ -18,3 +24,12 @@ def poll_until( if timeout_s is not None and (time.time() - start) > timeout_s: raise TimeoutError("Timed out") time.sleep(interval_s) + + +# Basic validation functions +def valid_string(v: str | None) -> bool: + return v is not None and str(v).strip() != "" + + +def valid_string_id(v: str | None) -> bool: + return v is not None and _STRING_ID_PATTERN.match(str(v)) is not None diff --git a/tests/units/test_project.py b/tests/units/test_project.py index da5f222b..7124bd24 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -1,7 +1,7 @@ from unittest.mock import Mock from tfe.resources.projects import Projects, _safe_str -from tfe.types import Project +from tfe.types import Project, ProjectCreateOptions, ProjectUpdateOptions class TestProjects: @@ -63,6 +63,7 @@ def test_create_project_success(self): """Test successful project creation""" organization = "test-org" project_name = "New Project" + options = ProjectCreateOptions(name=project_name) # Mock API response mock_response = Mock() @@ -75,7 +76,7 @@ def test_create_project_success(self): } self.mock_transport.request.return_value = mock_response - result = self.projects_service.create(organization, project_name) + result = self.projects_service.create(organization, options) # Assertions assert isinstance(result, Project) @@ -124,6 +125,7 @@ def test_update_project_success(self): """Test successful project update""" project_id = "prj-123" new_name = "Updated Project" + options = ProjectUpdateOptions(name=new_name) # Mock API response mock_response = Mock() @@ -137,7 +139,7 @@ def test_update_project_success(self): } self.mock_transport.request.return_value = mock_response - result = self.projects_service.update(project_id, new_name) + result = self.projects_service.update(project_id, options) # Assertions assert isinstance(result, Project) From 8c68e39eb5082e3cb8849bbea7455fe6ae5cfe39 Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Tue, 16 Sep 2025 15:49:37 +0530 Subject: [PATCH 09/10] CRUD operation for Project of Python TFE --- examples/project_integration_test_example.py | 225 ------------------- 1 file changed, 225 deletions(-) delete mode 100644 examples/project_integration_test_example.py diff --git a/examples/project_integration_test_example.py b/examples/project_integration_test_example.py deleted file mode 100644 index c4876f88..00000000 --- a/examples/project_integration_test_example.py +++ /dev/null @@ -1,225 +0,0 @@ -""" -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"]) From 3effdd84517d030a564de6046ce2d3a577df355e Mon Sep 17 00:00:00 2001 From: KshitijaChoudhari Date: Tue, 16 Sep 2025 15:57:32 +0530 Subject: [PATCH 10/10] CRUD operation for Project of Python TFE 1 --- src/tfe/utils.py | 6 +----- 1 file changed, 1 insertion(+), 5 deletions(-) diff --git a/src/tfe/utils.py b/src/tfe/utils.py index 874bc92a..303431a4 100644 --- a/src/tfe/utils.py +++ b/src/tfe/utils.py @@ -1,13 +1,10 @@ -"""Basic utility functions for the TFE client.""" - from __future__ import annotations import re import time from collections.abc import Callable -# Pattern for validating string IDs -_STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9\-_]+$") +_STRING_ID_PATTERN = re.compile(r"^[a-zA-Z0-9][a-zA-Z0-9_-]{2,}$") def poll_until( @@ -26,7 +23,6 @@ def poll_until( time.sleep(interval_s) -# Basic validation functions def valid_string(v: str | None) -> bool: return v is not None and str(v).strip() != ""