[test]: unit-tests for batch edit

specifyweb/backend/stored_queries/tests/test_batch_edit.py

@@ -2342,4 +2342,55 @@ def test_column_key_collision(self):
         (headers, rows, packs, plan, order) = run_batch_edit_query(props)
 
         self.assertEqual(headers, expected_captions)
+
+    # Tests to verify batch edit relationships are not editable. 
+    def _make_determination(self):
+        return models.Determination.objects.create(
+            collectionobject=self.collectionobjects[0],
+            remarks="Remarks for collection object 1, det 1",
+        )
+
+    def _run_omitting_relationships(self, omit_relationships: bool):
+        base_table = "collectionobject"
+        query_paths = [
+            ["catalognumber"],
+            ["cataloger", "firstname"],
+            ["determinations", "remarks"],
+        ]
+        added = [(base_table, *path) for path in query_paths]
+        query_fields = [
+            BatchEditPack._query_field(QueryFieldSpec.from_path(path), 0)
+            for path in added
+        ]
+        props = self.build_props(query_fields, base_table)
+        props["omit_relationships"] = omit_relationships
+        (headers, rows, packs, plan, order) = run_batch_edit_query(props)
+        return headers, plan["uploadable"]["uploadTable"]
+
+    @patch(OBJ_FORMATTER_PATH, new=fake_obj_formatter)
+    def test_relationships_are_editable_by_default(self):
+        self._make_determination()
+        (headers, upload_table) = self._run_omitting_relationships(False)
+        self.assertIn("cataloger", upload_table["toOne"])
+        self.assertIn("determinations", upload_table["toMany"])
+
+    @patch(OBJ_FORMATTER_PATH, new=fake_obj_formatter)
+    def test_omitting_relationships_removes_them_from_the_upload_plan(self):
+        self._make_determination()
+        (headers, upload_table) = self._run_omitting_relationships(True)
+        self.assertEqual(upload_table["toOne"], {})
+        self.assertEqual(upload_table["toMany"], {})
+
+    @patch(OBJ_FORMATTER_PATH, new=fake_obj_formatter)
+    def test_base_table_fields_stay_editable_without_relationships(self):
+        self._make_determination()
+        (headers, upload_table) = self._run_omitting_relationships(True)
+        self.assertIn("catalognumber", upload_table["wbcols"])
+
+    @patch(OBJ_FORMATTER_PATH, new=fake_obj_formatter)
+    def test_relationship_columns_are_still_shown(self):
+        self._make_determination()
+        (shown_headers, _) = self._run_omitting_relationships(False)
+        (omitted_headers, _) = self._run_omitting_relationships(True)
+        self.assertEqual(omitted_headers, shown_headers)
 

specifyweb/backend/workbench/upload/tests/test_batch_edit_rollback.py

@@ -0,0 +1,125 @@
+import json
+from specifyweb.backend.stored_queries.batch_edit import make_dataset, run_batch_edit_query # type: ignore
+from specifyweb.backend.stored_queries.queryfield import QueryField
+from specifyweb.backend.stored_queries.queryfieldspec import QueryFieldSpec
+from specifyweb.backend.stored_queries.tests.test_batch_edit import props_builder
+from specifyweb.backend.stored_queries.tests.tests import SQLAlchemySetup
+from specifyweb.backend.workbench.models import Spdataset
+from specifyweb.backend.workbench.upload.tests.base import UploadTestsBase
+from specifyweb.backend.workbench.upload.upload import (
+    do_upload_dataset, 
+    rollback_batch_edit,
+)
+from specifyweb.backend.workbench.views import regularize_rows
+from specifyweb.specify.models import Collectionobject
+from specifyweb.backend.workbench.tasks import unupload
+
+"""
+Creating a batch edit data set also creates a backup through the make_dataset 
+function. Rolling back (rollback_batch_edit) re-uploads the backed up values over 
+the committed records, calling upon unupload to roll back the data set
+"""
+
+ORIGINAL_REMARKS = "Remarks before batch edit"
+EDITED_REMARKS = "Remarks changed by batch edit"
+
+class BatchEditRollbackTests(SQLAlchemySetup, UploadTestsBase):
+    def setUp(self):
+        super().setUp()
+        self.build_props = props_builder(
+            self, BatchEditRollbackTests.test_session_context
+        )
+        Collectionobject.objects.all().delete()
+        self.co = Collectionobject.objects.create(
+            catalognumber="7924".zfill(9),
+            remarks=ORIGINAL_REMARKS,
+            collection=self.collection,
+        )
+
+    def _query_field(self, path) -> QueryField:
+        return QueryField(
+            fieldspec=QueryFieldSpec.from_path(("Collectionobject", *path)),
+            op_num=8,
+            value=None,
+            negate=False,
+            display=True,
+            format_name=None,
+            sort_type=0,
+        )
+
+    def _commit_edit(self) -> Spdataset:
+        props = self.build_props(
+            [self._query_field(["catalognumber"]), self._query_field(["remarks"])],
+            "Collectionobject",
+        )
+        headers, rows, packs, plan_json, visual_order = run_batch_edit_query(props)
+        rows_with_packs = [
+            [*row, json.dumps({"batch_edit": pack})] for row, pack in zip(rows, packs)
+        ]
+        original_rows = regularize_rows(len(headers), rows_with_packs, skip_empty=False)
+
+        dataset_id, _ = make_dataset(
+            user=self.specifyuser,
+            collection=self.collection,
+            name="rollback",
+            headers=headers,
+            regularized_rows=original_rows,
+            agent=self.agent,
+            json_upload_plan=plan_json,
+            visual_order=visual_order,
+        )
+        dataset = Spdataset.objects.get(id=dataset_id)
+
+        row_index = next(
+            index
+            for index, pack in enumerate(packs)
+            if pack["self"]["id"] == self.co.id
+        )
+        remarks_column = next(
+            index
+            for index, header in enumerate(headers)
+            if header.lower().endswith("remarks")
+        )
+        edited_rows = [list(row) for row in dataset.data]
+        edited_rows[row_index][remarks_column] = EDITED_REMARKS
+        dataset.data = edited_rows
+        dataset.save()
+
+        do_upload_dataset(
+            self.collection,
+            self.agent.id,
+            dataset,
+            no_commit=False,
+            allow_partial=False,
+        )
+        dataset.refresh_from_db()
+        return dataset
+
+    def test_commit_applies_the_edit(self):
+        self._commit_edit()
+        self.co.refresh_from_db()
+        self.assertEqual(self.co.remarks, EDITED_REMARKS)
+
+    def test_rollback_restores_the_original_values(self):
+        dataset = self._commit_edit()
+        rollback_batch_edit(dataset, self.collection, self.agent)
+        self.co.refresh_from_db()
+        self.assertEqual(self.co.remarks, ORIGINAL_REMARKS)
+
+    def test_rollback_does_not_delete_edited_records(self):
+        dataset = self._commit_edit()
+        rollback_batch_edit(dataset, self.collection, self.agent)
+        self.assertTrue(Collectionobject.objects.filter(id=self.co.id).exists())
+
+    def test_rollback_makes_the_data_set_rolled_back(self):
+        dataset = self._commit_edit()
+        dataset.uploaderstatus = {"operation": "unuploading", "taskid": None}
+        dataset.save()
+        unupload(self.collection.id, dataset.id, self.agent.id)
+        dataset.refresh_from_db()
+        self.assertTrue(dataset.rolledback)
+        self.assertIsNone(dataset.uploaderstatus)
+        self.assertTrue(dataset.get_dataset_as_dict()["rolledback"])
+        self.assertTrue(dataset.get_dataset_as_dict()["isupdate"])
+        self.co.refresh_from_db()
+        self.assertEqual(self.co.remarks, ORIGINAL_REMARKS)

specifyweb/backend/workbench/upload/tests/test_batch_edit_table.py

@@ -1019,6 +1019,188 @@ def test_batch_edit_attachment_dataset_commit_updates_original_record(self):
         self.assertEqual(Attachment.objects.count(), initial_attachment_count)
         self.enforce_in_log(self.co_1_attachment.id, "attachment", "UPDATE")
 
