Add workbench validation for cotype

specifyweb/specify/tree_views.py

@@ -35,7 +35,7 @@
 COMMON_TREES: Tuple[TREE_TABLE, ...] = ['Taxon', 'Storage',
                                         'Geography']
 
-ALL_TRESS: Tuple[TREE_TABLE, ...] = [
+ALL_TREES: Tuple[TREE_TABLE, ...] = [
     *COMMON_TREES, 'Geologictimeperiod', 'Lithostrat', *GEO_TREES]
 
 
@@ -457,7 +457,7 @@ def tree_rank_item_count(request, tree, rankid):
                         "schema": {
                             "type": "object",
                             'properties': {
-                                tree: {"$ref": "#/components/schemas/tree_info"} for tree in ALL_TRESS
+                                tree: {"$ref": "#/components/schemas/tree_info"} for tree in ALL_TREES
                             }
                         }
                     }
@@ -499,7 +499,7 @@ def has_tree_read_permission(tree: TREE_TABLE) -> bool:
     is_paleo_or_geo_discipline = request.specify_collection.discipline.is_paleo_geo()
 
     accessible_trees = tuple(filter(
-        has_tree_read_permission, ALL_TRESS if is_paleo_or_geo_discipline else COMMON_TREES))
+        has_tree_read_permission, ALL_TREES if is_paleo_or_geo_discipline else COMMON_TREES))
 
     result = {}
 

specifyweb/workbench/upload/treerecord.py

@@ -335,6 +335,37 @@ def process_row_item(row_key: str, row_value: Any) -> List[TreeRankCell]:
 
     def rescope_tree_from_row(self, row: Row) -> Tuple["ScopedTreeRecord", Optional["WorkBenchParseFailure"]]:
         """Rescope tree from row data."""
+
+        def validate_trees_with_cotype(row: Row, treedefs_in_row: Set[int]):
+            # TODO: Need a better way to do this
+            # Find a way to send cotype column when ScopedTreeRecord instance is created?
+            COL_NAMES = ["Type", "Collection Object Type"]
+            def find_cotype_in_row(row: Row):
+                for col_name, value in row.items():
+                    if col_name in COL_NAMES:
+                        return col_name, value
+
+                return None
+
+            def get_cotype_tree_def(cotype_name: str):
+                # TODO: Need to scope by collection?
+                cotypes = models.Collectionobjecttype.objects.filter(name=cotype_name)
+                return cotypes[0].taxontreedef.id if len(cotypes) > 0 else None
+
+            cotype = find_cotype_in_row(row)
+            if not cotype: return None
+
+            cotype_column, cotype_value = cotype
+
+            cotype_treedef = get_cotype_tree_def(cotype_value)
+            if not cotype_treedef: return None
+
+            # Check only the first treedef assuming all ranks belong to same tree
+            # Validation for multiple ranks is done elsewhere
+            if len(treedefs_in_row) > 0 and cotype_treedef == list(treedefs_in_row)[0]:
+                return None
+
+            return self, WorkBenchParseFailure('Invalid type for selected tree rank(s)', {}, cotype_column)
 
         # Get models based on the name
         def get_models(name: str):
@@ -420,16 +451,10 @@ def check_root(root):
         result = handle_multiple_or_no_treedefs(unique_treedef_ids, targeted_treedefids, ranks_columns_in_row_not_null)
         if result:
             return result
-
-        # Determine the target treedef based on the columns that are not null
-        targeted_treedefids = {rank_column.treedef_id for rank_column in ranks_columns_in_row_not_null}
-        if not targeted_treedefids:
-            # return self, WorkBenchParseFailure('noRanksInRow', {}, None)
-            return self, None
-        elif len(targeted_treedefids) > 1 and len(unique_treedef_ids) > 1:
-            logger.warning(f"Multiple treedefs found in row: {targeted_treedefids}")
-            error_col_name = list(ranks_columns_in_row_not_null)[0].column_fullname
-            return self, WorkBenchParseFailure("multipleRanksInRow", {}, error_col_name)
+
+        result = validate_trees_with_cotype(row, targeted_treedefids)
+        if result:
+            return result
 
         target_rank_treedef_id = targeted_treedefids.pop()
         target_rank_treedef = get_target_rank_treedef(tree_def_model, target_rank_treedef_id)

specifyweb/workbench/upload/upload.py

@@ -15,6 +15,7 @@
 from specifyweb.specify.auditlog import auditlog
 from specifyweb.specify.datamodel import Table
 from specifyweb.specify.tree_extras import renumber_tree, set_fullnames
+from specifyweb.specify.tree_views import ALL_TREES
 from specifyweb.workbench.upload.upload_table import DeferredScopeUploadTable, ScopedUploadTable
 
 from . import disambiguation
@@ -290,9 +291,9 @@ def fixup_trees(upload_plan: ScopedUploadable, results: List[UploadResult]) -> N
     treedefs = upload_plan.get_treedefs()
 
     to_fix = [
-        tree
-        for tree in ('taxon', 'geography', 'geologictimeperiod', 'lithostrat', 'storage', 'tectonicunit')
-        if any(changed_tree(tree, r) for r in results)
+        tree.lower()
+        for tree in ALL_TREES
+        if any(changed_tree(tree.lower(), r) for r in results)
     ]
 
     for tree in to_fix: