Remove base Plugin testcase
Change-Id: I5d01233fac24f837402957440d37724135252b59
This commit is contained in:
parent
f238d2f036
commit
83d135a669
@ -16,17 +16,27 @@
|
|||||||
import json
|
import json
|
||||||
import os
|
import os
|
||||||
from moniker.notification_handler.base import Handler
|
from moniker.notification_handler.base import Handler
|
||||||
from moniker.tests.test_plugins import PluginTestCase
|
from moniker.tests import TestCase
|
||||||
|
|
||||||
FIXTURES_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__),
|
FIXTURES_PATH = os.path.abspath(os.path.join(os.path.dirname(__file__),
|
||||||
'..',
|
'..',
|
||||||
'sample_notifications'))
|
'sample_notifications'))
|
||||||
|
|
||||||
|
|
||||||
class NotificationHandlerTestCase(PluginTestCase):
|
class NotificationHandlerTestCase(TestCase):
|
||||||
__test__ = False
|
__test__ = False
|
||||||
__plugin_base__ = Handler
|
__plugin_base__ = Handler
|
||||||
|
|
||||||
|
def setUp(self):
|
||||||
|
super(NotificationHandlerTestCase, self).setUp()
|
||||||
|
|
||||||
|
self.central_service = self.get_central_service()
|
||||||
|
self.central_service.start()
|
||||||
|
|
||||||
|
def tearDown(self):
|
||||||
|
self.central_service.stop()
|
||||||
|
super(NotificationHandlerTestCase, self).tearDown()
|
||||||
|
|
||||||
def get_notification_fixture(self, service, name):
|
def get_notification_fixture(self, service, name):
|
||||||
filename = os.path.join(FIXTURES_PATH, service, '%s.json' % name)
|
filename = os.path.join(FIXTURES_PATH, service, '%s.json' % name)
|
||||||
|
|
||||||
@ -43,20 +53,3 @@ class NotificationHandlerTestCase(PluginTestCase):
|
|||||||
|
|
||||||
with self.assertRaises(ValueError):
|
with self.assertRaises(ValueError):
|
||||||
self.plugin.process_notification(event_type, 'payload')
|
self.plugin.process_notification(event_type, 'payload')
|
||||||
|
|
||||||
def pre_invoke(self):
|
|
||||||
self.central_service = self.get_central_service()
|
|
||||||
self.invoke_args = [self.central_service]
|
|
||||||
|
|
||||||
|
|
||||||
class AddressHandlerTestCase(NotificationHandlerTestCase):
|
|
||||||
"""
|
|
||||||
Test something that receives notifications with regards to addresses
|
|
||||||
"""
|
|
||||||
def pre_invoke(self):
|
|
||||||
super(AddressHandlerTestCase, self).pre_invoke()
|
|
||||||
|
|
||||||
domain = self.create_domain()
|
|
||||||
self.domain_id = str(domain['id'])
|
|
||||||
|
|
||||||
return {'domain_id': self.domain_id}
|
|
||||||
|
@ -14,14 +14,23 @@
|
|||||||
# License for the specific language governing permissions and limitations
|
# License for the specific language governing permissions and limitations
|
||||||
# under the License.
|
# under the License.
|
||||||
from moniker.openstack.common import log as logging
|
from moniker.openstack.common import log as logging
|
||||||
from moniker.tests.test_notification_handler import AddressHandlerTestCase
|
from moniker.notification_handler.nova import NovaFixedHandler
|
||||||
|
from moniker.tests.test_notification_handler import NotificationHandlerTestCase
|
||||||
|
|
||||||
LOG = logging.getLogger(__name__)
|
LOG = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class NovaTestCase(AddressHandlerTestCase):
|
class NovaFixedHandlerTest(NotificationHandlerTestCase):
|
||||||
__test__ = True
|
__test__ = True
|
||||||
__plugin_name__ = 'nova_fixed'
|
|
||||||
|
def setUp(self):
|
||||||
|
super(NovaFixedHandlerTest, self).setUp()
|
||||||
|
|
||||||
|
domain = self.create_domain()
|
||||||
|
self.domain_id = domain['id']
|
||||||
|
self.config(domain_id=domain['id'], group='handler:nova_fixed')
|
||||||
|
|
||||||
|
self.plugin = NovaFixedHandler(self.central_service)
|
||||||
|
|
||||||
def test_instance_create_end(self):
|
def test_instance_create_end(self):
|
||||||
event_type = 'compute.instance.create.end'
|
event_type = 'compute.instance.create.end'
|
||||||
|
@ -14,14 +14,23 @@
|
|||||||
# License for the specific language governing permissions and limitations
|
# License for the specific language governing permissions and limitations
|
||||||
# under the License.
|
# under the License.
|
||||||
from moniker.openstack.common import log as logging
|
from moniker.openstack.common import log as logging
|
||||||
from moniker.tests.test_notification_handler import AddressHandlerTestCase
|
from moniker.notification_handler.quantum import QuantumFloatingHandler
|
||||||
|
from moniker.tests.test_notification_handler import NotificationHandlerTestCase
|
||||||
|
|
||||||
LOG = logging.getLogger(__name__)
|
LOG = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
class QuantumFloatingTestCase(AddressHandlerTestCase):
|
class QuantumFloatingHandlerTest(NotificationHandlerTestCase):
|
||||||
__test__ = True
|
__test__ = True
|
||||||
__plugin_name__ = 'quantum_floatingip'
|
|
||||||
|
def setUp(self):
|
||||||
|
super(QuantumFloatingHandlerTest, self).setUp()
|
||||||
|
|
||||||
|
domain = self.create_domain()
|
||||||
|
self.domain_id = domain['id']
|
||||||
|
self.config(domain_id=domain['id'], group='handler:quantum_floatingip')
|
||||||
|
|
||||||
|
self.plugin = QuantumFloatingHandler(self.central_service)
|
||||||
|
|
||||||
def test_floatingip_associate(self):
|
def test_floatingip_associate(self):
|
||||||
event_type = 'floatingip.update.end'
|
event_type = 'floatingip.update.end'
|
||||||
|
@ -1,44 +0,0 @@
|
|||||||
from moniker.tests import TestCase
|
|
||||||
|
|
||||||
|
|
||||||
class PluginTestCase(TestCase):
|
|
||||||
__test__ = False
|
|
||||||
|
|
||||||
__plugin_base__ = None
|
|
||||||
__plugin_name__ = None
|
|
||||||
|
|
||||||
def setUp(self):
|
|
||||||
super(PluginTestCase, self).setUp()
|
|
||||||
|
|
||||||
self.invoke_args = []
|
|
||||||
self.invoke_kwds = {}
|
|
||||||
|
|
||||||
# NOTE: In case overrider of _pre_invoke forgets to return {}
|
|
||||||
plugin_opts = self.pre_invoke() or {}
|
|
||||||
|
|
||||||
# NOTE: Load plugin and register it's opts
|
|
||||||
plugin_cls = self.get_plugin()
|
|
||||||
|
|
||||||
self.config(group=plugin_cls.get_canonical_name(), **plugin_opts)
|
|
||||||
|
|
||||||
self.plugin = self.get_plugin(
|
|
||||||
invoke_args=self.invoke_args,
|
|
||||||
invoke_kwds=self.invoke_kwds,
|
|
||||||
invoke_on_load=True)
|
|
||||||
|
|
||||||
def get_plugin(self, **kw):
|
|
||||||
"""
|
|
||||||
Override me
|
|
||||||
"""
|
|
||||||
print self.__plugin_base__.__plugin_ns__, self.__plugin_name__
|
|
||||||
return self.__plugin_base__.get_plugin(self.__plugin_name__, **kw)
|
|
||||||
|
|
||||||
def pre_invoke(self):
|
|
||||||
"""
|
|
||||||
Do something before invoking the actual plugin, returned hash will
|
|
||||||
be passed to self.config()
|
|
||||||
|
|
||||||
Also you can manipulate self.invoke_args and self.invoke_kwds to your
|
|
||||||
likings here
|
|
||||||
"""
|
|
||||||
return {}
|
|
Loading…
x
Reference in New Issue
Block a user