+    def _dataset_editing_a_preparation(self):
+        query_paths = [
+            ["catalognumber"],
+            ["preparations", "countamt"],
+            ["preparations", "text1"],
+        ]
+        query_fields = [
+            self.make_query(QueryFieldSpec.from_path(("Collectionobject", *path)), 0)
+            for path in query_paths
+        ]
+        props = self._build_props(query_fields, "Collectionobject")
+
+        (headers, rows, packs, plan_json, visual_order) = run_batch_edit_query(props)
+
+        mapped_rows = [
+            [*row, json.dumps({"batch_edit": pack})] for (row, pack) in zip(rows, packs)
+        ]
+        regularized_rows = regularize_rows(len(headers), mapped_rows, skip_empty=False)
+        row_index = next(
+            index
+            for index, pack in enumerate(packs)
+            if pack["self"]["id"] == self.co_1.id
+        )
+
+        dataset_rows = [row[:] for row in regularized_rows]
+        dataset_rows[row_index][headers.index("Preparation text1")] = "Edited by batch edit"
+
+        dataset_id, _ = make_dataset(
+            user=self.specifyuser,
+            collection=self.collection,
+            name="validate-batch-edit",
+            headers=headers,
+            regularized_rows=dataset_rows,
+            agent=self.agent,
+            json_upload_plan=plan_json,
+            visual_order=visual_order,
+        )
+        return Spdataset.objects.get(id=dataset_id), row_index
+
+    def test_validating_reports_changes_to_related_records(self):
+        dataset, row_index = self._dataset_editing_a_preparation()
+
+        results = do_upload_dataset(
+            self.collection, self.agent.id, dataset, no_commit=True, allow_partial=False
+        )
+
+        self.assertIsInstance(
+            results[row_index].toMany["preparations"][0].record_result, Updated
+        )
+
+    def test_validating_does_not_save_changes(self):
+        dataset, _ = self._dataset_editing_a_preparation()
+
+        do_upload_dataset(
+            self.collection, self.agent.id, dataset, no_commit=True, allow_partial=False
+        )
+
+        self.co_1_prep_1.refresh_from_db()
+        self.assertEqual(self.co_1_prep_1.text1, "Value for preparation")
+
+    def test_validating_does_not_mark_the_data_set_uploaded(self):
+        dataset, _ = self._dataset_editing_a_preparation()
+
+        do_upload_dataset(
+            self.collection, self.agent.id, dataset, no_commit=True, allow_partial=False
+        )
+
+        dataset.refresh_from_db()
+        self.assertIsNotNone(dataset.rowresults)
+        self.assertIsNone(dataset.uploadresult)
+        self.assertFalse(dataset.was_uploaded())
+
+    def test_a_validated_data_set_can_be_committed(self):
+        dataset, _ = self._dataset_editing_a_preparation()
+
+        do_upload_dataset(
+            self.collection, self.agent.id, dataset, no_commit=True, allow_partial=False
+        )
+        do_upload_dataset(
+            self.collection, self.agent.id, dataset, no_commit=False, allow_partial=False
+        )
+
+        self.co_1_prep_1.refresh_from_db()
+        self.assertEqual(self.co_1_prep_1.text1, "Edited by batch edit")
+
+
+    def _make_plant_tree(self):
+        plant_tree = get_table("Taxontreedef").objects.create(
+            name="Plant ttd", discipline=self.discipline
+        )
+        plant_tree.treedefitems.create(name="Taxonomy Root", rankid=0)
+        plant_tree.treedefitems.create(name="Kingdom", rankid=10)
+        plant_tree.treedefitems.create(name="Genus", rankid=180)
+        plant_tree.treedefitems.create(name="Species", rankid=220)
+        plant_tree.treedefitems.create(name="Variety", rankid=240)
+        return plant_tree
+
+    def _tree_ranks_for_genus_query(self, treedefsfilter):
+        query_fields = fields_from_json(
+            [
+                {
+                    "tablelist": "1",
+                    "stringid": "1.collectionobject.catalogNumber",
+                    "fieldname": "catalogNumber",
+                    "isrelfld": False,
+                    "sorttype": 0,
+                    "position": 0,
+                    "isdisplay": True,
+                    "operstart": 8,
+                    "startvalue": "",
+                    "isnot": False,
+                },
+                {
+                    "tablelist": "1,9-determinations,4",
+                    "stringid": "1,9-determinations,4.taxon.Genus",
+                    "fieldname": "Genus",
+                    "isrelfld": False,
+                    "sorttype": 0,
+                    "position": 1,
+                    "isdisplay": True,
+                    "operstart": 8,
+                    "startvalue": "",
+                    "isnot": False,
+                },
+            ]
+        )
+        get_table("Determination").objects.create(
+            collectionobject=self.co_1, remarks="A determination"
+        )
+        props = self._build_props(query_fields, "Collectionobject")
+        props["treedefsfilter"] = treedefsfilter
+
+        (headers, rows, packs, plan_json, visual_order) = run_batch_edit_query(props)
+
+        found = set()
+
+        def collect(node):
+            if isinstance(node, dict):
+                for rank_key in node.get("treeRecord", {}).get("ranks", {}):
+                    tree_name, rank_name = rank_key.split(RANK_KEY_DELIMITER)[:2]
+                    found.add((tree_name, rank_name))
+                for value in node.values():
+                    collect(value)
+            elif isinstance(node, list):
+                for value in node:
+                    collect(value)
+
+        collect(plan_json)
+        return found
+
+    def _default_tree_ranks(self):
+        return {
+            (self.taxontreedef.name, rank)
+            for rank in ["Genus", "Subgenus", "Species", "Subspecies"]
+        }
+
+    def _plant_tree_ranks(self):
+        return {("Plant ttd", rank) for rank in ["Genus", "Species", "Variety"]}
+
+    def test_only_the_selected_trees_ranks_are_added(self):
+        plant_tree = self._make_plant_tree()
+
+        ranks = self._tree_ranks_for_genus_query({"taxon": [plant_tree.id]})
+
+        self.assertEqual(ranks, self._plant_tree_ranks())
+
+    def test_every_selected_trees_ranks_are_added(self):
+        plant_tree = self._make_plant_tree()
+
+        ranks = self._tree_ranks_for_genus_query(
+            {"taxon": [self.taxontreedef.id, plant_tree.id]}
+        )
+
+        self.assertEqual(ranks, self._default_tree_ranks() | self._plant_tree_ranks())
+
+    def test_no_trees_selected_adds_every_tree(self):
+        self._make_plant_tree()
+
+        ranks = self._tree_ranks_for_genus_query({})
+
+        self.assertEqual(ranks, self._default_tree_ranks() | self._plant_tree_ranks())
+
     def enforce_in_log(
         self,
         record_id,