fix test by skipping unsupported test for version 5.0 of the API

Author: lalepee

tests/unit/coal/test_cosmotech_api/test_cosmotech_api_run_template.py

@@ -7,8 +7,7 @@
 
 import pathlib
 import pytest
-from unittest.mock import MagicMock, patch, mock_open
-from io import BytesIO
+from unittest.mock import MagicMock, patch
 from zipfile import BadZipfile, ZipFile
 
 import cosmotech_api
@@ -17,8 +16,12 @@
 from cosmotech_api.exceptions import ServiceException
 
 from cosmotech.coal.cosmotech_api.run_template import load_run_template_handlers
+from cosmotech.coal.utils.semver import semver_of
 
 
+@pytest.mark.skipif(
+    semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+)
 class TestRunTemplateFunctions:
     """Tests for top-level functions in the run_template module."""
 
@@ -66,7 +69,6 @@ def test_load_run_template_handlers_success(self):
             patch("cosmotech.coal.cosmotech_api.run_template.WorkspaceApi", return_value=mock_workspace_api),
             patch("cosmotech.coal.cosmotech_api.run_template.SolutionApi", return_value=mock_solution_api),
             patch("cosmotech.coal.cosmotech_api.run_template.ZipFile", return_value=mock_zipfile_context),
-            patch("cosmotech.coal.cosmotech_api.run_template.BytesIO") as mock_bytesio,
             patch("cosmotech.coal.cosmotech_api.run_template.pathlib.Path") as mock_path_class,
         ):
             mock_path_class.return_value = mock_path

tests/unit/coal/test_cosmotech_api/test_cosmotech_api_twin_data_layer.py

@@ -14,21 +14,29 @@
 
 import pytest
 import requests
