Sp6 to Sp7 add missing constraints and tables

specifyweb/backend/locality_update_tool/tests/test_localityupdate_status.py

@@ -36,7 +36,7 @@ def test_localityupdate_not_exist(self):
         self._assertStatusCodeEqual(response, http.HttpResponseNotFound.status_code)
         self.assertEqual(response.content.decode(), f"The localityupdate with task id '{task_id}' was not found")
 
-    @patch("specifyweb.specify.views.update_locality_task.AsyncResult")
+    @patch("specifyweb.backend.locality_update_tool.views.update_locality_task.AsyncResult")
     def test_failed(self, AsyncResult: Mock):
         mock_result = Mock()
         mock_result.state = CELERY_TASK_STATE.FAILURE
@@ -70,7 +70,7 @@ def test_failed(self, AsyncResult: Mock):
             }
         )
 
-    @patch("specifyweb.specify.views.update_locality_task.AsyncResult")
+    @patch("specifyweb.backend.locality_update_tool.views.update_locality_task.AsyncResult")
     def test_parse_failed(self, AsyncResult: Mock):
         mock_result = Mock()
         mock_result.state = CELERY_TASK_STATE.SUCCESS
@@ -98,7 +98,7 @@ def test_parse_failed(self, AsyncResult: Mock):
             }
         )
 
-    @patch("specifyweb.specify.views.update_locality_task.AsyncResult")
+    @patch("specifyweb.backend.locality_update_tool.views.update_locality_task.AsyncResult")
     def test_parsed(self, AsyncResult: Mock):
         mock_result = Mock()
         mock_result.state = CELERY_TASK_STATE.SUCCESS
@@ -149,7 +149,7 @@ def test_parsed(self, AsyncResult: Mock):
             }
         )
 
-    @patch("specifyweb.specify.views.update_locality_task.AsyncResult")
+    @patch("specifyweb.backend.locality_update_tool.views.update_locality_task.AsyncResult")
     def test_succeeded(self, AsyncResult: Mock):
         mock_result = Mock()
         mock_result.state = LocalityUpdateStatus.SUCCEEDED
@@ -181,7 +181,7 @@ def test_succeeded(self, AsyncResult: Mock):
             }
         )
 
-    @patch("specifyweb.specify.views.update_locality_task.AsyncResult")
+    @patch("specifyweb.backend.locality_update_tool.views.update_locality_task.AsyncResult")
     def test_succeeded_locality_rows(self, AsyncResult: Mock):
         mock_result = Mock()
         mock_result.state = LocalityUpdateStatus.SUCCEEDED

specifyweb/backend/locality_update_tool/tests/test_parse_field.py

@@ -33,7 +33,7 @@ def test_no_ui_formatter(self):
 
         self.assertEqual(parsed_with_value, parsed_with_value_result)
 
-    @patch("specifyweb.specify.update_locality.get_uiformatter")
+    @patch("specifyweb.backend.locality_update_tool.update_locality.get_uiformatter")
     def test_cnn_formatter(self, get_uiformatter: Mock):
 
         get_uiformatter.return_value = UIFormatter(

specifyweb/backend/locality_update_tool/update_locality.py

@@ -379,11 +379,19 @@ def parse_locality_set(collection, raw_headers: list[str], data: list[list[str]]
         locality_id: int | None = None if len(
             locality_query) != 1 else locality_query[0].id
 
-        parsed_locality_fields = [parse_field(
-            collection, 'Locality', dict['field'], dict['value'], locality_id, row_number) for dict in locality_values if dict['value'].strip() != ""]
+        parsed_locality_fields = [
+            parse_field(
+                collection, 'Locality', d['field'], d['value'], locality_id, row_number
+            )
+            for d in locality_values
+        ]
 
-        parsed_geocoorddetail_fields = [parse_field(
-            collection, 'Geocoorddetail', dict["field"], dict['value'], locality_id, row_number) for dict in geocoorddetail_values if dict['value'].strip() != ""]
+        parsed_geocoorddetail_fields = [
+            parse_field(
+                collection, 'Geocoorddetail', d['field'], d['value'], locality_id, row_number
+            )
+            for d in geocoorddetail_values
+        ]
 
         parsed_row, parsed_errors = merge_parse_results(
             [*parsed_locality_fields, *parsed_geocoorddetail_fields], locality_id, row_number)

specifyweb/backend/merge/record_merging.py

@@ -195,7 +195,8 @@ def record_merge_fx(model_name: str, old_model_ids: list[int], new_model_id: int
 
     # Get all of the columns in all of the tables of specify the are foreign keys referencing model ID
     foreign_key_cols = []
-    for table in spmodels.datamodel.tables:
+    # for table in spmodels.datamodel.tables:
+    for table in (t for t in spmodels.datamodel.tables if not getattr(t, "skip", False)):
         for relationship in table.relationships:
             if relationship.relatedModelName.lower() == model_name.lower() and not relationship.type.endswith('to-many'):
                 foreign_key_cols.append((table.name, relationship.name))

specifyweb/backend/stored_queries/build_models.py

@@ -42,7 +42,7 @@ def make_table(datamodel: Datamodel, tabledef: Table):
 
 def make_foreign_key(datamodel: Datamodel, reldef: Relationship):
     remote_tabledef = datamodel.get_table(reldef.relatedModelName) # TODO: this could be a method of relationship
-    if remote_tabledef is None:
+    if remote_tabledef is None or getattr(remote_tabledef, "skip", False):
         return
 
     fk_target = '.'.join((remote_tabledef.table, remote_tabledef.idColumn))
@@ -84,7 +84,7 @@ def make_column(flddef: Field):
 }
 
 def make_tables(datamodel: Datamodel):
-    return {td.table: make_table(datamodel, td) for td in datamodel.tables}
+    return {td.table: make_table(datamodel, td) for td in iter_included_tables(datamodel)}
 
 def make_classes(datamodel: Datamodel):
     def make_class(tabledef):
@@ -97,7 +97,7 @@ def make_class(tabledef):
             },
         )
 
