import datetime import os import unittest from sys import version_info import yaml if version_info > (3, 4): from owlrl import DeductiveClosure, RDFS_Semantics from rdflib import URIRef, Literal from rdflib.namespace import Namespace, RDF, RDFS, XSD from rdflib.plugins.sparql import prepareQuery import odml from odml.format import Format from odml.tools.rdf_converter import ODML_NS, RDFWriter from .test_samplefile import SampleFileCreator from .test_samplefile import parse ODMLNS = Format.namespace() class TestRDFWriter(unittest.TestCase): def setUp(self): doc = SampleFileCreator().create_document() doc1 = SampleFileCreator().create_document() self.doc = doc self.doc1 = doc1 def test_convert_to_rdf(self): rdf_writer = RDFWriter([self.doc, self.doc1]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subjects(predicate=RDF.type, object=URIRef(ODMLNS.Document)) self.assertEqual(len(list(check)), 2) def test_adding_doc_to_the_hub(self): rdf_writer = RDFWriter([self.doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.objects(subject=rdf_writer.hub_root, predicate=ODMLNS.hasDocument) self.assertEqual(len(list(check)), 1) def test_adding_repository(self): rdf_writer = RDFWriter([self.doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.objects(subject=rdf_writer.hub_root, predicate=ODMLNS.hasTerminology) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.objects(subject=URIRef(ODMLNS + rdf_writer.docs[0].id), predicate=ODMLNS.hasTerminology) self.assertEqual(len(list(check)), 0) url = "terminology_url" self.doc.repository = url rdf_writer = RDFWriter([self.doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subjects(predicate=RDF.type, object=URIRef(url)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.objects(subject=rdf_writer.hub_root, predicate=ODMLNS.hasTerminology) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.objects(subject=URIRef(ODMLNS + rdf_writer.docs[0].id), predicate=ODMLNS.hasTerminology) self.assertEqual(len(list(check)), 1) def test_adding_sections(self): doc = odml.Document() rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=ODMLNS.hasSection) self.assertEqual(len(list(check)), 0) rdf_writer = RDFWriter([self.doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=ODMLNS.hasSection) self.assertEqual(len(list(check)), 9) rdf_writer = RDFWriter([self.doc, self.doc1]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=ODMLNS.hasSection) self.assertEqual(len(list(check)), 18) def test_adding_properties(self): doc = parse(""" s1[t1] - s11[t1] s2[t2] """) rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=ODMLNS.hasProperty) self.assertEqual(len(list(check)), 0) rdf_writer = RDFWriter([self.doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=ODMLNS.hasProperty) self.assertEqual(len(list(check)), 12) rdf_writer = RDFWriter([self.doc, self.doc1]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=ODMLNS.hasProperty) self.assertEqual(len(list(check)), 24) def test_adding_values(self): doc = parse(""" s1[t1] """) rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=RDF.li) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.subject_objects(predicate=URIRef("%s_1" % str(RDF))) self.assertEqual(len(list(check)), 0) doc = parse(""" s1[t1] - p1 """) rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subjects(predicate=RDF.li, object=Literal("val")) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.subjects(predicate=URIRef("%s_1" % str(RDF)), object=Literal("val")) self.assertEqual(len(list(check)), 1) doc.sections[0].properties[0].append("val2") rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subject_objects(predicate=RDF.li) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.subjects(predicate=RDF.li, object=Literal("val")) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.subjects(predicate=RDF.li, object=Literal("val2")) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.subjects(predicate=URIRef("%s_1" % str(RDF)), object=Literal("val")) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=URIRef("%s_2" % str(RDF)), object=Literal("val2")) self.assertEqual(len(list(check)), 1) doc = parse(""" s1[t1] - p1 s2[t2] - p1 - p2 """) rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subjects(predicate=RDF.li, object=Literal("val")) self.assertEqual(len(list(check)), 0) check = rdf_writer.graph.subjects(predicate=URIRef("%s_1" % str(RDF)), object=Literal("val")) self.assertEqual(len(list(check)), 3) def test_section_subclass(self): file_path = os.path.join(odml.__path__[0], 'resources', 'section_subclasses.yaml') with open(file_path, "r") as subclass_file: subclass = yaml.safe_load(subclass_file) doc = odml.Document() subclass_key = next(iter(subclass)) sec = odml.Section("S", type=subclass_key) doc.append(sec) rdf_writer = RDFWriter(doc) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subjects(predicate=RDF.type, object=URIRef(ODMLNS[subclass[subclass_key]])) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=RDF.type, object=URIRef(ODMLNS.Section)) self.assertEqual(len(list(check)), 0) def test_adding_other_entities_properties(self): doc = parse(""" s1[t1] - p1 """) version = "v1" date = datetime.date(1979, 10, 12) author = "nice person" s_def = "comment" s_ref = "reference" p_unit = "u1" p_name = "p1" p_def = "p comment" p_uncertainty = 13.0 p_dtype = "string" p_value_origin = "value" p_ref = "p_ref" doc.version = version doc.date = date doc.author = author doc.sections[0].definition = s_def doc.sections[0].reference = s_ref doc.sections[0].properties[0].name = p_name doc.sections[0].properties[0].unit = p_unit doc.sections[0].properties[0].definition = p_def doc.sections[0].properties[0].uncertainty = p_uncertainty doc.sections[0].properties[0].dtype = p_dtype doc.sections[0].properties[0].value_origin = p_value_origin doc.sections[0].properties[0].reference = p_ref rdf_writer = RDFWriter([doc]) rdf_writer.convert_to_rdf() check = rdf_writer.graph.subjects(predicate=ODMLNS.hasDocVersion, object=Literal(version)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasDate, object=Literal(date, datatype=XSD.date)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasAuthor, object=Literal(author)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasName, object=Literal("s1")) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasType, object=Literal("t1")) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasDefinition, object=Literal(s_def)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasReference, object=Literal(s_ref)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasName, object=Literal(p_name)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasUnit, object=Literal(p_unit)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasDefinition, object=Literal(p_def)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasUncertainty, object=Literal(p_uncertainty)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasDtype, object=Literal(p_dtype)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasValueOrigin, object=Literal(p_value_origin)) self.assertEqual(len(list(check)), 1) check = rdf_writer.graph.subjects(predicate=ODMLNS.hasReference, object=Literal(p_ref)) self.assertEqual(len(list(check)), 1) def test_get_rdf_string(self): rdf_writer = RDFWriter([self.doc1]) rdf_writer.get_rdf_str() with self.assertRaises(ValueError): rdf_writer.get_rdf_str("abc") def test_rdf_subclassing_switch(self): """ Test the RDF section subclassing switch. """ # Section type term defined in odml/resources/section_subclasses.yaml that will # be converted to an RDF Section Subclass of Class "Cell". sub_class_term = "cell" # Create minimal document doc = odml.Document() _ = odml.Section(name="test_subclassing", type=sub_class_term, parent=doc) # Test default subclassing rdf_writer = RDFWriter([doc]) result = rdf_writer.get_rdf_str() self.assertIn("odml:Cell", result) # Test inactivation of subclassing feature rdf_writer = RDFWriter([doc], rdf_subclassing=False) result = rdf_writer.get_rdf_str() self.assertNotIn("odml:Cell", result) def test_rdf_custom_subclasses(self): """ Test collection of the odml RDF subclassing feature. Tests that the resulting output RDF document contains any required additional RDF subclasses. """ sub_class_term = "cell" # Create minimal document doc = odml.Document() _ = odml.Section(name="test_subclassing", type=sub_class_term, parent=doc) # Test None dict rdf_writer = RDFWriter([doc], custom_subclasses=None) self.assertIn("odml:Cell", rdf_writer.get_rdf_str()) # Test invalid dict rdf_writer = RDFWriter([doc], custom_subclasses=["invalid"]) self.assertIn("odml:Cell", rdf_writer.get_rdf_str()) # Test value whitespace inval_a = "This should" inval_b = "fail\nin" inval_c = "the\tmost" inval_d = "complete\rway" invalid_dict = {"type_1": inval_a, "type_2": inval_b, "type_3": inval_c, "type_4": inval_d} with self.assertRaises(ValueError): _ = RDFWriter([doc], custom_subclasses=invalid_dict) # Test custom subclassing type_custom_class = "species" custom_class_dict = {type_custom_class: "Species"} doc = odml.Document() _ = odml.Section(name="test_subclassing", type="cell", parent=doc) _ = odml.Section(name="test_custom_subclassing", type=type_custom_class, parent=doc) rdf_writer = RDFWriter([doc], custom_subclasses=custom_class_dict) self.assertIn("odml:Cell", rdf_writer.get_rdf_str()) self.assertIn("odml:Species", rdf_writer.get_rdf_str()) # Test custom subclassing overwrite sub_class_type = "cell" custom_class_dict = {sub_class_type: "Neuron"} doc = odml.Document() _ = odml.Section(name="test_subclassing", type=sub_class_type, parent=doc) if version_info > (3, 4): with self.assertWarns(UserWarning): rdf_writer = RDFWriter([doc], custom_subclasses=custom_class_dict) self.assertNotIn("odml:Cell", rdf_writer.get_rdf_str()) self.assertIn("odml:Neuron", rdf_writer.get_rdf_str()) def test_rdf_subclassing_definitions(self): """ Test that RDF Subclass definitions are written to the resulting graph. """ # -- Test default subclassing doc = odml.Document() _ = odml.Section(name="test_subclassing", type="cell", parent=doc) rdf_writer = RDFWriter([doc]) curr_str = " ".join(rdf_writer.get_rdf_str().split()) self.assertIn("odml:Cell a rdfs:Class ; rdfs:subClassOf odml:Section", curr_str) self.assertIn("odml:Section a rdfs:Class", curr_str) # -- Test multiple entries; a definition should only occur once in an RDF document doc = odml.Document() sec = odml.Section(name="test_subclassing", type="cell", parent=doc) sub_sec = odml.Section(name="test_subclassing", type="cell", parent=sec) _ = odml.Section(name="test_subclassing", type="cell", parent=sub_sec) rdf_writer = RDFWriter([doc]) curr_str = " ".join(rdf_writer.get_rdf_str().split()) self.assertIn("odml:Cell a rdfs:Class ; rdfs:subClassOf odml:Section", curr_str) self.assertIs(curr_str.count("odml:Cell a rdfs:Class ; rdfs:subClassOf odml:Section"), 1) self.assertIn("odml:Section a rdfs:Class", curr_str) self.assertIs(curr_str.count("odml:Section a rdfs:Class"), 1) # -- Test custom subclassing type_custom_class = "species" custom_class_dict = {type_custom_class: "Species"} doc = odml.Document() _ = odml.Section(name="test_subclassing", type="cell", parent=doc) _ = odml.Section(name="test_custom_subclassing", type=type_custom_class, parent=doc) rdf_writer = RDFWriter([doc], custom_subclasses=custom_class_dict) curr_str = " ".join(rdf_writer.get_rdf_str().split()) self.assertIn("odml:Cell a rdfs:Class ; rdfs:subClassOf odml:Section", curr_str) self.assertIn("odml:Species a rdfs:Class ; rdfs:subClassOf odml:Section", curr_str) self.assertIn("odml:Section a rdfs:Class", curr_str) # -- Test inactive subclassing doc = odml.Document() _ = odml.Section(name="test_subclassing", type="cell", parent=doc) rdf_writer = RDFWriter([doc], rdf_subclassing=False) curr_str = " ".join(rdf_writer.get_rdf_str().split()) self.assertNotIn("odml:Section a rdfs:Class", curr_str) self.assertNotIn("odml:Cell a rdfs:Class ; rdfs:subClassOf odml:Section", curr_str) def test_rdf_subclassing_queries(self): """ Test the proper implementation of the RDF subclassing feature. Tests ensure, that queries relying on RDF Subclasses return appropriate results. """ if version_info > (3, 4): namespace_map = {"odml": Namespace(ODML_NS), "rdf": RDF, "rdfs": RDFS} doc = odml.Document() _ = odml.Section(name="test_subclass", type="cell", parent=doc) _ = odml.Section(name="test_regular_class", type="regular", parent=doc) rdf_writer = RDFWriter([doc]) _ = rdf_writer.get_rdf_str() use_graph = rdf_writer.graph DeductiveClosure(RDFS_Semantics).expand(use_graph) q_string = "SELECT * WHERE {?s rdf:type odml:Section .}" curr_query = prepareQuery(q_string, initNs=namespace_map) # Make sure the query finds two sections self.assertIs(len(use_graph.query(curr_query)), 2) # Make sure the query finds result_section = [] for row in use_graph.query(curr_query): result_section.append(row.s) q_string = "SELECT * WHERE {?s rdf:type odml:Cell .}" curr_query = prepareQuery(q_string, initNs=namespace_map) self.assertIs(len(use_graph.query(curr_query)), 1) for row in use_graph.query(curr_query): self.assertIn(row.s, result_section) # -- Test custom subclassing queries type_custom_class = "species" type_overwrite_class = "cell" custom_class_dict = {type_custom_class: "Species", type_overwrite_class: "Neuron"} doc = odml.Document() sec = odml.Section(name="test_subclass", type="species", parent=doc) _ = odml.Section(name="test_subclass_overwrite", type="cell", parent=sec) _ = odml.Section(name="test_regular_class", type="regular", parent=sec) rdf_writer = RDFWriter([doc], custom_subclasses=custom_class_dict) _ = rdf_writer.get_rdf_str() use_graph = rdf_writer.graph DeductiveClosure(RDFS_Semantics).expand(use_graph) q_string = "SELECT * WHERE {?s rdf:type odml:Section .}" curr_query = prepareQuery(q_string, initNs=namespace_map) # Make sure the query finds three sections self.assertIs(len(use_graph.query(curr_query)), 3) # Make sure the query finds result_section = [] for row in use_graph.query(curr_query): result_section.append(row.s) # Custom class 'Species' should be found. q_string = "SELECT * WHERE {?s rdf:type odml:Species .}" curr_query = prepareQuery(q_string, initNs=namespace_map) self.assertIs(len(use_graph.query(curr_query)), 1) for row in use_graph.query(curr_query): self.assertIn(row.s, result_section) # Custom class 'Neuron' should be found. q_string = "SELECT * WHERE {?s rdf:type odml:Neuron .}" curr_query = prepareQuery(q_string, initNs=namespace_map) self.assertIs(len(use_graph.query(curr_query)), 1) for row in use_graph.query(curr_query): self.assertIn(row.s, result_section) # Default class 'Cell' was replaced and should not return any result. q_string = "SELECT * WHERE {?s rdf:type odml:Cell .}" curr_query = prepareQuery(q_string, initNs=namespace_map) self.assertIs(len(use_graph.query(curr_query)), 0) # -- Test inactivated subclassing doc = odml.Document() _ = odml.Section(name="test_regular_class", type="regular", parent=doc) _ = odml.Section(name="test_subclass", type="cell", parent=doc) rdf_writer = RDFWriter([doc], rdf_subclassing=False) _ = rdf_writer.get_rdf_str() use_graph = rdf_writer.graph DeductiveClosure(RDFS_Semantics).expand(use_graph) q_string = "SELECT * WHERE {?s rdf:type odml:Section .}" curr_query = prepareQuery(q_string, initNs=namespace_map) self.assertIs(len(use_graph.query(curr_query)), 2) q_string = "SELECT * WHERE {?s rdf:type odml:Cell .}" curr_query = prepareQuery(q_string, initNs=namespace_map) self.assertIs(len(use_graph.query(curr_query)), 0)