aboutsummaryrefslogtreecommitdiffstats
path: root/test/tools/test_builder.py
diff options
context:
space:
mode:
authorMatthias Baumgartner <dev@igsor.net>2023-03-05 19:22:58 +0100
committerMatthias Baumgartner <dev@igsor.net>2023-03-05 19:22:58 +0100
commita35b33f4f1ddcf6f1bb8ab0f41b87bf2b847f11d (patch)
treefb220da28bb7248ebf37ce09af5de88f2c1aaad4 /test/tools/test_builder.py
parent7582c280ad5324a2f0427999911c7e7abc14a6ab (diff)
parentaf81318ae9311fd0b0e16949cef3cfaf7996970b (diff)
downloadbsie-release.tar.gz
bsie-release.tar.bz2
bsie-release.zip
Merge branch 'develop'HEADv0.23.03releasemain
Diffstat (limited to 'test/tools/test_builder.py')
-rw-r--r--test/tools/test_builder.py246
1 files changed, 0 insertions, 246 deletions
diff --git a/test/tools/test_builder.py b/test/tools/test_builder.py
deleted file mode 100644
index 62c637c..0000000
--- a/test/tools/test_builder.py
+++ /dev/null
@@ -1,246 +0,0 @@
-"""
-
-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 ##