TST: Add unit tests for force merge

unittests/test_apiutils.py

-# -*- encoding: utf-8 -*-
 #
 # This file is a part of the CaosDB Project.
 #
-# Copyright (C) 2018 Research Group Biomedical Physics,
-# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
 # Copyright (C) 2020 Timm Fitschen <t.fitschen@indiscale.com>
+# Copyright (C) 2022 Florian Spreckelsen <f.spreckelsen@indiscale.com>
 # Copyright (C) 2020-2022 IndiScale GmbH <info@indiscale.com>
+# Copyright (C) 2018 Research Group Biomedical Physics,
+# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as
@@ -20,7 +20,6 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with this program. If not, see <https://www.gnu.org/licenses/>.
 #
-# ** end header
 #
 # Test apiutils
 # A. Schlemmer, 02/2018
@@ -466,3 +465,82 @@ def test_empty_diff():
         "RefType"), value=[ref_rec_b, ref_rec_b])
     assert not empty_diff(rec_a, rec_b)
     assert empty_diff(rec_a, rec_b, compare_referenced_records=True)
+
+
+def test_force_merge():
+    """Test whether a forced merge overwrites existing properties correctly."""
+
+    # name overwrite
+    recA = db.Record(name="A")
+    recB = db.Record(name="B")
+
+    with pytest.raises(RuntimeError) as re:
+        merge_entities(recA, recB)
+    assert "Merge conflict" in str(re.value)
+
+    merge_entities(recA, recB, force=True)
+    assert recA.name == recB.name
+
+    # description overwrite
+    recA = db.Record(name="A")
+    recA.description = "something"
+    recB = db.Record(name="B")
+    recB.description = "something else"
+
+    with pytest.raises(RuntimeError) as re:
+        merge_entities(recA, recB)
+    assert "Merge conflict" in str(re.value)
+
+    merge_entities(recA, recB, force=True)
+    assert recA.description == recB.description
+
+    # property overwrite
+    recA = db.Record(name="A")
+    recA.add_property(name="propA", value="something")
+    recB = db.Record(name="A")
+    recB.add_property(name="propA", value="something else")
+
+    with pytest.raises(RuntimeError) as re:
+        merge_entities(recA, recB)
+    assert "Merge conflict" in str(re.value)
+
+    merge_entities(recA, recB, force=True)
+    assert recA.get_property("propA").value == recB.get_property("propA").value
+
+    # don't overwrite a property that's not in recB
+    recA = db.Record(name="A")
+    recA.add_property(name="propA", value="something")
+    recA.add_property(name="propB", value=5.0)
+    recB = db.Record(name="A")
+    recB.add_property(name="propA", value="something else")
+
+    merge_entities(recA, recB, force=True)
+    assert recA.get_property("propA").value == recB.get_property("propA").value
+    assert recA.get_property("propB").value == 5.0
+
+    # also overwrite datatypes ...
+    rtA = db.RecordType(name="A")
+    rtA.add_property(name="propA", datatype=db.INTEGER)
+    rtB = db.RecordType(name="B")
+    rtB.add_property(name="propA", datatype=db.TEXT)
+
+    with pytest.raises(RuntimeError) as re:
+        merge_entities(rtA, rtB)
+    assert "Merge conflict" in str(re.value)
+
+    merge_entities(rtA, rtB, force=True)
+    assert rtA.get_property(
+        "propA").datatype == rtB.get_property("propA").datatype
+
+    # ... and units
+    recA = db.Record(name="A")
+    recA.add_property(name="propA", value=5, unit="m")
+    recB = db.Record(name="A")
+    recB.add_property(name="propA", value=500, unit="cm")
+
+    with pytest.raises(RuntimeError) as re:
+        merge_entities(recA, recB)
+    assert "Merge conflict" in str(re.value)
+    merge_entities(recA, recB, force=True)
+    assert recA.get_property("propA").value == recB.get_property("propA").value
+    assert recA.get_property("propA").unit == recB.get_property("propA").unit