test(damicore-distance): add unit tests for all modules

models/damicore_distance/tests/conftest.py

@@ -1,7 +0,0 @@
-import pytest
-
-
-@pytest.fixture
-def sample_texts() -> tuple[str, str]:
-    """Provide a pair of texts for distance tests."""
-    return ("hello world", "hello earth")

models/damicore_distance/tests/test_compressors.py

@@ -0,0 +1,24 @@
+import pytest
+
+from damicore_distance.compressors import Compressor, get_compressed_size
+
+_COMPRESSIBLE = b"a" * 1000
+
+
+@pytest.mark.unit
+def test_get_compressed_size_returns_positive_integer() -> None:
+    result = get_compressed_size(_COMPRESSIBLE, Compressor.GZIP)
+    assert result > 0
+
+
+@pytest.mark.unit
+def test_get_compressed_size_level_9_smaller_than_level_1() -> None:
+    size_level_1 = get_compressed_size(_COMPRESSIBLE, Compressor.GZIP, compression_level=1)
+    size_level_9 = get_compressed_size(_COMPRESSIBLE, Compressor.GZIP, compression_level=9)
+    assert size_level_9 < size_level_1
+
+
+@pytest.mark.unit
+def test_get_compressed_size_raises_for_unsupported_compressor() -> None:
+    with pytest.raises(ValueError, match="Unsupported compressor"):
+        get_compressed_size(b"data", "bzip2")  # type: ignore[arg-type]

models/damicore_distance/tests/test_core.py

@@ -0,0 +1,126 @@
+import json
+import pathlib
+
+import numpy as np
+import pytest
+
+from damicore_distance.compressors import Compressor
+from damicore_distance.core import (
+    AlgorithmType,
+    DistanceMatrixInput,
+    DistanceMatrixOutput,
+    MetricStrategy,
+    StatusType,
+    _ncd,  # pyright: ignore[reportPrivateUsage]
+    compute_distance_matrix,
+    ncd_matrix,
+)
+
+
+@pytest.mark.unit
+def test_ncd_returns_zero_for_identical_sizes() -> None:
+    assert _ncd(100, 100, 100) == 0.0
+
+
+@pytest.mark.unit
+def test_ncd_clamps_to_one_when_result_exceeds_range() -> None:
+    assert _ncd(10, 10, 1000) == 1.0
+
+
+@pytest.mark.unit
+def test_ncd_clamps_to_zero_when_result_is_negative() -> None:
+    assert _ncd(100, 100, 50) == 0.0
+
+
+@pytest.mark.unit
+def test_ncd_returns_zero_when_max_c_is_zero() -> None:
+    assert _ncd(0, 0, 0) == 0.0
+
+
+@pytest.mark.unit
+def test_ncd_matrix_returns_n_by_n_symmetric_matrix(tmp_path: pathlib.Path) -> None:
+    (tmp_path / "a.txt").write_bytes(b"hello world" * 10)
+    (tmp_path / "b.txt").write_bytes(b"hello earth" * 10)
+    (tmp_path / "c.txt").write_bytes(b"x" * 100)
+
+    matrix, filenames = ncd_matrix(str(tmp_path), Compressor.GZIP)
+
+    assert matrix.shape == (3, 3)
+    assert len(filenames) == 3
+    np.testing.assert_array_equal(matrix, matrix.T)
+
+
+@pytest.mark.unit
+def test_ncd_matrix_diagonal_is_zero(tmp_path: pathlib.Path) -> None:
+    (tmp_path / "a.txt").write_bytes(b"hello world" * 10)
+    (tmp_path / "b.txt").write_bytes(b"hello earth" * 10)
+
+    matrix, _ = ncd_matrix(str(tmp_path), Compressor.GZIP)
+
+    np.testing.assert_array_equal(np.diag(matrix), np.zeros(2, dtype=np.float32))
+
+
+@pytest.mark.unit
+def test_compute_distance_matrix_accepts_pydantic_input(tmp_path: pathlib.Path) -> None:
+    (tmp_path / "a.txt").write_bytes(b"hello world" * 10)
+    (tmp_path / "b.txt").write_bytes(b"hello earth" * 10)
+    output_path = str(tmp_path / "output.csv")
+
+    input_contract = DistanceMatrixInput(
+        input_directory=str(tmp_path),
+        metric_strategy=MetricStrategy(
+            algorithm=AlgorithmType.NCD,
+            compressor=Compressor.GZIP,
+        ),
+        output_destination=output_path,
+    )
+
+    result = compute_distance_matrix(input_contract)
+
+    assert isinstance(result, DistanceMatrixOutput)
+    assert result.status == StatusType.SUCCESS
+    assert result.total_files_analyzed == 2
+    assert result.matrix_dimensions == "2x2"
+    assert result.metric_used == AlgorithmType.NCD
+    assert result.compressor_used == Compressor.GZIP
+    assert result.output_file_path == output_path
+
+
+@pytest.mark.unit
+def test_compute_distance_matrix_accepts_json_string(tmp_path: pathlib.Path) -> None:
+    (tmp_path / "a.txt").write_bytes(b"hello world" * 10)
+    (tmp_path / "b.txt").write_bytes(b"hello earth" * 10)
+    output_path = str(tmp_path / "output.csv")
+
+    json_input = json.dumps(
+        {
+            "input_directory": str(tmp_path),
+            "metric_strategy": {
+                "algorithm": "ncd",
+                "compressor": "gzip",
+                "compression_level": 9,
+            },
+            "output_destination": output_path,
+        }
+    )
+
+    result = compute_distance_matrix(json_input)
+
+    assert result.status == StatusType.SUCCESS
+
+
+@pytest.mark.unit
+def test_compute_distance_matrix_raises_for_unsupported_algorithm(tmp_path: pathlib.Path) -> None:
+    strategy = MetricStrategy.model_construct(
+        algorithm="unsupported_algo",  # type: ignore[arg-type]  # bypass Pydantic validation to reach the defensive case _ branch
+        compressor=Compressor.GZIP,
+        compression_level=9,
+    )
+    input_contract = DistanceMatrixInput.model_construct(
+        input_directory=str(tmp_path),
+        metric_strategy=strategy,
+        output_destination=str(tmp_path / "out.csv"),
+    )
+
+    with pytest.raises(NotImplementedError):
+        compute_distance_matrix(input_contract)

