<commit_before>from __future__ import division, absolute_import, print_function

__author__ = 'tamasgal'

from km3pipe.testing import *
from km3pipe.core import Pipeline, Module, Blob


class TestPipeline(TestCase):
    """Tests for the main pipeline"""

    def test_attach(self):
        pl = Pipeline()
        pl.attach(Module, 'module1')
        pl.attach(Module, 'module2')
        self.assertEqual('module1', pl.modules[0].name)
        self.assertEqual('module2', pl.modules[1].name)

    def test_drain_calls_process_method_on_each_attached_module(self):
        pl = Pipeline(blob=1, cycles=1)
        pl.attach(Module, 'module1')
        pl.attach(Module, 'module2')
        for module in pl.modules:
            module.process = MagicMock(return_value=1)
        pl.drain()
        for module in pl.modules:
            module.process.assert_called_once_with(1)

    def test_finish(self):
        pl = Pipeline()
        pl.finish()

    def test_drain_calls_finish_on_each_attached_module(self):
        pl = Pipeline(cycles=4)
        pl.attach(Module, 'module1')
        pl.attach(Module, 'module2')
        for module in pl.modules:
            module.finish = MagicMock()
        pl.drain()
        for module in pl.modules:
            module.finish.assert_called_once_with()


class TestModule(TestCase):
    """Tests for the pipeline module"""

    def test_name_can_be_set_on_init(self):
        name = 'foo'
        module = Module(name=name)
        self.assertEqual(name, module.name)

    def test_name_is_read_only(self):
        module = Module(name='foo')
        with self.assertRaises(AttributeError):
            module.name = 'narf'

    def test_process(self):
        blob = Blob()
        module = Module(name='foo')
        processed_blob = module.process(blob)
        self.assertIs(blob, processed_blob)

    def test_add_parameter(self):
        module = Module()
        module.add('foo', 'default', 'help')
        self.assertDictEqual({'foo': 'default'}, module.parameters)

    def test_get_parameter(self):
        module = Module()
        module.add('foo', 'default', 'help')
        self.assertEqual('default', module.get('foo'))

    def test_default_parameter_value_can_be_overwritten(self):
        class Foo(Module):
            def __init__(self, **context):
                super(self.__class__, self).__init__(**context)
                self.foo = self.get('foo') or 'default_foo'
        module = Foo()
        self.assertEqual('default_foo', module.foo)
        module = Foo(foo='overwritten')
        self.assertEqual('overwritten', module.foo)

    def test_finish(self):
        module = Module()
        module.finish()


class TestBlob(TestCase):
    """Tests for the blob holding the data"""

    def test_init(self):
        blob = Blob()

    def test_field_can_be_added(self):
        blob = Blob()
        blob['foo'] = 1
        self.assertEqual(1, blob['foo'])<commit_msg>Pylint fixes<commit_after>from __future__ import division, absolute_import, print_function

__author__ = 'tamasgal'

from km3pipe.testing import *
from km3pipe.core import Pipeline, Module, Blob


class TestPipeline(TestCase):
    """Tests for the main pipeline"""

    def test_attach(self):
        pl = Pipeline()
        pl.attach(Module, 'module1')
        pl.attach(Module, 'module2')
        self.assertEqual('module1', pl.modules[0].name)
        self.assertEqual('module2', pl.modules[1].name)

    def test_drain_calls_process_method_on_each_attached_module(self):
        pl = Pipeline(blob=1, cycles=1)
        pl.attach(Module, 'module1')
        pl.attach(Module, 'module2')
        for module in pl.modules:
            module.process = MagicMock(return_value=1)
        pl.drain()
        for module in pl.modules:
            module.process.assert_called_once_with(1)

    def test_finish(self):
        pl = Pipeline()
        pl.finish()

    def test_drain_calls_finish_on_each_attached_module(self):
        pl = Pipeline(cycles=4)
        pl.attach(Module, 'module1')
        pl.attach(Module, 'module2')
        for module in pl.modules:
            module.finish = MagicMock()
        pl.drain()
        for module in pl.modules:
            module.finish.assert_called_once_with()


class TestModule(TestCase):
    """Tests for the pipeline module"""

    def test_name_can_be_set_on_init(self):
        name = 'foo'
        module = Module(name=name)
        self.assertEqual(name, module.name)

    def test_name_is_read_only(self):
        module = Module(name='foo')
        with self.assertRaises(AttributeError):
            module.name = 'narf'

    def test_process(self):
        blob = Blob()
        module = Module(name='foo')
        processed_blob = module.process(blob)
        self.assertIs(blob, processed_blob)

    def test_add_parameter(self):
        module = Module()
        module.add('foo', 'default')
        self.assertDictEqual({'foo': 'default'}, module.parameters)

    def test_get_parameter(self):
        module = Module()
        module.add('foo', 'default')
        self.assertEqual('default', module.get('foo'))

    def test_default_parameter_value_can_be_overwritten(self):
        class Foo(Module):
            def __init__(self, **context):
                super(self.__class__, self).__init__(**context)
                self.foo = self.get('foo') or 'default_foo'
        module = Foo()
        self.assertEqual('default_foo', module.foo)
        module = Foo(foo='overwritten')
        self.assertEqual('overwritten', module.foo)

    def test_finish(self):
        module = Module()
        module.finish()


class TestBlob(TestCase):
    """Tests for the blob holding the data"""

    def test_init(self):
        blob = Blob()

    def test_field_can_be_added(self):
        blob = Blob()
        blob['foo'] = 1
        self.assertEqual(1, blob['foo'])
{'fraction_non_alphanumeric': 0.119693094629156, 'fraction_numerical': 0.006479113384484228, 'mean_word_length': 2.6571072319201994, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2820442', 'n_tokens_mistral': 1781, 'n_tokens_neox': 1737, 'n_words': 360}