Skip to content
Snippets Groups Projects
Commit 4da6f3a6 authored by Alexander Schlemmer's avatar Alexander Schlemmer
Browse files

TST: fixed detection of insertions

parent d87a9eed
No related branches found
No related tags found
1 merge request!53Release 0.1
......@@ -28,6 +28,7 @@
module description
"""
from caosdb import EmptyUniqueQueryError
import argparse
import sys
from argparse import RawTextHelpFormatter
......@@ -38,9 +39,10 @@ import pytest
from caosadvancedtools.models.parser import parse_model_from_yaml
import yaml
import os
from caosadvancedtools.testutils import clear_database, set_test_key
set_test_key("10b128cf8a1372f30aa3697466bb55e76974e0c16a599bb44ace88f19c8f61e2")
TEST_KEY = "10b128cf8a1372f30aa3697466bb55e76974e0c16a599bb44ace88f19c8f61e2"
import os
def rfp(*pathcomponents):
"""
......@@ -50,65 +52,6 @@ def rfp(*pathcomponents):
return os.path.join(os.path.dirname(__file__), *pathcomponents)
def not_test_record(r: db.Record):
global TEST_KEY
if r.name == "PyTestInfo" or r.name == "TestIdentification":
return False
if r.get_property("TestIdentification") is not None:
if (r.get_property("TestIdentification").value ==
TEST_KEY):
return False
return True
def register_test():
info = db.Info()
for info_element in info.messages:
if info_element.startswith("<Counts>"):
counts = yaml.load(info_element[8:-9], Loader=yaml.SafeLoader)
for count_element in ['records',
'properties',
'recordTypes',
'files',
'fssize']:
if counts[count_element] > 0:
raise RuntimeError("This instance of CaosDB contains entities already."
"It must be empty in order to gegister a new test.")
break
else:
raise RuntimeError("Response from server for Info could not be interpreted.")
answer = input("This method will register your current test with key {} with the currently"
" running instance of CaosDB. Do you want to continue (y/N)?".format(
TEST_KEY))
if answer != "y":
raise RuntimeError("Test registration aborted by user.")
model = parse_model_from_yaml(rfp("test_model.yml"))
model.sync_data_model(noquestion=True, verbose=False)
r = db.Record()
r.add_parent("PyTestInfo")
r.add_property("TestIdentification", TEST_KEY)
r.insert()
@pytest.fixture
def clear_database():
"""First remove Records, then RecordTypes, then Properties, finally
files. Since there may be no entities, execute all deletions
without raising errors.
"""
global TEST_KEY
test_record = db.execute_query("FIND Record PyTestInfo", unique=True)
if test_record.get_property("TestIdentification").value != TEST_KEY:
raise RuntimeError("Test database has not been setup for this test.")
c = db.execute_query("FIND Record")
c.extend(db.execute_query("FIND RecordType"))
c.extend(db.execute_query("FIND Property"))
c.extend(db.execute_query("FIND File"))
c = db.Container().extend([r for r in c if not_test_record(r)])
c.delete(raise_exception_on_error=False)
@pytest.fixture
......
......@@ -375,12 +375,19 @@ class Crawler(object):
del flat[i]
resolved_references = True
resolved_references = True
while resolved_references and len(flat) > 0:
resolved_references = False
for i in reversed(range(len(flat))):
record = flat[i]
# e.g. references an identifiable that does not exist remotely
elif self.all_references_are_existing_already(record):
if self.all_references_are_existing_already(record):
to_be_inserted.append(record)
self.add_identified_record_to_local_cache(record)
del flat[i]
resolved_references = True
if len(flat) > 0:
raise RuntimeError("Could not resolve all Entity references. Circular Dependency?")
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment