import mock
from neutron import service
-from neutron.tests import base
+from neutron.tests.unit import test_wsgi
-class TestRpcWorker(base.BaseTestCase):
+class TestRpcWorker(test_wsgi.TestServiceBase):
- @mock.patch("neutron.policy.refresh")
- @mock.patch("neutron.common.config.setup_logging")
- def test_reset(self, setup_logging_mock, refresh_mock):
+ def test_reset(self):
_plugin = mock.Mock()
-
rpc_worker = service.RpcWorker(_plugin)
- rpc_worker.reset()
-
- setup_logging_mock.assert_called_once_with()
- refresh_mock.assert_called_once_with()
+ self._test_reset(rpc_worker)
return opener.open(*args, **kwargs)
-class TestWorkerService(base.BaseTestCase):
+class TestServiceBase(base.BaseTestCase):
+ """Service tests base."""
+
+ @mock.patch("neutron.policy.refresh")
+ @mock.patch("neutron.common.config.setup_logging")
+ def _test_reset(self, worker_service, setup_logging_mock, refresh_mock):
+ worker_service.reset()
+
+ setup_logging_mock.assert_called_once_with()
+ refresh_mock.assert_called_once_with()
+
+
+class TestWorkerService(TestServiceBase):
"""WorkerService tests."""
@mock.patch('neutron.db.api.get_engine')
workerservice.start()
self.assertFalse(apimock.called)
- @mock.patch("neutron.policy.refresh")
- @mock.patch("neutron.common.config.setup_logging")
- def test_reset(self, setup_logging_mock, refresh_mock):
+ def test_reset(self):
_service = mock.Mock()
_app = mock.Mock()
worker_service = wsgi.WorkerService(_service, _app)
- worker_service.reset()
-
- setup_logging_mock.assert_called_once_with()
- refresh_mock.assert_called_once_with()
+ self._test_reset(worker_service)
class TestWSGIServer(base.BaseTestCase):