From 653af8f33be80bd4e62ed8252a353d09eae12a68 Mon Sep 17 00:00:00 2001 From: Brian Summa Date: Tue, 13 May 2025 14:29:39 -0500 Subject: [PATCH 1/2] feat(assets): add asset history API methods - Add fetch method to retrieve paginated asset list - Add fetch_asset_history_entities to get history for an asset - Add create_history_entity to record new history events - Implement comprehensive test suite for all new methods --- pythonik/specs/assets.py | 71 +++++++++- pythonik/tests/test_assets_history.py | 185 ++++++++++++++++++++++++++ 2 files changed, 255 insertions(+), 1 deletion(-) create mode 100644 pythonik/tests/test_assets_history.py diff --git a/pythonik/specs/assets.py b/pythonik/specs/assets.py index a048c82..c6a47f4 100644 --- a/pythonik/specs/assets.py +++ b/pythonik/specs/assets.py @@ -12,7 +12,7 @@ AssetVersionFromAssetCreate, AssetVersion, ) -from pythonik.models.base import Response +from pythonik.models.base import Response, PaginatedResponse from pythonik.specs.base import Spec from pythonik.specs.collection import CollectionSpec @@ -511,3 +511,72 @@ def delete_version( VERSION_URL.format(asset_id, version_id), params=params, **kwargs ) return self.parse_response(response, None) + + def fetch(self, **kwargs) -> Response: + """ + Get list of assets. + + Args: + **kwargs: Additional kwargs to pass to the request + + Returns: + Response(model=PaginatedResponse) containing paginated asset list + """ + resp = self._get(self.gen_url("assets/"), **kwargs) + return self.parse_response(resp, PaginatedResponse) + + def fetch_asset_history_entities(self, asset_id: str, **kwargs) -> Response: + """ + Get list of history entities for asset. + + Args: + asset_id: ID of the asset + **kwargs: Additional kwargs to pass to the request + + Returns: + Response(model=PaginatedResponse) containing history entities + """ + resp = self._get(self.gen_url(f"assets/{asset_id}/history/"), **kwargs) + return self.parse_response(resp, PaginatedResponse) + + def create_history_entity( + self, asset_id: str, operation_description: str, + operation_type: str, + **kwargs + ) -> Response: + """ + Create an asset history entity. + + Args: + asset_id: ID of the asset + operation_description: Description of the operation + operation_type: Type of operation (e.g., VERSION_CREATE, ADD_FORMAT) + **kwargs: Additional kwargs to pass to the request + + Returns: + Response with history entry creation status + + Raises: + ValueError: If operation_type is not a valid operation type + """ + operation_types = [ + "EXPORT", "TRANSCODE", "ANALYZE", "ADD_FORMAT", "DELETE_FORMAT", + "RESTORE_FORMAT", "DELETE_FILESET", "DELETE_FILE", + "RESTORE_FILESET", "MODIFY_FILESET", "APPROVE", "REJECT", + "DOWNLOAD", "METADATA", "CUSTOM", "TRANSCRIPTION", "VERSION_CREATE", + "VERSION_DELETE", "VERSION_UPDATE", "VERSION_PROMOTE", "RESTORE", + "RESTORE_FROM_GLACIER", "ARCHIVE", "RESTORE_ARCHIVE", "DELETE", + "TRANSFER", "UNLINK_SUBCLIP", "FACE_RECOGNITION" + ] + if operation_type not in operation_types: + raise ValueError( + f"operation_type must be one of: {'|'.join(operation_types)}" + ) + body = { + "operation_description": operation_description, + "operation_type": operation_type + } + resp = self._post( + self.gen_url(f"assets/{asset_id}/history/"), json=body, **kwargs + ) + return self.parse_response(resp, None) \ No newline at end of file diff --git a/pythonik/tests/test_assets_history.py b/pythonik/tests/test_assets_history.py new file mode 100644 index 0000000..d3a191f --- /dev/null +++ b/pythonik/tests/test_assets_history.py @@ -0,0 +1,185 @@ +# pythonik/tests/test_assets_history.py +import uuid +import pytest +import requests_mock + +from pythonik.client import PythonikClient +from pythonik.models.base import PaginatedResponse +from pythonik.specs.assets import AssetSpec + + +def test_fetch(): + """Test fetching a list of assets.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + + # Mock response data structure + response_data = { + "objects": [ + {"id": str(uuid.uuid4()), "title": "Test Asset 1", "status": "ACTIVE"}, + {"id": str(uuid.uuid4()), "title": "Test Asset 2", "status": "ACTIVE"}, + ], + "page": 1, + "pages": 1, + "per_page": 2, + "total": 2, + } + + # Setup mock endpoint + mock_address = AssetSpec.gen_url("assets/") + m.get(mock_address, json=response_data) + + # Create client and call the method + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.assets().fetch() + + # Verify response + assert result.response.ok + assert isinstance(result.data, PaginatedResponse) + assert len(result.data.objects) == 2 + assert result.data.page == 1 + assert result.data.pages == 1 + assert result.data.total == 2 + + +def test_fetch_with_params(): + """Test fetching a list of assets with query parameters.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + + # Mock response data + response_data = { + "objects": [ + {"id": str(uuid.uuid4()), "title": "Test Asset 1", "status": "ACTIVE"} + ], + "page": 1, + "pages": 1, + "per_page": 1, + "total": 1, + } + + # Setup mock endpoint with params matcher + mock_address = AssetSpec.gen_url("assets/") + m.get( + mock_address, + json=response_data, + # Add request matcher to ensure params are passed correctly + additional_matcher=lambda req: req.qs == {"page": ["1"], "per_page": ["1"]}, + ) + + # Create client and call method with params + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + params = {"page": 1, "per_page": 1} + result = client.assets().fetch(params=params) + + # Verify response + assert result.response.ok + assert isinstance(result.data, PaginatedResponse) + assert len(result.data.objects) == 1 + + +def test_fetch_asset_history_entities(): + """Test fetching history entities for an asset.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + + # Mock response data + response_data = { + "objects": [ + { + "id": str(uuid.uuid4()), + "operation_type": "METADATA", + "operation_description": "Updated metadata", + "date_created": "2025-05-13T10:00:00Z", + "created_by_user": "user123", + }, + { + "id": str(uuid.uuid4()), + "operation_type": "VERSION_CREATE", + "operation_description": "Created new version", + "date_created": "2025-05-12T15:30:00Z", + "created_by_user": "user123", + }, + ], + "page": 1, + "pages": 1, + "per_page": 10, + "total": 2, + } + + # Setup mock endpoint + mock_address = AssetSpec.gen_url(f"assets/{asset_id}/history/") + m.get(mock_address, json=response_data) + + # Create client and call the method + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.assets().fetch_asset_history_entities(asset_id) + + # Verify response + assert result.response.ok + assert isinstance(result.data, PaginatedResponse) + assert len(result.data.objects) == 2 + assert result.data.objects[0]["operation_type"] == "METADATA" + assert result.data.objects[1]["operation_type"] == "VERSION_CREATE" + + +def test_create_history_entity(): + """Test creating a history entity for an asset.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + operation_description = "Test history entry" + operation_type = "CUSTOM" + + # Setup mock endpoint + mock_address = AssetSpec.gen_url(f"assets/{asset_id}/history/") + m.post(mock_address, status_code=201) + + # Create client and call the method + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.assets().create_history_entity( + asset_id=asset_id, + operation_description=operation_description, + operation_type=operation_type, + ) + + # Verify response + assert result.response.ok + assert result.response.status_code == 201 + + # Verify the correct request body was sent + assert m.last_request.json() == { + "operation_description": operation_description, + "operation_type": operation_type, + } + + +def test_create_history_entity_with_invalid_operation_type(): + """Test creating a history entity with an invalid operation type.""" + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + operation_description = "Test history entry" + operation_type = "INVALID_TYPE" + + # Create client + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + + # Check that an error is raised for invalid operation type + with pytest.raises(ValueError) as excinfo: + client.assets().create_history_entity( + asset_id=asset_id, + operation_description=operation_description, + operation_type=operation_type, + ) + + # Verify the error message + assert "operation_type must be one of:" in str(excinfo.value) + assert "EXPORT" in str(excinfo.value) + assert "CUSTOM" in str(excinfo.value) + assert "VERSION_CREATE" in str(excinfo.value) From d193a423f65ee3075cc6ea33e95f2140d79cd862 Mon Sep 17 00:00:00 2001 From: Brian Summa Date: Tue, 20 May 2025 15:45:34 -0500 Subject: [PATCH 2/2] Refactor to use `list_` prefix instead of `fetch_` when getting a list of objects. Refactor `fetch()` be `list_all()` to avoid conflict with `list` built-in. --- pythonik/specs/assets.py | 4 ++-- pythonik/tests/test_assets_history.py | 6 +++--- 2 files changed, 5 insertions(+), 5 deletions(-) diff --git a/pythonik/specs/assets.py b/pythonik/specs/assets.py index c6a47f4..41bca90 100644 --- a/pythonik/specs/assets.py +++ b/pythonik/specs/assets.py @@ -512,7 +512,7 @@ def delete_version( ) return self.parse_response(response, None) - def fetch(self, **kwargs) -> Response: + def list_all(self, **kwargs) -> Response: """ Get list of assets. @@ -525,7 +525,7 @@ def fetch(self, **kwargs) -> Response: resp = self._get(self.gen_url("assets/"), **kwargs) return self.parse_response(resp, PaginatedResponse) - def fetch_asset_history_entities(self, asset_id: str, **kwargs) -> Response: + def list_asset_history_entities(self, asset_id: str, **kwargs) -> Response: """ Get list of history entities for asset. diff --git a/pythonik/tests/test_assets_history.py b/pythonik/tests/test_assets_history.py index d3a191f..5089c58 100644 --- a/pythonik/tests/test_assets_history.py +++ b/pythonik/tests/test_assets_history.py @@ -32,7 +32,7 @@ def test_fetch(): # Create client and call the method client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) - result = client.assets().fetch() + result = client.assets().list_all() # Verify response assert result.response.ok @@ -72,7 +72,7 @@ def test_fetch_with_params(): # Create client and call method with params client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) params = {"page": 1, "per_page": 1} - result = client.assets().fetch(params=params) + result = client.assets().list_all(params=params) # Verify response assert result.response.ok @@ -117,7 +117,7 @@ def test_fetch_asset_history_entities(): # Create client and call the method client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) - result = client.assets().fetch_asset_history_entities(asset_id) + result = client.assets().list_asset_history_entities(asset_id) # Verify response assert result.response.ok