Fixed automated tests

Author: Lucashsmello

datamint/api/endpoints/resources_api.py

@@ -54,6 +54,8 @@ def _open_io(file_path: str | Path | IO, mode: str = 'rb') -> IO:
 class ResourcesApi(CreatableEntityApi[Resource], DeletableEntityApi[Resource]):
     """API handler for resource-related endpoints."""
 
+    _ENDPOINT_BASE = 'resources'
+
     def __init__(self,
                  config: ApiConfig,
                  client: httpx.Client | None = None,
@@ -68,7 +70,7 @@ def __init__(self,
         """
         from .annotations_api import AnnotationsApi
         from .projects_api import ProjectsApi
-        super().__init__(config, Resource, 'resources', client)
+        super().__init__(config, Resource, ResourcesApi._ENDPOINT_BASE, client)
         nest_asyncio.apply()
         self.annotations_api = AnnotationsApi(
             config, client, resources_api=self) if annotations_api is None else annotations_api

pyproject.toml

@@ -1,7 +1,7 @@
 [project]
 name = "datamint"
 description = "A library for interacting with the Datamint API, designed for efficient data management, processing and Deep Learning workflows."
-version = "2.7.1"
+version = "2.8.0"
 dynamic = ["dependencies"]
 requires-python = ">=3.10"
 readme = "README.md"
@@ -66,6 +66,7 @@ pytest = "^7.0.0"
 pytest-cov = "^4.0.0"
 responses = "^0.20.0"
 aioresponses = "^0.7.0"
+respx = ">=0.22.0"
 
 # Extra dependencies for docs
 [tool.poetry.extras]

tests/test_api_handler.py

@@ -1,20 +1,21 @@
 import pytest
 from unittest.mock import patch
-from datamint.apihandler.api_handler import APIHandler
+from datamint.api.client import Api
+from datamint.exceptions import DatamintException
 import datamint
-import responses
+import respx
 from aioresponses import aioresponses, CallbackResult
 import pydicom
 from pydicom.data import get_testdata_files
 import datamint.configs
 from medimgkit.dicom_utils import to_bytesio
-from datamint.apihandler.base_api_handler import DatamintException
 import json
 from aiohttp import FormData
 from typing import IO
 import os
 import numpy as np
 from copy import deepcopy
+import httpx
 
 # pytest tests --log-cli-level=INFO
 
@@ -103,38 +104,33 @@ def get_projects_sample(self) -> dict:
 
     @pytest.fixture
     def get_resources_sample(self) -> dict:
-        res = {'id': 'f0fea905-6823-48c3-838a-b0c62c5a0466',
-               'resource_uri': '/resources/f0fea905-6823-48c3-838a-b0c62c5a0466/file',
+        res = {'id': 'cd69c126-02ee-44af-8672-13d61b09eee4',
+               'resource_uri': '/resources/cd69c126-02ee-44af-8672-13d61b09eee4/file',
                'storage': 'ImageResource',
-               'location': 'resources/f5a18308-ab64-4c1e-8a1f-db4bccd77f4a/f0fea905-6823-48c3-838a-b0c62c5a0466',
+               'location': 'resources/79113ed1-0535-4f53-9359-7fe3fa9f28a8/cd69c126-02ee-44af-8672-13d61b09eee4',
                'upload_channel': 'Unknown',
-               'filename': 'IMG0002121.jpg',
-               'modality': 'DX',
-               'mimetype': 'image/jpeg',
-               'size': 15662,
-               'upload_mechanism': 'api',
-               'customer_id': 'f5a18308-ab64-4c1e-8a1f-db4bccd77f4a',
-               'status': 'published',
-               'created_at': '2025-02-05T14:32:30.378Z',
+               'filename': 'normal (95).png',
+               'modality': 'US',
+               'mimetype': 'image/png',
+               'size': 444584, 'upload_mechanism': 'api',
+               'customer_id': '79113ed1-0535-4f53-9359-7fe3fa9f28a8',
+               'status': 'inbox',
+               'created_at': '2025-12-24T11:34:18.036Z',
                'created_by': 'datamint-dev@mail.com',
                'published': False,
-               'published_on': '2025-02-05T14:47:47.942Z',
-               'published_by': 'datamint-dev@mail.com',
+               'published_on': None,
+               'published_by': None,
                'publish_transforms': None,
                'deleted': False,
                'deleted_at': None,
                'deleted_by': None,
-               'metadata': {},
-               'source_filepath': 'images/IMG0002121.jpg',
-               'tags': ['non_fractured', 'leg'],
-               'segmentations': None,
-               'measurements': None,
-               'categories': None,
-               'labels': [],
-               'user_info': {'firstname': 'data', 'lastname': 'mint'},
-               'projects': []}
+               'metadata': {'width': 693, 'height': 582},
+               'source_filepath': '/tmp/normal/normal (95).png',
+               'tags': ['ultrasound', 'split:train'],
+               'user_info': {'firstname': None, 'lastname': None},
+               'projects': [], 'labels': []}
 
-        return {'data': [{'resources': [res]}]}
+        return {'data': [{'resources': [res]}], 'totalCount': 1}
 
     @pytest.fixture
     def sample_2dmask1(self) -> np.ndarray:
@@ -145,214 +141,26 @@ def sample_2dmask1(self) -> np.ndarray:
         rng = np.random.RandomState(42)
         return rng.randint(0, 2, (32, 32), dtype=np.uint8)
 
-    @responses.activate
+    @respx.mock
     @patch('os.getenv')
     def test_api_handler_init(self, mock_getenv, get_projects_sample: dict):
-        def mock_getenv_side_effect(key):
-            if key == datamint.configs.get_env_var_name(datamint.configs.APIKEY_KEY):
-                return 'test_api_key'
-            if key == datamint.configs.get_env_var_name(datamint.configs.APIURL_KEY):
-                return _TEST_URL
-            return None
-
-        mock_getenv.side_effect = mock_getenv_side_effect
-        api_handler = APIHandler(check_connection=False)
-        assert api_handler.api_key == 'test_api_key'
-        assert api_handler.root_url == _TEST_URL
-
-        responses.get(
-            f"{_TEST_URL}/projects",
-            status=200,
-            json=get_projects_sample
-        )
-
-        api_handler = APIHandler(check_connection=True)
+        api_handler = Api(check_connection=False)
 
         ### Test wrong url ###
         with pytest.raises(DatamintException):
-            api_handler = APIHandler('wrong', check_connection=True)
-
-    def test_upload_dicoms_resources(self, sample_dicom1):
-        def _callback1(url, data, **kwargs):
-            dicom_bytes, data = _get_request_data(data)
-            ds = pydicom.dcmread(dicom_bytes)
-
-            assert str(sample_dicom1.PatientName) == str(ds.PatientName)
-            return CallbackResult(status=201, payload={"id": "newdicomid"})
-
-        with aioresponses() as mock_aioresp:
-            # check that post request has data 'batch_id'
-            mock_aioresp.post(
-                f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-                callback=_callback1,
-                repeat=2
-            )
-
-            api_handler = APIHandler(_TEST_URL, 'test_api_key', check_connection=False)
-            new_dicoms_id = api_handler.upload_resources(files_path=to_bytesio(sample_dicom1, 'sample_dicom1'),
-                                                         channel='mychannel',
-                                                         anonymize=False)
-            assert new_dicoms_id == 'newdicomid'
-
-            ### Same thing, but in a list ###
-            new_dicoms_id = api_handler.upload_resources(files_path=[to_bytesio(sample_dicom1, 'sample_dicom1')],
-                                                         channel='mychannel',
-                                                         anonymize=False)
-            assert len(new_dicoms_id) == 1 and new_dicoms_id[0] == 'newdicomid'
-            ######
-
-    def test_upload_dicoms_mungfilename(self, sample_dicom1):
-        from builtins import open
-
-        def my_open_mock(file, *args, **kwargs):
-            if str(file).endswith('.dcm'):
-                return to_bytesio(sample_dicom1, file)
-            return open(file, *args, **kwargs)
-
-        def _request_callback(url, data, **kwargs):
-            dicom_bytes, data = _get_request_data(data)
-            assert '__data_test_dicom.dcm' in data
-            return CallbackResult(status=201, payload={"id": "newdicomid"})
-
-        def _request_callback2(url, data, **kwargs):
-            if isinstance(data, FormData):
-                data = str(data._fields)
-            else:
-                data = str(data)
-            assert 'test_dicom.dcm' in data and 'data_test' not in data
-            return CallbackResult(status=201, payload={"id": "newdicomid"})
-
-        def _request_callback3(url, data, **kwargs):
-            dicom_bytes, data = _get_request_data(data)
-            assert 'me_data_test_dicom.dcm' in data
-            return CallbackResult(status=201, payload={"id": "newdicomid"})
-
-        api_handler = APIHandler(_TEST_URL, 'test_api_key', check_connection=False)
-
-        with patch('builtins.open', new=my_open_mock):
-            with aioresponses() as mock_aioresp:
-                mock_aioresp.post(
-                    f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-                    callback=_request_callback
-                )
-                new_dicoms_id = api_handler.upload_resources(files_path=os.path.join('..', 'data', 'test_dicom.dcm'),
-                                                             anonymize=False,
-                                                             mung_filename='all')
-                assert new_dicoms_id == 'newdicomid'
-
-            with aioresponses() as mock_aioresp:
-                mock_aioresp.post(
-                    f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-                    callback=_request_callback2
-                )
-                new_dicoms_id = api_handler.upload_resources(files_path=os.path.join('data', 'test_dicom.dcm'),
-                                                             anonymize=False,
-                                                             mung_filename=None)
-                assert new_dicoms_id == 'newdicomid'
-
-            with aioresponses() as mock_aioresp:
-                mock_aioresp.post(
-                    f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-                    callback=_request_callback3
-                )
-                new_dicoms_id = api_handler.upload_resources(files_path=os.path.join('home', 'me', 'data', 'test_dicom.dcm'),
-                                                             anonymize=False,
-                                                             mung_filename=[2, 3])
-                assert new_dicoms_id == 'newdicomid'
-
-    @responses.activate
-    def test_upload_resources_video(self):
-        def _callback(url, data, *args, **kwargs):
-            assert kwargs['headers']['apikey'] == 'test_api_key'
-
-            return CallbackResult(status=201, payload={"id": "new_resource_id"})
-
-        with aioresponses() as mock_aioresp:
-            api_handler = APIHandler(_TEST_URL, 'test_api_key', check_connection=False)
-            mock_aioresp.post(
-                api_handler._get_endpoint_url(APIHandler.ENDPOINT_RESOURCES),
-                callback=_callback
-            )
-
-            new_resources_id = api_handler.upload_resources(files_path=MP4_TEST_FILE,
-                                                            tags=['label1', 'label2'],
-                                                            channel='mychannel',
-                                                            anonymize=False)
-            new_resources_id == 'new_resource_id'
-
-    def test_upload_resources_assembling_dicoms(self, sample_dicom1: pydicom.Dataset):
-        sample_dicom2 = deepcopy(sample_dicom1)
-        sample_dicom2.InstanceNumber = 2
-        sample_dicom2.SOPInstanceUID = pydicom.uid.generate_uid()
-        api_handler = APIHandler(_TEST_URL, 'test_api_key', check_connection=False)
-
-        with aioresponses() as mock_aioresp:
-            mock_aioresp.post(
-                f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-                status=201,
-                payload={"id": "new_res_id"},
-                repeat=2
-            )
-
-            new_res_ids = api_handler.upload_resources(files_path=[to_bytesio(sample_dicom1, 'sample_dicom1'),
-                                                                   to_bytesio(sample_dicom2, 'sample_dicom2'),
-                                                                   MP4_TEST_FILE,
-                                                                   ],
-                                                       channel='mychannel',
-                                                       assemble_dicoms=True,
-                                                       anonymize=False)
-            assert len(new_res_ids) == 2
-
-        ### Same thing, but assemble_dicoms=False ###
-        with aioresponses() as mock_aioresp:
-            mock_aioresp.post(
-                f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-                status=201,
-                payload={"id": "new_res_id"},
-                repeat=3
-            )
-
-            new_res_ids = api_handler.upload_resources(files_path=[to_bytesio(sample_dicom1, 'sample_dicom1'),
-                                                                   to_bytesio(sample_dicom2, 'sample_dicom2'),
-                                                                   MP4_TEST_FILE,
-                                                                   ],
-                                                       channel='mychannel',
-                                                       assemble_dicoms=False,
-                                                       anonymize=False)
-            assert len(new_res_ids) == 3
-
-    @responses.activate
-    def test_get_channels(self, get_channels_sample: dict):
-        def _request_callback(request):
-            return (200, "", json.dumps(get_channels_sample))
-
-        # Mocking the response from the server
-        responses.add_callback(
-            responses.GET,
-            f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}/channels",
-            content_type='application/json',
-            callback=_request_callback
-        )
-
-        api_handler = APIHandler(_TEST_URL, 'test_api_key', check_connection=False)
-        channels_info = list(api_handler.get_channels())
-        assert len(channels_info) == 2  # two channels
+            api_handler = Api(server_url='wrong', check_connection=True)
 
-    @responses.activate
+    @respx.mock
     def test_get_resources(self, get_resources_sample: dict):
-        def _request_callback(request):
-            return (200, "", json.dumps(get_resources_sample))
+        from datamint.api.endpoints.resources_api import ResourcesApi
 
         # Mocking the response from the server
-        responses.add_callback(
-            responses.GET,
-            f"{_TEST_URL}/{APIHandler.ENDPOINT_RESOURCES}",
-            content_type='application/json',
-            callback=_request_callback
+        respx.get(f"{_TEST_URL}/{ResourcesApi._ENDPOINT_BASE}").mock(
+            return_value=httpx.Response(200, json=get_resources_sample)
         )
 
-        api_handler = APIHandler(_TEST_URL, 'test_api_key', check_connection=False)
-        resources = list(api_handler.get_resources())
+        api = Api(_TEST_URL, 'test_api_key', check_connection=False)
+        resources = api.resources.get_list()
         assert len(resources) == 1
         r = resources[0]
-        assert r['id'] == 'f0fea905-6823-48c3-838a-b0c62c5a0466'
+        assert r.filename == 'normal (95).png'

tests/test_datamint_config.py

@@ -4,11 +4,7 @@
 """
 import pytest
 import sys
-import subprocess
-import tempfile
-import os
-from unittest.mock import patch, MagicMock
-from pathlib import Path
+from unittest.mock import patch
 import logging
 
 _LOGGER = logging.getLogger(__name__)