diff --git a/CHANGELOG.md b/CHANGELOG.md
index c26449b8be417077cfa69f129e8e9e62c257d290..9f22064aa509b1d5d4a8f1cd673ceef6d2e79543 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
in caosdb server which checks if quoted datetimes in queries are
evaluated correctly.
* Test for [caosdb-server#99](https://gitlab.com/caosdb/caosdb-server/-/issues/99).
+* Tests for retrieving older versions.
### Changed (for changes in existing functionality)
diff --git a/tests/test_query_version.py b/tests/test_query_version.py
new file mode 100644
index 0000000000000000000000000000000000000000..c951fa3c4fa57dea30b22627746760575229c03c
--- /dev/null
+++ b/tests/test_query_version.py
@@ -0,0 +1,669 @@
+# encoding: utf-8
+#
+# ** header v3.0
+# 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
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# 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
+#
+import pytest
+import caosdb as db
+from caosdb import execute_query as query
+
+CONTAINER = db.Container()
+NAME_PROPERTY = db.Property("name").retrieve()
+TEST_PROP_TEXT = db.Property("TestPropertyText", datatype=db.TEXT)
+TEST_PROP_DOUBLE = db.Property("TestPropertyDouble", datatype=db.DOUBLE)
+TEST_PROP_INTEGER = db.Property("TestPropertyInteger", datatype=db.INTEGER)
+TEST_PROP_DATETIME = db.Property("TestPropertyDatetime", datatype=db.DATETIME)
+TEST_PROP_DATE = db.Property("TestPropertyDate", datatype=db.DATETIME)
+TEST_PROP_BOOLEAN = db.Property("TestPropertyBoolean", datatype=db.BOOLEAN)
+TEST_PROP_NAME = db.Property("TestPropertyName",
+ datatype=db.TEXT).add_parent(NAME_PROPERTY)
+
+TEST_REF_RT = db.RecordType("TestRefRecordType")
+TEST_REF_1 = db.Record("TestRef1", description="ref1").add_parent(TEST_REF_RT)
+TEST_REF_2 = db.Record("TestRef2", description="ref2").add_parent(TEST_REF_RT)
+TEST_REF_3 = db.Record("TestRef3", description="ref3").add_parent(TEST_REF_RT)
+
+TEST_RT_1 = db.RecordType("TestRecordType1")
+
+TEST_REC_1 = db.Record(
+ "TestRecord1",
+ description="v1").add_parent(
+ TEST_RT_1).add_property(
+ TEST_PROP_TEXT,
+ "val1").add_property(
+ TEST_PROP_DOUBLE,
+ 1.14).add_property(
+ TEST_PROP_INTEGER,
+ 100).add_property(
+ TEST_PROP_DATETIME,
+ "2021-12-24T20:15:00").add_property(
+ TEST_PROP_DATE,
+ "2021-12").add_property(
+ TEST_REF_RT,
+ TEST_REF_1).add_property(
+ TEST_PROP_BOOLEAN,
+ True).add_property(
+ TEST_PROP_NAME,
+ "Alias1")
+
+
+TEST_PROP_TEXT_2 = db.Property("TestPropertyText2", datatype=db.TEXT)
+TEST_BACKREF_RT = db.RecordType("TestBackRefRecordType")
+TEST_BACKREF_0 = db.Record(
+ "TestBackRef0",
+ description="bref0").add_parent(TEST_BACKREF_RT).add_property(
+ TEST_RT_1,
+ "TestRecord1").add_property(
+ TEST_PROP_TEXT_2,
+ "val0")
+TEST_BACKREF_1 = db.Record("TestBackRef1", description="bref1").add_parent(
+ TEST_BACKREF_RT).add_property(TEST_PROP_TEXT_2, "val1")
+TEST_BACKREF_2 = db.Record("TestBackRef2", description="bref2").add_parent(
+ TEST_BACKREF_RT).add_property(TEST_PROP_TEXT_2, "val2")
+TEST_BACKREF_3 = db.Record("TestBackRef3", description="bref3").add_parent(
+ TEST_BACKREF_RT).add_property(TEST_PROP_TEXT_2, "val3")
+
+
+def teardown_module():
+ CONTAINER.delete()
+
+
+def setup_module():
+ CONTAINER.extend([TEST_RT_1, TEST_REC_1, TEST_PROP_TEXT, TEST_PROP_DOUBLE,
+ TEST_PROP_INTEGER, TEST_PROP_DATETIME, TEST_PROP_DATE,
+ TEST_REF_RT, TEST_REF_1, TEST_REF_2, TEST_REF_3,
+ TEST_BACKREF_RT, TEST_BACKREF_0, TEST_PROP_TEXT_2,
+ TEST_PROP_BOOLEAN, TEST_PROP_NAME,
+ ]).insert()
+ CONTAINER.append(TEST_BACKREF_1)
+ TEST_BACKREF_1.add_property(
+ TEST_RT_1, str(
+ TEST_REC_1.id) + "@HEAD").insert()
+
+ # update test_rec_1 two times
+ TEST_REC_1.description = "v2"
+ TEST_REC_1.get_property(TEST_PROP_TEXT).value = "val2"
+ TEST_REC_1.get_property(TEST_PROP_DOUBLE).value = 2.14
+ TEST_REC_1.get_property(TEST_PROP_INTEGER).value = 200
+ TEST_REC_1.get_property(TEST_PROP_DATETIME).value = "2022-12-24T20:15:00"
+ TEST_REC_1.get_property(TEST_PROP_DATE).value = "2022-12"
+ TEST_REC_1.get_property(TEST_REF_RT).value = TEST_REF_2
+ TEST_REC_1.get_property(TEST_PROP_NAME).value = "Alias2"
+ TEST_REC_1.get_property(TEST_PROP_BOOLEAN).value = False
+ TEST_REC_1.update()
+
+ CONTAINER.append(TEST_BACKREF_2)
+ TEST_BACKREF_2.add_property(
+ TEST_RT_1, str(
+ TEST_REC_1.id) + "@HEAD").insert()
+
+ TEST_REC_1.description = "v3"
+ TEST_REC_1.get_property(TEST_PROP_TEXT).value = "val3"
+ TEST_REC_1.get_property(TEST_PROP_DOUBLE).value = 3.14
+ TEST_REC_1.get_property(TEST_PROP_INTEGER).value = 300
+ TEST_REC_1.get_property(TEST_PROP_DATETIME).value = "2023-12-24T20:15:00"
+ TEST_REC_1.get_property(TEST_PROP_DATE).value = "2023-12"
+ TEST_REC_1.get_property(TEST_REF_RT).value = TEST_REF_3
+ TEST_REC_1.get_property(TEST_PROP_NAME).value = "Alias3"
+ TEST_REC_1.get_property(TEST_PROP_BOOLEAN).value = True
+ TEST_REC_1.update()
+
+ CONTAINER.append(TEST_BACKREF_3)
+ TEST_BACKREF_3.add_property(
+ TEST_RT_1, str(
+ TEST_REC_1.id) + "@HEAD").insert()
+
+
+def test_normal_find_record():
+ result = query("FIND RECORD TestRecord1")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecordType1")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyText = val3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyText = val1")
+ assert len(result) == 0
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDouble = 3.14")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDouble = 1.14")
+ assert len(result) == 0
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyInteger = 300")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyInteger = 100")
+ assert len(result) == 0
+
+ result = query(
+ "FIND RECORD TestRecord1 WITH TestPropertyDatetime = 2023-12-24T20:15:00")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND RECORD TestRecord1 WITH TestPropertyDatetime IN 2023-12")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND RECORD TestRecord1 WITH TestPropertyDatetime = 2021-12-24T20:15:00")
+ assert len(result) == 0
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDatetime IN 2021")
+ assert len(result) == 0
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDate = 2023-12")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDate = 2021-12")
+ assert len(result) == 0
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDate in 2023")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestPropertyDate in 2021")
+ assert len(result) == 0
+
+ result = query("FIND RECORD TestRecord1 WITH TestRefRecordType = TestRef3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRecord1 WITH TestRefRecordType = TestRef1")
+ assert len(result) == 0
+
+ result = query(
+ "FIND RECORD TestRecord1 WITH TestRefRecordType -> TestRef3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND RECORD TestRecord1 WITH TestRefRecordType -> TestRef1")
+ assert len(result) == 0
+
+ result = query("FIND RECORD WITH TestRefRecordType AND TestPropertyText")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD TestRefRecordType WITH NOT TestPropertyText")
+ assert len(result) == 3
+ assert set(["ref1", "ref2", "ref3"]) == set(
+ [r.description for r in result])
+
+ result = query("FIND RECORD TestRecord1 WITH NOT TestPropertyText")
+ assert len(result) == 0
+
+ result = query("FIND RECORD WITH NOT TestPropertyText")
+ assert len(result) == 7
+ assert set(["ref1", "ref2", "ref3", "bref0", "bref1", "bref2",
+ "bref3", ]) == set([r.description for r in result])
+
+ result = query("FIND RECORD WHICH IS REFERENCED BY TestBackRefRecordType")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND RECORD WHICH IS REFERENCED BY TestBackRef0")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND RECORD WHICH IS REFERENCED BY TestBackRefRecordType WITH TestPropertyText2=val0")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+
+@pytest.mark.xfail(
+ reason=("Find queries which respect the version of the references are not supported yet and "
+ "return false positives."))
+def test_normal_find_with_versioned_reference():
+ # current version is not referenced
+ result = query("FIND RECORD WHICH IS REFERENCED BY TestBackRef1")
+ assert len(result) == 0
+
+
+def test_find_any_version_of_record():
+ result = query("FIND ANY VERSION OF RECORD")
+ assert len(result) == 10
+ assert set(["v3", "v2", "v1", "ref1", "ref2", "ref3", "bref0", "bref1",
+ "bref2", "bref3"]
+ ) == {r.description for r in result}
+
+
+def test_find_any_version_of_entity_by_name():
+ result = query("FIND ANY VERSION OF ENTITY TestRecord1")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_record_by_name():
+ result = query("FIND ANY VERSION OF RECORD TestRecord1")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_record_by_parent():
+ result = query("FIND ANY VERSION OF RECORD TestRecordType1")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_named_entity_with_id_filter():
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WITH id = " + str(TEST_REC_1.id))
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_entity_with_id_filter():
+ result = query("FIND ANY VERSION OF ENTITY WITH id = " +
+ str(TEST_REC_1.id))
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_record_with_id_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH id = " +
+ str(TEST_REC_1.id))
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_record_with_simple_pov_text_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyText")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyText = val3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyText = val2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyText").value == "val2"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyText = val1")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_pov_name_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyName")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyName = Alias3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyName = Alias2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyName").value == "Alias2"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyName = Alias1")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_pov_boolean_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyBoolean")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyBoolean = TRUE AND TestPropertyText = val3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyBoolean = FALSE")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyBoolean").value is False
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyBoolean = TRUE AND TestPropertyText = val1")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_pov_double_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDouble")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDouble = 3.14")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDouble = 2.14")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyDouble").value == 2.14
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDouble = 1.14")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_pov_integer_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyInteger")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyInteger = 300")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyInteger = 200")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyInteger").value == 200
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyInteger = 100")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_pov_datetime_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDatetime")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDatetime = 2023-12-24T20:15:00")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDatetime = 2022-12-24T20:15:00")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property(
+ "TestPropertyDatetime").value.startswith("2022")
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDatetime = 2021-12-24T20:15:00")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDatetime IN 2023")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDatetime IN 2022")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property(
+ "TestPropertyDatetime").value.startswith("2022")
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDatetime IN 2021")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_pov_date_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDate")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDate = 2023-12")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDate = 2022-12")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyDate").value.startswith("2022")
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestPropertyDate = 2021-12")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDate IN 2023")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDate IN 2022")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestPropertyDate").value.startswith("2022")
+
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyDate IN 2021")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_simple_reference_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestRefRecordType")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestRefRecordType").value == TEST_REF_2.id
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef1")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType -> TestRef3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType -> TestRef2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+ assert result[0].get_property("TestRefRecordType").value == TEST_REF_2.id
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType -> TestRef1")
+ assert len(result) == 1
+ assert result[0].description == "v1"
+
+
+def test_find_any_version_of_record_with_conjunction_filter():
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType AND TestPropertyText")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef3 AND TestPropertyText")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef3 AND TestPropertyText = val3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef2 AND TestPropertyText = val2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef3 AND TestPropertyText = val2")
+ assert len(result) == 0
+
+
+def test_find_any_version_of_record_with_disjunction_filter():
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType OR TestPropertyText")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef3 OR TestPropertyText")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef2 OR TestPropertyText")
+ assert len(result) == 3
+ assert set(["v3", "v2", "v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef3 OR TestPropertyText = val3")
+ assert len(result) == 1
+ assert result[0].description == "v3"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef2 OR TestPropertyText = val2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WITH TestRefRecordType = TestRef2 OR TestPropertyText = val2")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef2 OR TestPropertyText = val3")
+ assert len(result) == 2
+ assert set(["v3", "v2"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH TestRefRecordType = TestRef2 OR TestPropertyText = val3 OR TestPropertyInteger = 200")
+ assert len(result) == 2
+ assert set(["v3", "v2"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_record_with_negation_filter():
+ result = query(
+ "FIND ANY VERSION OF RECORD WHICH DOES NOT HAVE A TestPropertyText")
+ assert len(result) == 7
+ assert set(["ref1", "ref2", "ref3", "bref0", "bref1", "bref2",
+ "bref3", ]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD WHICH DOES NOT HAVE A TestPropertyText=val2")
+ assert len(result) == 9
+ assert set(["ref1", "ref2", "ref3", "bref0", "bref1", "bref2", "bref3",
+ "v1", "v3"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD TestRecord1 WHICH DOES NOT HAVE A TestPropertyText=val2")
+ assert len(result) == 2
+ assert set(["v1", "v3"]) == {r.description for r in result}
+
+
+def test_find_any_version_of_record_with_complex_filter():
+ result = query("FIND ANY VERSION OF RECORD WITH TestPropertyText=val2 OR (TestPropertyInteger=200 AND NOT TestPropertyInteger=400 AND TestRefRecordType = TestRef2)")
+ assert len(result) == 1
+ assert result[0].description == "v2"
+
+
+def test_find_any_version_of_record_with_backref_filter():
+ result = query(
+ "FIND ANY VERSION OF RECORD WHICH IS REFERENCED BY TestBackRefRecordType")
+ assert len(result) == 3
+ assert set(["v1", "v2", "v3"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF RECORD TestRecord1 WHICH IS REFERENCED BY TestBackRefRecordType")
+ assert len(result) == 3
+ assert set(["v1", "v2", "v3"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WHICH IS REFERENCED BY TestBackRefRecordType")
+ assert len(result) == 3
+ assert set(["v1", "v2", "v3"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WHICH IS REFERENCED BY TestBackRef0")
+ assert len(result) == 3
+ assert set(["v1", "v2", "v3"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WHICH IS REFERENCED BY TestBackRef0")
+ assert len(result) == 3
+ assert set(["v1", "v2", "v3"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WHICH IS REFERENCED BY TestBackRef1")
+ assert len(result) == 1
+ assert set(["v1"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WHICH IS REFERENCED BY TestBackRef2")
+ assert len(result) == 1
+ assert set(["v2"]) == {r.description for r in result}
+
+ result = query(
+ "FIND ANY VERSION OF TestRecord1 WHICH IS REFERENCED BY TestBackRef3")
+ assert len(result) == 1
+ assert set(["v3"]) == {r.description for r in result}
+
+
+@pytest.mark.xfail(
+ reason="versioned queries with subqueries are not supported yet")
+def test_find_any_version_of_record_with_backref_and_subquery():
+ result = query(
+ "FIND ANY VERSION OF RECORD WHICH IS REFERENCED BY TestBackRefRecordType WITH name = TestBackRef0")
+ assert len(result) == 1
+ assert set(["v3"]) == {r.description for r in result}
+
+
+@pytest.mark.xfail(
+ reason="versioned queries with subqueries are not supported yet")
+def test_find_any_version_of_record_with_ref_and_subquery():
+ result = query(
+ "FIND ANY VERSION OF RECORD WHICH REFERENCES TestRefRecordType WITH name = TestRef2")
+ assert len(result) == 1
+ assert set(["v3"]) == {r.description for r in result}
+
+
+@pytest.mark.xfail(
+ reason="versioned queries with aggregate functions are not supported yet")
+def test_find_any_version_of_record_with_greatest_filter():
+ result = query(
+ "FIND ANY VERSION OF RECORD WITH THE GREATEST TestPropertyInteger")
+ assert len(result) == 1
+ assert set(["v3"]) == {r.description for r in result}
diff --git a/tests/test_version.py b/tests/test_version.py
index 7a9bf8b4e71647418c110b05dad7e7a8a9190519..4a02b7df4a70446b3956d9bc77c22cff01faac04 100644
--- a/tests/test_version.py
+++ b/tests/test_version.py
@@ -40,8 +40,11 @@ def test_version_object():
from caosdb.common.versioning import Version
-def insertion(name="TestRT"):
- rt = c.RecordType(name, description="TestDescription1").insert()
+def insertion(name="TestRT", parents=[]):
+ rt = c.RecordType(name, description="TestDescription1")
+ for par in parents:
+ rt.add_parent(par)
+ rt.insert()
assert rt.version is not None
assert rt.version.id is not None
assert rt.version.date is not None
@@ -92,8 +95,9 @@ def test_update_description():
def test_update_parent():
- par1 = insertion("TestRTParent1")
- par2 = insertion("TestRTParent2")
+ superpar = insertion("TestRTParentSuper")
+ par1 = insertion("TestRTParent1", parents=[superpar])
+ par2 = insertion("TestRTParent2", parents=[superpar])
rt = insertion("TestRTChild")
assert len(rt.get_parents()) == 0
@@ -119,13 +123,16 @@ def test_update_parent():
assert len(c.execute_query("FIND TestRTParent2")) == 2
assert c.execute_query("FIND TestRTChild", unique=True).id == rt.id
- rt_head = c.Container().retrieve(query=str(rt.id), sync=False)[0]
+ rt_head = c.Container().retrieve(
+ query=str(
+ rt.id), sync=False, flags={
+ "cache": "false"})[0]
rt_v1 = c.Container().retrieve(query=str(rt.id) + "@" + first_version.id,
- sync=False)[0]
+ sync=False, flags={"cache": "false"})[0]
rt_v2 = c.Container().retrieve(query=str(rt.id) + "@" + second_version.id,
- sync=False)[0]
+ sync=False, flags={"cache": "false"})[0]
rt_v3 = c.Container().retrieve(query=str(rt.id) + "@" + third_version.id,
- sync=False)[0]
+ sync=False, flags={"cache": "false"})[0]
assert rt_head.version == third_version
assert rt_v1.version.id == first_version.id
@@ -133,9 +140,11 @@ def test_update_parent():
assert rt_v3.version.id == third_version.id
assert len(rt_v3.get_parents()) == 1
+ print(rt_v3)
assert rt_v3.get_parent("TestRTParent1") is None
assert rt_v3.get_parent("TestRTParent2") is not None
+ print(rt_v2)
assert len(rt_v2.get_parents()) == 1
assert rt_v2.get_parent("TestRTParent1") is not None
assert rt_v2.get_parent("TestRTParent2") is None
@@ -390,6 +399,37 @@ def test_bug_cached_parent_name_in_old_version():
assert old_rec.get_parent(del_rt_name) is None
+@mark.xfail(reason="bug fix needed")
+def test_bug_parent_name_in_old_version():
+ """ mysql backend return current name even if this historically not
+ correct."""
+ OLD_NAME = "TestRT1_old"
+ NEW_NAME = "TestRT1_cur"
+ rt = c.RecordType(name=OLD_NAME).insert()
+
+ rec = c.Record(name="TestRec").add_parent(rt)
+ rec.insert()
+ rec.description = "v2"
+ rec.update()
+
+ rec_v1 = c.Container().retrieve(str(rec.id) + "@HEAD~1",
+ sync=False,
+ flags={"cache": "false"},
+ raise_exception_on_error=False)[0]
+ rec_v2 = c.Container().retrieve(str(rec.id) + "@HEAD",
+ sync=False,
+ flags={"cache": "false"},
+ raise_exception_on_error=False)[0]
+
+ assert [OLD_NAME] == [p.name for p in rec_v2.get_parents()]
+ assert [OLD_NAME] == [p.name for p in rec_v1.get_parents()]
+
+ # change parent
+ rt.name = NEW_NAME
+ assert [NEW_NAME] == [p.name for p in rec_v2.get_parents()]
+ assert [OLD_NAME] == [p.name for p in rec_v1.get_parents()]
+
+
def test_reference_deleted_in_old_version():
ref_rt = insertion("TestReferencedObject")
rt = insertion("TestRT")