Skip to content
Snippets Groups Projects

Compare revisions

Changes are shown as if the source revision was being merged into the target revision. Learn more about comparing revisions.

Source

Select target project
No results found
Select Git revision
Loading items

Target

Select target project
  • caosdb/src/caosdb-pyinttest
1 result
Select Git revision
Loading items
Show changes
Commits on Source (112)
Showing
with 2002 additions and 2279 deletions
FROM debian:latest
FROM debian:bullseye
RUN apt-get update && \
apt-get install \
curl \
......
......@@ -44,8 +44,8 @@ stages:
# | +-(caosdb-server)-------------------+ |
# | | | |
# | | /opt/caosdb | |
# | .------->| + /git/caosdb-server/scripting/ | |
# | | .----->| + /git/caosdb-server/authtoken/ | |
# | .------->| + /git/caosdb-server/authtoken/ | |
# | | .----->| + /git/caosdb-server/scripting/ | |
# | | | .--->| + /mnt/extroot | |
# | | | | .->| + /cert | |
# | | | | | | | |
......
## Summary
*Please give a short summary of what the issue is.*
## Expected Behavior
*What did you expect how the software should behave?*
## Actual Behavior
*What did the software actually do?*
## Steps to Reproduce the Problem
*Please describe, step by step, how others can reproduce the problem. Please try these steps for yourself on a clean system.*
1.
2.
3.
## Specifications
- Version: *Which version of this software?*
- Platform: *Which operating system, which other relevant software versions?*
## Possible fixes
*Do you have ideas how the issue can be resolved?*
# Summary
*Insert a meaningful description for this merge request here: What is the new/changed behavior?
Which bug has been fixed? Are there related issues?*
# Focus
*Point the reviewer to the core of the code change. Where should they start reading? What should
they focus on (e.g. security, performance, maintainability, user-friendliness, compliance with the
specs, finding more corner cases, concrete questions)?*
# Test Environment
*How to set up a test environment for manual testing?*
# Check List for the Author
Please, prepare your MR for a review. Be sure to write a summary and a focus and create gitlab
comments for the reviewer. They should guide the reviewer through the changes, explain your changes
and also point out open questions. For further good practices have a look at [our review
guidelines](https://gitlab.com/caosdb/caosdb/-/blob/dev/REVIEW_GUIDELINES.md)
- [ ] All automated tests pass
- [ ] Reference related issues
- [ ] Up-to-date CHANGELOG.md (or not necessary)
- [ ] Up-to-date JSON schema (or not necessary)
- [ ] Appropriate user and developer documentation (or not necessary)
- Update / write published documentation (`make doc`).
- How do I use the software? Assume "stupid" users.
- How do I develop or debug the software? Assume novice developers.
- [ ] Annotations in code (Gitlab comments)
- Intent of new code
- Problems with old code
- Why this implementation?
# Check List for the Reviewer
- [ ] I understand the intent of this MR
- [ ] All automated tests pass
- [ ] Up-to-date CHANGELOG.md (or not necessary)
- [ ] Appropriate user and developer documentation (or not necessary), also in published
documentation.
- [ ] The test environment setup works and the intended behavior is reproducible in the test
environment
- [ ] In-code documentation and comments are up-to-date.
- [ ] Check: Are there specifications? Are they satisfied?
For further good practices have a look at [our review guidelines](https://gitlab.com/caosdb/caosdb/-/blob/dev/REVIEW_GUIDELINES.md).
/assign me
/target_branch dev
......@@ -41,6 +41,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
* Test for [caosdb-server#136](https://gitlab.com/caosdb/caosdb-server/-/issues/136)
* Test for [caosdb-server#141](https://gitlab.com/caosdb/caosdb-server/-/issues/141)
* Test for [caosdb-server#145](https://gitlab.com/caosdb/caosdb-server/-/issues/145)
* Tests for [caosdb-pylib#90](https://gitlab.com/caosdb/caosdb-pylib/-/issues/90): `Entity.get_parents_recursively()` did not work for unretrieved
parents.
* Test for [caosdb-server#192](https://gitlab.com/caosdb/caosdb-server/-/issues/192)
* Test for miscellaneous *-too-long errors.
### Changed (for changes in existing functionality)
......@@ -51,6 +55,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
xfail until
[caosdb-pylib#34](https://gitlab.com/caosdb/caosdb-pylib/-/issues/34)
is fixed.
* Test for [caosdb-server#197](https://gitlab.com/caosdb/caosdb-server/-/issues/197).
### Deprecated (for soon-to-be removed features)
......
cff-version: 1.2.0
message: "If you use this software, please cite it as below."
authors:
- family-names: Fitschen
given-names: Timm
orcid: https://orcid.org/0000-0002-4022-432X
- family-names: Schlemmer
given-names: Alexander
orcid: https://orcid.org/0000-0003-4124-9649
- family-names: Hornung
given-names: Daniel
orcid: https://orcid.org/0000-0002-7846-6375
- family-names: tom Wörden
given-names: Henrik
orcid: https://orcid.org/0000-0002-5549-578X
- family-names: Spreckelsen
given-names: Florian
orcid: https://orcid.org/0000-0002-6856-2910
- family-names: Parlitz
given-names: Ulrich
orcid: https://orcid.org/0000-0003-3058-1435
- family-names: Luther
given-names: Stefan
orcid: https://orcid.org/0000-0001-7214-8125
title: "CaosDB - Server"
version: 0.8.1
doi: 10.3390/data4020083
date-released: 2022-11-07
......@@ -16,9 +16,12 @@ CaosDB project.
- Modify `pycaosdb.ini.template` and save it as `pycaosdb.ini`, taking care of the following points:
- Certificates must be valid and be specified in `pycaosdb.ini`.
- Server-side scripting paths must be given, otherwise server-side scripting will be omitted.
- Paths for the file tests must exist, or be creatable by the testing script and the server.
- Run the tests with `pytest` or `pytest-3` (depending on your system).
- If you want to run just a single test, you can also select a single test file:
`pytest-3 tests/test_issues.py`
or a test matching a regular expression:
`pytest-3 tests/test_issues.py -k issue_123`
## Requirements ##
......@@ -63,7 +66,7 @@ when creating the merge request. This allows our team to work with you on your r
* Copyright (C) 2018 Research Group Biomedical Physics, Max Planck Institute
for Dynamics and Self-Organization Göttingen.
* Copyright (C) 2020-2021 Indiscale GmbH <info@indiscale.com>
* Copyright (C) 2020-2023 Indiscale GmbH <info@indiscale.com>
All files in this repository are licensed under the [GNU Affero General Public
License](LICENCE.md) (version 3 or later).
......@@ -59,7 +59,7 @@ def setup_function(function):
def teardown_function(function):
db.execute_query("FIND Test*").delete()
db.execute_query("FIND ENTITY Test*").delete()
def test_affiliation_there():
......
......@@ -46,7 +46,7 @@ def setup_function(function):
# deactivate anonymous user
db.administration.set_server_property("AUTH_OPTIONAL", "FALSE")
d = db.execute_query("FIND Test*")
d = db.execute_query("FIND ENTITY Test*")
if len(d) > 0:
d.delete()
......@@ -307,18 +307,18 @@ def test_crud_with_one_time_token():
# CREATE
rt = db.RecordType(name="TestRT")
rt.insert()
assert rt.id == db.execute_query("FIND TestRT", unique=True).id
assert rt.id == db.execute_query("FIND ENTITY TestRT", unique=True).id
# UPDATE
assert db.execute_query("FIND TestRT", unique=True).description is None
assert db.execute_query("FIND ENTITY TestRT", unique=True).description is None
rt.description = "new desc"
rt.update()
assert rt.description == db.execute_query(
"FIND TestRT", unique=True).description
"FIND ENTITY TestRT", unique=True).description
# RETRIEVE
rt.retrieve()
assert rt.id == db.execute_query("FIND TestRT", unique=True).id
assert rt.id == db.execute_query("FIND ENTITY TestRT", unique=True).id
rt.delete()
assert len(db.execute_query("FIND TestRT")) == 0
assert len(db.execute_query("FIND ENTITY TestRT")) == 0
......@@ -27,102 +27,78 @@
"""
import caosdb as h
# @UnresolvedImport
from nose.tools import assert_true, assert_equal, assert_false
from caosdb.exceptions import TransactionError
from pytest import raises
def test_property():
try:
def teardown_module():
try:
h.execute_query("FIND Simple*").delete()
h.execute_query("FIND ENTITY").delete()
except BaseException:
pass
def setup_function(function):
teardown_module()
def test_property():
p = h.Property(
name="SimpleBooleanProperty",
datatype=h.BOOLEAN).insert()
assert_true(p.is_valid())
assert_equal(p.datatype, h.BOOLEAN)
assert p.is_valid()
assert p.datatype == h.BOOLEAN
p2 = h.Property(id=p.id).retrieve()
assert_true(p2.is_valid)
assert_equal(p2.datatype, h.BOOLEAN)
finally:
try:
p.delete()
except BaseException:
pass
assert p2.is_valid
assert p2.datatype == h.BOOLEAN
def test_recordType():
try:
p = h.Property(
name="SimpleBooleanProperty",
datatype=h.BOOLEAN).insert()
assert_true(p.is_valid())
assert p.is_valid()
rt = h.RecordType(name="SimpleRecordType").add_property(p).insert()
assert_true(rt.is_valid())
assert_equal(
rt.get_property("SimpleBooleanProperty").datatype,
h.BOOLEAN)
assert rt.is_valid()
assert rt.get_property("SimpleBooleanProperty").datatype == h.BOOLEAN
rt2 = h.RecordType(id=rt.id).retrieve()
assert_true(rt2.is_valid())
assert_equal(
rt2.get_property("SimpleBooleanProperty").datatype,
h.BOOLEAN)
finally:
try:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
assert rt2.is_valid()
assert rt2.get_property("SimpleBooleanProperty").datatype == h.BOOLEAN
def test_record():
try:
p = h.Property(
name="SimpleBooleanProperty",
datatype=h.BOOLEAN).insert()
assert_true(p.is_valid())
assert p.is_valid()
rt = h.RecordType(name="SimpleRecordType").add_property(p).insert()
assert_true(rt.is_valid())
assert rt.is_valid()
rec1 = h.Record(name="SimpleRecord1").add_parent(
rt).add_property(p, value="TRUE").insert()
assert_true(rec1.is_valid())
assert_equal(
rec1.get_property("SimpleBooleanProperty").datatype,
h.BOOLEAN)
assert_equal(rec1.get_property("SimpleBooleanProperty").value, True)
assert rec1.is_valid()
assert rec1.get_property("SimpleBooleanProperty").datatype == h.BOOLEAN
assert rec1.get_property("SimpleBooleanProperty").value is True
rec1c = h.Record(id=rec1.id).retrieve()
assert_true(rec1c.is_valid())
assert_equal(rec1c.get_property(
"SimpleBooleanProperty").datatype, h.BOOLEAN)
assert_equal(rec1c.get_property(
"SimpleBooleanProperty").value, True)
assert rec1c.is_valid()
assert rec1c.get_property("SimpleBooleanProperty").datatype == h.BOOLEAN
assert rec1c.get_property("SimpleBooleanProperty").value is True
rec2 = h.Record(name="SimpleRecord2").add_parent(
rt).add_property(p, value=True).insert()
assert_true(rec2.is_valid())
assert_equal(
rec2.get_property("SimpleBooleanProperty").datatype,
h.BOOLEAN)
assert_equal(rec2.get_property("SimpleBooleanProperty").value, True)
assert rec2.is_valid()
assert rec2.get_property("SimpleBooleanProperty").datatype == h.BOOLEAN
assert rec2.get_property("SimpleBooleanProperty").value is True
rec2c = h.Record(id=rec2.id).retrieve()
assert_true(rec2c.is_valid())
assert_equal(rec2c.get_property(
"SimpleBooleanProperty").datatype, h.BOOLEAN)
assert_equal(rec2c.get_property(
"SimpleBooleanProperty").value, True)
assert rec2c.is_valid()
assert rec2c.get_property("SimpleBooleanProperty").datatype == h.BOOLEAN
assert rec2c.get_property("SimpleBooleanProperty").value is True
rec3 = h.Record(
name="SimpleRecord3").add_parent(rt).add_property(
......@@ -130,31 +106,7 @@ def test_record():
with raises(TransactionError):
rec3.insert()
assert_false(rec3.is_valid())
assert_equal(
rec3.get_property(
"SimpleBooleanProperty").get_errors()[0].description,
"Cannot parse value to boolean (either 'true' or 'false', "
"case insensitive).")
finally:
try:
rec3.delete()
except BaseException:
pass
try:
rec2.delete()
except BaseException:
pass
try:
rec1.delete()
except BaseException:
pass
try:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
assert not rec3.is_valid()
assert rec3.get_property("SimpleBooleanProperty").get_errors()[
0].description, ("Cannot parse value to boolean "
"(either 'true' or 'false' == " "case insensitive).")
......@@ -31,16 +31,13 @@ from pytest import raises
def setup_function(function):
try:
db.execute_query("FIND Test*").delete()
db.execute_query("FIND ENTITY").delete()
except BaseException:
pass
def teardown_function(function):
try:
db.execute_query("FIND Test*").delete()
except BaseException:
pass
setup_function(function)
def test_override_with_non_existing_ref():
......@@ -83,7 +80,7 @@ def test_reference_datatype_sequencial():
assert p.is_valid() is True
dt = db.execute_query("FIND TestProp", unique=True).datatype
dt = db.execute_query("FIND ENTITY TestProp", unique=True).datatype
assert dt == "TestRT"
rt2 = db.RecordType(
......
......@@ -25,7 +25,6 @@ from caosdb import Container, Property, RecordType, Record, execute_query
import caosdb
import caosdb as db
from caosdb.connection.connection import get_connection
from nose.tools import assert_is_not_none, assert_true, assert_equal
from caosdb.exceptions import TransactionError
from pytest import raises
......@@ -37,11 +36,7 @@ def setup_function(function):
def teardown_function(function):
try:
db.execute_query("FIND test_*").delete()
except BaseException:
pass
try:
db.execute_query("FIND Test*").delete()
db.execute_query("FIND ENTITY").delete()
except BaseException:
pass
......@@ -49,104 +44,81 @@ def teardown_function(function):
def test_default_datatype_for_recordtypes():
rt1 = db.RecordType(name="TestRT1").insert()
rt2 = db.RecordType(name="TestRT2").add_property(name="TestRT1").insert()
p = db.execute_query("FIND TestRT2", unique=True).get_properties()[0]
assert_equal(p.id, rt1.id)
assert_equal(p.name, rt1.name)
assert_equal(p.datatype, rt1.name)
p = db.execute_query("FIND ENTITY TestRT2", unique=True).get_properties()[0]
assert p.id == rt1.id
assert p.name == rt1.name
assert p.datatype == rt1.name
p = rt2.get_properties()[0]
assert_equal(p.id, rt1.id)
assert_equal(p.name, rt1.name)
assert_equal(p.datatype, rt1.name)
assert p.id == rt1.id
assert p.name == rt1.name
assert p.datatype == rt1.name
def test_datatype_inheritance():
insert = '<Insert><Property id="-1" name="test_property" description="bla" datatype="Text"/><RecordType name="test_rt" description="bla"> <Property id="-1" importance="obligatory" /></RecordType></Insert>'
print(insert)
con = get_connection()
http_response = con.insert(
entity_uri_segment=["Entity"], body=insert)
c = Container._response_to_entities(http_response)
print(c)
for e in c:
assert_true(e.is_valid())
assert_is_not_none(e.id)
assert_true(int(e.id) >= 100)
c.delete()
assert e.is_valid()
assert e.id is not None
assert int(e.id) >= 100
def test_datatype_overriding():
try:
p = Property(
name="DatatypeOverridingDoubleProperty",
description="DoubleDesc",
datatype="DOUBLE").insert()
assert_true(p.is_valid())
assert p.is_valid()
rt = RecordType(
name="DatatypeOverridingRT").add_property(
p, datatype="TEXT")
rt.insert()
assert_true(rt.is_valid())
assert_is_not_none(rt.get_properties())
assert_equal(1, len(rt.get_properties()))
assert_equal(
"DatatypeOverridingDoubleProperty",
rt.get_properties()[0].name)
assert_equal("DoubleDesc", rt.get_properties()[0].description)
assert_equal(
str("TEXT").lower(),
rt.get_properties()[0].datatype.lower())
assert rt.is_valid()
assert rt.get_properties() is not None
assert 1 == len(rt.get_properties())
assert "DatatypeOverridingDoubleProperty" == rt.get_properties()[0].name
assert "DoubleDesc" == rt.get_properties()[0].description
assert str("TEXT").lower() == rt.get_properties()[0].datatype.lower()
# retrieve again:
rt = execute_query("FIND DatatypeOverridingRT", unique=True)
assert_true(rt.is_valid())
assert_is_not_none(rt.get_properties())
assert_equal(1, len(rt.get_properties()))
assert_equal(
"DatatypeOverridingDoubleProperty",
rt.get_properties()[0].name)
assert_equal("DoubleDesc", rt.get_properties()[0].description)
assert_equal(
str("TEXT").lower(),
rt.get_properties()[0].datatype.lower())
finally:
try:
execute_query("FIND DatatypeOverriding*").delete()
except BaseException:
pass
rt = execute_query("FIND ENTITY DatatypeOverridingRT", unique=True)
assert rt.is_valid()
assert rt.get_properties() is not None
assert 1 == len(rt.get_properties())
assert "DatatypeOverridingDoubleProperty" == rt.get_properties()[0].name
assert "DoubleDesc" == rt.get_properties()[0].description
assert str("TEXT").lower() == rt.get_properties()[0].datatype.lower()
def test_datatype_overriding_update():
try:
p = Property(name="DoubleProperty", datatype="DOUBLE").insert()
assert_true(p.is_valid())
assert p.is_valid()
rt = RecordType(
name="DatatypeOverridingRT").add_property(
p, datatype="TEXT")
rt.insert()
assert_true(rt.is_valid())
assert_is_not_none(rt.get_properties())
assert_equal(1, len(rt.get_properties()))
assert_equal("DoubleProperty", rt.get_properties()[0].name)
assert_equal(
str("TEXT").lower(),
rt.get_properties()[0].datatype.lower())
assert rt.is_valid()
assert rt.get_properties() is not None
assert 1 == len(rt.get_properties())
assert "DoubleProperty" == rt.get_properties()[0].name
assert str("TEXT").lower() == rt.get_properties()[0].datatype.lower()
# retrieve again:
rt = execute_query("FIND DatatypeOverridingRT", unique=True)
assert_true(rt.is_valid())
assert_is_not_none(rt.get_properties())
assert_equal(1, len(rt.get_properties()))
assert_equal("DoubleProperty", rt.get_properties()[0].name)
assert_equal(
str("TEXT").lower(),
rt.get_properties()[0].datatype.lower())
rt = execute_query("FIND ENTITY DatatypeOverridingRT", unique=True)
assert rt.is_valid()
assert rt.get_properties() is not None
assert 1 == len(rt.get_properties())
assert "DoubleProperty" == rt.get_properties()[0].name
assert str("TEXT").lower() == rt.get_properties()[0].datatype.lower()
with raises(TransactionError) as te:
p.datatype = "INT"
......@@ -156,72 +128,50 @@ def test_datatype_overriding_update():
p.datatype = "INTEGER"
p.update()
assert_true(p.is_valid())
assert_equal(str("INTEGER").lower(), p.datatype.lower())
assert p.is_valid()
assert str("INTEGER").lower() == p.datatype.lower()
# retrieve again:
p = execute_query("FIND DoubleProperty", unique=True)
assert_true(p.is_valid())
assert_equal(str("INTEGER").lower(), p.datatype.lower())
p = execute_query("FIND ENTITY DoubleProperty", unique=True)
assert p.is_valid()
assert str("INTEGER").lower() == p.datatype.lower()
# retrieve rt again:
rt = execute_query("FIND DatatypeOverridingRT", unique=True)
assert_true(rt.is_valid())
assert_is_not_none(rt.get_properties())
assert_equal(1, len(rt.get_properties()))
assert_equal("DoubleProperty", rt.get_properties()[0].name)
rt = execute_query("FIND ENTITY DatatypeOverridingRT", unique=True)
assert rt.is_valid()
assert rt.get_properties() is not None
assert 1 == len(rt.get_properties())
assert "DoubleProperty" == rt.get_properties()[0].name
# overriding still ok?
assert_equal(
str("TEXT").lower(),
rt.get_properties()[0].datatype.lower())
assert str("TEXT").lower() == rt.get_properties()[0].datatype.lower()
rt.get_properties()[0].datatype = "DATETIME"
rt.update()
assert_true(rt.is_valid())
assert_is_not_none(rt.get_properties())
assert_equal(1, len(rt.get_properties()))
assert_equal("DoubleProperty", rt.get_properties()[0].name)
assert_equal(
str("DATETIME").lower(),
rt.get_properties()[0].datatype.lower())
finally:
try:
if len(execute_query("FIND DatatypeOverridingRT")) > 0:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
assert rt.is_valid()
assert rt.get_properties() is not None
assert 1 == len(rt.get_properties())
assert "DoubleProperty" == rt.get_properties()[0].name
assert str("DATETIME").lower() == rt.get_properties()[0].datatype.lower()
def test_recordtype_to_record():
try:
rt = RecordType(name="SimpleTextRecordType")
rt.datatype = "TEXT"
rt.insert()
assert_true(rt.is_valid())
assert_is_not_none(rt.datatype)
assert_equal(str("TEXT").lower(), rt.datatype.lower())
assert rt.is_valid()
assert rt.datatype is not None
assert str("TEXT").lower() == rt.datatype.lower()
rt = RecordType(id=rt.id).retrieve()
assert_true(rt.is_valid())
assert_is_not_none(rt.datatype)
assert_equal(str("TEXT").lower(), rt.datatype.lower())
assert rt.is_valid()
assert rt.datatype is not None
assert str("TEXT").lower() == rt.datatype.lower()
rec = Record().add_parent(name="SimpleTextRecordType").insert()
assert_true(rec.is_valid())
assert rec.is_valid()
# TODO
# assert_is_not_none(rec.datatype)
# assert_equal(str("TEXT").lower(),rec.datatype.lower())
finally:
try:
rt = execute_query("FIND SimpleTextRecordType").delete()
except BaseException:
pass
# assert rec.datatype is not None
# assert str("TEXT").lower() == rec.datatype.lower()
def test_concrete_property():
......@@ -230,11 +180,11 @@ def test_concrete_property():
p = Property(
name="DataTypeTestDoubleProperty",
datatype=datatype.upper()).insert()
assert_true(p.is_valid())
assert_equal(p.datatype.upper(), datatype.upper())
assert p.is_valid()
assert p.datatype.upper() == datatype.upper()
p2 = Property(id=p.id).retrieve()
assert_equal(p.datatype, p2.datatype)
assert p.datatype == p2.datatype
return p
finally:
......@@ -242,7 +192,6 @@ def test_concrete_property():
p.delete()
except BaseException:
pass
try:
p_double = test("DOUBLE")
p_datetime = test("DATETIME")
p_integer = test("INTEGER")
......@@ -251,89 +200,26 @@ def test_concrete_property():
p_file = test("FILE")
p_reference = test("REFERENCE")
finally:
try:
p_reference.delete()
except BaseException:
pass
try:
p_file.delete()
except BaseException:
pass
try:
p_text.delete()
except BaseException:
pass
try:
p_timespan.delete()
except BaseException:
pass
try:
p_integer.delete()
except BaseException:
pass
try:
p_datetime.delete()
except BaseException:
pass
try:
p_double.delete()
except BaseException:
pass
def test_reference_with_null_value():
try:
try:
execute_query("FIND RT2").delete()
except BaseException:
pass
try:
execute_query("FIND RT1").delete()
except BaseException:
pass
RT1 = RecordType(name="RT1").insert()
assert_true(RT1.is_valid())
assert RT1.is_valid()
RT2 = RecordType(name="RT2").add_property(RT1).insert()
assert_true(RT2.is_valid())
assert_equal(RT2.get_property("RT1").datatype, "RT1")
assert RT2.is_valid()
assert RT2.get_property("RT1").datatype == "RT1"
RT2c = RecordType(name="RT2").retrieve()
assert_true(RT2c.is_valid())
assert_equal(RT2c.get_property("RT1").datatype, "RT1")
finally:
try:
RT2.delete()
except BaseException:
pass
try:
RT1.delete()
except BaseException:
pass
assert RT2c.is_valid()
assert RT2c.get_property("RT1").datatype == "RT1"
def test_duplicate_properties():
try:
p = Property(
name="SimpleTestProperty",
datatype=caosdb.DOUBLE).insert()
assert_true(p.is_valid())
assert p.is_valid()
rt = RecordType(
name="SimpleTestRecordType").add_property(
p,
datatype=caosdb.TEXT).add_property(
p,
datatype=caosdb.INTEGER).insert()
assert_true(rt.is_valid())
finally:
try:
rt.delete()
except BaseException:
pass
try:
p.delete()
except BaseException:
pass
name="SimpleTestRecordType").add_property(p, datatype=caosdb.TEXT).add_property(p, datatype=caosdb.INTEGER).insert()
assert rt.is_valid()
This diff is collapsed.
......@@ -26,7 +26,7 @@ import caosdb as db
def setup_module():
clean = db.execute_query("FIND Test*")
clean = db.execute_query("FIND ENTITY")
if len(clean) > 0:
clean.delete()
......@@ -40,9 +40,9 @@ def test_nan():
rt = db.RecordType(name="TestRecordType").add_property(name="TestDoubleProperty",
value="NaN").insert()
test1 = db.execute_query("FIND TestRecordType", unique=True)
test1 = db.execute_query("FIND ENTITY TestRecordType", unique=True)
assert math.isnan(test1.get_property("TestDoubleProperty").value)
test2 = db.execute_query(
"FIND TestRecordType WITH TestDoubleProperty = NaN", unique=True)
"FIND ENTITY TestRecordType WITH TestDoubleProperty = NaN", unique=True)
assert math.isnan(test1.get_property("TestDoubleProperty").value)
......@@ -34,7 +34,7 @@ def setup_function(function):
def teardown_function(function):
try:
db.execute_query("FIND Test*").delete()
db.execute_query("FIND ENTITY").delete()
except Exception as e:
print(e)
......@@ -50,8 +50,8 @@ def test_empty_string():
"TestProp").value == ""
# query language works for empty string
assert db.execute_query(
"FIND TestRT with TestProp=''", unique=True).id == r1.id
assert db.execute_query('FIND TestRT with TestProp=""',
"FIND ENTITY TestRT with TestProp=''", unique=True).id == r1.id
assert db.execute_query('FIND ENTITY TestRT with TestProp=""',
unique=True).get_property("TestProp").value == ""
r2 = db.Record()
......@@ -60,10 +60,10 @@ def test_empty_string():
r2.insert()
assert db.execute_query(
"FIND TestRT with TestProp='not empty'", unique=True).id == r2.id
"FIND ENTITY TestRT with TestProp='not empty'", unique=True).id == r2.id
# query language work while other records with non empty values are present
assert db.execute_query(
"FIND TestRT with TestProp=''", unique=True).id == r1.id
"FIND ENTITY TestRT with TestProp=''", unique=True).id == r1.id
assert len(db.execute_query("FIND Record TestRT")) == 2
......@@ -78,7 +78,7 @@ def test_null_value():
unique=True).get_property("TestProp").value is None
# query language works with null value
assert db.execute_query(
"FIND TestRT WHERE TestProp IS NULL", unique=True).id == r1.id
"FIND ENTITY TestRT WHERE TestProp IS NULL", unique=True).id == r1.id
# add a bit of noise
r2 = db.Record()
......@@ -87,10 +87,10 @@ def test_null_value():
r2.insert()
assert db.execute_query(
"FIND TestRT with TestProp='null'", unique=True).id == r2.id
"FIND ENTITY TestRT with TestProp='null'", unique=True).id == r2.id
# query language works while other record with non-null values are present
assert db.execute_query(
"FIND TestRT WHERE TestProp IS NULL", unique=True).id == r1.id
"FIND ENTITY TestRT WHERE TestProp IS NULL", unique=True).id == r1.id
assert len(db.execute_query("FIND Record TestRT")) == 2
......@@ -105,7 +105,7 @@ def test_list_with_empty_string():
unique=True).get_property("TestProp").value == [""]
# query language works
assert db.execute_query(
"FIND TestRT with TestProp=''", unique=True).id == r1.id
"FIND ENTITY TestRT with TestProp=''", unique=True).id == r1.id
r2 = db.Record()
r2.add_parent("TestRT")
......@@ -114,9 +114,9 @@ def test_list_with_empty_string():
assert db.execute_query("FIND Record TestRT with TestProp='leer'",
unique=True).get_property("TestProp").value == ["leer"]
assert db.execute_query(
"FIND TestRT with TestProp='leer'", unique=True).id == r2.id
"FIND ENTITY TestRT with TestProp='leer'", unique=True).id == r2.id
assert db.execute_query(
"FIND TestRT with TestProp=''", unique=True).id == r1.id
"FIND ENTITY TestRT with TestProp=''", unique=True).id == r1.id
def test_null_list():
......@@ -129,7 +129,7 @@ def test_null_list():
assert db.execute_query("FIND Record TestRT",
unique=True).get_property("TestProp").value is None
assert db.execute_query(
"FIND TestRT WHERE TestProp IS NULL", unique=True).id == r1.id
"FIND ENTITY TestRT WHERE TestProp IS NULL", unique=True).id == r1.id
@pytest.mark.xfail(reason="""this is the confirmation for
......
......@@ -30,7 +30,7 @@ Created on 19.02.2015.
@author: tf
"""
import caosdb as h
import linkahead as db
from caosdb.exceptions import (AmbiguousEntityError,
EntityDoesNotExistError, EntityError,
EntityHasNoDatatypeError,
......@@ -42,7 +42,7 @@ import pytest
def setup_function(function):
try:
h.execute_query("FIND Test*").delete()
db.execute_query("FIND ENTITY").delete()
except BaseException:
pass
......@@ -54,7 +54,7 @@ def teardown_function(function):
def test_retrieval_no_exception_raised():
"""Test whether retrieval fails but error is suppressed."""
p = h.Property(name="TestNon-ExsistentProperty").retrieve(
p = db.Property(name="TestNon-ExsistentProperty").retrieve(
unique=True, raise_exception_on_error=False)
assert not p.is_valid()
assert (p.id is None or p.id < 0)
......@@ -67,7 +67,7 @@ def test_retrieval_exception_raised():
"""
propname = "TestNon-ExistentProperty"
with pytest.raises(TransactionError) as te:
h.Property(name="TestNon-ExistentProperty").retrieve(unique=True,
db.Property(name="TestNon-ExistentProperty").retrieve(unique=True,
raise_exception_on_error=True)
assert len(te.value.errors) == 1
ee = te.value.errors[0]
......@@ -89,19 +89,19 @@ def test_ambiguous_retrieval():
raised correctly if there are two possible candidates.
"""
h.RecordType(name="TestType").insert()
h.Record(name="TestRec").add_parent(name="TestType").insert()
db.RecordType(name="TestType").insert()
db.Record(name="TestRec").add_parent(name="TestType").insert()
# Insert twice, so unique=False
h.Record(name="TestRec").add_parent(name="TestType").insert(unique=False)
db.Record(name="TestRec").add_parent(name="TestType").insert(unique=False)
with pytest.raises(TransactionError) as te:
h.Record(name="TestRec").retrieve()
db.Record(name="TestRec").retrieve()
assert te.value.has_error(AmbiguousEntityError)
assert te.value.errors[0].entity.name == "TestRec"
def test_insertion_no_exception_raised():
"""Test whether insertion fails but no error is raised."""
p = h.Property(name="TestNoTypeProperty").insert(
p = db.Property(name="TestNoTypeProperty").insert(
raise_exception_on_error=False)
assert not p.is_valid()
assert (p.id is None or p.id < 0)
......@@ -109,7 +109,7 @@ def test_insertion_no_exception_raised():
def test_insertion_exception_raised():
"""Test insertion of a property with missing datatype."""
p = h.Property(name="TestNoTypeProperty")
p = db.Property(name="TestNoTypeProperty")
with pytest.raises(TransactionError) as te:
p.insert(raise_exception_on_error=True)
assert te.value.has_error(EntityHasNoDatatypeError)
......@@ -117,7 +117,7 @@ def test_insertion_exception_raised():
def test_insertion_with_invalid_parents():
with pytest.raises(TransactionError) as te:
p = h.Property(
p = db.Property(
name="TestNoTypeProperty",
datatype="Text").add_parent(
id=-1)
......@@ -136,7 +136,7 @@ def test_insertion_with_invalid_parents():
def test_insertion_with_invalid_properties():
with pytest.raises(TransactionError) as te:
p = h.Property(
p = db.Property(
name="TestNoTypeProperty",
datatype="Text").add_property(
id=-1)
......@@ -157,11 +157,11 @@ def test_entity_does_not_exist():
EntityDoesNotExistErrors.
"""
p1 = h.Property(name="TestNon-ExistentProperty1").retrieve(
p1 = db.Property(name="TestNon-ExistentProperty1").retrieve(
raise_exception_on_error=False)
p2 = h.Property(name="TestNon-ExistentProperty2").retrieve(
p2 = db.Property(name="TestNon-ExistentProperty2").retrieve(
raise_exception_on_error=False)
p3 = h.Property(name="TestNon-ExistentProperty3").retrieve(
p3 = db.Property(name="TestNon-ExistentProperty3").retrieve(
raise_exception_on_error=False)
# None of them should exist
assert not p1.is_valid()
......@@ -171,17 +171,17 @@ def test_entity_does_not_exist():
assert not p3.is_valid()
assert (p3.id is None or p3.id < 0)
pe = h.Property(name="TestExistentProperty", datatype="text").insert()
pe = db.Property(name="TestExistentProperty", datatype="text").insert()
c = h.Container().extend(
c = db.Container().extend(
[
h.Property(
db.Property(
name="TestNon-ExistentProperty1"),
h.Property(
db.Property(
name="TestNon-ExistentProperty2"),
h.Property(
db.Property(
name="TestNon-ExistentProperty3"),
h.Property(
db.Property(
name="TestExistentProperty")])
with pytest.raises(TransactionError) as te:
......@@ -199,11 +199,11 @@ def test_insert_existent_entity():
UniqueNamesError.
"""
p1 = h.Property(name="TestNon-ExistentProperty1").retrieve(
p1 = db.Property(name="TestNon-ExistentProperty1").retrieve(
raise_exception_on_error=False)
p2 = h.Property(name="TestNon-ExistentProperty2").retrieve(
p2 = db.Property(name="TestNon-ExistentProperty2").retrieve(
raise_exception_on_error=False)
p3 = h.Property(name="TestNon-ExistentProperty3").retrieve(
p3 = db.Property(name="TestNon-ExistentProperty3").retrieve(
raise_exception_on_error=False)
# None of them should exist
assert not p1.is_valid()
......@@ -213,21 +213,21 @@ def test_insert_existent_entity():
assert not p3.is_valid()
assert (p3.id is None or p3.id < 0)
pe = h.Property(name="TestExistentProperty", datatype="text").insert()
pe = db.Property(name="TestExistentProperty", datatype="text").insert()
assert pe.is_valid()
c = h.Container().extend(
c = db.Container().extend(
[
h.Property(
db.Property(
name="TestNon-ExistentProperty1",
datatype="text"),
h.Property(
db.Property(
name="TestNon-ExistentProperty2",
datatype="text"),
h.Property(
db.Property(
name="TestNon-ExistentProperty3",
datatype="text"),
h.Property(
db.Property(
name="TestExistentProperty",
datatype="text")])
......@@ -243,31 +243,31 @@ def test_insert_existent_entity():
def test_double_insertion():
c1 = h.Container()
c1 = db.Container()
c1.append(
h.Property(
db.Property(
name="TestSimpleTextProperty",
description="simple text property (from test_error_stuff.py)",
datatype='text'))
c1.append(
h.Property(
db.Property(
name="TestSimpleDoubleProperty",
description="simple double property (from test_error_stuff.py)",
datatype='double'))
c1.append(
h.Property(
db.Property(
name="TestSimpleIntegerProperty",
description="simple integer property (from test_error_stuff.py)",
datatype='integer'))
c1.append(
h.Property(
db.Property(
name="TestSimpleDatetimeProperty",
description="simple datetime property (from test_error_stuff.py)",
datatype='datetime'))
c1.append(
h.RecordType(
db.RecordType(
name="TestSimpleRecordType",
description="simple recordType (from test_error_stuff.py)").add_property(
name='TestSimpleTextProperty').add_property(
......@@ -277,30 +277,30 @@ def test_double_insertion():
c1.insert()
c2 = h.Container()
c2 = db.Container()
c2.append(
h.Property(
db.Property(
name="TestSimpleTextProperty",
description="simple text property (from test_error_stuff.py)",
datatype='text'))
c2.append(
h.Property(
db.Property(
name="TestSimpleDoubleProperty",
description="simple double property (from test_error_stuff.py)",
datatype='double'))
c2.append(
h.Property(
db.Property(
name="TestSimpleIntegerProperty",
description="simple integer property (from test_error_stuff.py)",
datatype='integer'))
c2.append(
h.Property(
db.Property(
name="TestSimpleDatetimeProperty",
description="simple datetime property (from test_error_stuff.py)",
datatype='datetime'))
c2.append(
h.RecordType(
db.RecordType(
name="TestSimpleRecordType",
description="simple recordType (from test_error_stuff.py)").add_property(
name='TestSimpleTextProperty').add_property(
......@@ -322,7 +322,7 @@ def test_update_acl_errors():
`Entity.update_acl`
"""
rec_ne = h.Record("TestRecordNonExisting")
rec_ne = db.Record("TestRecordNonExisting")
with pytest.raises(TransactionError) as te:
......@@ -331,12 +331,36 @@ def test_update_acl_errors():
assert te.value.has_error(EntityDoesNotExistError)
assert te.value.errors[0].entity.name == rec_ne.name
rt = h.RecordType(name="TestType").insert()
rec = h.Record(name="TestRecord").add_parent(rt).insert()
h.Record(name=rec.name).add_parent(rt).insert(unique=False)
rt = db.RecordType(name="TestType").insert()
rec = db.Record(name="TestRecord").add_parent(rt).insert()
db.Record(name=rec.name).add_parent(rt).insert(unique=False)
with pytest.raises(TransactionError) as te:
h.Record(name=rec.name).update_acl()
db.Record(name=rec.name).update_acl()
assert te.value.has_error(AmbiguousEntityError)
def test_URI_too_long():
"""Some tests for variours URI too long commands.
See for example https://gitlab.indiscale.com/caosdb/src/caosdb-pylib/-/issues/180
"""
short = 100
uri_long = 819
header_long = 815
with pytest.raises(db.TransactionError) as excinfo:
db.execute_query("0123456789" * short)
assert "Parsing" in excinfo.value.msg
with pytest.raises(db.HTTPURITooLongError) as excinfo:
db.execute_query("0123456789" * uri_long)
assert "414" in excinfo.value.msg
with pytest.raises(db.HTTPURITooLongError) as excinfo:
db.execute_query("0123456789" * header_long)
assert "431" in excinfo.value.msg
......@@ -5,8 +5,8 @@
#
# Copyright (C) 2018 Research Group Biomedical Physics,
# Max-Planck-Institute for Dynamics and Self-Organization Göttingen
# Copyright (C) 2019 IndiScale GmbH (info@indiscale.com)
# Copyright (C) 2019 Daniel Hornung (d.hornung@indiscale.com)
# Copyright (C) 2019-2023 IndiScale GmbH (info@indiscale.com)
# Copyright (C) 2019-2023 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
......@@ -536,7 +536,7 @@ def test_insert_files_in_dir_with_symlink():
assert_is_not_none(c.messages["Info", 0])
assert_equal(c.messages["Info", 0][0],
"Files count in linked_subfolder/: 1")
assert_equal(c[0].name, "test2.dat")
assert c[0].name is None
assert_equal(c[0].path, "/linked_subfolder/test2.dat")
d = models.Container()
......@@ -551,7 +551,7 @@ def test_insert_files_in_dir_with_symlink():
assert_equal(d.messages["Info", 0][0],
"Files count in linked_subfolder/: 1")
assert_true(d[0].is_valid())
assert_equal(d[0].name, "test2.dat")
assert d[0].name is None
assert_equal(d[0].path, "/linked_subfolder/test2.dat")
finally:
......@@ -633,7 +633,7 @@ def test_insert_files_in_dir():
# only the new file is given back...
assert_equal(1, len(e))
assert_false(e[0].is_valid())
assert_equal(e[0].name, "test3.dat")
assert e[0].name is None
assert_equal(e[0].path, "/testfolder/test3.dat")
f = models.Container()
......@@ -648,7 +648,7 @@ def test_insert_files_in_dir():
# only the new file is given back...
assert_equal(1, len(f))
assert_true(f[0].is_valid())
assert_equal(f[0].name, "test3.dat")
assert f[0].name is None
assert_equal(f[0].path, "/testfolder/test3.dat")
finally:
......@@ -700,7 +700,7 @@ def test_insert_files_in_dir_regex():
# the other has been inserted
assert len(c) == 1
assert c[0].name == "test_insert.dat"
assert c[0].name is None
finally:
try:
......
......@@ -27,7 +27,7 @@ def setup_function(function):
def teardown_module():
clean = db.execute_query("FIND Test*")
clean = db.execute_query("FIND ENTITY")
if len(clean) > 0:
clean.delete()
......
......@@ -200,7 +200,7 @@ def test_inheritance_unit():
assert rt.is_valid()
assert rt.get_property("SimpleIntProperty").unit == "km"
rt2 = db.execute_query("FIND SimpleRecordType", True)
rt2 = db.execute_query("FIND ENTITY SimpleRecordType", True)
assert rt2.id == rt.id
assert rt2.get_property("SimpleIntProperty").unit == "km"
......@@ -212,7 +212,7 @@ def test_inheritance_unit():
assert rt3.get_property("SimpleIntProperty") is not None
assert rt3.get_property("SimpleIntProperty").unit == "km"
rt4 = db.execute_query("FIND SimpleRecordType2", True)
rt4 = db.execute_query("FIND ENTITY SimpleRecordType2", True)
assert rt4.is_valid()
assert rt4.id == rt3.id
assert rt4.get_property("SimpleIntProperty").unit == "km"
......
# -*- coding: utf-8 -*-
# This file is a part of the LinkAhead Project.
#
# Copyright (c) 2023 IndiScale GmbH <info@indiscale.com>
# Copyright (c) 2023 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/>.
"""Tests for issues on gitlab.com, project linkahead-pylib.
Tests should be named:
``test_gitlab_com_{issue_id}``
"""
import math
import os
import tempfile
import time
import linkahead as db
import pytest
from linkahead import administration as admin
from linkahead.exceptions import (TransactionError, HTTPClientError)
CURATOR_ROLE = "curator"
def setup_module():
db.configure_connection()
try:
db.execute_query("FIND ENTITY WITH ID > 99").delete()
except Exception as delete_exc:
print(delete_exc)
try:
admin._delete_user("TestUser")
except Exception as delete_exc:
print(delete_exc)
try:
admin._delete_role(CURATOR_ROLE)
except Exception as delete_exc:
print(delete_exc)
def setup_function(function):
"""No setup required."""
setup_module()
def teardown_function(function):
"""Deleting entities again."""
setup_module()
# ########################### Issue tests start here #####################
# @pytest.mark.xfail(reason="Entities with many, long, properties: "
# "https://gitlab.com/linkahead/linkahead-pylib/-/issues/108")
def test_gitlab_com_108():
"""Create RT and a list of properties, then insert and retrieve.
This is another instance of bugs caused by caosdb/src/caosdb-mysqlbackend#48, but was originally
reported as https://gitlab.com/linkahead/linkahead-pylib/-/issues/108
"""
cont = db.Container()
long = "Long" * 50
first_RT = db.RecordType(name="TestRecord_first")
for index in range(20):
this_RT = db.RecordType(name=f"TestRecord_{long}_{index:02d}")
first_RT.add_property(this_RT)
cont.append(this_RT)
cont.append(first_RT)
cont.insert()
cont.retrieve()
print("retrieved")
# Incidentally, the following lines seem to trigger another, unrelated problem
tests = db.execute_query("FIND ENTITY test*", cache=False)
tests.delete()
print("deleted")