Unit tests & better warning / error messages

darwin/importer/importer.py

@@ -983,7 +983,7 @@ def process_local_file(
         local_path: Path,
         remote_files: Dict[str, Dict[str, Any]],
         annotation_format: str,
-    ) -> Tuple[Path, List[str], List[str]]:
+    ) -> Tuple[Path, List[str], List[str]]:  # type: ignore
         imported_files: Union[List[dt.AnnotationFile], dt.AnnotationFile, None] = (
             importer(local_path)
         )
@@ -1058,7 +1058,7 @@ def process_local_file(
                         )
                     )
 
-            return local_path, file_slot_warnings, file_slot_errors
+            return local_path, file_slot_warnings, file_slot_errors  # type: ignore
 
     def _validate_slot_alignment_and_import(
         file: dt.AnnotationFile, remote_file: Dict[str, Any], annotation_format: str
@@ -1086,7 +1086,7 @@ def _validate_slot_alignment_and_import(
             ):
                 try:
                     local_path, file_slot_warnings, file_slot_errors = future.result()
-                    slot_warnings, slot_errors = handle_slot_warnings_and_errors(
+                    slot_warnings, slot_errors = _handle_slot_warnings_and_errors(
                         local_path,
                         file_slot_warnings,
                         file_slot_errors,
@@ -1103,7 +1103,7 @@ def _validate_slot_alignment_and_import(
             local_path, file_slot_warnings, file_slot_errors = process_local_file(
                 local_path, remote_files, annotation_format
             )
-            slot_warnings, slot_errors = handle_slot_warnings_and_errors(
+            slot_warnings, slot_errors = _handle_slot_warnings_and_errors(
                 local_path,
                 file_slot_warnings,
                 file_slot_errors,
@@ -1582,7 +1582,7 @@ def _display_slot_alignment_errors(
 
     if slot_warnings:
         console.print(
-            "WARNING: Although they imported successfully, the following files had the following non-blocking warnings:",
+            f"WARNING: Although they imported successfully, the following {len(slot_warnings)} file(s) had the following non-blocking warnings:",
             style="warning",
         )
         for file in slot_warnings:
@@ -1592,11 +1592,11 @@ def _display_slot_alignment_errors(
 
     if slot_errors:
         console.print(
-            f"WARNING: {len(slot_errors)} file(s) have the following blocking issues and could not be. Please resolve these issues and re-import them.",
+            f"WARNING: {len(slot_errors)} file(s) have the following blocking issues and could not be imported. Please resolve these issues and re-import them.",
             style="warning",
         )
         for file in slot_errors:
-            console.print(f"- File: {file}, issues:", style="info")
+            console.print(f"- File: {file}, errors:", style="info")
             for warning in slot_errors[file]:
                 console.print(f"  - {warning}")
 
@@ -1627,7 +1627,7 @@ def _check_annotation_format_slot_compatibility(
         return
     for local_file in local_files:
         remote_file = remote_files[local_file.full_path]
-        if len(remote_file["slot_names"] > 1):
+        if len(remote_file["slot_names"]) > 1:
             raise TypeError(
                 f"You are attempting to import annotations to multi-slotted or multi-channeled items using an annotation format ({annotation_format}) that doesn't support them. To import annotations to multi-slotted or multi-channeled items, please use the Darwin JSON 2.0 format: https://docs.v7labs.com/reference/darwin-json"
             )
@@ -1684,7 +1684,7 @@ def _verify_slot_annotation_alignment(
                 annotation_slot = annotation.slot_names[0]
             except IndexError:
                 file_slot_warnings.append(
-                    f"Annotation imported to multi-slotted item not assigned slot. Uploading to the default slot: {base_slot}"
+                    f"Annotation imported to multi-slotted item {local_file.full_path} not assigned slot. Uploading to the default slot: {base_slot}"
                 )
 
     elif layout_version == 3:  # Multi-channeled item
@@ -1693,27 +1693,50 @@ def _verify_slot_annotation_alignment(
                 annotation_slot = annotation.slot_names[0]
             except IndexError:
                 file_slot_warnings.append(
-                    f"Annotation imported to multi-channeled item not assigned a slot. Uploading to the base slot: {base_slot}"
+                    f"Annotation imported to multi-channeled item {local_file.full_path} not assigned a slot. Uploading to the base slot: {base_slot}"
                 )
                 annotation.slot_names = [base_slot]
                 continue
             if annotation_slot != base_slot:
                 file_slot_errors.append(
-                    f"Annotation is linked to slot {annotation_slot} of the multi-channeled item {local_file.full_path}. Annotations uploaded to multi-channeled items have to be uploaded to the base slot, which for this item is {base_slot}."
+                    f"Annotation linked to slot {annotation_slot} of the multi-channeled item {local_file.full_path}. Annotations uploaded to multi-channeled items have to be uploaded to the base slot, which for this item is {base_slot}."
                 )
     else:
         raise Exception(f"Unknown layout version: {layout_version}")
 
     return local_file, file_slot_warnings, file_slot_errors
 
 
-def handle_slot_warnings_and_errors(
+def _handle_slot_warnings_and_errors(
     local_path,
     file_slot_warnings,
     file_slot_errors,
     slot_warnings,
     slot_errors,
 ):
+    """
+    Adds warnings and errors for a specific local annotation file to the slot_warnings and slot_errors dictionaries
+
+    Parameters
+    ---------
+    local_path : str
+        Path to the local annotation file
+    file_slot_warnings : List[str]
+        A list of non-blocking warnings for the annotation file
+    file_slot_errors : List[str]
+        A list of blocking errors for the annotation file
+    slot_warnings : Dict[str, List[str]]
+        A dictionary of all non-blocking warnings for a specific local annotation file
+    slot_errors : Dict[str, List[str]]
+        A dictionary of all blocking errors for a specific local annotation file
+
+    Returns
+    -------
+    slot_warnings : Dict[str, List[str]]
+        A dictionary of all non-blocking warnings for a specific local annotation file
+    slot_errors : Dict[str, List[str]]
+        A dictionary of all blocking errors for a specific local annotation file
+    """
     if file_slot_warnings:
         if local_path not in slot_warnings:
             slot_warnings[local_path] = []

tests/darwin/importer/importer_test.py

@@ -6,23 +6,28 @@
 from zipfile import ZipFile
 
 import pytest
+from rich.console import Console
 from rich.theme import Theme
 
 from darwin import datatypes as dt
 from darwin.importer import get_importer
 from darwin.importer.importer import (
     _build_attribute_lookup,
     _build_main_annotations_lookup_table,
+    _check_annotation_format_slot_compatibility,
     _console_theme,
+    _display_slot_alignment_errors,
     _find_and_parse,
     _get_annotation_format,
     _get_remote_files,
     _get_slot_names,
+    _handle_slot_warnings_and_errors,
     _import_annotations,
     _is_skeleton_class,
     _overwrite_warning,
     _parse_empty_masks,
     _resolve_annotation_classes,
+    _verify_slot_annotation_alignment,
 )
 
 
@@ -910,3 +915,234 @@ def test__get_annotation_format():
     assert _get_annotation_format(get_importer("nifti")) == "nifti"
     assert _get_annotation_format(get_importer("pascal_voc")) == "pascal_voc"
     assert _get_annotation_format(get_importer("superannotate")) == "superannotate"
+
+
+def test__verify_slot_annotation_alignment_passes_on_single_slotted_items():
+    bounding_box_class = dt.AnnotationClass(
+        name="class1", annotation_type="bounding_box"
+    )
+    local_file = dt.AnnotationFile(
+        path=Path("path/to/file"),
+        filename="file",
+        annotation_classes=set(),
+        annotations=[
+            dt.Annotation(
+                annotation_class=bounding_box_class,
+                data={"x": 5, "y": 10, "w": 5, "h": 10},
+                slot_names=[],
+            ),
+            dt.Annotation(
+                annotation_class=bounding_box_class,
+                data={"x": 15, "y": 20, "w": 15, "h": 20},
+                slot_names=[],
+            ),
+        ],
+        remote_path="/",
+    )
+    remote_file = {"slot_names": ["0"], "layout": {"version": 1}}
+
+    result_file, slot_warnings, slot_errors = _verify_slot_annotation_alignment(
+        local_file, remote_file
+    )
+
+    assert result_file == local_file
+    assert slot_warnings == []
+    assert slot_errors == []
+
+
+def test__verify_slot_annotation_alignment_returns_warning_for_multi_slotted_no_slot():
+    annotation_class = dt.AnnotationClass(name="class1", annotation_type="bounding_box")
+    local_file = dt.AnnotationFile(
+        path=Path("path/to/file"),
+        filename="file",
+        annotation_classes=set(),
+        annotations=[
+            dt.Annotation(
+                annotation_class=annotation_class,
+                data={},
+                slot_names=[],
+            ),
+            dt.Annotation(
+                annotation_class=annotation_class,
+                data={},
+                slot_names=[],
+            ),
+        ],
+        remote_path="/",
+    )
+    remote_file = {"slot_names": ["0", "1"], "layout": {"version": 1}}
+
+    result_file, slot_warnings, slot_errors = _verify_slot_annotation_alignment(
+        local_file, remote_file
+    )
+
+    assert result_file == local_file
+    assert slot_warnings == [
+        "Annotation imported to multi-slotted item /file not assigned slot. Uploading to the default slot: 0",
+        "Annotation imported to multi-slotted item /file not assigned slot. Uploading to the default slot: 0",
+    ]
+    assert slot_errors == []
+
+
+def test__verify_slot_annotation_alignment_returns_warning_for_multi_channeled_no_slot():
+    annotation_class = dt.AnnotationClass(name="class1", annotation_type="bounding_box")
+    local_file = dt.AnnotationFile(
+        path=Path("path/to/file"),
+        filename="file",
+        annotation_classes=set(),
+        annotations=[
+            dt.Annotation(
+                annotation_class=annotation_class,
+                data={},
+                slot_names=[],
+            ),
+            dt.Annotation(
+                annotation_class=annotation_class,
+                data={},
+                slot_names=[],
+            ),
+        ],
+        remote_path="/",
+    )
+    remote_file = {"slot_names": ["0", "1"], "layout": {"version": 3}}
+
+    result_file, slot_warnings, slot_errors = _verify_slot_annotation_alignment(
+        local_file, remote_file
+    )
+
+    assert result_file.annotations[0].slot_names == ["0"]
+    assert result_file.annotations[1].slot_names == ["0"]
+    assert slot_warnings == [
+        "Annotation imported to multi-channeled item /file not assigned a slot. Uploading to the base slot: 0",
+        "Annotation imported to multi-channeled item /file not assigned a slot. Uploading to the base slot: 0",
+    ]
+    assert slot_errors == []
+
+
+def test__verify_slot_annotation_alignment_returns_error_for_multi_channeled_not_base_slot():
+    annotation_class = dt.AnnotationClass(name="class1", annotation_type="bounding_box")
+    local_file = dt.AnnotationFile(
+        path=Path("path/to/file"),
+        filename="file",
+        annotation_classes=set(),
+        annotations=[
+            dt.Annotation(
+                annotation_class=annotation_class,
+                data={},
+                slot_names=["1"],
+            ),
+            dt.Annotation(
+                annotation_class=annotation_class,
+                data={},
+                slot_names=["2"],
+            ),
+        ],
+        remote_path="/",
+    )
+    remote_file = {"slot_names": ["0", "1"], "layout": {"version": 3}}
+
+    result_file, slot_warnings, slot_errors = _verify_slot_annotation_alignment(
+        local_file, remote_file
+    )
+
+    assert result_file == local_file
+    assert slot_warnings == []
+    assert slot_errors == [
+        "Annotation linked to slot 1 of the multi-channeled item /file. Annotations uploaded to multi-channeled items have to be uploaded to the base slot, which for this item is 0.",
+        "Annotation linked to slot 2 of the multi-channeled item /file. Annotations uploaded to multi-channeled items have to be uploaded to the base slot, which for this item is 0.",
+    ]
+
+
+def test__handle_slot_warnings_and_errors():
+    slot_warnings = {"path/to/file": ["warning1", "warning2"]}
+    slot_errors = {}
+    local_path = "path/to/file"
+    file_slot_warnings = ["warning3", "warning4"]
+    file_slot_errors = ["error1", "error2"]
+
+    slot_warnings, slot_errors = _handle_slot_warnings_and_errors(
+        local_path, file_slot_warnings, file_slot_errors, slot_warnings, slot_errors
+    )
+
+    assert slot_warnings == {
+        local_path: ["warning1", "warning2", "warning3", "warning4"]
+    }
+    assert slot_errors == {local_path: ["error1", "error2"]}
+
+
+def test__display_slot_alignment_errors():
+    console = Console()
+    slot_warnings = {
+        "path/to/file_1": ["warning1", "warning2"],
+        "path/to/file_2": ["warning3", "warning4"],
+    }
+    slot_errors = {"path/to/file_1": ["error1", "error2"]}
+
+    with patch.object(console, "print") as mock_print:
+        _display_slot_alignment_errors(slot_warnings, slot_errors, console)
+
+    mock_print.assert_any_call(
+        "WARNING: 1 file(s) have the following blocking issues and could not be imported. Please resolve these issues and re-import them.",
+        style="warning",
+    )
+    mock_print.assert_any_call("- File: path/to/file_1, errors:", style="info")
+    mock_print.assert_any_call("  - error1")
+    mock_print.assert_any_call("  - error2")
+
+    mock_print.assert_any_call(
+        "WARNING: Although they imported successfully, the following 1 file(s) had the following non-blocking warnings:",
+        style="warning",
+    )
+    mock_print.assert_any_call("- File: path/to/file_2, warnings:", style="info")
+    mock_print.assert_any_call("  - warning3")
+    mock_print.assert_any_call("  - warning4")
+
+    # Check that warnings for files with errors are not printed
+    assert "warning1" not in [call.args[0] for call in mock_print.call_args_list]
+    assert "warning2" not in [call.args[0] for call in mock_print.call_args_list]
+
+
+class TestCheckAnnotationFormatSlotCompatibility:
+    @pytest.fixture
+    def local_files(self):
+        return [
+            dt.AnnotationFile(
+                path=Path("/"),
+                filename="path/to/file1.json",
+                annotation_classes=set(),
+                annotations=[],
+                remote_path="/",
+            )
+        ]
+
+    @pytest.fixture
+    def remote_files(self):
+        return {
+            "/path/to/file1.json": {
+                "slot_names": ["0", "1"],
+                "layout": {"version": 1},
+            }
+        }
+
+    def test_darwin_format(self, local_files, remote_files):
+        try:
+            _check_annotation_format_slot_compatibility(
+                local_files, remote_files, "darwin"
+            )
+        except Exception as e:
+            pytest.fail(f"Unexpected exception raised: {e}")
+
+    def test_non_darwin_format_multiple_slots(self, local_files, remote_files):
+        with pytest.raises(TypeError):
+            _check_annotation_format_slot_compatibility(
+                local_files, remote_files, "coco"
+            )
+
+    def test_non_darwin_format_single_slot(self, local_files, remote_files):
+        remote_files["/path/to/file1.json"]["slot_names"] = ["0"]
+        try:
+            _check_annotation_format_slot_compatibility(
+                local_files, remote_files, "coco"
+            )
+        except Exception as e:
+            pytest.fail(f"Unexpected exception raised: {e}")