clean tests and make sure all test code is executed (TGSAI#685)

* clean tests

* pre-commit added a space

* Remove DEBUG_MODE

* Fix pre-commit issues

* pre-commit

---------

Co-authored-by: Altay Sansal <[email protected]>

Author: dmitriyrepin

tests/conftest.py

@@ -3,12 +3,13 @@
 from __future__ import annotations
 
 import warnings
-from pathlib import Path
+from typing import TYPE_CHECKING
 from urllib.request import urlretrieve
 
 import pytest
 
-DEBUG_MODE = False
+if TYPE_CHECKING:
+    from pathlib import Path
 
 # Suppress Dask's chunk balancing warning
 warnings.filterwarnings(
@@ -22,10 +23,6 @@
 @pytest.fixture(scope="session")
 def fake_segy_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
     """Make a temp file for the fake SEG-Y files we are going to create."""
-    if DEBUG_MODE:
-        tmp_dir = Path("tmp/fake_segy")
-        tmp_dir.mkdir(parents=True, exist_ok=True)
-        return tmp_dir
     return tmp_path_factory.mktemp(r"fake_segy")
 
 
@@ -38,11 +35,7 @@ def segy_input_uri() -> str:
 @pytest.fixture(scope="session")
 def segy_input(segy_input_uri: str, tmp_path_factory: pytest.TempPathFactory) -> Path:
     """Download teapot dome dataset for testing."""
-    if DEBUG_MODE:
-        tmp_dir = Path("tmp/segy")
-        tmp_dir.mkdir(parents=True, exist_ok=True)
-    else:
-        tmp_dir = tmp_path_factory.mktemp("segy")
+    tmp_dir = tmp_path_factory.mktemp("segy")
     tmp_file = tmp_dir / "teapot.segy"
     urlretrieve(segy_input_uri, tmp_file)  # noqa: S310
     return tmp_file
@@ -51,25 +44,17 @@ def segy_input(segy_input_uri: str, tmp_path_factory: pytest.TempPathFactory) ->
 @pytest.fixture(scope="module")
 def zarr_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
     """Make a temp file for the output MDIO."""
-    if DEBUG_MODE:
-        return Path("tmp/mdio")
     return tmp_path_factory.mktemp(r"mdio")
 
 
 @pytest.fixture(scope="module")
-def zarr_tmp2(tmp_path_factory: pytest.TempPathFactory) -> Path:
+def zarr_tmp2(tmp_path_factory: pytest.TempPathFactory) -> Path:  # pragma: no cover - used by disabled test
     """Make a temp file for the output MDIO."""
-    if DEBUG_MODE:
-        return Path("tmp/mdio2")
     return tmp_path_factory.mktemp(r"mdio2")
 
 
 @pytest.fixture(scope="session")
 def segy_export_tmp(tmp_path_factory: pytest.TempPathFactory) -> Path:
     """Make a temp file for the round-trip IBM SEG-Y."""
-    if DEBUG_MODE:
-        tmp_dir = Path("tmp/segy")
-        tmp_dir.mkdir(parents=True, exist_ok=True)
-    else:
-        tmp_dir = tmp_path_factory.mktemp("segy")
+    tmp_dir = tmp_path_factory.mktemp("segy")
     return tmp_dir / "teapot_roundtrip.segy"

tests/integration/test_import_streamer_grid_overrides.py

@@ -33,7 +33,7 @@
 @pytest.mark.skip(reason="NonBinned and HasDuplicates haven't been properly implemented yet.")
 @pytest.mark.parametrize("grid_override", [{"NonBinned": True}, {"HasDuplicates": True}])
 @pytest.mark.parametrize("chan_header_type", [StreamerShotGeometryType.C])
-class TestImport4DNonReg:
+class TestImport4DNonReg:  # pragma: no cover - tests is skipped
     """Test for 4D segy import with grid overrides."""
 
     def test_import_4d_segy(  # noqa: PLR0913
@@ -161,7 +161,7 @@ def test_import_4d_segy(  # noqa: PLR0913
 @pytest.mark.skip(reason="AutoShotWrap requires a template that is not implemented yet.")
 @pytest.mark.parametrize("grid_override", [{"AutoChannelWrap": True}, {"AutoShotWrap": True}, None])
 @pytest.mark.parametrize("chan_header_type", [StreamerShotGeometryType.A, StreamerShotGeometryType.B])
-class TestImport6D:
+class TestImport6D:  # pragma: no cover - tests is skipped
     """Test for 6D segy import with grid overrides."""
 
     def test_import_6d_segy(  # noqa: PLR0913

tests/integration/test_segy_import_export_masked.py

@@ -21,7 +21,6 @@
 from segy.schema import HeaderField
 from segy.schema import SegySpec
 from segy.standards import get_segy_standard
-from tests.conftest import DEBUG_MODE
 
 from mdio import mdio_to_segy
 from mdio.api.io import open_mdio
@@ -241,17 +240,6 @@ def mock_nd_segy(path: str, grid_conf: GridConfig, segy_factory_conf: SegyFactor
         headers["group_coord_y"] = (y_origin + dim_grids["shot_point"] * y_step + dim_grids[cable_key] * y_step).ravel()
         headers["gun"] = np.tile((1, 2, 3), num_traces // 3)
 
-    # for field in ["cdp_x", "source_coord_x", "group_coord_x"]:
-    #     start = 700000
-    #     step = 100
-    #     stop = start + step * (trace_numbers.size - 0)
-    #     headers[field] = np.arange(start=start, stop=stop, step=step)
-    # for field in ["cdp_y", "source_coord_y", "group_coord_y"]:
-    #     start = 4000000
-    #     step = 100
-    #     stop = start + step * (trace_numbers.size - 0)
-    #     headers[field] = np.arange(start=start, stop=stop, step=step)
-
     samples[:] = trace_numbers[..., None]
 
     with fsspec.open(path, mode="wb") as fp:
@@ -289,8 +277,6 @@ def export_masked_path(tmp_path_factory: pytest.TempPathFactory, raw_headers_env
     raw_headers_enabled = os.getenv("MDIO__IMPORT__RAW_HEADERS") in ("1", "true", "yes", "on")
     path_suffix = "with_raw_headers" if raw_headers_enabled else "without_raw_headers"
 
-    if DEBUG_MODE:
-        return Path(f"tmp/export_masked_{path_suffix}")
     return tmp_path_factory.getbasetemp() / f"export_masked_{path_suffix}"
 
 

tests/test_main.py

@@ -19,7 +19,9 @@ def runner() -> CliRunner:
 # https://github.com/TGSAI/mdio-python/issues/646
 @pytest.mark.skip(reason="CLI hasn't been updated to work with v1 yet.")
 @pytest.mark.dependency
-def test_main_succeeds(runner: CliRunner, segy_input: Path, zarr_tmp: Path) -> None:
+def test_main_succeeds(
+    runner: CliRunner, segy_input: Path, zarr_tmp: Path
+) -> None:  # pragma: no cover - test is skipped
     """It exits with a status code of zero."""
     cli_args = ["segy", "import", str(segy_input), str(zarr_tmp)]
     cli_args.extend(["--header-locations", "181,185"])
@@ -29,8 +31,11 @@ def test_main_succeeds(runner: CliRunner, segy_input: Path, zarr_tmp: Path) -> N
     assert result.exit_code == 0
 
 
+@pytest.mark.skip(reason="CLI hasn't been updated to work with v1 yet.")
 @pytest.mark.dependency(depends=["test_main_succeeds"])
-def test_main_cloud(runner: CliRunner, segy_input_uri: str, zarr_tmp: Path) -> None:
+def test_main_cloud(
+    runner: CliRunner, segy_input_uri: str, zarr_tmp: Path
+) -> None:  # pragma: no cover - tests is skipped
     """It exits with a status code of zero."""
     os.environ["MDIO__IMPORT__CLOUD_NATIVE"] = "true"
     cli_args = ["segy", "import", segy_input_uri, str(zarr_tmp)]
@@ -42,8 +47,9 @@ def test_main_cloud(runner: CliRunner, segy_input_uri: str, zarr_tmp: Path) -> N
     assert result.exit_code == 0
 
 
+@pytest.mark.skip(reason="CLI hasn't been updated to work with v1 yet.")
 @pytest.mark.dependency(depends=["test_main_succeeds"])
-def test_main_info_succeeds(runner: CliRunner, zarr_tmp: Path) -> None:
+def test_main_info_succeeds(runner: CliRunner, zarr_tmp: Path) -> None:  # pragma: no cover - tests is skipped
     """It exits with a status code of zero."""
     cli_args = ["info"]
     cli_args.extend([str(zarr_tmp)])
@@ -52,8 +58,9 @@ def test_main_info_succeeds(runner: CliRunner, zarr_tmp: Path) -> None:
     assert result.exit_code == 0
 
 
+@pytest.mark.skip(reason="CLI hasn't been updated to work with v1 yet.")
 @pytest.mark.dependency(depends=["test_main_succeeds"])
-def test_main_copy(runner: CliRunner, zarr_tmp: Path, zarr_tmp2: Path) -> None:
+def test_main_copy(runner: CliRunner, zarr_tmp: Path, zarr_tmp2: Path) -> None:  # pragma: no cover - tests is skipped
     """It exits with a status code of zero."""
     cli_args = ["copy", str(zarr_tmp), str(zarr_tmp2), "-headers", "-traces"]
 

tests/unit/conftest.py

@@ -1,7 +1,5 @@
 """Helper methods used in unit tests."""
 
-from pathlib import Path
-
 from mdio.builder.dataset_builder import MDIODatasetBuilder
 from mdio.builder.dataset_builder import _BuilderState
 from mdio.builder.dataset_builder import _get_named_dimension
@@ -66,7 +64,7 @@ def validate_variable(
     elif isinstance(container, Dataset):
         var_list = container.variables
         global_coord_list = _get_all_coordinates(container)
-    else:
+    else:  # pragma: no cover
         err_msg = f"Expected MDIODatasetBuilder or Dataset, got {type(container)}"
         raise TypeError(err_msg)
 
@@ -105,7 +103,7 @@ def _get_coordinate(
     in the global coordinate list.
     If the coordinate is stored as a Coordinate object, it is returned directly.
     """
-    if coordinates_or_references is None:
+    if coordinates_or_references is None:  # pragma: no cover
         return None
 
     for c in coordinates_or_references:
@@ -115,7 +113,7 @@ def _get_coordinate(
             # Find the Coordinate in the global list and return it.
             if global_coord_list is not None:
                 cc = next((cc for cc in global_coord_list if cc.name == name), None)
-            if cc is None:
+            if cc is None:  # pragma: no cover
                 msg = f"Pre-existing coordinate named {name!r} is not found"
                 raise ValueError(msg)
             return cc
@@ -124,7 +122,7 @@ def _get_coordinate(
             # Return it.
             return c
 
-    return None
+    return None  # pragma: no cover
 
 
 def _get_all_coordinates(dataset: Dataset) -> list[Coordinate]:
@@ -138,19 +136,6 @@ def _get_all_coordinates(dataset: Dataset) -> list[Coordinate]:
     return list(all_coords.values())
 
 
-def output_path(file_dir: Path, file_name: str, debugging: bool = False) -> Path:
-    """Generate the output path for the test file-system output.
-
-    Note:
-        Use debugging=True, if you need to retain the created files for debugging
-        purposes. Otherwise, the files will be created in-memory and not saved to disk.
-    """
-    if debugging:
-        return file_dir / f"mdio-tests/{file_name}.zarr"
-
-    return file_dir / f"{file_name}.zarr"
-
-
 def make_seismic_poststack_3d_acceptance_dataset(dataset_name: str) -> Dataset:
     """Create in-memory Seismic PostStack 3D Acceptance dataset."""
     ds = MDIODatasetBuilder(

tests/unit/v1/helpers.py

@@ -46,15 +46,7 @@ def _name(self) -> str:
         return self.template_name
 
     def _load_dataset_attributes(self) -> None:
-        return None  # Mock implementation
-
-    def create_dataset(self) -> str:
-        """Create a mock dataset.
-
-        Returns:
-            str: A message indicating the dataset creation.
-        """
-        return f"Mock dataset created by {self.template_name}"
+        return None  # pragma: no cover - Mock implementation
 
 
 def _assert_default_templates(template_names: list[str]) -> None:
@@ -92,12 +84,9 @@ def test_singleton_thread_safety(self) -> None:
         errors = []
 
         def create_instance() -> None:
-            try:
-                instance = TemplateRegistry()
-                instances.append(instance)
-                time.sleep(0.001)  # Small delay to increase contention
-            except Exception as e:
-                errors.append(e)
+            instance = TemplateRegistry()
+            instances.append(instance)
+            time.sleep(0.001)  # Small delay to increase contention
 
         # Create multiple threads trying to create instances
         threads = [threading.Thread(target=create_instance) for _ in range(10)]
@@ -311,13 +300,10 @@ def test_concurrent_registration(self) -> None:
         errors = []
 
         def register_template_worker(template_id: int) -> None:
-            try:
-                template = MockDatasetTemplate(f"template_{template_id}")
-                name = registry.register(template)
-                results.append((template_id, name))
-                time.sleep(0.001)  # Small delay
-            except Exception as e:
-                errors.append((template_id, e))
+            template = MockDatasetTemplate(f"template_{template_id}")
+            name = registry.register(template)
+            results.append((template_id, name))
+            time.sleep(0.001)  # Small delay
 
         # Create multiple threads registering different templates
         threads = [threading.Thread(target=register_template_worker, args=(i,)) for i in range(10)]
@@ -351,28 +337,24 @@ def test_concurrent_access_mixed_operations(self) -> None:
         errors = []
 
         def mixed_operations_worker(worker_id: int) -> None:
-            try:
-                operations_results = []
-
-                # Get existing template
-                if worker_id % 2 == 0:
-                    template = registry.get("initial_0")
-                    operations_results.append(("get", template.template_name))
+            operations_results = []
 
-                # Register new template
-                if worker_id % 3 == 0:
-                    new_template = MockDatasetTemplate(f"worker_{worker_id}")
-                    name = registry.register(new_template)
-                    operations_results.append(("register", name))
+            # Get existing template
+            if worker_id % 2 == 0:
+                template = registry.get("initial_0")
+                operations_results.append(("get", template.template_name))
 
-                # List templates
-                templates = registry.list_all_templates()
-                operations_results.append(("list", len(templates)))
+            # Register new template
+            if worker_id % 3 == 0:
+                new_template = MockDatasetTemplate(f"worker_{worker_id}")
+                name = registry.register(new_template)
+                operations_results.append(("register", name))
 
-                results.append((worker_id, operations_results))
+            # List templates
+            templates = registry.list_all_templates()
+            operations_results.append(("list", len(templates)))
 
-            except Exception as e:
-                errors.append((worker_id, e))
+            results.append((worker_id, operations_results))
 
         # Run concurrent operations
         threads = [threading.Thread(target=mixed_operations_worker, args=(i,)) for i in range(15)]