Skip to content
Snippets Groups Projects

FIX: Better implementation of split_restricted_path that fixes an error. Also...

1 file
+ 76
1
Compare changes
  • Side-by-side
  • Inline
+ 76
1
@@ -114,7 +114,7 @@ def test_issue_23(clear_database):
@@ -114,7 +114,7 @@ def test_issue_23(clear_database):
assert rec_crawled.get_property("identifying_prop").value == "identifier"
assert rec_crawled.get_property("identifying_prop").value == "identifier"
assert rec_crawled.get_property("prop_b") is not None
assert rec_crawled.get_property("prop_b") is not None
assert rec_crawled.get_property("prop_b").value == "something_else"
assert rec_crawled.get_property("prop_b").value == "something_else"
# no interaction with the database yet, so the rrecord shouldn't have a prop_a yet
# no interaction with the database yet, so the record shouldn't have a prop_a yet
assert rec_crawled.get_property("prop_a") is None
assert rec_crawled.get_property("prop_a") is None
# synchronize with database and update the record
# synchronize with database and update the record
@@ -133,3 +133,78 @@ def test_issue_23(clear_database):
@@ -133,3 +133,78 @@ def test_issue_23(clear_database):
"identifying_prop").value == rec_crawled.get_property("identifying_prop").value
"identifying_prop").value == rec_crawled.get_property("identifying_prop").value
assert rec_retrieved.get_property(
assert rec_retrieved.get_property(
"prop_b").value == rec_crawled.get_property("prop_b").value
"prop_b").value == rec_crawled.get_property("prop_b").value
 
 
 
def test_issue_83(clear_database):
 
"""https://gitlab.com/linkahead/linkahead-crawler/-/issues/83. Test that
 
names don't need to be unique for referenced entities if they are not part
 
of the identifiable.
 
 
"""
 
 
# Very simple data model
 
identifying_prop = db.Property(name="IdentifyingProp", datatype=db.INTEGER).insert()
 
referenced_type = db.RecordType(name="ReferencedType").add_property(
 
name=identifying_prop.name, importance=db.OBLIGATORY).insert()
 
referencing_type = db.RecordType(name="ReferencingType").add_property(
 
name=referenced_type.name, datatype=db.LIST(referenced_type.name)).insert()
 
 
# Define identifiables. ReferencingType by name, ReferencedType by
 
# IdentifyingProp and not by name.
 
ident = CaosDBIdentifiableAdapter()
 
ident.register_identifiable(referenced_type.name, db.RecordType().add_parent(
 
name=referenced_type.name).add_property(name=identifying_prop.name))
 
ident.register_identifiable(referencing_type.name, db.RecordType().add_parent(
 
name=referencing_type.name).add_property(name="name"))
 
 
crawler = Crawler(identifiableAdapter=ident)
 
 
ref_target1 = db.Record(name="RefTarget").add_parent(
 
name=referenced_type.name).add_property(name=identifying_prop.name, value=1)
 
ref_target2 = db.Record(name="RefTarget").add_parent(
 
name=referenced_type.name).add_property(name=identifying_prop.name, value=2)
 
 
referencing1 = db.Record(name="Referencing1").add_parent(
 
name=referencing_type.name).add_property(name=referenced_type.name, value=[ref_target1])
 
referencing2 = db.Record(name="Referencing2").add_parent(
 
name=referencing_type.name).add_property(name=referenced_type.name, value=[ref_target2])
 
referencing3 = db.Record(name="Referencing3").add_parent(name=referencing_type.name).add_property(
 
name=referenced_type.name, value=[ref_target1, ref_target2])
 
 
records = db.Container().extend(
 
[ref_target1, ref_target2, referencing1, referencing2, referencing3])
 
 
ins, ups = crawler.synchronize(crawled_data=records, unique_names=False)
 
assert len(ins) == len(records)
 
assert len(ups) == 0
 
 
retrieved_target1 = db.execute_query(
 
f"FIND {referenced_type.name} WITH {identifying_prop.name}=1", unique=True)
 
retrieved_target2 = db.execute_query(
 
f"FIND {referenced_type.name} WITH {identifying_prop.name}=2", unique=True)
 
assert retrieved_target2.name == retrieved_target1.name
 
assert retrieved_target1.name == ref_target1.name
 
assert retrieved_target1.id != retrieved_target2.id
 
 
retrieved_referencing1 = db.execute_query(
 
f"FIND {referencing_type.name} WITH name={referencing1.name}", unique=True)
 
assert retrieved_referencing1.get_property(referenced_type.name) is not None
 
assert retrieved_referencing1.get_property(referenced_type.name).value == [
 
retrieved_target1.id]
 
assert retrieved_referencing1.get_property(referenced_type.name).value != [
 
retrieved_target2.id]
 
 
retrieved_referencing2 = db.execute_query(
 
f"FIND {referencing_type.name} WITH name={referencing2.name}", unique=True)
 
assert retrieved_referencing2.get_property(referenced_type.name) is not None
 
assert retrieved_referencing2.get_property(referenced_type.name).value == [
 
retrieved_target2.id]
 
assert retrieved_referencing2.get_property(referenced_type.name).value != [
 
retrieved_target1.id]
 
 
retrieved_referencing3 = db.execute_query(
 
f"FIND {referencing_type.name} WITH name={referencing3.name}", unique=True)
 
assert retrieved_referencing3.get_property(referenced_type.name) is not None
 
assert len(retrieved_referencing3.get_property(referenced_type.name).value) == 2
 
assert retrieved_target1.id in retrieved_referencing3.get_property(referenced_type.name).value
 
assert retrieved_target2.id in retrieved_referencing3.get_property(referenced_type.name).value
Loading