-from cosmotech_api import DatasetApi, RunnerApi, DatasetTwinGraphQuery
-
-from cosmotech.coal.cosmotech_api.twin_data_layer import (
-    get_dataset_id_from_runner,
-    send_files_to_tdl,
-    load_files_from_tdl,
-    CSVSourceFile,
-    _process_csv_file,
-    _get_node_properties,
-    _get_relationship_properties,
-    _execute_queries,
-    _write_files,
-    ID_COLUMN,
-    SOURCE_COLUMN,
-    TARGET_COLUMN,
+
+from cosmotech.coal.utils.semver import semver_of
+from cosmotech_api import DatasetApi, RunnerApi
+if semver_of('cosmotech_api').major < 5:
+    from cosmotech_api import DatasetTwinGraphQuery
+
+    from cosmotech.coal.cosmotech_api.twin_data_layer import (
+        get_dataset_id_from_runner,
+        send_files_to_tdl,
+        load_files_from_tdl,
+        CSVSourceFile,
+        _process_csv_file,
+        _get_node_properties,
+        _get_relationship_properties,
+        _execute_queries,
+        _write_files,
+        ID_COLUMN,
+        SOURCE_COLUMN,
+        TARGET_COLUMN,
+    )
+
+pytestmark = pytest.mark.skipif(
+    semver_of('cosmotech_api').major >= 5, reason='not supported under version 5'
 )
 
 

tests/unit/coal/test_cosmotech_api/test_cosmotech_api_twin_data_layer_auth.py

@@ -14,18 +14,27 @@
 
 import pytest
 import requests
-from cosmotech_api import DatasetApi, RunnerApi, DatasetTwinGraphQuery
+
+from cosmotech.coal.utils.semver import semver_of
+from cosmotech_api import DatasetApi, RunnerApi
+if semver_of('cosmotech_api').major < 5:
+    from cosmotech_api import DatasetTwinGraphQuery
+    from cosmotech.coal.cosmotech_api.twin_data_layer import (
+        send_files_to_tdl,
+        load_files_from_tdl,
+        _process_csv_file,
+        _get_node_properties,
+        _get_relationship_properties,
+    )
 
 from cosmotech.orchestrator.utils.translate import T
-from cosmotech.coal.cosmotech_api.twin_data_layer import (
-    send_files_to_tdl,
-    load_files_from_tdl,
-    _process_csv_file,
-    _get_node_properties,
-    _get_relationship_properties,
+
+skip_under_v5 = pytest.mark.skipif(
+    semver_of('cosmotech_api').major >= 5, reason='not supported under version 5'
 )
 
 
+@skip_under_v5
 class TestTwinDataLayerAuth:
     """Tests for authentication in the twin_data_layer module."""
 

tests/unit/coal/test_cosmotech_api/test_cosmotech_api_twin_data_layer_coverage.py

@@ -17,18 +17,28 @@
 
 import pytest
 import requests
-from cosmotech_api import DatasetApi, RunnerApi, DatasetTwinGraphQuery
-
-from cosmotech.coal.cosmotech_api.twin_data_layer import (
-    get_dataset_id_from_runner,
-    send_files_to_tdl,
-    load_files_from_tdl,
-    _process_csv_file,
-    _write_files,
-    BATCH_SIZE_LIMIT,
+
+from cosmotech.coal.utils.semver import semver_of
+from cosmotech_api import DatasetApi, RunnerApi
+
+if semver_of('cosmotech_api').major < 5:
+    from cosmotech_api import DatasetTwinGraphQuery
+
+    from cosmotech.coal.cosmotech_api.twin_data_layer import (
+        get_dataset_id_from_runner,
+        send_files_to_tdl,
+        load_files_from_tdl,
+        _process_csv_file,
+        _write_files,
+        BATCH_SIZE_LIMIT,
+    )
+
+skip_under_v5 = pytest.mark.skipif(
+    semver_of('cosmotech_api').major >= 5, reason='not supported under version 5'
 )
 
 
+@skip_under_v5
 class TestTwinDataLayerCoverage:
     """Additional tests for the twin_data_layer module to improve coverage."""
 

tests/unit/coal/test_cosmotech_api/test_cosmotech_api_twin_data_layer_edge_cases.py

@@ -14,18 +14,26 @@
 
 import pytest
 import requests
-from cosmotech_api import DatasetApi, RunnerApi, DatasetTwinGraphQuery
 
+from cosmotech.coal.utils.semver import semver_of
+from cosmotech_api import DatasetApi, RunnerApi
+if semver_of('cosmotech_api').major < 5:
+    from cosmotech_api import DatasetTwinGraphQuery
+    from cosmotech.coal.cosmotech_api.twin_data_layer import (
+        send_files_to_tdl,
+        load_files_from_tdl,
+        _process_csv_file,
+        _get_node_properties,
+        _get_relationship_properties,
+    )
 from cosmotech.orchestrator.utils.translate import T
-from cosmotech.coal.cosmotech_api.twin_data_layer import (
-    send_files_to_tdl,
-    load_files_from_tdl,
-    _process_csv_file,
-    _get_node_properties,
-    _get_relationship_properties,
+
+skip_under_v5 = pytest.mark.skipif(
+    semver_of('cosmotech_api').major >= 5, reason='not supported under version 5'
 )
 
 
+@skip_under_v5
 class TestTwinDataLayerEdgeCases:
     """Tests for edge cases in the twin_data_layer module."""
 

tests/unit/coal/test_cosmotech_api/test_dataset/test_download/test_download_twingraph.py

@@ -11,14 +11,21 @@
 
 import pytest
 import cosmotech_api
-from cosmotech_api import DatasetApi, TwingraphApi
-
-from cosmotech.coal.cosmotech_api.dataset.download.twingraph import (
-    download_twingraph_dataset,
-    download_legacy_twingraph_dataset,
+from cosmotech_api import DatasetApi
+from cosmotech.coal.utils.semver import semver_of
+if semver_of('cosmotech_api').major < 5:
+    from cosmotech_api import TwingraphApi
+    from cosmotech.coal.cosmotech_api.dataset.download.twingraph import (
+        download_twingraph_dataset,
+        download_legacy_twingraph_dataset,
+    )
+
+skip_under_v5 = pytest.mark.skipif(
+    semver_of('cosmotech_api').major >= 5, reason='not supported under version 5'
 )
 
 
+@skip_under_v5
 class TestTwingraphFunctions:
     """Tests for top-level functions in the twingraph module."""
 

tests/unit/coal/test_cosmotech_api/test_runner/test_runner_datasets.py

@@ -5,13 +5,10 @@
 # etc., to any person is prohibited unless it has been previously and
 # specifically authorized by written means by Cosmo Tech.
 
-import multiprocessing
-import tempfile
 from pathlib import Path
 from unittest.mock import MagicMock, patch, call
-
 import pytest
-from azure.identity import DefaultAzureCredential
+
 from cosmotech_api import DatasetApi
 
 from cosmotech.coal.cosmotech_api.runner.datasets import (
@@ -22,6 +19,7 @@
     download_datasets,
     dataset_to_file,
 )
+from cosmotech.coal.utils.semver import semver_of
 
 
 class TestDatasetsFunctions:
@@ -60,6 +58,9 @@ def test_get_dataset_ids_from_runner(self):
 
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.download_adt_dataset")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_dataset_adt(self, mock_download_adt, mock_get_api_client):
         """Test the download_dataset function with ADT dataset."""
         # Arrange
@@ -106,6 +107,9 @@ def test_download_dataset_adt(self, mock_download_adt, mock_get_api_client):
 
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.download_legacy_twingraph_dataset")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_dataset_legacy_twingraph(self, mock_download_legacy, mock_get_api_client):
         """Test the download_dataset function with legacy twin graph dataset."""
         # Arrange
@@ -153,6 +157,9 @@ def test_download_dataset_legacy_twingraph(self, mock_download_legacy, mock_get_
 
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.download_file_dataset")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_dataset_storage(self, mock_download_file, mock_get_api_client):
         """Test the download_dataset function with storage dataset."""
         # Arrange
@@ -205,6 +212,9 @@ def test_download_dataset_storage(self, mock_download_file, mock_get_api_client)
 
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.download_file_dataset")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_dataset_workspace_file(self, mock_download_file, mock_get_api_client):
         """Test the download_dataset function with workspace file dataset."""
         # Arrange
@@ -260,6 +270,9 @@ def test_download_dataset_workspace_file(self, mock_download_file, mock_get_api_
 
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.download_twingraph_dataset")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_dataset_twingraph(self, mock_download_twingraph, mock_get_api_client):
         """Test the download_dataset function with twin graph dataset."""
         # Arrange
@@ -309,6 +322,9 @@ def test_download_dataset_twingraph(self, mock_download_twingraph, mock_get_api_
     @patch("multiprocessing.Process")
     @patch("multiprocessing.Manager")
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_datasets_parallel(self, mock_get_api_client, mock_manager, mock_process, mock_download_dataset):
         """Test the download_datasets_parallel function."""
         # Arrange
@@ -392,14 +408,12 @@ def test_download_datasets_sequential(self, mock_get_api_client, mock_download_d
                     workspace_id=workspace_id,
                     dataset_id="dataset-1",
                     read_files=True,
-                    credentials=None,
                 ),
                 call(
                     organization_id=organization_id,
                     workspace_id=workspace_id,
                     dataset_id="dataset-2",
                     read_files=True,
-                    credentials=None,
                 ),
             ]
         )
@@ -436,7 +450,6 @@ def test_download_datasets_parallel_mode(self, mock_sequential, mock_parallel):
             workspace_id=workspace_id,
             dataset_ids=dataset_ids,
             read_files=True,
-            credentials=None,
         )
         mock_sequential.assert_not_called()
         assert len(result) == 2
@@ -472,7 +485,6 @@ def test_download_datasets_sequential_mode(self, mock_sequential, mock_parallel)
             workspace_id=workspace_id,
             dataset_ids=dataset_ids,
             read_files=True,
-            credentials=None,
         )
         mock_parallel.assert_not_called()
         assert len(result) == 2
@@ -507,7 +519,6 @@ def test_download_datasets_single_dataset(self, mock_sequential, mock_parallel):
             workspace_id=workspace_id,
             dataset_ids=dataset_ids,
             read_files=True,
-            credentials=None,
         )
         mock_parallel.assert_not_called()
         assert len(result) == 1

tests/unit/coal/test_cosmotech_api/test_runner/test_runner_datasets_additional.py

@@ -5,12 +5,10 @@
 # etc., to any person is prohibited unless it has been previously and
 # specifically authorized by written means by Cosmo Tech.
 
-import multiprocessing
 from pathlib import Path
-from unittest.mock import MagicMock, patch, call
+from unittest.mock import MagicMock, patch
 
 import pytest
-from azure.identity import DefaultAzureCredential
 from cosmotech_api import DatasetApi
 
 from cosmotech.coal.cosmotech_api.runner.datasets import (
@@ -19,12 +17,16 @@
     download_datasets,
     dataset_to_file,
 )
+from cosmotech.coal.utils.semver import semver_of
 
 
 class TestDatasetsAdditional:
     """Additional tests for the datasets module to improve coverage."""
 
     @patch("cosmotech.coal.cosmotech_api.runner.datasets.get_api_client")
+    @pytest.mark.skipif(
+        semver_of('cosmotech_api').major >= 5, reason='not supported in version 5'
+    )
     def test_download_dataset_no_connector(self, mock_get_api_client):
         """Test the download_dataset function with a dataset that has no connector."""
         # Arrange
@@ -189,12 +191,11 @@ def test_download_datasets_parallel_start_join(self, mock_get_api_client, mock_m
         mock_process.side_effect = [mock_process_instance1, mock_process_instance2]
 
         # Act
-        with patch("cosmotech.coal.cosmotech_api.runner.datasets.download_dataset") as mock_download_dataset:
-            download_datasets_parallel(
-                organization_id=organization_id,
-                workspace_id=workspace_id,
-                dataset_ids=dataset_ids,
-            )
+        download_datasets_parallel(
+            organization_id=organization_id,
+            workspace_id=workspace_id,
+            dataset_ids=dataset_ids,
+        )
 
         # Assert
         # Check that start and join were called for each process