"""Tests for Lab 01 — a financial knowledge graph from scratch. pytest test_lab.py -v LAB_MODULE=solution pytest test_lab.py -v # the reference — must be green """ import importlib import os import pytest lab = importlib.import_module(os.environ.get("LAB_MODULE", "lab")) def graph(): return lab.Graph() def built(): g = lab.Graph() lab.build_ontology(g) lab.build_facts(g) return g def materialized(): g = built() lab.materialize(g) return g # ====================================================================================== # 1. Terms and prefixes # ====================================================================================== def test_iri_and_literal_are_distinct_types(): assert lab.IRI("x") != lab.Literal("x") def test_literal_carries_a_datatype(): assert lab.Literal("5", "xsd:integer").datatype == "xsd:integer" assert lab.Literal("x").datatype == "xsd:string" def test_prefix_expansion_round_trips(): p = lab.PrefixMap() iri = p.expand("bank:controls") assert iri.value.startswith("https://bank.example.ae/ontology/") assert p.shorten(iri) == "bank:controls" def test_unknown_prefix_is_an_error(): with pytest.raises(KeyError): lab.PrefixMap().expand("nope:thing") def test_a_full_iri_passes_through(): p = lab.PrefixMap() assert p.expand("https://example.org/x").value == "https://example.org/x" def test_shorten_prefers_the_longest_matching_namespace(): p = lab.PrefixMap({"fibo-be-le": "https://spec.edmcouncil.org/fibo/ontology/BE/LegalEntities/x/"}) iri = lab.IRI("https://spec.edmcouncil.org/fibo/ontology/BE/LegalEntities/x/Thing") assert p.shorten(iri) == "fibo-be-le:Thing" # ====================================================================================== # 2. The triple store # ====================================================================================== def test_a_graph_is_a_set_not_a_list(): g = graph() assert g.add_curies("ent:A", "bank:controls", "ent:B") is True assert g.add_curies("ent:A", "bank:controls", "ent:B") is False assert len(g) == 1 def test_contains_uses_triple_identity(): g = graph() g.add_curies("ent:A", "bank:controls", "ent:B") p = g.prefixes assert lab.Triple(p.expand("ent:A"), p.expand("bank:controls"), p.expand("ent:B")) in g def test_match_with_every_position_free(): g = built() assert len(g.match()) == len(g) def test_match_by_subject_predicate_and_object(): g = graph() g.add_curies("ent:A", "bank:controls", "ent:B") g.add_curies("ent:A", "bank:legalName", "A Ltd", literal=True) p = g.prefixes assert len(g.match(subject=p.expand("ent:A"))) == 2 assert len(g.match(predicate=p.expand("bank:controls"))) == 1 assert len(g.match(obj=p.expand("ent:B"))) == 1 assert len(g.match(subject=p.expand("ent:A"), predicate=p.expand("bank:controls"))) == 1 def test_match_returns_sorted_output(): g = built() triples = g.match() keys = [(str(t.subject), str(t.predicate), str(t.object)) for t in triples] assert keys == sorted(keys) def test_objects_and_subjects_accessors(): g = graph() g.add_curies("ent:A", "bank:controls", "ent:B") p = g.prefixes assert g.objects(p.expand("ent:A"), p.expand("bank:controls")) == [p.expand("ent:B")] assert g.subjects(p.expand("bank:controls"), p.expand("ent:B")) == [p.expand("ent:A")] def test_matching_an_absent_term_is_empty_not_an_error(): assert graph().match(subject=lab.IRI("nope")) == [] def test_literals_can_be_objects_but_are_matched_exactly(): g = graph() g.add_curies("ent:A", "bank:legalName", "Acme", literal=True) assert g.match(obj=lab.Literal("Acme")) != [] assert g.match(obj=lab.Literal("acme")) == [] # ====================================================================================== # 3. Entailment # ====================================================================================== def test_subclass_is_transitive(): g = graph() g.add_curies("bank:Bank", "rdfs:subClassOf", "fibo-be:LegalEntity") g.add_curies("fibo-be:LegalEntity", "rdfs:subClassOf", "fibo-fnd:AutonomousAgent") lab.materialize(g) p = g.prefixes assert lab.Triple(p.expand("bank:Bank"), lab.RDFS_SUBCLASS, p.expand("fibo-fnd:AutonomousAgent")) in g def test_type_propagates_up_the_class_hierarchy(): g = materialized() p = g.prefixes types = {p.shorten(t) for t in g.objects(p.expand("ent:Acme"), lab.RDF_TYPE)} assert {"bank:Corporation", "fibo-be:LegalEntity", "fibo-fnd:AutonomousAgent"} <= types def test_subproperty_assertions_become_superproperty_assertions(): g = materialized() p = g.prefixes assert lab.Triple(p.expand("ent:Northgate"), p.expand("bank:controls"), p.expand("ent:Acme")) in g def test_transitivity_composes_with_subproperty(): """majorityOwns => controls (rdfs7), then controls is transitive (owl). Nobody asserted Meridian controls Acme; two rules composed to derive it. """ g = materialized() p = g.prefixes assert lab.Triple(p.expand("ent:Meridian"), p.expand("bank:controls"), p.expand("ent:Acme")) in g def test_transitive_closure_reaches_the_top_of_the_chain(): g = materialized() p = g.prefixes assert lab.Triple(p.expand("ent:Opaque"), p.expand("bank:controls"), p.expand("ent:Acme")) in g def test_inverse_properties(): g = materialized() p = g.prefixes controllers = {p.shorten(t) for t in g.objects(p.expand("ent:Acme"), p.expand("bank:controlledBy"))} assert "ent:Northgate" in controllers def test_symmetric_properties(): g = materialized() p = g.prefixes assert lab.Triple(p.expand("ent:Northgate"), p.expand("bank:isCounterpartyOf"), p.expand("ent:Acme")) in g def test_domain_and_range_confer_types(): g = graph() g.add_curies("bank:obligor", "rdfs:domain", "bank:Obligation") g.add_curies("bank:obligor", "rdfs:range", "fibo-be:LegalEntity") g.add_curies("ent:L1", "bank:obligor", "ent:A") lab.materialize(g) p = g.prefixes assert lab.Triple(p.expand("ent:L1"), lab.RDF_TYPE, p.expand("bank:Obligation")) in g assert lab.Triple(p.expand("ent:A"), lab.RDF_TYPE, p.expand("fibo-be:LegalEntity")) in g def test_range_does_not_type_a_literal(): """A literal is never a subject, so it can never be given a type.""" g = graph() g.add_curies("bank:legalName", "rdfs:range", "xsd:string") g.add_curies("ent:A", "bank:legalName", "Acme", literal=True) lab.materialize(g) assert g.match(subject=lab.Literal("Acme")) == [] def test_materialization_reaches_a_fixed_point(): g = built() first = lab.materialize(g) second = lab.materialize(g) assert first.added > 0 assert second.added == 0 assert second.rounds == 1 def test_entailment_is_monotone(): """Adding an unrelated fact never retracts an existing conclusion.""" g = materialized() before = set(g.triples()) g.add_curies("ent:Unrelated", "bank:legalName", "Nothing To Do With It", literal=True) lab.materialize(g) assert before <= set(g.triples()) def test_entailment_terminates_on_a_cycle(): g = graph() g.add_curies("bank:controls", "rdf:type", "owl:TransitiveProperty") g.add_curies("ent:A", "bank:controls", "ent:B") g.add_curies("ent:B", "bank:controls", "ent:A") report = lab.materialize(g) assert report.rounds < 50 p = g.prefixes assert lab.Triple(p.expand("ent:A"), p.expand("bank:controls"), p.expand("ent:A")) in g # ====================================================================================== # 4. SHACL # ====================================================================================== def entity_shape(**kwargs): kwargs.setdefault("name", "S") kwargs.setdefault("target_class", "fibo-be:LegalEntity") return lab.NodeShape(**kwargs) def test_a_conforming_graph_conforms(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") g.add_curies("ent:A", "bank:legalName", "A Ltd", literal=True) shape = entity_shape(properties=( lab.PropertyShape(path="bank:legalName", min_count=1, max_count=1),)) assert lab.validate(g, [shape]).conforms def test_min_count_catches_a_missing_property(): """The question OWL cannot answer: this record is INCOMPLETE.""" g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") shape = entity_shape(properties=(lab.PropertyShape(path="bank:hasLEI", min_count=1),)) report = lab.validate(g, [shape]) assert not report.conforms assert report.results[0].path == "bank:hasLEI" def test_max_count_catches_a_duplicate(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") g.add_curies("ent:A", "bank:hasLEI", "AAA", literal=True) g.add_curies("ent:A", "bank:hasLEI", "BBB", literal=True) shape = entity_shape(properties=(lab.PropertyShape(path="bank:hasLEI", max_count=1),)) assert not lab.validate(g, [shape]).conforms def test_pattern_constraint(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") g.add_curies("ent:A", "bank:hasLEI", "NOT-AN-LEI", literal=True) shape = entity_shape(properties=( lab.PropertyShape(path="bank:hasLEI", pattern=r"^[A-Z0-9]{18}[0-9]{2}$"),)) report = lab.validate(g, [shape]) assert not report.conforms assert report.results[0].value == "NOT-AN-LEI" def test_datatype_constraint(): g = graph() g.add_curies("ent:L", "rdf:type", "bank:Obligation") g.add_curies("ent:L", "bank:amountAED", "250000", literal=True) # xsd:string shape = lab.NodeShape("S", "bank:Obligation", properties=( lab.PropertyShape(path="bank:amountAED", datatype="xsd:integer"),)) assert not lab.validate(g, [shape]).conforms def test_node_kind_constraint(): g = graph() g.add_curies("ent:L", "rdf:type", "bank:Obligation") g.add_curies("ent:L", "bank:obligor", "Acme", literal=True) # should be an IRI shape = lab.NodeShape("S", "bank:Obligation", properties=( lab.PropertyShape(path="bank:obligor", node_kind="IRI"),)) assert not lab.validate(g, [shape]).conforms def test_in_values_constraint(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") g.add_curies("ent:A", "bank:status", "ZOMBIE", literal=True) shape = entity_shape(properties=( lab.PropertyShape(path="bank:status", in_values=("ACTIVE", "DORMANT")),)) assert not lab.validate(g, [shape]).conforms def test_severity_below_violation_still_conforms(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") shape = entity_shape(properties=( lab.PropertyShape(path="bank:hasLEI", min_count=1, severity=lab.Severity.WARNING),)) report = lab.validate(g, [shape]) assert report.conforms # warnings do not fail validation assert len(report.by_severity(lab.Severity.WARNING)) == 1 def test_a_closed_shape_rejects_undeclared_predicates(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") g.add_curies("ent:A", "bank:legalName", "A", literal=True) g.add_curies("ent:A", "bank:secretField", "x", literal=True) shape = entity_shape(closed=True, properties=( lab.PropertyShape(path="bank:legalName", min_count=1),)) report = lab.validate(g, [shape]) assert not report.conforms assert any("not allowed" in v.message for v in report.results) def test_a_closed_shape_permits_ignored_properties(): g = graph() g.add_curies("ent:A", "rdf:type", "fibo-be:LegalEntity") g.add_curies("ent:A", "bank:legalName", "A", literal=True) g.add_curies("ent:A", "bank:internal", "x", literal=True) shape = entity_shape(closed=True, ignored_properties=("bank:internal",), properties=(lab.PropertyShape(path="bank:legalName", min_count=1),)) assert lab.validate(g, [shape]).conforms def test_shapes_target_entailed_types(): """A node typed only by inference is still validated — which is why materialization runs before validation.""" g = graph() g.add_curies("bank:Corporation", "rdfs:subClassOf", "fibo-be:LegalEntity") g.add_curies("ent:A", "rdf:type", "bank:Corporation") shape = entity_shape(properties=(lab.PropertyShape(path="bank:hasLEI", min_count=1),)) assert lab.validate(g, [shape]).conforms # not yet a LegalEntity lab.materialize(g) assert not lab.validate(g, [shape]).conforms # now it is, and it fails def test_validation_report_is_deterministic(): g = materialized() a = lab.validate(g, [lab.LEGAL_ENTITY_SHAPE, lab.OBLIGATION_SHAPE]) b = lab.validate(g, [lab.LEGAL_ENTITY_SHAPE, lab.OBLIGATION_SHAPE]) assert [str(v) for v in a.results] == [str(v) for v in b.results] def test_the_worked_graph_has_the_expected_violations(): report = lab.validate(materialized(), [lab.LEGAL_ENTITY_SHAPE, lab.OBLIGATION_SHAPE]) nodes = {v.focus_node for v in report.results} assert "ent:Meridian" in nodes # no LEI at all assert "ent:Northgate" in nodes # malformed LEI assert "ent:Acme" not in nodes # well-formed # ====================================================================================== # 5. SPARQL # ====================================================================================== def V(name): return lab.Variable(name) def test_a_single_pattern_binds_a_variable(): g = materialized() rows = lab.execute(g, lab.Query( select=("x",), where=(lab.Pattern("ent:Northgate", lab.PathStep("bank:majorityOwns"), V("x")),))) assert rows == [{"x": "ent:Acme"}] def test_shared_variables_are_the_join(): g = materialized() rows = lab.execute(g, lab.Query( select=("mid",), where=( lab.Pattern("ent:Meridian", lab.PathStep("bank:majorityOwns"), V("mid")), lab.Pattern(V("mid"), lab.PathStep("bank:majorityOwns"), "ent:Acme"), ))) assert rows == [{"mid": "ent:Northgate"}] def test_a_one_or_more_path_finds_the_whole_chain(): g = materialized() rows = lab.execute(g, lab.Query( select=("owner",), where=(lab.Pattern(V("owner"), lab.PathStep("bank:majorityOwns", "+"), "ent:Acme"),))) owners = {r["owner"] for r in rows} assert {"ent:Northgate", "ent:Meridian"} <= owners def test_a_zero_or_more_path_includes_the_start(): g = materialized() rows = lab.execute(g, lab.Query( select=("x",), where=(lab.Pattern("ent:Meridian", lab.PathStep("bank:majorityOwns", "*"), V("x")),))) assert "ent:Meridian" in {r["x"] for r in rows} def test_an_exact_path_does_not_transit(): g = graph() g.add_curies("ent:A", "bank:majorityOwns", "ent:B") g.add_curies("ent:B", "bank:majorityOwns", "ent:C") rows = lab.execute(g, lab.Query( select=("x",), where=(lab.Pattern("ent:A", lab.PathStep("bank:majorityOwns"), V("x")),))) assert [r["x"] for r in rows] == ["ent:B"] def test_an_inverse_path(): g = materialized() rows = lab.execute(g, lab.Query( select=("obligation",), where=(lab.Pattern("ent:Acme", lab.PathStep("bank:obligor", inverse=True), V("obligation")),))) assert [r["obligation"] for r in rows] == ["ent:LoanA"] def test_optional_is_a_left_join(): """A missing optional value leaves the variable unbound; the row survives.""" g = materialized() rows = lab.execute(g, lab.Query( select=("owner", "name"), where=( lab.Pattern(V("owner"), lab.PathStep("bank:controls", "+"), "ent:Acme"), lab.Optional_((lab.Pattern(V("owner"), lab.PathStep("bank:legalName"), V("name")),)), ))) by_owner = {r["owner"]: r["name"] for r in rows} assert by_owner["ent:Opaque"] == "" # no legalName, row still present assert by_owner["ent:Northgate"] == "Northgate Holdings Ltd" def test_a_required_pattern_removes_the_row(): g = materialized() rows = lab.execute(g, lab.Query( select=("owner",), where=( lab.Pattern(V("owner"), lab.PathStep("bank:controls", "+"), "ent:Acme"), lab.Pattern(V("owner"), lab.PathStep("bank:legalName"), V("name")), ))) assert "ent:Opaque" not in {r["owner"] for r in rows} def test_filter_narrows_bindings(): g = materialized() rows = lab.execute(g, lab.Query( select=("amount",), where=( lab.Pattern(V("o"), lab.PathStep("bank:amountAED"), V("amount")), lab.Filter(lambda b: int(str(b["amount"])) > 1_000_000), ))) assert rows == [] def test_the_sanctions_question(): """Three hops, one query — and no vector index can answer it.""" g = materialized() rows = lab.execute(g, lab.Query( select=("owner",), where=( lab.Pattern(V("owner"), lab.PathStep("bank:controls", "+"), "ent:Acme"), lab.Pattern(V("owner"), lab.PathStep("bank:onSanctionsList"), lab.Literal("true", "xsd:boolean")), ))) assert [r["owner"] for r in rows] == ["ent:Sanctioned"] def test_distinct_removes_duplicate_rows(): g = materialized() query = lab.Query( select=("owner",), where=(lab.Pattern(V("owner"), lab.PathStep("bank:controls", "+"), "ent:Acme"),)) rows = lab.execute(g, query) assert len(rows) == len({r["owner"] for r in rows}) def test_limit_and_order(): g = materialized() rows = lab.execute(g, lab.Query( select=("owner",), where=(lab.Pattern(V("owner"), lab.PathStep("bank:controls", "+"), "ent:Acme"),), limit=2)) assert len(rows) == 2 assert [r["owner"] for r in rows] == sorted(r["owner"] for r in rows) def test_a_query_matching_nothing_returns_no_rows(): g = materialized() rows = lab.execute(g, lab.Query( select=("x",), where=(lab.Pattern("ent:Nonexistent", lab.PathStep("bank:controls"), V("x")),))) assert rows == [] def test_a_path_traversal_terminates_on_a_cycle(): g = graph() g.add_curies("ent:A", "bank:controls", "ent:B") g.add_curies("ent:B", "bank:controls", "ent:A") rows = lab.execute(g, lab.Query( select=("x",), where=(lab.Pattern("ent:A", lab.PathStep("bank:controls", "+"), V("x")),))) assert {r["x"] for r in rows} == {"ent:A", "ent:B"} def test_results_are_deterministic(): g = materialized() query = lab.Query( select=("owner",), where=(lab.Pattern(V("owner"), lab.PathStep("bank:controls", "+"), "ent:Acme"),)) assert lab.execute(g, query) == lab.execute(g, query) # ====================================================================================== # 6. Graph-grounded retrieval # ====================================================================================== def test_neighbourhood_expansion_reaches_further_with_more_hops(): g = materialized() one = lab.expand_neighbourhood(g, "ent:Acme", hops=1, predicates=("bank:controlledBy",)) two = lab.expand_neighbourhood(g, "ent:Acme", hops=2, predicates=("bank:controlledBy",)) assert set(one.entities) <= set(two.entities) def test_zero_hops_is_just_the_seed(): g = materialized() context = lab.expand_neighbourhood(g, "ent:Acme", hops=0) assert context.entities == ("ent:Acme",) assert context.paths == () def test_expansion_can_be_restricted_to_predicates(): g = materialized() context = lab.expand_neighbourhood(g, "ent:Acme", hops=1, predicates=("bank:legalName",)) assert "Acme Trading FZE" in context.entities assert not any("controlledBy" in p for p in context.paths) def test_expansion_records_the_path_it_took(): g = materialized() context = lab.expand_neighbourhood(g, "ent:Acme", hops=1, predicates=("bank:controlledBy",)) assert any("controlledBy" in p for p in context.paths) assert context.rationale def test_expansion_terminates_on_a_cycle(): g = graph() g.add_curies("ent:A", "bank:controls", "ent:B") g.add_curies("ent:B", "bank:controls", "ent:A") context = lab.expand_neighbourhood(g, "ent:A", hops=10) assert set(context.entities) == {"ent:A", "ent:B"} def test_negative_hops_is_an_error(): with pytest.raises(ValueError): lab.expand_neighbourhood(materialized(), "ent:Acme", hops=-1) def test_expansion_is_deterministic(): g = materialized() a = lab.expand_neighbourhood(g, "ent:Acme", hops=2) b = lab.expand_neighbourhood(g, "ent:Acme", hops=2) assert a == b