amqtt/tests/plugins/test_manager.py

88 wiersze
2.7 KiB
Python
Czysty Zwykły widok Historia

2015-08-16 19:37:18 +00:00
# Copyright (c) 2015 Nicolas JOUANIN
#
# See the file license.txt for copying permission.
import unittest
import logging
import asyncio
2015-08-20 19:43:18 +00:00
from hbmqtt.plugins.manager import PluginManager, BaseContext
2015-08-16 19:37:18 +00:00
2015-08-20 19:43:18 +00:00
formatter = "[%(asctime)s] %(name)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s"
2015-08-17 19:52:26 +00:00
logging.basicConfig(level=logging.INFO, format=formatter)
2015-08-16 19:37:18 +00:00
class TestPlugin:
2015-08-20 19:43:18 +00:00
def __init__(self, context):
self.context = context
2015-08-16 19:37:18 +00:00
class EventTestPlugin:
2015-08-20 19:43:18 +00:00
def __init__(self, context):
self.context = context
2015-08-16 19:37:18 +00:00
self.test_flag = False
self.coro_flag = False
@asyncio.coroutine
def on_test(self):
self.test_flag = True
2015-08-20 19:43:18 +00:00
self.context.logger.info("on_test")
2015-08-16 19:37:18 +00:00
@asyncio.coroutine
def test_coro(self):
self.coro_flag = True
@asyncio.coroutine
def ret_coro(self):
return "TEST"
2015-08-16 19:37:18 +00:00
class TestPluginManager(unittest.TestCase):
def setUp(self):
self.loop = asyncio.new_event_loop()
def test_load_plugin(self):
manager = PluginManager("hbmqtt.test.plugins", context=None)
self.assertTrue(len(manager._plugins) > 0)
def test_fire_event(self):
@asyncio.coroutine
def fire_event():
yield from manager.fire_event("test")
2015-08-19 11:54:09 +00:00
yield from asyncio.sleep(1, loop=self.loop)
yield from manager.close()
manager = PluginManager("hbmqtt.test.plugins", context=None, loop=self.loop)
self.loop.run_until_complete(fire_event())
plugin = manager.get_plugin("event_plugin")
self.assertTrue(plugin.object.test_flag)
def test_fire_event_wait(self):
@asyncio.coroutine
def fire_event():
yield from manager.fire_event("test", wait=True)
yield from manager.close()
2015-08-16 19:37:18 +00:00
manager = PluginManager("hbmqtt.test.plugins", context=None, loop=self.loop)
self.loop.run_until_complete(fire_event())
plugin = manager.get_plugin("event_plugin")
self.assertTrue(plugin.object.test_flag)
def test_map_coro(self):
@asyncio.coroutine
def call_coro():
yield from manager.map_plugin_coro('test_coro')
manager = PluginManager("hbmqtt.test.plugins", context=None, loop=self.loop)
self.loop.run_until_complete(call_coro())
plugin = manager.get_plugin("event_plugin")
self.assertTrue(plugin.object.test_coro)
def test_map_coro_return(self):
@asyncio.coroutine
def call_coro():
return (yield from manager.map_plugin_coro('ret_coro'))
manager = PluginManager("hbmqtt.test.plugins", context=None, loop=self.loop)
ret = self.loop.run_until_complete(call_coro())
plugin = manager.get_plugin("event_plugin")
self.assertEqual(ret[plugin], "TEST")