HORNET / tests / test_plugin.py

# -*- coding: utf-8 -*-
#
# Copyright 2009 Vanderbilt University
# 
# Licensed under the Apache License, Version 2.0 (the "License"); 
# you may not use this file except in compliance with the License. 
# You may obtain a copy of the License at 
# 
#     http://www.apache.org/licenses/LICENSE-2.0 
# 
# Unless required by applicable law or agreed to in writing, software 
# distributed under the License is distributed on an "AS IS" BASIS, 
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
# See the License for the specific language governing permissions and 
# limitations under the License. 

from helpers import FakeEvent, FakePlugin, cause_error
from hornet.application import Config
from hornet.event import Event, EventNotAccepted
from hornet.plugin import *
from mock import Mock
from unittest import TestCase
import helpers
import hornet.plugin
import os
import tempfile

config = Config()
config.output_dir = tempfile.gettempdir()

class FakeHornet(object):
    config = config

class BadEvent(hornet.event.Event):
    pass

class LoggingPlugin(Plugin):
    def __init__(self):
        super(LoggingPlugin, self).__init__()
        self._id = 'no-op'
    
    @log
    def run(self):
        """Some docstring"""
        return 'ret value'

class NotifyingPlugin(Plugin):
    def __init__(self):
        super(NotifyingPlugin, self).__init__()
        self._id = 'no-op'
        self.done = False
    
    @notify_on_finish
    def run(self):
        """Some docstring"""
        self.done = True
        return FakeEvent('this is the result')

class FunctionsTestCase(TestCase):
    def test_valid_plugin(self):
        valid_plugin(FakePlugin())
        
    def test_valid_plugin_not_valid(self):
        class C(object): pass
        self.assertRaises(InvalidPlugin, valid_plugin, C())
    
    def test_log(self):
        plugin = LoggingPlugin()
        self.assertEqual('Some docstring', plugin.run.__doc__)
        self.assertEqual('run', plugin.run.__name__)
        self.assertEqual('ret value', plugin.run())
    
    def test_notify_on_finish(self):
        listener = FakePlugin()
        
        runner = NotifyingPlugin()
        runner.register(listener)
        
        self.assertFalse(listener.notify_called)
        runner.run()
        self.assertTrue(listener.notify_called)
        self.assertTrue(runner.done)
  
    
class PluginTestCase(TestCase):
    def setUp(self):
        self.config = Config()
        self.config.output_dir = 'output'
        self.config.input_dir = 'input'
        self.hornet =  FakeHornet()
        self.hornet.config = self.config
        self.plugin = FakePlugin()
        self.plugin._id = 'testPlugin'
        self.plugin._parent = self.hornet
        self.feeder_plugin = FakePlugin()
        self.feeder_plugin._id = 'rootPlugin' 

    def test_output(self):
        expected = ''.join(['output', os.sep, 'test.csv'])
        self.assertEqual(expected, self.plugin.output('test.csv'))
    
    def test_type(self):
        self.assertEqual('FakePlugin', self.plugin.type[1])
        
    def test_id(self):
        self.assertEqual('testPlugin', self.plugin.id)
     
    def test_parent(self):
        self.assertEqual(self.hornet, self.plugin.parent)

    def test_register(self):
        self.feeder_plugin.register(self.plugin)
        self.assertEqual(1, len(self.feeder_plugin.listeners()))
        self.assertTrue(self.plugin in self.feeder_plugin.listeners())
    
    def test_unregister(self):
        self.feeder_plugin.register(self.plugin)
        self.feeder_plugin.unregister(self.plugin)
        self.assertEqual(0, len(self.feeder_plugin.listeners()))

    def test_notify_listeners(self):
        self.feeder_plugin.register(self.plugin)
        self.feeder_plugin.notify_listeners(FakeEvent('event data'))
        self.assertTrue(self.plugin.notify_called)
        
        result = self.plugin.notify_event
        self.assertEqual('event data', result.data)
        self.assertEqual('rootPlugin', result.caller)

    def test_notify_listeners_event_not_accepted(self):
        self.feeder_plugin.register(self.plugin)
        self.failUnlessRaises(hornet.event.EventNotAccepted, 
                              self.feeder_plugin.notify_listeners, 
                              BadEvent('event data'))
    
    def test_initialize(self):
        self.plugin.listen_to.append(self.feeder_plugin)
        self.plugin.setup()
        self.assertTrue(self.plugin.setup_called)
        self.assertTrue(self.plugin in self.feeder_plugin.listeners())
    
    def test_teardown(self):
        self.plugin.listen_to.append(self.feeder_plugin)
        self.plugin.setup()
        self.plugin.teardown()
        self.assertTrue(self.plugin.teardown_called)
        self.assertEqual(0, len(self.feeder_plugin.listeners()))

    def test_stop_teardown_exception(self):
        self.plugin.teardown_error = True
        try:
            self.plugin.teardown()
        except:
            self.fail('exception should be caught')

    def test_stop_unregister_exception(self):
        announcer = FakePlugin()
        announcer.unregister = cause_error
        
        self.plugin.listen_to = [announcer]
        self.plugin.setup()
        
        try:
            self.plugin.teardown()
        except:
            self.fail('exception should be caught')

    def test_validate_event(self):
        n = self.plugin.notify
        self.assertEqual('notify', n.__name__)
        self.assertEqual('some docstring', n.__doc__)
        self.assertEqual(helpers.__name__, n.__module__)
        
    def test_notify_accepted(self):
        self.plugin.notify(FakeEvent())
        self.assertTrue(self.plugin.notify_called)
        
    def test_notify_accepted_not_set(self):
        self.failUnlessRaises(EventNotAccepted, self.plugin.notify, Event())

class DefaultPluginTestCase(TestCase):
    def setUp(self):
        self._plugin_logger = hornet.plugin.logger
        hornet.plugin.logger = Mock()
        
        self.plugin = Plugin()

    def tearDown(self):
        hornet.plugin.logger = self._plugin_logger
         
    def test_setup(self):
        self.plugin.setup()
        self.assertTrue(hornet.plugin.logger.debug.called)

    def test_teardown(self):
        self.plugin.teardown()
        self.assertTrue(hornet.plugin.logger.debug.called)
    
    def test_notify_none(self):
        self.failUnlessRaises(EventNotAccepted, self.plugin.notify, None)
    
    def test_notify_accepted_not_set(self):
        self.failUnlessRaises(EventNotAccepted, self.plugin.notify, Event())


class ArgPlugin(Plugin):
    def __init__(self, *args, **kw):
        self.args = args
        self.kw = kw
        
def test_plugin_with_no_args():
    plugin = ArgPlugin()
    assert len(plugin.args) == 0
    assert len(plugin.kw) == 0
    
def test_plugin_with_init_args():
    plugin = ArgPlugin('a')
    assert len(plugin.args) == 1
    assert len(plugin.kw) == 0
    
def test_plugin_with_init_kw():
    plugin = ArgPlugin(thing='a')
    assert len(plugin.args) == 0
    assert len(plugin.kw) == 1

def test_plugin_with_init_both():
    plugin = ArgPlugin(1, 4, 5, foo='bar', thing='a')
    assert len(plugin.args) == 3
    assert len(plugin.kw) == 2
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.