diff --git a/tests/units/test_base_service.py b/tests/units/test_base_service.py new file mode 100644 index 00000000..5de72287 --- /dev/null +++ b/tests/units/test_base_service.py @@ -0,0 +1,426 @@ +"""Unit tests for the BaseService abstract class.""" + +from unittest.mock import Mock + +import pytest + +from tfe.base_service import BaseService + + +# Mock model class for testing +class MockResource: + """Mock resource model for testing.""" + + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def __str__(self): + return ( + f"MockResource({', '.join(f'{k}={v}' for k, v in self.__dict__.items())})" + ) + + +# Mock service implementation for testing +class MockService(BaseService[MockResource]): + """Mock service implementation for testing BaseService.""" + + def __init__(self, client): + super().__init__(client) + self.resource_type = "mock-resources" + + def list(self, **kwargs): + """List resources.""" + self._log_request("GET", "mock-resources", **kwargs) + params = self._build_query_params(**kwargs) + response = self._make_request("GET", "mock-resources", params=params) + return self._handle_response(response, MockResource) + + def get(self, resource_id: str, **kwargs): + """Get a single resource.""" + self._log_request("GET", f"mock-resources/{resource_id}", **kwargs) + params = self._build_query_params(**kwargs) + response = self._make_request( + "GET", f"mock-resources/{resource_id}", params=params + ) + return self._handle_response(response, MockResource) + + def create(self, data: dict): + """Create a resource.""" + self._log_request("POST", "mock-resources", data=data) + jsonapi_data = self._prepare_jsonapi_data(data, self.resource_type) + response = self._make_request("POST", "mock-resources", json=jsonapi_data) + return self._handle_response(response, MockResource) + + def update(self, resource_id: str, data: dict): + """Update a resource.""" + self._log_request("PATCH", f"mock-resources/{resource_id}", data=data) + jsonapi_data = self._prepare_jsonapi_data(data, self.resource_type) + response = self._make_request( + "PATCH", f"mock-resources/{resource_id}", json=jsonapi_data + ) + return self._handle_response(response, MockResource) + + def delete(self, resource_id: str): + """Delete a resource.""" + self._log_request("DELETE", f"mock-resources/{resource_id}") + response = self._make_request("DELETE", f"mock-resources/{resource_id}") + return response.status_code == 204 + + +@pytest.fixture +def mock_client(): + """Provide a mock client for testing.""" + client = Mock() + client.base_url = "https://app.terraform.io/api/v2/" + return client + + +@pytest.fixture +def mock_http_client(): + """Provide a mock HTTP client for testing.""" + return Mock() + + +@pytest.fixture +def mock_service(mock_client, mock_http_client): + """Provide a mock service instance for testing.""" + mock_client.config.http_client = mock_http_client + return MockService(mock_client) + + +class TestBaseService: + """Test the BaseService abstract class.""" + + def test_initialization(self, mock_client): + """Test service initialization.""" + service = MockService(mock_client) + assert service.client == mock_client + assert service.resource_type == "mock-resources" + + def test_make_request_get(self, mock_service, mock_http_client): + """Test GET request handling.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_http_client.get.return_value = mock_response + + # Make request + response = mock_service._make_request("GET", "test-endpoint") + + # Verify + assert response == mock_response + mock_http_client.get.assert_called_once_with( + "https://app.terraform.io/api/v2/test-endpoint" + ) + + def test_make_request_post(self, mock_service, mock_http_client): + """Test POST request handling.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 201 + mock_http_client.post.return_value = mock_response + + # Make request + response = mock_service._make_request( + "POST", "test-endpoint", json={"test": "data"} + ) + + # Verify + assert response == mock_response + mock_http_client.post.assert_called_once_with( + "https://app.terraform.io/api/v2/test-endpoint", json={"test": "data"} + ) + + def test_make_request_put(self, mock_service, mock_http_client): + """Test PUT request handling.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_http_client.put.return_value = mock_response + + # Make request + response = mock_service._make_request( + "PUT", "test-endpoint", json={"test": "data"} + ) + + # Verify + assert response == mock_response + mock_http_client.put.assert_called_once_with( + "https://app.terraform.io/api/v2/test-endpoint", json={"test": "data"} + ) + + def test_make_request_patch(self, mock_service, mock_http_client): + """Test PATCH request handling.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_http_client.patch.return_value = mock_response + + # Make request + response = mock_service._make_request( + "PATCH", "test-endpoint", json={"test": "data"} + ) + + # Verify + assert response == mock_response + mock_http_client.patch.assert_called_once_with( + "https://app.terraform.io/api/v2/test-endpoint", json={"test": "data"} + ) + + def test_make_request_delete(self, mock_service, mock_http_client): + """Test DELETE request handling.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 204 + mock_http_client.delete.return_value = mock_response + + # Make request + response = mock_service._make_request("DELETE", "test-endpoint") + + # Verify + assert response == mock_response + mock_http_client.delete.assert_called_once_with( + "https://app.terraform.io/api/v2/test-endpoint" + ) + + def test_make_request_invalid_method(self, mock_service): + """Test invalid HTTP method handling.""" + with pytest.raises(ValueError, match="Unsupported HTTP method: INVALID"): + mock_service._make_request("INVALID", "test-endpoint") + + def test_make_request_url_building(self, mock_service, mock_http_client): + """Test URL building with different path combinations.""" + mock_response = Mock() + mock_response.status_code = 200 + mock_http_client.get.return_value = mock_response + + # Test with leading slash in path + mock_service._make_request("GET", "/test-endpoint") + mock_http_client.get.assert_called_with( + "https://app.terraform.io/api/v2/test-endpoint" + ) + + # Test with trailing slash in base_url + mock_service.client.base_url = "https://app.terraform.io/api/v2/" + mock_service._make_request("GET", "test-endpoint") + mock_http_client.get.assert_called_with( + "https://app.terraform.io/api/v2/test-endpoint" + ) + + def test_build_query_params(self, mock_service): + """Test query parameter building.""" + params = mock_service._build_query_params(page=1, per_page=10, search="test") + assert params["page"] == "1" + assert params["per_page"] == "10" + assert params["search"] == "test" + + def test_prepare_jsonapi_data(self, mock_service): + """Test JSONAPI data preparation.""" + data = {"name": "test", "description": "test description"} + jsonapi_data = mock_service._prepare_jsonapi_data(data, "workspaces") + + assert jsonapi_data["data"]["type"] == "workspaces" + assert jsonapi_data["data"]["attributes"]["name"] == "test" + + def test_handle_response_no_content(self, mock_service): + """Test handling of 204 No Content responses.""" + mock_response = Mock() + mock_response.status_code = 204 + + result = mock_service._handle_response(mock_response) + assert result is None + + def test_handle_response_jsonapi(self, mock_service): + """Test handling of JSONAPI responses.""" + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + } + } + + result = mock_service._handle_response(mock_response, MockResource) + assert isinstance(result, MockResource) + assert result.id == "ws-123" + assert result.name == "test-workspace" + + def test_handle_response_regular_json(self, mock_service): + """Test handling of regular JSON responses.""" + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = {"name": "test", "id": "123"} + + result = mock_service._handle_response(mock_response, MockResource) + assert isinstance(result, MockResource) + assert result.name == "test" + assert result.id == "123" + + def test_handle_response_parse_error(self, mock_service): + """Test handling of JSON parse errors.""" + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.side_effect = ValueError("Invalid JSON") + mock_response.text = "raw response text" + + result = mock_service._handle_response(mock_response) + assert result == "raw response text" + + def test_log_request(self, mock_service, caplog): + """Test logging functionality.""" + with caplog.at_level("DEBUG"): + mock_service._log_request("GET", "test-endpoint", page=1) + + assert "Making GET request to test-endpoint" in caplog.text + assert "Request parameters: {'page': 1}" in caplog.text + + +class TestMockService: + """Test the concrete mock service implementation.""" + + def test_list_resources(self, mock_service, mock_http_client): + """Test listing resources.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "data": [ + { + "id": "ws-1", + "type": "workspaces", + "attributes": {"name": "workspace-1"}, + } + ] + } + mock_http_client.get.return_value = mock_response + + # Call service method + result = mock_service.list(page=1, per_page=10) + + # Verify + assert len(result) == 1 + assert result[0].id == "ws-1" + assert result[0].name == "workspace-1" + + # Verify HTTP call + mock_http_client.get.assert_called_once() + call_args = mock_http_client.get.call_args + assert call_args[0][0] == "https://app.terraform.io/api/v2/mock-resources" + assert call_args[1]["params"] == {"page": "1", "per_page": "10"} + + def test_get_resource(self, mock_service, mock_http_client): + """Test getting a single resource.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + } + } + mock_http_client.get.return_value = mock_response + + # Call service method + result = mock_service.get("ws-123") + + # Verify + assert result.id == "ws-123" + assert result.name == "test-workspace" + + # Verify HTTP call + mock_http_client.get.assert_called_once_with( + "https://app.terraform.io/api/v2/mock-resources/ws-123", params={} + ) + + def test_create_resource(self, mock_service, mock_http_client): + """Test creating a resource.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 201 + mock_response.json.return_value = { + "data": { + "id": "ws-new", + "type": "workspaces", + "attributes": {"name": "new-workspace"}, + } + } + mock_http_client.post.return_value = mock_response + + # Call service method + data = {"name": "new-workspace"} + result = mock_service.create(data) + + # Verify + assert result.id == "ws-new" + assert result.name == "new-workspace" + + # Verify HTTP call + mock_http_client.post.assert_called_once() + call_args = mock_http_client.post.call_args + assert call_args[1]["json"]["data"]["type"] == "mock-resources" + assert call_args[1]["json"]["data"]["attributes"]["name"] == "new-workspace" + + def test_update_resource(self, mock_service, mock_http_client): + """Test updating a resource.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "updated-workspace"}, + } + } + mock_http_client.patch.return_value = mock_response + + # Call service method + data = {"name": "updated-workspace"} + result = mock_service.update("ws-123", data) + + # Verify + assert result.id == "ws-123" + assert result.name == "updated-workspace" + + # Verify HTTP call + mock_http_client.patch.assert_called_once() + call_args = mock_http_client.patch.call_args + assert ( + call_args[0][0] == "https://app.terraform.io/api/v2/mock-resources/ws-123" + ) + + def test_delete_resource(self, mock_service, mock_http_client): + """Test deleting a resource.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 204 + mock_http_client.delete.return_value = mock_response + + # Call service method + result = mock_service.delete("ws-123") + + # Verify + assert result is True + + # Verify HTTP call + mock_http_client.delete.assert_called_once_with( + "https://app.terraform.io/api/v2/mock-resources/ws-123" + ) + + def test_delete_resource_failure(self, mock_service, mock_http_client): + """Test deleting a resource with failure response.""" + # Mock response + mock_response = Mock() + mock_response.status_code = 404 + mock_http_client.delete.return_value = mock_response + + # Call service method + result = mock_service.delete("ws-123") + + # Verify + assert result is False diff --git a/tests/units/test_utils.py b/tests/units/test_utils.py new file mode 100644 index 00000000..331715df --- /dev/null +++ b/tests/units/test_utils.py @@ -0,0 +1,413 @@ +"""Unit tests for the utility functions.""" + +from tfe.utils import build_query_params, deserialize_jsonapi, prepare_jsonapi_data + + +# Mock model class for testing +class MockResource: + """Mock resource model for testing.""" + + def __init__(self, **kwargs): + for key, value in kwargs.items(): + setattr(self, key, value) + + def __str__(self): + return ( + f"MockResource({', '.join(f'{k}={v}' for k, v in self.__dict__.items())})" + ) + + +class TestBuildQueryParams: + """Test the build_query_params function.""" + + def test_basic_parameters(self): + """Test building query parameters with basic values.""" + params = build_query_params(page=1, per_page=10, search="test") + + assert params["page"] == "1" + assert params["per_page"] == "10" + assert params["search"] == "test" + assert len(params) == 3 + + def test_none_values_filtered(self): + """Test that None values are filtered out.""" + params = build_query_params(page=1, per_page=None, search="test", filter=None) + + assert params["page"] == "1" + assert params["search"] == "test" + assert "per_page" not in params + assert "filter" not in params + assert len(params) == 2 + + def test_list_values(self): + """Test handling of list values.""" + params = build_query_params(include=["users", "teams"], tags=["dev", "prod"]) + + assert params["include"] == ["users", "teams"] + assert params["tags"] == ["dev", "prod"] + assert len(params) == 2 + + def test_tuple_values(self): + """Test handling of tuple values.""" + params = build_query_params(include=("users", "teams")) + + assert params["include"] == ("users", "teams") + assert len(params) == 1 + + def test_empty_parameters(self): + """Test with no parameters.""" + params = build_query_params() + + assert params == {} + assert len(params) == 0 + + def test_mixed_types(self): + """Test with mixed parameter types.""" + params = build_query_params( + page=1, + per_page=25, + search="terraform", + active=True, + include=["users"], + tags=None, + filter="", + ) + + assert params["page"] == "1" + assert params["per_page"] == "25" + assert params["search"] == "terraform" + assert params["active"] == "True" + assert params["include"] == ["users"] + assert params["filter"] == "" + assert "tags" not in params + assert len(params) == 6 + + def test_zero_values(self): + """Test that zero values are included.""" + params = build_query_params(page=0, per_page=0, search="") + + assert params["page"] == "0" + assert params["per_page"] == "0" + assert params["search"] == "" + assert len(params) == 3 + + +class TestPrepareJsonapiData: + """Test the prepare_jsonapi_data function.""" + + def test_basic_data(self): + """Test preparing basic data for JSONAPI format.""" + data = {"name": "test-workspace", "description": "A test workspace"} + jsonapi_data = prepare_jsonapi_data(data, "workspaces") + + assert jsonapi_data["data"]["type"] == "workspaces" + assert jsonapi_data["data"]["attributes"]["name"] == "test-workspace" + assert jsonapi_data["data"]["attributes"]["description"] == "A test workspace" + assert "id" not in jsonapi_data["data"] + + def test_with_id(self): + """Test that ID is moved to top level.""" + data = {"id": "ws-123", "name": "test-workspace"} + jsonapi_data = prepare_jsonapi_data(data, "workspaces") + + assert jsonapi_data["data"]["id"] == "ws-123" + assert jsonapi_data["data"]["type"] == "workspaces" + assert jsonapi_data["data"]["attributes"]["name"] == "test-workspace" + assert "id" not in jsonapi_data["data"]["attributes"] + + def test_empty_data(self): + """Test with empty data dictionary.""" + data = {} + jsonapi_data = prepare_jsonapi_data(data, "workspaces") + + assert jsonapi_data["data"]["type"] == "workspaces" + assert jsonapi_data["data"]["attributes"] == {} + assert "id" not in jsonapi_data["data"] + + def test_nested_data(self): + """Test with nested data structures.""" + data = { + "name": "test-workspace", + "settings": {"execution_mode": "remote", "auto_apply": True}, + "tags": ["dev", "test"], + } + jsonapi_data = prepare_jsonapi_data(data, "workspaces") + + assert jsonapi_data["data"]["type"] == "workspaces" + assert jsonapi_data["data"]["attributes"]["name"] == "test-workspace" + assert ( + jsonapi_data["data"]["attributes"]["settings"]["execution_mode"] == "remote" + ) + assert jsonapi_data["data"]["attributes"]["settings"]["auto_apply"] is True + assert jsonapi_data["data"]["attributes"]["tags"] == ["dev", "test"] + + def test_complex_resource_type(self): + """Test with complex resource type names.""" + data = {"name": "test"} + jsonapi_data = prepare_jsonapi_data(data, "terraform-runs") + + assert jsonapi_data["data"]["type"] == "terraform-runs" + assert jsonapi_data["data"]["attributes"]["name"] == "test" + + def test_boolean_values(self): + """Test handling of boolean values.""" + data = {"auto_apply": True, "locked": False} + jsonapi_data = prepare_jsonapi_data(data, "workspaces") + + assert jsonapi_data["data"]["attributes"]["auto_apply"] is True + assert jsonapi_data["data"]["attributes"]["locked"] is False + + +class TestDeserializeJsonapi: + """Test the deserialize_jsonapi function.""" + + def test_single_resource(self): + """Test deserializing a single resource.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": { + "name": "test-workspace", + "description": "A test workspace", + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert isinstance(result, MockResource) + assert result.id == "ws-123" + assert result.name == "test-workspace" + assert result.description == "A test workspace" + + def test_list_of_resources(self): + """Test deserializing a list of resources.""" + jsonapi_data = { + "data": [ + { + "id": "ws-1", + "type": "workspaces", + "attributes": {"name": "workspace-1"}, + }, + { + "id": "ws-2", + "type": "workspaces", + "attributes": {"name": "workspace-2"}, + }, + ] + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert isinstance(result, list) + assert len(result) == 2 + + assert result[0].id == "ws-1" + assert result[0].name == "workspace-1" + assert result[1].id == "ws-2" + assert result[1].name == "workspace-2" + + def test_with_relationships(self): + """Test deserializing resources with relationships.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + "relationships": { + "organization": { + "data": {"id": "org-456", "type": "organizations"} + }, + "teams": { + "data": [ + {"id": "team-1", "type": "teams"}, + {"id": "team-2", "type": "teams"}, + ] + }, + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert result.organization_id == "org-456" + assert result.teams_ids == ["team-1", "team-2"] + + def test_single_relationship(self): + """Test deserializing with single relationship.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + "relationships": { + "organization": {"data": {"id": "org-456", "type": "organizations"}} + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert result.organization_id == "org-456" + assert not hasattr(result, "teams_ids") + + def test_empty_relationships(self): + """Test deserializing with empty relationships.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + "relationships": {}, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert result.id == "ws-123" + assert result.name == "test-workspace" + # No relationship attributes should be added + + def test_no_attributes(self): + """Test deserializing resource without attributes.""" + jsonapi_data = {"data": {"id": "ws-123", "type": "workspaces"}} + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + # Should return the data as-is since no attributes + assert result == jsonapi_data["data"] + + def test_non_jsonapi_data(self): + """Test handling of non-JSONAPI data.""" + regular_data = {"name": "test", "id": "123"} + result = deserialize_jsonapi(regular_data, MockResource) + + # Should return as-is + assert result == regular_data + + def test_none_data(self): + """Test handling of None data.""" + result = deserialize_jsonapi(None, MockResource) + + assert result is None + + def test_empty_data(self): + """Test handling of empty data.""" + empty_data = {} + result = deserialize_jsonapi(empty_data, MockResource) + + # Should return None for empty data + assert result is None + + def test_empty_data_list(self): + """Test handling of empty data list.""" + empty_list_data = {"data": []} + result = deserialize_jsonapi(empty_list_data, MockResource) + + assert result == [] + + def test_malformed_relationships(self): + """Test handling of malformed relationship data.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + "relationships": { + "organization": { + "data": "invalid-data" # Should be dict or list + } + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + # Should still create the resource, but relationship handling may fail + assert result.name == "test-workspace" + + def test_relationship_without_id(self): + """Test relationship data without ID.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": {"name": "test-workspace"}, + "relationships": { + "organization": { + "data": {"type": "organizations"} # No ID + } + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert result.name == "test-workspace" + # No organization_id should be set since no ID in relationship data + + +class TestDeserializeSingleResource: + """Test the deserialize_single_resource function indirectly through deserialize_jsonapi.""" + + def test_resource_without_id(self): + """Test resource without ID attribute.""" + jsonapi_data = { + "data": {"type": "workspaces", "attributes": {"name": "test-workspace"}} + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert result.name == "test-workspace" + assert not hasattr(result, "id") + + def test_resource_with_extra_attributes(self): + """Test resource with extra attributes.""" + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": { + "name": "test-workspace", + "created_at": "2023-01-01T00:00:00Z", + "updated_at": "2023-01-02T00:00:00Z", + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, MockResource) + + assert result.id == "ws-123" + assert result.name == "test-workspace" + assert result.created_at == "2023-01-01T00:00:00Z" + assert result.updated_at == "2023-01-02T00:00:00Z" + + def test_model_creation_failure_fallback(self): + """Test fallback when model creation fails.""" + + # Create a model class that will fail on certain attributes + class FailingModel: + def __init__(self, **kwargs): + if "failing_attr" in kwargs: + raise ValueError("Model creation failed") + for key, value in kwargs.items(): + setattr(self, key, value) + + jsonapi_data = { + "data": { + "id": "ws-123", + "type": "workspaces", + "attributes": { + "name": "test-workspace", + "failing_attr": "this will cause failure", + }, + } + } + + result = deserialize_jsonapi(jsonapi_data, FailingModel) + + # Should return attributes dict as fallback + assert isinstance(result, dict) + assert result["name"] == "test-workspace" + assert result["failing_attr"] == "this will cause failure" diff --git a/tfe/__init__.py b/tfe/__init__.py index 60fbccb2..0c252b9a 100644 --- a/tfe/__init__.py +++ b/tfe/__init__.py @@ -6,7 +6,8 @@ workspaces, runs, state files, and other TFE/TFC resources. """ +from tfe.base_service import BaseService from tfe.client import Client, TFEClientError from tfe.config import Config -__all__ = ["Client", "TFEClientError", "Config"] +__all__ = ["Client", "TFEClientError", "Config", "BaseService"] diff --git a/tfe/base_service.py b/tfe/base_service.py new file mode 100644 index 00000000..4a239e95 --- /dev/null +++ b/tfe/base_service.py @@ -0,0 +1,139 @@ +""" +Base service class for Terraform Enterprise/Cloud API services. + +This module provides an abstract base class that all TFE API services +should inherit from. It provides common functionality for HTTP requests, +response handling, and JSONAPI processing. +""" + +import logging +from abc import ABC, abstractmethod +from typing import Any, Generic, TypeVar + +from tfe.utils import build_query_params, deserialize_jsonapi, prepare_jsonapi_data + +logger = logging.getLogger(__name__) + +# Generic type for model classes +T = TypeVar("T") + + +class BaseService(ABC, Generic[T]): + """ + Abstract base class for all TFE API services. + + Generic type T represents the model class that this service manages. + For example, OrganizationsService would be BaseService[Organization]. + """ + + def __init__(self, client: Any) -> None: + """client: The main TFE client instance that provides HTTP access""" + self.client = client + + def _make_request(self, method: str, path: str, **kwargs: Any) -> Any: + """ + Make an HTTP request using the client's HTTP client. + + Args: + method: HTTP method (GET, POST, PUT, PATCH, DELETE) + path: API path (will be joined with base_url) + **kwargs: Additional arguments to pass to the HTTP client + + Returns: + The HTTP response + """ + # Build full URL + url = self.client.base_url.rstrip("/") + "/" + path.lstrip("/") + + # Get the HTTP client from the config + http_client = self.client.config.http_client + + # Log the request + self._log_request(method, path, **kwargs) + method = method.upper() + + # Make the request + if method == "GET": + return http_client.get(url, **kwargs) + elif method == "POST": + return http_client.post(url, **kwargs) + elif method == "PUT": + return http_client.put(url, **kwargs) + elif method == "PATCH": + return http_client.patch(url, **kwargs) + elif method == "DELETE": + return http_client.delete(url, **kwargs) + else: + raise ValueError(f"Unsupported HTTP method: {method}") + + def _handle_response(self, response: Any, model_class: type | None = None) -> Any: + """ + Handle the API response and optionally deserialize to a model. + + Args: + response: The HTTP response from the API + model_class: Optional model class to deserialize the response to + + Returns: + The deserialized response or raw response data + """ + if response.status_code == 204: # No Content + return None + + try: + data = response.json() + + if model_class and data: + # Handle JSONAPI format + if "data" in data: + return deserialize_jsonapi(data, model_class) + else: + # Handle regular JSON response + return model_class(**data) if isinstance(data, dict) else data + + return data + + except ValueError as e: + logger.warning(f"Failed to parse JSON response: {e}") + return response.text + + def _prepare_jsonapi_data( + self, data: dict[str, Any], resource_type: str + ) -> dict[str, Any]: + """Prepare data for JSONAPI format.""" + return prepare_jsonapi_data(data, resource_type) + + def _build_query_params(self, **kwargs: Any) -> dict[str, Any]: + """Build query parameters for API requests.""" + return build_query_params(**kwargs) + + def _log_request(self, method: str, path: str, **kwargs: Any) -> None: + """Log API request details for debugging.""" + logger.debug(f"Making {method} request to {path}") + if kwargs: + logger.debug(f"Request parameters: {kwargs}") + + @abstractmethod + def list(self, **kwargs: Any) -> list[T]: + """List resources instances.""" + pass + + @abstractmethod + def get(self, resource_id: str, **kwargs: Any) -> T | None: + """Get a single resource by ID.""" + pass + + @abstractmethod + def create(self, data: dict[str, Any]) -> T: + """Create a new resource.""" + pass + + @abstractmethod + def update(self, resource_id: str, data: dict[str, Any]) -> T: + """Update an existing resource.""" + pass + + @abstractmethod + def delete(self, resource_id: str) -> bool: + """Delete a resource.""" + pass diff --git a/tfe/utils.py b/tfe/utils.py new file mode 100644 index 00000000..2c85ebdc --- /dev/null +++ b/tfe/utils.py @@ -0,0 +1,86 @@ +""" +Utility functions for Terraform Enterprise/Cloud API. + +This module provides utility functions for common operations +like JSONAPI handling and response processing. +""" + +from typing import Any + + +def deserialize_jsonapi(data: dict[str, Any], model_class: type) -> Any: + """Deserialize JSONAPI response to a model instance.""" + if not data: + return None + + if "data" not in data: + return data + + jsonapi_data = data["data"] + + if isinstance(jsonapi_data, list): + # Handle list of resources + return [deserialize_single_resource(item, model_class) for item in jsonapi_data] + else: + # Handle single resource + return deserialize_single_resource(jsonapi_data, model_class) + + +def deserialize_single_resource( + resource_data: dict[str, Any], model_class: type +) -> Any: + """Deserialize a single JSONAPI resource to a model instance.""" + if "attributes" not in resource_data: + return resource_data + + # Extract attributes and ID + attributes = resource_data.get("attributes", {}) + resource_id = resource_data.get("id") + + # Add ID to attributes if it exists + if resource_id: + attributes["id"] = resource_id + + # Handle relationships if they exist + relationships = resource_data.get("relationships", {}) + for rel_name, rel_data in relationships.items(): + if "data" in rel_data: + rel_value = rel_data["data"] + if isinstance(rel_value, dict) and "id" in rel_value: + attributes[f"{rel_name}_id"] = rel_value["id"] + elif isinstance(rel_value, list): + attributes[f"{rel_name}_ids"] = [ + item.get("id") for item in rel_value if "id" in item + ] + + try: + return model_class(**attributes) + except Exception: + # If model creation fails, return the attributes + return attributes + + +def prepare_jsonapi_data(data: dict[str, Any], resource_type: str) -> dict[str, Any]: + """Prepare data for JSONAPI format.""" + jsonapi_data = {"data": {"type": resource_type, "attributes": {}}} + + # Extract ID if present + if "id" in data: + jsonapi_data["data"]["id"] = data.pop("id") + + # Move all remaining data to attributes + jsonapi_data["data"]["attributes"] = data + + return jsonapi_data + + +def build_query_params(**kwargs: Any) -> dict[str, Any]: + """Build query parameters for API requests.""" + params: dict[str, Any] = {} + for key, value in kwargs.items(): + if value is not None: + if isinstance(value, list | tuple): + params[key] = value + else: + params[key] = str(value) + return params