Added tests for upload_update and upload_ignore

Author: Claire Pajot

nucleus/annotation.py

@@ -23,6 +23,7 @@
     POLYGON_TYPE,
     POSITION_KEY,
     REFERENCE_ID_KEY,
+    TAXONOMY_NAME_KEY,
     TYPE_KEY,
     VERTICES_KEY,
     WIDTH_KEY,
@@ -319,9 +320,9 @@ def to_payload(self) -> dict:
 @dataclass
 class CategoryAnnotation(Annotation):
     label: str
+    taxonomy_name: str
     reference_id: Optional[str] = None
     item_id: Optional[str] = None
-    annotation_id: Optional[str] = None
     metadata: Optional[Dict] = None
 
     def __post_init__(self):
@@ -330,22 +331,21 @@ def __post_init__(self):
 
     @classmethod
     def from_json(cls, payload: dict):
-        # TODO: Remove? geometry = payload.get(GEOMETRY_KEY, {})
         return cls(
             label=payload.get(LABEL_KEY, 0),
+            taxonomy_name=payload.get(TAXONOMY_NAME_KEY, None),
             reference_id=payload.get(REFERENCE_ID_KEY, None),
             item_id=payload.get(DATASET_ITEM_ID_KEY, None),
-            annotation_id=payload.get(ANNOTATION_ID_KEY, None),
             metadata=payload.get(METADATA_KEY, {}),
         )
 
     def to_payload(self) -> dict:
         return {
             LABEL_KEY: self.label,
+            TAXONOMY_NAME_KEY: self.taxonomy_name,
             TYPE_KEY: CATEGORY_TYPE,
             GEOMETRY_KEY: {},
             REFERENCE_ID_KEY: self.reference_id,
-            ANNOTATION_ID_KEY: self.annotation_id,
             METADATA_KEY: self.metadata,
         }
 

nucleus/constants.py

@@ -84,6 +84,7 @@
 STATUS_CODE_KEY = "status_code"
 STATUS_KEY = "status"
 SUCCESS_STATUS_CODES = [200, 201, 202]
+TAXONOMY_NAME_KEY = "taxonomy_name"
 TYPE_KEY = "type"
 UPDATED_ITEMS = "updated_items"
 UPDATE_KEY = "update"

tests/helpers.py

@@ -162,7 +162,7 @@ def reference_id_from_url(url):
     {
         "label": f"[Pytest] Category Annotation ${i}",
         "reference_id": reference_id_from_url(TEST_IMG_URLS[i]),
-        "annotation_id": f"[Pytest] Category Annotation Annotation Id{i}",
+        "taxonomy_name": f"[Pytest] Category Taxonomy{i}",
     }
     for i in range(len(TEST_IMG_URLS))
 ]
@@ -278,9 +278,8 @@ def assert_category_annotation_matches_dict(
     annotation_instance, annotation_dict
 ):
     assert annotation_instance.label == annotation_dict["label"]
-
     assert (
-        annotation_instance.annotation_id == annotation_dict["annotation_id"]
+        annotation_instance.taxonomy_name == annotation_dict["taxonomy_name"]
     )
 
 

tests/test_annotation.py

@@ -66,7 +66,9 @@ def dataset(CLIENT):
 
 def test_box_gt_upload(dataset):
     annotation = BoxAnnotation(**TEST_BOX_ANNOTATIONS[0])
+    print(annotation)
     response = dataset.annotate(annotations=[annotation])
+    print(response)
 
     assert response["dataset_id"] == dataset.id
     assert response["annotations_processed"] == 1
@@ -324,6 +326,63 @@ def test_polygon_gt_upload_ignore(dataset):
         response_annotation, TEST_POLYGON_ANNOTATIONS[0]
     )
 
+
+def test_category_gt_upload_update(dataset):
+    annotation = CategoryAnnotation.from_json(TEST_CATEGORY_ANNOTATIONS[0])
+    response = dataset.annotate(annotations=[annotation])
+
+    assert response["annotations_processed"] == 1
+
+    # Copy so we don't modify the original.
+    annotation_update_params = dict(TEST_CATEGORY_ANNOTATIONS[1])
+    annotation_update_params["reference_id"] = TEST_CATEGORY_ANNOTATIONS[0][
+        "reference_id"
+    ]
+
+    annotation_update = CategoryAnnotation.from_json(annotation_update_params)
+    response = dataset.annotate(annotations=[annotation_update], update=True)
+
+    assert response["annotations_processed"] == 1
+    assert response["annotations_ignored"] == 0
+
+    response = dataset.refloc(annotation.reference_id)["annotations"][
+        "category"
+    ]
+    assert len(response) == 1
+    response_annotation = response[0]
+    assert_category_annotation_matches_dict(
+        response_annotation, annotation_update_params
+    )
+
+
+def test_category_gt_upload_ignore(dataset):
+    annotation = CategoryAnnotation.from_json(TEST_CATEGORY_ANNOTATIONS[0])
+    response = dataset.annotate(annotations=[annotation])
+
+    assert response["annotations_processed"] == 1
+
+    # Copy so we don't modify the original.
+    annotation_update_params = dict(TEST_CATEGORY_ANNOTATIONS[1])
+    annotation_update_params["reference_id"] = TEST_CATEGORY_ANNOTATIONS[0][
+        "reference_id"
+    ]
+
+    annotation_update = CategoryAnnotation.from_json(annotation_update_params)
+    # Default behavior is ignore.
+    response = dataset.annotate(annotations=[annotation_update])
+
+    assert response["annotations_processed"] == 0
+    assert response["annotations_ignored"] == 1
+
+    response = dataset.refloc(annotation.reference_id)["annotations"][
+        "category"
+    ]
+    assert len(response) == 1
+    response_annotation = response[0]
+    assert_category_annotation_matches_dict(
+        response_annotation, TEST_CATEGORY_ANNOTATIONS[0]
+    )
+
     @pytest.mark.integration
     def test_box_gt_deletion(dataset):
         annotation = BoxAnnotation(**TEST_BOX_ANNOTATIONS[0])