-    return {td.name: make_class(td) for td in datamodel.tables}
+    return {td.name: make_class(td) for td in iter_included_tables(datamodel)}
 
 def map_classes(datamodel: Datamodel, tables: list[Table], classes):
 
@@ -106,22 +106,75 @@ def map_class(tabledef):
         table = tables[ tabledef.table ]
 
         def make_relationship(reldef):
-            if not hasattr(reldef, 'column') or not reldef.column or reldef.relatedModelName not in classes:
+            has_back_populates = False
+            if reldef.relatedModelName not in classes:
                 return
 
-            remote_class = classes[ reldef.relatedModelName ]
-            column = getattr(table.c, reldef.column)
-
-            relationship_args = {'foreign_keys': column}
-            if remote_class is cls:
-                relationship_args['remote_side'] = table.c[ tabledef.idColumn ]
-
-            if hasattr(reldef, 'otherSideName') and reldef.otherSideName:
-                backref_args = {'uselist': reldef.type != 'one-to-one'}
-
-                relationship_args['backref'] = orm.backref(reldef.otherSideName, **backref_args)
-
-            return reldef.name, orm.relationship(remote_class, **relationship_args)
+            remote_class = classes[reldef.relatedModelName]
+            remote_tabledef = datamodel.get_table(reldef.relatedModelName)
+            remote_table = tables.get(remote_tabledef.table) if remote_tabledef else None
+
+            # Handle standard to-one relationships with an explicit column.
+            if getattr(reldef, "column", None):
+                column = getattr(table.c, reldef.column)
+                relationship_args = {"foreign_keys": column}
+
+                if remote_class is cls:
+                    relationship_args["remote_side"] = table.c[tabledef.idColumn]
+
+                # If the remote side declares a one-to-many back-link without a column,
+                # wire the two sides together.
+                reverse_one_to_many = None
+                if remote_tabledef:
+                    reverse_one_to_many = next(
+                        (
+                            r
+                            for r in remote_tabledef.relationships
+                            if r.relatedModelName == tabledef.name
+                            and r.type == "one-to-many"
+                        ),
+                        None,
+                    )
+                if reverse_one_to_many is not None:
+                    relationship_args["back_populates"] = reverse_one_to_many.name
+                    has_back_populates = True
+
+                if (not has_back_populates) and getattr(reldef, "otherSideName", None):
+                    backref_args = {"uselist": reldef.type != "one-to-one"}
+                    relationship_args["backref"] = orm.backref(
+                        reldef.otherSideName, **backref_args
+                    )
+
+                return reldef.name, orm.relationship(remote_class, **relationship_args)
+
+            # Handle one-to-many relationships defined on the parent side (no column specified).
+            if reldef.type == "one-to-many" and remote_table is not None:
+                # Find a reverse many-to-one pointing back to this table with a FK column.
+                reverse_rel = next(
+                    (
+                        r
+                        for r in remote_tabledef.relationships
+                        if r.relatedModelName == tabledef.name
+                        and getattr(r, "column", None)
+                        and r.type in ("many-to-one", "one-to-one")
+                    ),
+                    None,
+                )
+                if reverse_rel is None:
+                    return
+
+                fk_column = remote_table.c[reverse_rel.column]
+                relationship_args = {
+                    "foreign_keys": fk_column,
+                    "primaryjoin": table.c[tabledef.idColumn] == fk_column,
+                }
+
+                # Keep both sides linked when possible.
+                relationship_args["back_populates"] = reverse_rel.name
+
+                return reldef.name, orm.relationship(remote_class, **relationship_args)
+
+            return
 
         id_column = table.c[tabledef.idColumn]
         properties = {
@@ -139,6 +192,11 @@ def make_relationship(reldef):
 
         orm.mapper(cls, table, properties=properties)
 
-    for tabledef in datamodel.tables:
+    for tabledef in iter_included_tables(datamodel):
         map_class(tabledef)
 
+def iter_included_tables(datamodel: Datamodel):
+    for td in datamodel.tables:
+        if getattr(td, "skip", False):
+            continue
+        yield td

specifyweb/backend/stored_queries/tests/tests.py

@@ -183,6 +183,10 @@ def validate_sqlalchemy_model(datamodel_table):
         known_fields = datamodel_table.all_fields
 
         for field in known_fields:
+            if field.is_relationship:
+                remote_td = spmodels.datamodel.get_table(field.relatedModelName)
+                if remote_td is not None and getattr(remote_td, "skip", False):
+                    continue
 
             in_sql = getattr(orm_table, field.name, None) or getattr(
                 orm_table, field.name.lower(), None
@@ -244,7 +248,8 @@ def validate_sqlalchemy_model(datamodel_table):
         return {key: value for key, value in table_errors.items() if len(value) > 0}
 
     def test_sqlalchemy_model_errors(self):
-        for table in spmodels.datamodel.tables:
+        # for table in spmodels.datamodel.tables:
+        for table in (t for t in spmodels.datamodel.tables if not getattr(t, "skip", False)):
             table_errors = SQLAlchemyModelTest.validate_sqlalchemy_model(table)
             self.assertTrue(
                 len(table_errors) == 0 or table.name in expected_errors,

specifyweb/backend/stored_queries/tests/tests_legacy.py

@@ -345,6 +345,10 @@ def validate_sqlalchemy_model(datamodel_table):
     known_fields = datamodel_table.all_fields
 
     for field in known_fields:
+        if field.is_relationship:
+            remote_td = spmodels.datamodel.get_table(field.relatedModelName)
+            if remote_td is not None and getattr(remote_td, "skip", False):
+                continue
 
         in_sql = getattr(orm_table, field.name, None) or getattr(orm_table, field.name.lower(), None)
 
@@ -385,7 +389,8 @@ def validate_sqlalchemy_model(datamodel_table):
 
 class SQLAlchemyModelTest(TestCase):
     def test_sqlalchemy_model_errors(self):
-        for table in spmodels.datamodel.tables:
+        # for table in spmodels.datamodel.tables:
+        for table in (t for t in spmodels.datamodel.tables if not getattr(t, "skip", False)):
             table_errors = validate_sqlalchemy_model(table)
             self.assertTrue(len(table_errors) == 0 or table.name in expected_errors, f"Did not find {table.name}. Has errors: {table_errors}")
             if 'not_found' in table_errors:

specifyweb/backend/trees/extras.py

@@ -411,18 +411,48 @@ def synonymize(node, into, agent, user=None, collection=None):
 
     # This check can be disabled by a remote pref
     import specifyweb.backend.context.app_resource as app_resource
-    collection_prefs_json, _, __ = app_resource.get_app_resource(collection, user, 'CollectionPreferences')
-    if collection_prefs_json is not None:
-            collection_prefs_dict = json.loads(collection_prefs_json)
 
-    treeManagement_pref = collection_prefs_dict.get('treeManagement', {})
+    collection_prefs_dict = {}  # always defined
+
+    res = app_resource.get_app_resource(collection, user, 'CollectionPreferences')
+    force_checks = (collection is None or user is None)
+    if res is not None:
+        collection_prefs_json, _, __ = res
+        if collection_prefs_json:
+            try:
+                collection_prefs_dict = json.loads(collection_prefs_json) or {}
+            except Exception:
+                collection_prefs_dict = {}
 
+    import specifyweb.backend.context.app_resource as app_resource
+
+    treeManagement_pref = collection_prefs_dict.get('treeManagement', {})
+    if force_checks and target.children.exists():
+        raise TreeBusinessRuleException(
+            f'Synonymizing "{node.fullname}" to "{into.fullname}" which has children',
+            {"tree": "Taxon",
+             "localizationKey": "nodeSynonimizeWithChildren",
+             "node": {
+                "id": node.id,
+                "rankid": node.rankid,
+                "fullName": node.fullname,
+                "children": list(node.children.values('id', 'fullname'))
+             },
+             "parent": {
+                "id": into.id,
+                "rankid": into.rankid,
+                "fullName": into.fullname,
+                "parentid": into.parent.id,
+                "children": list(into.children.values('id', 'fullname'))
+             }}
+        )
+    force_checks = (collection is None or user is None)
     synonymized = treeManagement_pref.get('synonymized', {}) \
         if isinstance(treeManagement_pref, dict) else {}
 
     add_synonym_enabled = synonymized.get(r'^sp7\.allow_adding_child_to_synonymized_parent\.' + node.specify_model.name + '=(.+)', False) if isinstance(synonymized, dict) else False
 
-    if node.children.count() > 0 and (add_synonym_enabled is True):
+    if node.children.count() > 0 and (force_checks or add_synonym_enabled is False):
         raise TreeBusinessRuleException(
             f'Synonymizing node "{node.fullname}" which has children',
             {"tree" : "Taxon",