models/damicore_distance/tests/test_io.py

@@ -0,0 +1,65 @@
+import csv
+import pathlib
+
+import numpy as np
+import pytest
+
+from damicore_distance.io import export_ncd_to_csv, get_file_data, get_files_from_directory
+
+
+@pytest.mark.unit
+def test_get_files_from_directory_returns_filenames(tmp_path: pathlib.Path) -> None:
+    (tmp_path / "a.txt").write_bytes(b"aaa")
+    (tmp_path / "b.txt").write_bytes(b"bbb")
+
+    result = get_files_from_directory(str(tmp_path))
+
+    assert set(result) == {"a.txt", "b.txt"}
+
+
+@pytest.mark.unit
+def test_get_files_from_directory_raises_for_nonexistent_path() -> None:
+    with pytest.raises(ValueError, match="does not exist"):
+        get_files_from_directory("/nonexistent/path/that/does/not/exist")
+
+
+@pytest.mark.unit
+def test_get_file_data_returns_bytes(tmp_path: pathlib.Path) -> None:
+    file = tmp_path / "sample.bin"
+    file.write_bytes(b"hello world")
+
+    assert get_file_data(str(file)) == b"hello world"
+
+
+@pytest.mark.unit
+def test_get_file_data_raises_for_nonexistent_file() -> None:
+    with pytest.raises(ValueError, match="does not exist"):
+        get_file_data("/nonexistent/file.txt")
+
+
+@pytest.mark.unit
+def test_export_ncd_to_csv_writes_correct_headers_and_values(tmp_path: pathlib.Path) -> None:
+    matrix = np.array([[0.0, 0.5], [0.5, 0.0]], dtype=np.float32)
+    filenames = ["a.txt", "b.txt"]
+    output_path = str(tmp_path / "matrix.csv")
+
+    export_ncd_to_csv(matrix, filenames, output_path)
+
+    with open(output_path) as f:
+        rows = list(csv.reader(f))
+
+    assert rows[0] == ["", "a.txt", "b.txt"]
+    assert rows[1][0] == "a.txt"
+    assert rows[2][0] == "b.txt"
+    assert float(rows[1][2]) == 0.5
+    assert float(rows[2][1]) == 0.5
+
+
+@pytest.mark.unit
+def test_export_ncd_to_csv_creates_intermediate_directories(tmp_path: pathlib.Path) -> None:
+    matrix = np.array([[0.0]], dtype=np.float32)
+    output_path = str(tmp_path / "nested" / "deep" / "matrix.csv")
+
+    export_ncd_to_csv(matrix, ["a.txt"], output_path)
+
+    assert pathlib.Path(output_path).exists()