Skip to content
Snippets Groups Projects
Verified Commit da28ee7e authored by Timm Fitschen's avatar Timm Fitschen
Browse files

Merge branch 'f-fsm' into f-fsm-v0.2

parents 002737f0 686a588f
Branches
Tags
1 merge request!3F fsm
...@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ...@@ -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 in caosdb server which checks if quoted datetimes in queries are
evaluated correctly. evaluated correctly.
* Test for [caosdb-server#99](https://gitlab.com/caosdb/caosdb-server/-/issues/99). * Test for [caosdb-server#99](https://gitlab.com/caosdb/caosdb-server/-/issues/99).
* Tests for retrieving older versions.
### Changed (for changes in existing functionality) ### Changed (for changes in existing functionality)
......
Makefile 0 → 100644
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2020 IndiScale GmbH <info@indiscale.com>
# Copyright (C) 2020 Daniel Hornung <d.hornung@indiscale.com>
#
# 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
# Just a collection of scripts
.PHONY: all autopep8 test
# Do the integration tests.
# Accepts freestyle arguments like so:
# `make test test_opts=test_foo.py -k test_function`
test:
cd tests && pytest $(test_opts)
# Run the tests through autopep8.
autopep8:
autopep8 -ri tests
# Meta target to call the other targets.
all: autopep8 test
# 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}
...@@ -156,6 +156,7 @@ def setup(): ...@@ -156,6 +156,7 @@ def setup():
def test_plugin_disabled(): def test_plugin_disabled():
"""Test the behavior when the state machine extension is disabled on the server."""
admin.set_server_property("EXT_ENTITY_STATE", "DISABLED") admin.set_server_property("EXT_ENTITY_STATE", "DISABLED")
rec = db.Record() rec = db.Record()
...@@ -184,6 +185,7 @@ def test_plugin_disabled(): ...@@ -184,6 +185,7 @@ def test_plugin_disabled():
def test_state_message(): def test_state_message():
"""State is constant between creation, insertion, retrieval."""
state_id = db.execute_query("FIND Record State1", unique=True).id state_id = db.execute_query("FIND Record State1", unique=True).id
rec = db.Record() rec = db.Record()
rec.add_parent("TestRT") rec.add_parent("TestRT")
...@@ -318,7 +320,7 @@ def test_transition_permissions(): ...@@ -318,7 +320,7 @@ def test_transition_permissions():
def test_transition_not_allowed(): def test_transition_not_allowed():
"""unallowed transitions return errors and do not update the entity""" """Unallowed transitions return errors and do not update the entity."""
rec = db.Record() rec = db.Record()
rec.add_parent("TestRT") rec.add_parent("TestRT")
rec.state = db.State(model="Model1", name="State1") rec.state = db.State(model="Model1", name="State1")
...@@ -423,7 +425,7 @@ def test_state_not_in_state_model(): ...@@ -423,7 +425,7 @@ def test_state_not_in_state_model():
0].description == "State does not exist in this StateModel." 0].description == "State does not exist in this StateModel."
def test_transition_with_out_state_change(): def test_transition_without_state_change():
rec = db.Record() rec = db.Record()
rec.description = "old description" rec.description = "old description"
rec.add_parent("TestRT") rec.add_parent("TestRT")
......
...@@ -40,8 +40,11 @@ def test_version_object(): ...@@ -40,8 +40,11 @@ def test_version_object():
from caosdb.common.versioning import Version from caosdb.common.versioning import Version
def insertion(name="TestRT"): def insertion(name="TestRT", parents=[]):
rt = c.RecordType(name, description="TestDescription1").insert() 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 is not None
assert rt.version.id is not None assert rt.version.id is not None
assert rt.version.date is not None assert rt.version.date is not None
...@@ -92,8 +95,9 @@ def test_update_description(): ...@@ -92,8 +95,9 @@ def test_update_description():
def test_update_parent(): def test_update_parent():
par1 = insertion("TestRTParent1") superpar = insertion("TestRTParentSuper")
par2 = insertion("TestRTParent2") par1 = insertion("TestRTParent1", parents=[superpar])
par2 = insertion("TestRTParent2", parents=[superpar])
rt = insertion("TestRTChild") rt = insertion("TestRTChild")
assert len(rt.get_parents()) == 0 assert len(rt.get_parents()) == 0
...@@ -119,13 +123,16 @@ def test_update_parent(): ...@@ -119,13 +123,16 @@ def test_update_parent():
assert len(c.execute_query("FIND TestRTParent2")) == 2 assert len(c.execute_query("FIND TestRTParent2")) == 2
assert c.execute_query("FIND TestRTChild", unique=True).id == rt.id 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, 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, 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, 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_head.version == third_version
assert rt_v1.version.id == first_version.id assert rt_v1.version.id == first_version.id
...@@ -133,9 +140,11 @@ def test_update_parent(): ...@@ -133,9 +140,11 @@ def test_update_parent():
assert rt_v3.version.id == third_version.id assert rt_v3.version.id == third_version.id
assert len(rt_v3.get_parents()) == 1 assert len(rt_v3.get_parents()) == 1
print(rt_v3)
assert rt_v3.get_parent("TestRTParent1") is None assert rt_v3.get_parent("TestRTParent1") is None
assert rt_v3.get_parent("TestRTParent2") is not None assert rt_v3.get_parent("TestRTParent2") is not None
print(rt_v2)
assert len(rt_v2.get_parents()) == 1 assert len(rt_v2.get_parents()) == 1
assert rt_v2.get_parent("TestRTParent1") is not None assert rt_v2.get_parent("TestRTParent1") is not None
assert rt_v2.get_parent("TestRTParent2") is None assert rt_v2.get_parent("TestRTParent2") is None
...@@ -390,6 +399,37 @@ def test_bug_cached_parent_name_in_old_version(): ...@@ -390,6 +399,37 @@ def test_bug_cached_parent_name_in_old_version():
assert old_rec.get_parent(del_rt_name) is None 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(): def test_reference_deleted_in_old_version():
ref_rt = insertion("TestReferencedObject") ref_rt = insertion("TestReferencedObject")
rt = insertion("TestRT") rt = insertion("TestRT")
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment