From 7ec907d53fb90a573a9b36958d91ef2dfb8e7c8f Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Tue, 24 Mar 2026 14:50:44 +0530 Subject: [PATCH 1/2] refactor(project): Updated create, update & Project models, updated the endpoint for list-effective-tag-bindings & delete-tag-bindings and added new examples for project --- examples/project.py | 1059 ++++++++----------------------- src/pytfe/models/__init__.py | 2 + src/pytfe/models/project.py | 91 ++- src/pytfe/resources/projects.py | 225 ++++--- 4 files changed, 467 insertions(+), 910 deletions(-) diff --git a/examples/project.py b/examples/project.py index 7702b09a..5ad8a167 100644 --- a/examples/project.py +++ b/examples/project.py @@ -1,849 +1,310 @@ -""" -Comprehensive Integration Test for python-tfe Projects CRUD Operations - -This file tests all CRUD operations: -- 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 -""" +from __future__ import annotations +import argparse import os import uuid -import pytest - -from pytfe._http import HTTPTransport -from pytfe.config import TFEConfig -from pytfe.errors import NotFound +from pytfe import TFEClient, TFEConfig from pytfe.models import ( ProjectAddTagBindingsOptions, ProjectCreateOptions, ProjectListOptions, + ProjectSettingOverwrites, ProjectUpdateOptions, TagBinding, ) -from pytfe.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 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}") +def _print_header(title: str) -> None: + print("\n" + "=" * 80) + print(title) + print("=" * 80) + + +def _org_display(project) -> str: + """Render organization safely for both string and object representations.""" + org = getattr(project, "organization", None) + if org is None: + return "" + if isinstance(org, str): + return org + return getattr(org, "id", str(org)) + + +def _parse_tag_pairs(tag_pairs: list[str] | None) -> list[TagBinding]: + """Convert --tag key=value args into TagBinding models.""" + if not tag_pairs: + return [] + + tags: list[TagBinding] = [] + for pair in tag_pairs: + if "=" in pair: + key, value = pair.split("=", 1) + key = key.strip() + value = value.strip() + if not key: + raise ValueError(f"Invalid tag format '{pair}'. Key is empty.") + tags.append(TagBinding(key=key, value=value)) 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}" - ) + key = pair.strip() + if not key: + raise ValueError(f"Invalid tag format '{pair}'.") + tags.append(TagBinding(key=key, value=None)) + return tags -def test_create_project_integration(integration_client): - """Test CREATE operation - Add new projects +def main() -> None: + parser = argparse.ArgumentParser(description="Projects demo for python-tfe SDK") - 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 + parser.add_argument( + "--address", + default=os.getenv("TFE_ADDRESS", "https://app.terraform.io"), + help="TFE/TFC address", + ) + parser.add_argument( + "--token", + default=os.getenv("TFE_TOKEN", ""), + help="TFE/TFC API token", + ) + parser.add_argument( + "--organization", + default=os.getenv("TFE_ORG", ""), + help="Organization name", + ) + parser.add_argument( + "--page-size", + type=int, + default=20, + help="Page size for project listing", + ) - # 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) + parser.add_argument("--list", action="store_true", help="List projects") + parser.add_argument("--create", action="store_true", help="Create a project") + parser.add_argument("--read", action="store_true", help="Read a project") + parser.add_argument("--update", action="store_true", help="Update a project") + parser.add_argument("--delete", action="store_true", help="Delete a project") - 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}") + parser.add_argument( + "--list-tag-bindings", + action="store_true", + help="List project tag bindings", + ) + parser.add_argument( + "--list-effective-tag-bindings", + action="store_true", + help="List project effective tag bindings", + ) + parser.add_argument( + "--add-tag-bindings", + action="store_true", + help="Add/replace tag bindings on project", + ) + parser.add_argument( + "--delete-tag-bindings", + action="store_true", + help="Delete all tag bindings from project", + ) - # 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) + parser.add_argument( + "--project-id", + help="Project ID for read/update/delete/tag operations", + ) + parser.add_argument("--name", help="Project name for create/update") + parser.add_argument("--description", help="Project description for create/update") + parser.add_argument( + "--tag", + action="append", + default=[], + help="Tag binding in key=value format (repeatable)", + ) + parser.add_argument( + "--create-random", + action="store_true", + help="Append a short random suffix to --name for create", + ) - 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") + args = parser.parse_args() - # 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) + if not args.token: + raise SystemExit("Error: --token or TFE_TOKEN is required") - 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}") + cfg = TFEConfig(address=args.address, token=args.token) + client = TFEClient(cfg) - except Exception as e: - pytest.fail(f"UPDATE operation failed: {e}") + has_org_op = args.list or args.create + has_project_op = ( + args.read + or args.update + or args.delete + or args.list_tag_bindings + or args.list_effective_tag_bindings + or args.add_tag_bindings + or args.delete_tag_bindings + ) - 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}") + if has_org_op and not args.organization: + raise SystemExit("Error: --organization or TFE_ORG is required") + if has_project_op and not args.project_id: + raise SystemExit("Error: --project-id is required for selected operation") -def test_delete_project_integration(integration_client): - """Test DELETE operation - Remove projects + # 1) List projects + if args.list: + _print_header(f"Listing projects for organization: {args.organization}") + list_options = ProjectListOptions(page_size=args.page_size) - Tests: projects.delete(project_id) - Creates a project, deletes it, verifies it's gone - """ - projects, org = integration_client + count = 0 + for project in client.projects.list(args.organization, list_options): + count += 1 + print(f"- {project.name} (ID: {project.id})") + print(f" Description: {project.description}") + print(f" Workspaces: {project.workspace_count}") + print(f" Default execution mode: {project.default_execution_mode}") + print( + f" Auto destroy activity duration: {project.auto_destroy_activity_duration}" + ) + print(f" Created at: {project.created_at}") + print(f" Updated at: {project.updated_at}") + print(f" Setting overwrites: {project.setting_overwrites}") + print(f" Default agent pool: {project.default_agent_pool}") + print(f" Organization: {_org_display(project)}") - unique_id = str(uuid.uuid4())[:8] - test_name = f"delete-test-{unique_id}" - project_id = None + print() - 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 + if count == 0: + print("No projects found.") + else: + print(f"Total: {count} projects") - assert created_project.name == test_name - assert created_project.description == test_description - print(f"CREATE: {project_id}") + # 2) Create project + if args.create: + if not args.name: + raise SystemExit("Error: --name is required for create") - # 2. READ - print("2 READ: Reading created project") - read_project = projects.read(project_id) + name = args.name + if args.create_random: + name = f"{name}-{uuid.uuid4().hex[:8]}" - assert read_project.id == project_id - assert read_project.name == test_name - assert read_project.description == test_description - print(f"READ: {read_project.name}") + _print_header(f"Creating 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}") + tags = _parse_tag_pairs(args.tag) 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: + name=name, + description=args.description, + auto_destroy_activity_duration="14d", + default_execution_mode="remote", + default_agent_pool_id=None, + setting_overwrites=ProjectSettingOverwrites( + execution_mode=False, + agent_pool=False, + ), + tag_bindings=tags, + ) + + project = client.projects.create(args.organization, create_options) + print(f"Created project: {project.id}") + print(f"Name: {project.name}") + print(f"Description: {project.description}") + print(f"Workspaces: {project.workspace_count}") + print(f"Default execution mode: {project.default_execution_mode}") 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: + f"Auto destroy activity duration: {project.auto_destroy_activity_duration}" + ) + print(f"Created at: {project.created_at}") + print(f"Updated at: {project.updated_at}") + print(f"Setting overwrites: {project.setting_overwrites}") + print(f"Default agent pool: {project.default_agent_pool}") + print(f"Organization: {_org_display(project)}") + + # 3) Read project + if args.read: + _print_header(f"Reading project: {args.project_id}") + project = client.projects.read(args.project_id) + print(f"ID: {project.id}") + print(f"Name: {project.name}") + print(f"Description: {project.description}") + print(f"Organization: {_org_display(project)}") + print(f"Created at: {project.created_at}") + print(f"Updated at: {project.updated_at}") + print(f"Workspace count: {project.workspace_count}") + print(f"Default execution mode: {project.default_execution_mode}") print( - f"Correctly handled delete error for non-existent project: {type(e).__name__}" + f"Auto destroy activity duration: {project.auto_destroy_activity_duration}" ) - assert "404" in str(e) or "not found" in str(e).lower() - print("All error handling scenarios tested successfully") - - -def test_project_tag_bindings_integration(integration_client): - """ - Integration test for project tag binding operations - - Note: Project tag bindings may not be available in all HCP Terraform plans. - This test gracefully handles unavailable features while testing what's available. - """ - projects, org = integration_client - - unique_id = str(uuid.uuid4())[:8] - test_name = f"tag-test-{unique_id}" - test_description = f"Project for testing tag bindings - {unique_id}" - project_id = None - - try: - # Create a test project for tagging operations - print(f"Setting up test project for tagging: {test_name}") - create_options = ProjectCreateOptions( - name=test_name, description=test_description - ) - created_project = projects.create(org, create_options) - project_id = created_project.id - print(f"Created test project: {project_id}") - - # Test 1: List tag bindings (this should work) - print("Testing LIST_TAG_BINDINGS") - try: - initial_tag_bindings = projects.list_tag_bindings(project_id) - assert isinstance(initial_tag_bindings, list), "Should return a list" - print(f"list_tag_bindings works: {len(initial_tag_bindings)} bindings") - list_tag_bindings_available = True - except Exception as e: - print(f"list_tag_bindings not available: {e}") - list_tag_bindings_available = False - - # Test 2: List effective tag bindings - print("Testing LIST_EFFECTIVE_TAG_BINDINGS") - try: - effective_bindings = projects.list_effective_tag_bindings(project_id) - assert isinstance(effective_bindings, list), "Should return a list" - print( - f"list_effective_tag_bindings works: {len(effective_bindings)} bindings" + # 4) Update project + if args.update: + if args.name is None and args.description is None and not args.tag: + raise SystemExit( + "Error: provide at least one of --name, --description or --tag for update" ) - effective_tag_bindings_available = True - except Exception as e: - print(f"list_effective_tag_bindings not available: {e}") - print("This feature may require a higher HCP Terraform plan") - effective_tag_bindings_available = False - - # Test 3: Add tag bindings (if basic listing works) - if list_tag_bindings_available: - print("Testing ADD_TAG_BINDINGS") - try: - test_tags = [ - TagBinding(key="environment", value="testing"), - TagBinding(key="integration-test", value="true"), - ] - add_options = ProjectAddTagBindingsOptions(tag_bindings=test_tags) - added_bindings = projects.add_tag_bindings(project_id, add_options) - - assert isinstance(added_bindings, list), "Should return a list" - assert len(added_bindings) == len(test_tags), ( - "Should return all added tags" - ) - print(f"add_tag_bindings works: added {len(added_bindings)} bindings") - - # Verify tags were actually added - current_bindings = projects.list_tag_bindings(project_id) - added_keys = {binding.key for binding in current_bindings} - for tag in test_tags: - assert tag.key in added_keys, ( - f"Tag {tag.key} not found after adding" - ) - print(f"Verified tags added: {len(current_bindings)} total bindings") - - add_tag_bindings_available = True - - # Test 4: Delete tag bindings - print("Testing DELETE_TAG_BINDINGS") - try: - result = projects.delete_tag_bindings(project_id) - assert result is None, "Delete should return None" - - # Verify deletion - final_bindings = projects.list_tag_bindings(project_id) - print( - f"delete_tag_bindings works: {len(final_bindings)} bindings remain" - ) - delete_tag_bindings_available = True - except Exception as e: - print(f"delete_tag_bindings not available: {e}") - delete_tag_bindings_available = False - - except Exception as e: - print(f"add_tag_bindings not available: {e}") - print("This feature may require a higher HCP Terraform plan") - add_tag_bindings_available = False - delete_tag_bindings_available = False - else: - add_tag_bindings_available = False - delete_tag_bindings_available = False - - # Summary - print("\n Project Tag Bindings API Availability Summary:") - features = [ - ("list_tag_bindings", list_tag_bindings_available), - ("list_effective_tag_bindings", effective_tag_bindings_available), - ("add_tag_bindings", add_tag_bindings_available), - ("delete_tag_bindings", delete_tag_bindings_available), - ] - - for feature_name, available in features: - status = "Available" if available else " Not Available" - print(f"{feature_name}: {status}") - - available_count = sum(available for _, available in features) - print( - f"\n {available_count}/4 tag binding features are available in this HCP Terraform organization" - ) - if available_count == 4: - print("All project tag binding operations work perfectly!") - elif available_count > 0: - print("Partial functionality available - basic operations work!") - else: - print("Tag binding features may require a higher HCP Terraform plan") + _print_header(f"Updating project: {args.project_id}") - except Exception as e: - pytest.fail( - f"Project tag binding integration test failed unexpectedly. " - f"This may indicate a configuration or connectivity issue. Error: {e}" - ) + tags = _parse_tag_pairs(args.tag) + update_options = ProjectUpdateOptions( + name=args.name, + description=args.description, + tag_bindings=tags if tags else None, + ) + + updated = client.projects.update(args.project_id, update_options) + print("Project updated successfully") + print(f"ID: {updated.id}") + print(f"Name: {updated.name}") + print(f"Description: {updated.description}") + + # 5) Delete project + if args.delete: + _print_header(f"Deleting project: {args.project_id}") + client.projects.delete(args.project_id) + print("Project deleted successfully") + + # 6) List tag bindings + if args.list_tag_bindings: + _print_header(f"Listing tag bindings for project: {args.project_id}") + bindings = client.projects.list_tag_bindings(args.project_id) + + if not bindings: + print("No tag bindings found.") + else: + for tag in bindings: + print(f"- {tag.key}={tag.value}") + print(f"Total: {len(bindings)} tag bindings") - finally: - # Clean up: Delete the test project - if project_id: - try: - print(f"🧹 Cleaning up test project: {project_id}") - projects.delete(project_id) - print("Test project deleted successfully") - except Exception as cleanup_error: - print( - f" Warning: Failed to clean up test project {project_id}: {cleanup_error}" - ) - - -def test_project_tag_bindings_error_scenarios(integration_client): - """ - Test error handling for project tag binding operations - - Tests various error conditions: - - Invalid project IDs - - Empty tag binding lists - - Non-existent projects - """ - projects, org = integration_client - - print("Testing tag binding error scenarios") - - # Test invalid project ID validation - print("Testing invalid project ID scenarios") - - invalid_project_ids = ["", "x", "invalid-id", None] - - for invalid_id in invalid_project_ids: - if invalid_id is None: - continue # Skip None as it will cause different error - - try: - projects.list_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised ValueError or NotFound for invalid project ID: {invalid_id}" - ) - except (ValueError, NotFound) as e: - print(f"Correctly rejected invalid project ID '{invalid_id}': {e}") - if isinstance(e, ValueError): - assert "Project ID is required and must be valid" in str(e) - - try: - projects.list_effective_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised ValueError or NotFound for invalid project ID: {invalid_id}" - ) - except (ValueError, NotFound) as e: - print(f"Correctly rejected invalid project ID '{invalid_id}': {e}") + # 7) List effective tag bindings + if args.list_effective_tag_bindings: + _print_header(f"Listing effective tag bindings for project: {args.project_id}") + bindings = client.projects.list_effective_tag_bindings(args.project_id) - try: - projects.delete_tag_bindings(invalid_id) - pytest.fail( - f"Should have raised ValueError or NotFound for invalid project ID: {invalid_id}" - ) - except (ValueError, NotFound) as e: - print(f"Correctly rejected invalid project ID '{invalid_id}': {e}") - - # Test empty tag binding list - print("Testing empty tag binding list") - try: - fake_project_id = "prj-fakefakefake123" - empty_options = ProjectAddTagBindingsOptions(tag_bindings=[]) - projects.add_tag_bindings(fake_project_id, empty_options) - pytest.fail("Should have raised ValueError for empty tag binding list") - except ValueError as e: - print(f"Correctly rejected empty tag binding list: {e}") - assert "At least one tag binding is required" in str(e) - - # Test non-existent project operations - print("Testing operations on non-existent project") - fake_project_id = "prj-doesnotexist123" - - # These should raise HTTP errors (404) from the API - for operation_name, operation_func in [ - ("list_tag_bindings", lambda: projects.list_tag_bindings(fake_project_id)), - ( - "list_effective_tag_bindings", - lambda: projects.list_effective_tag_bindings(fake_project_id), - ), - ("delete_tag_bindings", lambda: projects.delete_tag_bindings(fake_project_id)), - ]: - try: - operation_func() - pytest.fail(f"{operation_name} should have failed for non-existent project") - except Exception as e: - print( - f"{operation_name} correctly failed for non-existent project: {type(e).__name__}" - ) - # Should be some kind of HTTP error (404, not found, etc.) - assert ( - "404" in str(e) - or "not found" in str(e).lower() - or "does not exist" in str(e).lower() + if not bindings: + print("No effective tag bindings found.") + else: + for tag in bindings: + print(f"- {tag.key}={tag.value}") + print(f"Total: {len(bindings)} effective tag bindings") + + # 8) Add tag bindings + if args.add_tag_bindings: + tags = _parse_tag_pairs(args.tag) + if not tags: + raise SystemExit( + "Error: at least one --tag key=value is required for --add-tag-bindings" ) - # Test add_tag_bindings on non-existent project - try: - test_tags = [TagBinding(key="test", value="value")] - add_options = ProjectAddTagBindingsOptions(tag_bindings=test_tags) - projects.add_tag_bindings(fake_project_id, add_options) - pytest.fail("add_tag_bindings should have failed for non-existent project") - except Exception as e: - print( - f"add_tag_bindings correctly failed for non-existent project: {type(e).__name__}" - ) - assert ( - "404" in str(e) - or "not found" in str(e).lower() - or "does not exist" in str(e).lower() - ) + _print_header(f"Adding tag bindings to project: {args.project_id}") + options = ProjectAddTagBindingsOptions(tag_bindings=tags) + updated_tags = client.projects.add_tag_bindings(args.project_id, options) + for tag in updated_tags: + print(f"- {tag.key}={tag.value}") + print(f"Total returned: {len(updated_tags)} tag bindings") - print("All tag binding error scenarios tested successfully") + # 9) Delete tag bindings + if args.delete_tag_bindings: + _print_header(f"Deleting all tag bindings from project: {args.project_id}") + client.projects.delete_tag_bindings(args.project_id) + print("Deleted all project tag bindings") 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"]) + main() diff --git a/src/pytfe/models/__init__.py b/src/pytfe/models/__init__.py index 1110cf57..1fc8828b 100644 --- a/src/pytfe/models/__init__.py +++ b/src/pytfe/models/__init__.py @@ -153,6 +153,7 @@ ProjectAddTagBindingsOptions, ProjectCreateOptions, ProjectListOptions, + ProjectSettingOverwrites, ProjectUpdateOptions, ) @@ -501,6 +502,7 @@ "ProjectCreateOptions", "ProjectListOptions", "ProjectUpdateOptions", + "ProjectSettingOverwrites", "DataRetentionPolicy", "DataRetentionPolicyChoice", "DataRetentionPolicyDeleteOlder", diff --git a/src/pytfe/models/project.py b/src/pytfe/models/project.py index 3f4b9c6c..082c6861 100644 --- a/src/pytfe/models/project.py +++ b/src/pytfe/models/project.py @@ -1,19 +1,36 @@ from __future__ import annotations -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field +from .agent import AgentPool from .common import TagBinding +from .organization import Organization class Project(BaseModel): + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + id: str - name: str | None = None - description: str = "" - organization: str | None = None - created_at: str = "" - updated_at: str = "" - workspace_count: int = 0 - default_execution_mode: str = "remote" + name: str | None = Field(default=None, alias="name") + description: str | None = Field(default=None, alias="description") + created_at: str | None = Field(default=None, alias="created-at") + updated_at: str | None = Field(default=None, alias="updated-at") + workspace_count: int = Field(default=0, alias="workspace-count") + default_execution_mode: str = Field( + default="remote", alias="default-execution-mode" + ) + auto_destroy_activity_duration: str | None = Field( + default=None, alias="auto-destroy-activity-duration" + ) + setting_overwrites: ProjectSettingOverwrites | None = Field( + default=None, alias="setting-overwrites" + ) + + # relations + default_agent_pool: AgentPool | None = Field( + default=None, alias="default-agent-pool" + ) + organization: Organization | None = Field(default=None, alias="organization") class ProjectListOptions(BaseModel): @@ -26,29 +43,81 @@ class ProjectListOptions(BaseModel): # 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""" + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + # Required: A name to identify the project name: str # Optional: A description for the project - description: str | None = None + description: str | None = Field(default=None, alias="description") + auto_destroy_activity_duration: str | None = Field( + default=None, + alias="auto-destroy-activity-duration", + ) + default_execution_mode: str | None = Field( + default="remote", alias="default-execution-mode" + ) + # Optional: DefaultAgentPoolID default agent pool for workspaces in the project, + # required when DefaultExecutionMode is set to `agent` + default_agent_pool_id: str | None = Field( + default=None, + alias="default-agent-pool-id", + ) + setting_overwrites: ProjectSettingOverwrites | None = Field( + default=None, + alias="setting-overwrites", + ) + tag_bindings: list[TagBinding] | None = Field( + default_factory=list, alias="tag-bindings" + ) class ProjectUpdateOptions(BaseModel): """Options for updating a project""" + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + # Optional: A name to identify the project name: str | None = None # Optional: A description for the project - description: str | None = None + description: str | None = Field(default=None, alias="description") + auto_destroy_activity_duration: str | None = Field( + default=None, + alias="auto-destroy-activity-duration", + ) + default_execution_mode: str | None = Field( + default="remote", alias="default-execution-mode" + ) + # Optional: DefaultAgentPoolID default agent pool for workspaces in the project, + # required when DefaultExecutionMode is set to `agent` + default_agent_pool_id: str | None = Field( + default=None, + alias="default-agent-pool-id", + ) + setting_overwrites: ProjectSettingOverwrites | None = Field( + default=None, + alias="setting-overwrites", + ) + tag_bindings: list[TagBinding] | None = Field( + default_factory=list, alias="tag-bindings" + ) class ProjectAddTagBindingsOptions(BaseModel): """Options for adding tag bindings to a project""" tag_bindings: list[TagBinding] = Field(default_factory=list) + + +class ProjectSettingOverwrites(BaseModel): + """Options for overwriting project settings""" + + model_config = ConfigDict(populate_by_name=True, validate_by_name=True) + + execution_mode: bool | None = Field(alias="default-execution-mode") + agent_pool: bool | None = Field(alias="default-agent-pool") diff --git a/src/pytfe/resources/projects.py b/src/pytfe/resources/projects.py index e64cb345..77f1ced0 100644 --- a/src/pytfe/resources/projects.py +++ b/src/pytfe/resources/projects.py @@ -5,10 +5,12 @@ from collections.abc import Iterator from typing import Any +from ..models.agent import AgentPool from ..models.common import ( EffectiveTagBinding, TagBinding, ) +from ..models.organization import Organization from ..models.project import ( Project, ProjectAddTagBindingsOptions, @@ -44,38 +46,56 @@ def valid_organization_name(org_name: str) -> bool: def validate_project_create_options( - organization: str, name: str, description: str | None = None + organization: str, options: ProjectCreateOptions ) -> 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): + if not valid_string(options.name): raise ValueError("Project name is required") - if not valid_project_name(name): + if not valid_project_name(options.name): raise ValueError("Project name contains invalid characters or is too long") - if description is not None and not valid_string(description): + if options.description is not None and not valid_string(options.description): raise ValueError("Description must be a valid string") + if ( + options.default_execution_mode + and options.default_execution_mode == "agent" + and not options.default_agent_pool_id + ): + raise ValueError( + "Default agent pool is required when default execution mode is set to 'agent'" + ) + def validate_project_update_options( - project_id: str, name: str | None = None, description: str | None = None + project_id: str, options: ProjectUpdateOptions ) -> 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): + if options.name is not None: + if not valid_string(options.name): raise ValueError("Project name cannot be empty") - if not valid_project_name(name): + if not valid_project_name(options.name): raise ValueError("Project name contains invalid characters or is too long") - if description is not None and not valid_string(description): + if options.description is not None and not valid_string(options.description): raise ValueError("Description must be a valid string") + if ( + options.default_execution_mode + and options.default_execution_mode == "agent" + and not options.default_agent_pool_id + ): + raise ValueError( + "Default agent pool is required when default execution mode is set to 'agent'" + ) + def validate_project_list_options( organization: str, query: str | None = None, name: str | None = None @@ -118,8 +138,6 @@ def list( params["q"] = options.query if options.name: params["filter[names]"] = options.name - if options.page_number: - params["page[number]"] = options.page_number if options.page_size: params["page[size]"] = options.page_size @@ -130,51 +148,49 @@ def list( for item in items_iter: # Extract project data - attr = item.get("attributes", {}) or {} - 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) + yield self._project_from(item) def create(self, organization: str, options: ProjectCreateOptions) -> Project: """Create a new project in an organization""" # Validate inputs - validate_project_create_options(organization, options.name, options.description) + validate_project_create_options(organization, options) path = f"/api/v2/organizations/{organization}/projects" - attributes = {"name": options.name} - if options.description: - attributes["description"] = options.description - - payload = {"data": {"type": "projects", "attributes": attributes}} + attributes = options.model_dump( + by_alias=True, + exclude_none=True, + exclude={"tag_bindings", "setting_overwrites"}, + ) + if options.setting_overwrites: + attributes["setting-overwrites"] = options.setting_overwrites.model_dump( + by_alias=True, exclude_none=True + ) + if options.tag_bindings: + relationships = {} + data = [ + { + "type": "tag-bindings", + "attributes": tag_binding.model_dump( + by_alias=True, exclude_none=True + ), + } + for tag_binding in options.tag_bindings + ] + relationships["tag-bindings"] = {"data": data} + payload = { + "data": { + "type": "projects", + "attributes": attributes, + "relationships": relationships, + } + } + else: + payload = {"data": {"type": "projects", "attributes": attributes}} response = self.t.request("POST", path, json_body=payload) data = response.json()["data"] - # Extract project data - 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) + return self._project_from(data) def read( self, project_id: str, include: builtins.list[str] | None = None @@ -196,67 +212,49 @@ def read( data = response.json()["data"] - # Extract organization from relationships - relationships = data.get("relationships", {}) - org_data = relationships.get("organization", {}).get("data", {}) - organization = _safe_str(org_data.get("id")) - - # Extract project data - 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) + return self._project_from(data) def update(self, project_id: str, options: ProjectUpdateOptions) -> Project: """Update a project's name and/or description""" # Validate inputs - validate_project_update_options(project_id, options.name, options.description) + validate_project_update_options(project_id, options) 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": attributes} - } + attributes = options.model_dump( + by_alias=True, + exclude_none=True, + exclude={"tag_bindings", "setting_overwrites"}, + ) + if options.setting_overwrites: + attributes["setting-overwrites"] = options.setting_overwrites.model_dump( + by_alias=True, exclude_none=True + ) + if options.tag_bindings: + relationships = {} + data = [ + { + "type": "tag-bindings", + "attributes": tag_binding.model_dump( + by_alias=True, exclude_none=True + ), + } + for tag_binding in options.tag_bindings + ] + relationships["tag-bindings"] = {"data": data} + payload = { + "data": { + "type": "projects", + "attributes": attributes, + "relationships": relationships, + } + } + else: + payload = {"data": {"type": "projects", "attributes": attributes}} response = self.t.request("PATCH", path, json_body=payload) data = response.json()["data"] - # Extract organization from relationships - relationships = data.get("relationships", {}) - org_data = relationships.get("organization", {}).get("data", {}) - organization = _safe_str(org_data.get("id")) - - # Extract project data - 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) + return self._project_from(data) def delete(self, project_id: str) -> None: """Delete a project""" @@ -297,7 +295,7 @@ def list_effective_tag_bindings( if not valid_string_id(project_id): raise ValueError("Project ID is required and must be valid") - path = f"/api/v2/projects/{project_id}/tag-bindings/effective" + path = f"/api/v2/projects/{project_id}/effective-tag-bindings" response = self.t.request("GET", path) data = response.json()["data"] @@ -374,5 +372,32 @@ def delete_tag_bindings(self, project_id: str) -> None: if not valid_string_id(project_id): raise ValueError("Project ID is required and must be valid") - path = f"/api/v2/projects/{project_id}/tag-bindings" - self.t.request("DELETE", path) + payload = { + "data": { + "type": "projects", + "relationships": {"tag-bindings": {"data": []}}, + } + } + + path = f"/api/v2/projects/{project_id}" + self.t.request("PATCH", path, json_body=payload) + + def _project_from(self, data: dict[str, Any]) -> Project: + """Helper method to create a Project object from API response data""" + attrs = data.get("attributes", {}) + attrs["id"] = data.get("id") + + relationships = data.get("relationships", {}) + org_data = relationships.get("organization", {}).get("data", {}) + organization = _safe_str(org_data.get("id")) if org_data else None + default_agent_pool_data = relationships.get("default-agent-pool", {}).get( + "data", {} + ) + attrs["organization"] = Organization(id=organization) if organization else None + attrs["default_agent_pool"] = ( + AgentPool(id=_safe_str(default_agent_pool_data.get("id"))) + if default_agent_pool_data + else None + ) + + return Project.model_validate(attrs) From 7b2e32db749e2ef2dbb5469eea732be0deb8cffb Mon Sep 17 00:00:00 2001 From: Sivaselvan32 Date: Tue, 24 Mar 2026 15:21:26 +0530 Subject: [PATCH 2/2] refactor(project): Updated unit testcases for project and removed the placeholder value at variablesets --- src/pytfe/resources/variable_sets.py | 2 -- tests/units/test_project.py | 39 ++++++++++++++++++++-------- tests/units/test_workspaces.py | 4 ++- 3 files changed, 31 insertions(+), 14 deletions(-) diff --git a/src/pytfe/resources/variable_sets.py b/src/pytfe/resources/variable_sets.py index da422c1f..cdd245e9 100644 --- a/src/pytfe/resources/variable_sets.py +++ b/src/pytfe/resources/variable_sets.py @@ -631,7 +631,6 @@ def _parse_variable_set(self, data: dict[str, Any]) -> VariableSet: { "id": proj["id"], "name": f"project-{proj['id']}", # Placeholder name - "organization": "placeholder-org", # Placeholder organization } ) parsed_data["projects"] = projects @@ -667,7 +666,6 @@ def _parse_variable_set(self, data: dict[str, Any]) -> VariableSet: "project": { "id": parent_data["id"], "name": f"project-{parent_data['id']}", - "organization": "placeholder-org", } } elif parent_data.get("type") == "organizations": diff --git a/tests/units/test_project.py b/tests/units/test_project.py index 801a29f8..c4832ec6 100644 --- a/tests/units/test_project.py +++ b/tests/units/test_project.py @@ -95,12 +95,12 @@ def test_list_projects_success(self): # Check first project assert result[0].id == "prj-123" assert result[0].name == "Test Project 1" - assert result[0].organization == organization + assert result[0].organization is None # Check second project assert result[1].id == "prj-456" assert result[1].name == "Test Project 2" - assert result[1].organization == organization + assert result[1].organization is None # Verify the correct API path was used expected_path = f"/api/v2/organizations/{organization}/projects" @@ -129,12 +129,18 @@ def test_create_project_success(self): assert isinstance(result, Project) assert result.id == "prj-123" assert result.name == project_name - assert result.organization == organization + assert result.organization is None # 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, + "default-execution-mode": "remote", + }, + } } self.mock_transport.request.assert_called_once_with( "POST", expected_path, json_body=expected_payload @@ -162,7 +168,8 @@ def test_read_project_success(self): assert isinstance(result, Project) assert result.id == project_id assert result.name == "Test Project" - assert result.organization == "test-org" + assert result.organization is not None + assert result.organization.id == "test-org" # Verify API call expected_path = f"/api/v2/projects/{project_id}" @@ -192,15 +199,18 @@ def test_update_project_success(self): assert isinstance(result, Project) assert result.id == project_id assert result.name == new_name - assert result.organization == "test-org" + assert result.organization is not None + assert result.organization.id == "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}, + "attributes": { + "name": new_name, + "default-execution-mode": "remote", + }, } } self.mock_transport.request.assert_called_once_with( @@ -268,7 +278,7 @@ def test_read_project_missing_organization(self): result = self.projects_service.read(project_id) - assert result.organization == "" # Should default to empty string + assert result.organization is None class TestProjectTagBindings: @@ -381,7 +391,7 @@ def test_list_effective_tag_bindings_success(self): # Verify API call self.mock_transport.request.assert_called_once_with( - "GET", f"/api/v2/projects/{self.project_id}/tag-bindings/effective" + "GET", f"/api/v2/projects/{self.project_id}/effective-tag-bindings" ) def test_list_effective_tag_bindings_invalid_project_id(self): @@ -527,7 +537,14 @@ def test_delete_tag_bindings_success(self): # Verify API call self.mock_transport.request.assert_called_once_with( - "DELETE", f"/api/v2/projects/{self.project_id}/tag-bindings" + "PATCH", + f"/api/v2/projects/{self.project_id}", + json_body={ + "data": { + "type": "projects", + "relationships": {"tag-bindings": {"data": []}}, + } + }, ) def test_delete_tag_bindings_invalid_project_id(self): diff --git a/tests/units/test_workspaces.py b/tests/units/test_workspaces.py index 77313a20..f1b81c1a 100644 --- a/tests/units/test_workspaces.py +++ b/tests/units/test_workspaces.py @@ -338,7 +338,9 @@ def test_create_workspace_with_project( sample_workspace_response ) - project = Project(id="prj-123", name="Test Project", organization="test-org") + project = Project( + id="prj-123", + ) options = WorkspaceCreateOptions(name="project-workspace", project=project)