""" Part of the bsie test suite. A copy of the license is provided with the project. Author: Matthias Baumgartner, 2022 """ # imports import logging import unittest # bsie imports from bsie import base from bsie.utils import bsfs # objects to test from bsie.tools.builder import ExtractorBuilder from bsie.tools.builder import PipelineBuilder from bsie.tools.builder import ReaderBuilder from bsie.tools.builder import _safe_load from bsie.tools.builder import _unpack_name ## code ## class TestUtils(unittest.TestCase): def test_safe_load(self): # invalid module self.assertRaises(base.errors.LoaderError, _safe_load, 'dBGHMSAYOoKeKMpywDoKZQycENFPvN', 'foobar') self.assertRaises(base.errors.LoaderError, _safe_load, 'dBGHMSAYOoKeKMpywDoKZQycENFPvN.bar', 'foobar') # partially valid module self.assertRaises(base.errors.LoaderError, _safe_load, 'os.foo', 'foobar') # invalid class self.assertRaises(base.errors.LoaderError, _safe_load, 'os.path', 'foo') # valid module and class cls = _safe_load('collections.abc', 'Container') import collections.abc self.assertEqual(cls, collections.abc.Container) def test_unpack_name(self): self.assertRaises(TypeError, _unpack_name, 123) self.assertRaises(TypeError, _unpack_name, None) self.assertRaises(ValueError, _unpack_name, '') self.assertRaises(ValueError, _unpack_name, 'path') self.assertRaises(ValueError, _unpack_name, '.Path') self.assertEqual(_unpack_name('path.Path'), ('path', 'Path')) self.assertEqual(_unpack_name('path.foo.bar.Path'), ('path.foo.bar', 'Path')) class TestReaderBuilder(unittest.TestCase): def test_build(self): builder = ReaderBuilder({'bsie.reader.path.Path': {}}) # build configured reader cls = builder.build('bsie.reader.path.Path') import bsie.reader.path self.assertIsInstance(cls, bsie.reader.path.Path) # build unconfigured reader cls = builder.build('bsie.reader.stat.Stat') import bsie.reader.stat self.assertIsInstance(cls, bsie.reader.stat.Stat) # re-build previous reader (test cache) self.assertEqual(cls, builder.build('bsie.reader.stat.Stat')) # test invalid self.assertRaises(TypeError, builder.build, 123) self.assertRaises(TypeError, builder.build, None) self.assertRaises(ValueError, builder.build, '') self.assertRaises(ValueError, builder.build, 'Path') self.assertRaises(base.errors.BuilderError, builder.build, 'path.Path') # invalid config builder = ReaderBuilder({'bsie.reader.stat.Stat': dict(foo=123)}) self.assertRaises(base.errors.BuilderError, builder.build, 'bsie.reader.stat.Stat') builder = ReaderBuilder({'bsie.reader.stat.Stat': 123}) self.assertRaises(TypeError, builder.build, 'bsie.reader.stat.Stat') # no instructions builder = ReaderBuilder({}) cls = builder.build('bsie.reader.stat.Stat') self.assertIsInstance(cls, bsie.reader.stat.Stat) class TestExtractorBuilder(unittest.TestCase): def test_iter(self): # no specifications self.assertListEqual(list(ExtractorBuilder([])), []) # some specifications builder = ExtractorBuilder([ {'bsie.extractor.generic.path.Path': {}}, {'bsie.extractor.generic.stat.Stat': {}}, {'bsie.extractor.generic.path.Path': {}}, ]) self.assertListEqual(list(builder), [0, 1, 2]) def test_build(self): # simple and repeated extractors builder = ExtractorBuilder([ {'bsie.extractor.generic.path.Path': {}}, {'bsie.extractor.generic.stat.Stat': {}}, {'bsie.extractor.generic.path.Path': {}}, ]) ext = [builder.build(0), builder.build(1), builder.build(2)] import bsie.extractor.generic.path import bsie.extractor.generic.stat self.assertListEqual(ext, [ bsie.extractor.generic.path.Path(), bsie.extractor.generic.stat.Stat(), bsie.extractor.generic.path.Path(), ]) # out-of-bounds raises KeyError self.assertRaises(IndexError, builder.build, 3) # building with args builder = ExtractorBuilder([ {'bsie.extractor.generic.constant.Constant': { 'schema': ''' bse:author rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:string ; bsfs:unique "true"^^xsd:boolean . bse:rating rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:integer ; bsfs:unique "true"^^xsd:boolean . ''', 'tuples': [ ('http://bsfs.ai/schema/Entity#author', 'Me, myself, and I'), ('http://bsfs.ai/schema/Entity#rating', 123), ], }}]) obj = builder.build(0) import bsie.extractor.generic.constant self.assertEqual(obj, bsie.extractor.generic.constant.Constant(''' bse:author rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:string ; bsfs:unique "true"^^xsd:boolean . bse:rating rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:integer ; bsfs:unique "true"^^xsd:boolean . ''', [ ('http://bsfs.ai/schema/Entity#author', 'Me, myself, and I'), ('http://bsfs.ai/schema/Entity#rating', 123), ])) # building with invalid args self.assertRaises(base.errors.BuilderError, ExtractorBuilder( [{'bsie.extractor.generic.path.Path': {'foo': 123}}]).build, 0) # non-dict build specification self.assertRaises(TypeError, ExtractorBuilder( [('bsie.extractor.generic.path.Path', {})]).build, 0) # multiple keys per build specification self.assertRaises(TypeError, ExtractorBuilder( [{'bsie.extractor.generic.path.Path': {}, 'bsie.extractor.generic.stat.Stat': {}}]).build, 0) # non-dict value for kwargs self.assertRaises(TypeError, ExtractorBuilder( [{'bsie.extractor.generic.path.Path': 123}]).build, 0) class TestPipelineBuilder(unittest.TestCase): def test_build(self): prefix = bsfs.URI('http://example.com/local/file#') c_schema = ''' bse:author rdfs:subClassOf bsfs:Predicate ; rdfs:domain bsfs:Entity ; rdfs:range xsd:string ; bsfs:unique "true"^^xsd:boolean . ''' c_tuples = [('http://bsfs.ai/schema/Entity#author', 'Me, myself, and I')] # prepare builders rbuild = ReaderBuilder({}) ebuild = ExtractorBuilder([ {'bsie.extractor.generic.path.Path': {}}, {'bsie.extractor.generic.stat.Stat': {}}, {'bsie.extractor.generic.constant.Constant': dict( schema=c_schema, tuples=c_tuples, )}, ]) # build pipeline builder = PipelineBuilder(prefix, rbuild, ebuild) pipeline = builder.build() # delayed import import bsie.reader.path import bsie.reader.stat import bsie.extractor.generic.path import bsie.extractor.generic.stat import bsie.extractor.generic.constant # check pipeline self.assertDictEqual(pipeline._ext2rdr, { bsie.extractor.generic.path.Path(): bsie.reader.path.Path(), bsie.extractor.generic.stat.Stat(): bsie.reader.stat.Stat(), bsie.extractor.generic.constant.Constant(c_schema, c_tuples): None, }) # fail to load extractor ebuild_err = ExtractorBuilder([ {'bsie.extractor.generic.foo.Foo': {}}, {'bsie.extractor.generic.path.Path': {}}, ]) with self.assertLogs(logging.getLogger('bsie.tools.builder'), logging.ERROR): pipeline = PipelineBuilder(prefix, rbuild, ebuild_err).build() self.assertDictEqual(pipeline._ext2rdr, { bsie.extractor.generic.path.Path(): bsie.reader.path.Path()}) # fail to build extractor ebuild_err = ExtractorBuilder([ {'bsie.extractor.generic.path.Path': {'foo': 123}}, {'bsie.extractor.generic.path.Path': {}}, ]) with self.assertLogs(logging.getLogger('bsie.tools.builder'), logging.ERROR): pipeline = PipelineBuilder(prefix, rbuild, ebuild_err).build() self.assertDictEqual(pipeline._ext2rdr, { bsie.extractor.generic.path.Path(): bsie.reader.path.Path()}) # fail to load reader with self.assertLogs(logging.getLogger('bsie.tools.builder'), logging.ERROR): # switch reader of an extractor old_reader = bsie.extractor.generic.path.Path.CONTENT_READER bsie.extractor.generic.path.Path.CONTENT_READER = 'bsie.reader.foo.Foo' # build pipeline with invalid reader reference pipeline = PipelineBuilder(prefix, rbuild, ebuild).build() self.assertDictEqual(pipeline._ext2rdr, { bsie.extractor.generic.stat.Stat(): bsie.reader.stat.Stat(), bsie.extractor.generic.constant.Constant(c_schema, c_tuples): None, }) # switch back bsie.extractor.generic.path.Path.CONTENT_READER = old_reader # fail to build reader rbuild_err = ReaderBuilder({'bsie.reader.stat.Stat': dict(foo=123)}) with self.assertLogs(logging.getLogger('bsie.tools.builder'), logging.ERROR): pipeline = PipelineBuilder(prefix, rbuild_err, ebuild).build() self.assertDictEqual(pipeline._ext2rdr, { bsie.extractor.generic.path.Path(): bsie.reader.path.Path(), bsie.extractor.generic.constant.Constant(c_schema, c_tuples): None, }) ## main ## if __name__ == '__main__': unittest.main() ## EOF ##