diff --git a/pythonik/specs/files.py b/pythonik/specs/files.py index 49f98e0..76ad8fb 100644 --- a/pythonik/specs/files.py +++ b/pythonik/specs/files.py @@ -2,7 +2,7 @@ from xml.dom.minidom import parseString from functools import wraps import warnings -from typing import Union, Dict, Any +from typing import Union, Dict, Any, Optional, Literal import requests @@ -12,7 +12,7 @@ S3_UPLOADID_KEY, ) from pythonik.exceptions import UnexpectedStorageMethodForProxy -from pythonik.models.base import Response, StorageMethod +from pythonik.models.base import Response, StorageMethod, PaginatedResponse from pythonik.models.files.file import ( File, FileSetsFilesResponse, @@ -256,7 +256,7 @@ def partial_update_keyframe( exclude_defaults: bool = True, **kwargs, ) -> Response: - "Partially update an asset keyframe using PATCH" + """Partially update an asset keyframe using PATCH""" json_data = self._prepare_model_data(body, exclude_defaults=exclude_defaults) response = self._patch( GET_ASSET_KEYFRAME.format(asset_id, keyframe_id), @@ -273,7 +273,7 @@ def update_keyframe( exclude_defaults: bool = True, **kwargs, ) -> Response: - "Update an asset keyframe using POST" + """Update an asset keyframe using POST""" json_data = self._prepare_model_data(body, exclude_defaults=exclude_defaults) response = self._post( GET_ASSET_KEYFRAME.format(asset_id, keyframe_id), @@ -867,3 +867,171 @@ def get_asset_files_by_version( **kwargs, ) return self.parse_response(response, Files) + + def create_storage_file( + self, + storage_id: str, + body: Union[File, Dict[str, Any]], + exclude_defaults: bool = True, + **kwargs + ) -> Response: + """ + Create file without associating it to an asset. + + Args: + storage_id: The ID of the storage to retrieve + body: Storage file creation parameters, either as File model or dict + exclude_defaults: Whether to exclude default values when dumping + Pydantic models + **kwargs: Additional arguments to pass to the request + + Returns: + Response(model=PaginatedResponse) with created file information + """ + json_data = self._prepare_model_data( + body, exclude_defaults=exclude_defaults + ) + resp = self._post( + self.gen_url(f"storages/{storage_id}/files/"), + json=json_data, + **kwargs + ) + return self.parse_response(resp, PaginatedResponse) + + def list_asset_format_components( + self, asset_id: str, format_id: str, **kwargs + ) -> Response: + """ + Get all components for a format in an asset. + + Args: + asset_id: The ID of the asset + format_id: The ID of the format to retrieve + **kwargs: Additional arguments to pass to the request + + Returns: + Response(model=PaginatedResponse) containing format components + """ + resp = self._get( + self.gen_url(f"assets/{asset_id}/formats/{format_id}/components/"), + **kwargs + ) + return self.parse_response(resp, PaginatedResponse) + + def list_storage_files(self, storage_id: str, **kwargs) -> Response: + """ + Get all files on a storage, or files in a storage folder. + + Args: + storage_id: The ID of the storage to retrieve + **kwargs: Additional arguments to pass to the request + + Returns: + Response(model=PaginatedResponse) containing storage files + """ + resp = self._get( + self.gen_url(f"storages/{storage_id}/files/"), **kwargs + ) + return self.parse_response(resp, PaginatedResponse) + + def _get_deleted_object_type( + self, object_type: Literal["file_sets", "formats"], **kwargs + ) -> Response: + """ + Get deleted object type. + + Args: + object_type: The type of object to retrieve + **kwargs: Additional arguments to pass to the request + + Returns: + Response(model=None) containing deleted objects + + Raises: + ValueError: If object_type is not 'file_sets' or 'formats' + """ + if object_type not in ["file_sets", "formats"]: + raise ValueError("object_type must be one of file_sets or formats") + resp = self._get(self.gen_url(f"delete_queue/{object_type}/"), **kwargs) + return self.parse_response(resp, None) + + def get_deleted_file_sets(self, **kwargs) -> Response: + """ + Get deleted file sets. + + Args: + **kwargs: Additional arguments to pass to the request + + Returns: + Response(model=None) containing deleted file sets + """ + return self._get_deleted_object_type("file_sets", **kwargs) + + # Create method alias + get_deleted_filesets = get_deleted_file_sets + + def get_deleted_formats(self, **kwargs) -> Response: + """ + Get deleted formats. + + Args: + **kwargs: Additional arguments to pass to the request + + Returns: + Response(model=None) containing deleted formats + """ + return self._get_deleted_object_type("formats", **kwargs) + + def create_mediainfo_job( + self, + asset_id: str, + file_id: str, + priority: Optional[int] = 5, + **kwargs + ) -> Response: + """ + Create a job for extracting mediainfo. + + Args: + asset_id: ID of the asset + file_id: ID of the file + priority: Job priority 1-10. Default is 5 + **kwargs: Additional kwargs to pass to the request + + Returns: + Response(model=None) with job creation status + """ + body = {"priority": priority} + resp = self._post( + self.gen_url(f"assets/{asset_id}/files/{file_id}/mediainfo"), + json=body, + **kwargs + ) + return self.parse_response(resp, None) + + def create_transcode_job( + self, + asset_id: str, + file_id: str, + priority: Optional[int] = 5, + **kwargs + ) -> Response: + """ + Create a transcode job for proxy and keyframes. + + Args: + asset_id: ID of the asset + file_id: ID of the file + priority: Job priority 1-10. Default is 5 + **kwargs: Additional kwargs to pass to the request + + Returns: + Response(model=None) with job creation status + """ + body = {"priority": priority} + resp = self._post( + self.gen_url(f"assets/{asset_id}/files/{file_id}/keyframes"), + json=body, + **kwargs + ) + return self.parse_response(resp, None) diff --git a/pythonik/tests/test_files_extensions.py b/pythonik/tests/test_files_extensions.py new file mode 100644 index 0000000..4b5f3c9 --- /dev/null +++ b/pythonik/tests/test_files_extensions.py @@ -0,0 +1,429 @@ +import uuid +import requests_mock +import pytest +from typing import Literal + +from pythonik.client import PythonikClient +from pythonik.models.base import PaginatedResponse +from pythonik.models.files.file import File, FileType, FileStatus + + +def test_create_storage_file(): + """Test creating a file directly on storage without associating it to an asset.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + storage_id = str(uuid.uuid4()) + file_name = f"test_file_{str(uuid.uuid4())[:8]}.mp4" + + # Prepare file model + model = File( + size=1024000, + name=file_name, + original_name=file_name, + type=FileType.FILE, + status=FileStatus.CLOSED, + ) + + # Mock response for paginated response + response_data = { + "objects": [model.model_dump()], + "total": 1, + "page": 1, + "pages": 1, + "per_page": 25, + } + + mock_address = ( + f"https://app.iconik.io/API/files/v1/storages/{storage_id}/files/" + ) + m.post(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.files().create_storage_file(storage_id, body=model) + + # Verify response + assert m.called + assert result.response.ok + assert isinstance(result.data, PaginatedResponse) + assert len(result.data.objects) == 1 + assert result.data.total == 1 + + +def test_fetch_asset_format_components(): + """Test fetching components for a format in an asset.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + format_id = str(uuid.uuid4()) + + # Mock response + response_data = { + "objects": [ + { + "id": str(uuid.uuid4()), + "name": "Component 1", + "type": "video", + "metadata": {"codec": "h264"}, + }, + { + "id": str(uuid.uuid4()), + "name": "Component 2", + "type": "audio", + "metadata": {"codec": "aac"}, + }, + ], + "total": 2, + "page": 1, + "pages": 1, + "per_page": 25, + } + + mock_address = f"https://app.iconik.io/API/files/v1/assets/{asset_id}/formats/{format_id}/components/" + m.get(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.files().list_asset_format_components(asset_id, format_id) + + # Verify response + assert m.called + assert result.response.ok + assert isinstance(result.data, PaginatedResponse) + assert len(result.data.objects) == 2 + assert result.data.total == 2 + + +def test_fetch_storage_files(): + """Test fetching files from storage.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + storage_id = str(uuid.uuid4()) + + # Mock response + response_data = { + "objects": [ + { + "id": str(uuid.uuid4()), + "name": "file1.mp4", + "size": 1024000, + "type": "FILE", + "status": "CLOSED", + }, + { + "id": str(uuid.uuid4()), + "name": "file2.jpg", + "size": 256000, + "type": "FILE", + "status": "CLOSED", + }, + ], + "total": 2, + "page": 1, + "pages": 1, + "per_page": 25, + } + + mock_address = ( + f"https://app.iconik.io/API/files/v1/storages/{storage_id}/files/" + ) + m.get(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.files().list_storage_files(storage_id) + + # Verify response + assert m.called + assert result.response.ok + assert isinstance(result.data, PaginatedResponse) + assert len(result.data.objects) == 2 + assert result.data.total == 2 + + +def test_get_deleted_object_type_internal(): + """Test the internal _get_deleted_object_type method with invalid input.""" + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + + # Test with invalid object type + with pytest.raises(ValueError) as excinfo: + client.files()._get_deleted_object_type("invalid_type") + + assert "object_type must be one of file_sets or formats" in str(excinfo.value) + + +def test_get_deleted_file_sets(): + """Test fetching deleted file sets.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + + # Mock response + response_data = { + "deleted_file_sets": [ + { + "id": str(uuid.uuid4()), + "asset_id": str(uuid.uuid4()), + "name": "Deleted File Set 1", + "date_deleted": "2023-01-01T12:00:00Z", + }, + { + "id": str(uuid.uuid4()), + "asset_id": str(uuid.uuid4()), + "name": "Deleted File Set 2", + "date_deleted": "2023-01-02T12:00:00Z", + }, + ] + } + + mock_address = "https://app.iconik.io/API/files/v1/delete_queue/file_sets/" + m.get(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + + # Test main method + result = client.files().get_deleted_file_sets() + assert m.called + assert result.response.ok + # Since the method uses parse_response(resp, None), the data field will be None + # Instead, check the raw response json + assert result.response.json() == response_data + + # Reset mock and test alias + m.reset() + m.get(mock_address, json=response_data) + result_alias = client.files().get_deleted_filesets() + assert m.called + assert result_alias.response.ok + assert result_alias.response.json() == response_data + + +def test_get_deleted_formats(): + """Test fetching deleted formats.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + + # Mock response + response_data = { + "deleted_formats": [ + { + "id": str(uuid.uuid4()), + "asset_id": str(uuid.uuid4()), + "name": "Deleted Format 1", + "date_deleted": "2023-01-01T12:00:00Z", + }, + { + "id": str(uuid.uuid4()), + "asset_id": str(uuid.uuid4()), + "name": "Deleted Format 2", + "date_deleted": "2023-01-02T12:00:00Z", + }, + ] + } + + mock_address = "https://app.iconik.io/API/files/v1/delete_queue/formats/" + m.get(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.files().get_deleted_formats() + + # Verify response + assert m.called + assert result.response.ok + # Since the method uses parse_response(resp, None), the data field will be None + # Instead, check the raw response json + assert result.response.json() == response_data + + +def test_create_mediainfo_job(): + """Test creating a mediainfo job.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + file_id = str(uuid.uuid4()) + + # Mock response + response_data = { + "job_id": str(uuid.uuid4()), + "status": "CREATED", + "message": "Mediainfo job created successfully", + } + + # Expected request body + expected_body = {"priority": 7} + + mock_address = f"https://app.iconik.io/API/files/v1/assets/{asset_id}/files/{file_id}/mediainfo" + m.post(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.files().create_mediainfo_job(asset_id, file_id, priority=7) + + # Verify response and request + assert m.called + assert result.response.ok + # Since the method uses parse_response(resp, None), the data field will be None + # Instead, check the raw response json + assert result.response.json() == response_data + assert m.last_request.json() == expected_body + + +def test_create_transcode_job(): + """Test creating a transcode job.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + file_id = str(uuid.uuid4()) + + # Mock response + response_data = { + "job_id": str(uuid.uuid4()), + "status": "CREATED", + "message": "Transcode job created successfully", + } + + # Expected request body with default priority + expected_body = {"priority": 5} + + mock_address = f"https://app.iconik.io/API/files/v1/assets/{asset_id}/files/{file_id}/keyframes" + m.post(mock_address, json=response_data) + + client = PythonikClient(app_id=app_id, auth_token=auth_token, timeout=3) + result = client.files().create_transcode_job(asset_id, file_id) + + # Verify response and request + assert m.called + assert result.response.ok + # Since the method uses parse_response(resp, None), the data field will be None + # Instead, check the raw response json + assert result.response.json() == response_data + assert m.last_request.json() == expected_body + + +def test_alternate_base_url(): + """Test that all new methods respect the base_url parameter.""" + with requests_mock.Mocker() as m: + app_id = str(uuid.uuid4()) + auth_token = str(uuid.uuid4()) + alternate_base = "https://custom.iconik.io" + + # Test parameters + storage_id = str(uuid.uuid4()) + asset_id = str(uuid.uuid4()) + format_id = str(uuid.uuid4()) + file_id = str(uuid.uuid4()) + + # Mock responses - simple empty responses are sufficient for this test + storage_file_response = { + "objects": [], + "total": 0, + "page": 1, + "pages": 1, + "per_page": 25, + } + format_components_response = { + "objects": [], + "total": 0, + "page": 1, + "pages": 1, + "per_page": 25, + } + storage_files_response = { + "objects": [], + "total": 0, + "page": 1, + "pages": 1, + "per_page": 25, + } + deleted_file_sets_response = {"deleted_file_sets": []} + deleted_formats_response = {"deleted_formats": []} + job_response = {"job_id": str(uuid.uuid4()), "status": "CREATED"} + + # Register mock responses with alternate base URL + m.post( + f"{alternate_base}/API/files/v1/storages/{storage_id}/files/", + json=storage_file_response, + ) + m.get( + f"{alternate_base}/API/files/v1/assets/{asset_id}/formats/{format_id}/components/", + json=format_components_response, + ) + m.get( + f"{alternate_base}/API/files/v1/storages/{storage_id}/files/", + json=storage_files_response, + ) + m.get( + f"{alternate_base}/API/files/v1/delete_queue/file_sets/", + json=deleted_file_sets_response, + ) + m.get( + f"{alternate_base}/API/files/v1/delete_queue/formats/", + json=deleted_formats_response, + ) + m.post( + f"{alternate_base}/API/files/v1/assets/{asset_id}/files/{file_id}/mediainfo", + json=job_response, + ) + m.post( + f"{alternate_base}/API/files/v1/assets/{asset_id}/files/{file_id}/keyframes", + json=job_response, + ) + + # Initialize client with alternate base URL + client = PythonikClient( + app_id=app_id, auth_token=auth_token, timeout=3, base_url=alternate_base + ) + files_spec = client.files() + + # Test all methods to ensure they use the alternate base URL + file_model = File( + name="test.mp4", + original_name="test.mp4", + size=1024, + type=FileType.FILE, + status=FileStatus.CLOSED, + ) + files_spec.create_storage_file(storage_id, body=file_model) + files_spec.list_asset_format_components(asset_id, format_id) + files_spec.list_storage_files(storage_id) + files_spec.get_deleted_file_sets() + files_spec.get_deleted_formats() + files_spec.create_mediainfo_job(asset_id, file_id) + files_spec.create_transcode_job(asset_id, file_id) + + # Verify that all mocks were called + assert m.call_count == 7, f"Expected 7 calls, got {m.call_count}" + + # Verify each mock individually to better diagnose which one failed + assert ( + m.request_history[0].url + == f"{alternate_base}/API/files/v1/storages/{storage_id}/files/" + ) + assert ( + m.request_history[1].url + == f"{alternate_base}/API/files/v1/assets/{asset_id}/formats/{format_id}/components/" + ) + assert ( + m.request_history[2].url + == f"{alternate_base}/API/files/v1/storages/{storage_id}/files/" + ) + assert ( + m.request_history[3].url + == f"{alternate_base}/API/files/v1/delete_queue/file_sets/" + ) + assert ( + m.request_history[4].url + == f"{alternate_base}/API/files/v1/delete_queue/formats/" + ) + assert ( + m.request_history[5].url + == f"{alternate_base}/API/files/v1/assets/{asset_id}/files/{file_id}/mediainfo" + ) + assert ( + m.request_history[6].url + == f"{alternate_base}/API/files/v1/assets/{asset_id}/files/{file_id}/keyframes" + )