openstack/deb-designate
Code Issues Proposed changes

Fix the use of private symbols from oslo.messaging

Browse Source 
Closes-Bug: #1412812

Change-Id: Ic940fd32005c17a14a523e1535bf5ca32bfe7cd2
This commit is contained in:
Endre Karlson
2015-03-18 18:10:31 +01:00
parent 09169d52ff
commit 0412ffa3a6
4 changed files with 49 additions and 103 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
designate/tests/__init__.py
View File

@@ -262,8 +262,6 @@ class TestCase(base.BaseTestCase):
self.config(notification_driver='test')
self.notifications = self.useFixture(fixtures.NotifierFixture())
self.useFixture(fixtures.RPCFixture(cfg.CONF))
self.config(
@@ -345,13 +343,6 @@ class TestCase(base.BaseTestCase):
# Set the rules
policy.set_rules(rules, default_rule, overwrite)
# Other Utility Methods
def get_notifications(self):
return self.notifications.get()
def reset_notifications(self):
self.notifications.clear()
def start_service(self, svc_name, *args, **kw):
"""
Convenience method for starting a service!

13
designate/tests/fixtures.py
View File

@@ -22,7 +22,6 @@ import fixtures
from oslo_log import log as logging
from oslo_utils import importutils
from oslo.config import cfg
from oslo.messaging.notify import _impl_test as test_notifier
from designate import policy
from designate import network_api
@@ -34,18 +33,6 @@ from designate.sqlalchemy import utils as sqlalchemy_utils
LOG = logging.getLogger(__name__)
class NotifierFixture(fixtures.Fixture):
def setUp(self):
super(NotifierFixture, self).setUp()
self.addCleanup(test_notifier.reset)
def get(self):
return test_notifier.NOTIFICATIONS
def clear(self):
return test_notifier.reset()
class RPCFixture(fixtures.Fixture):
def __init__(self, conf):

19
designate/tests/test_api/test_middleware.py
View File

@@ -14,7 +14,9 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import mock
from oslo.config import cfg
from oslo_messaging.notify import notifier
from designate.tests.test_api import ApiTestCase
from designate import context
@@ -196,7 +198,8 @@ class NormalizeURIMiddlewareTest(ApiTestCase):
class FaultMiddlewareTest(ApiTestCase):
def test_notify_of_fault(self):
@mock.patch.object(notifier.Notifier, "error")
def test_notify_of_fault(self, mock_notifier):
self.config(notify_api_faults=True)
rpc.init(cfg.CONF)
app = middleware.FaultWrapperMiddleware({})
@@ -213,12 +216,8 @@ class FaultMiddlewareTest(ApiTestCase):
# Process the request
app(request)
notifications = self.get_notifications()
self.assertEqual(1, len(notifications))
ctxt, message, priority, retry = notifications.pop()
self.assertEqual('ERROR', message['priority'])
self.assertEqual('dns.api.fault', message['event_type'])
self.assertIn('timestamp', message)
self.assertIn('publisher_id', message)
self.assertEqual(mock_notifier.call_count, 1)
mock_notifier.call_args(
ctxt,
'dns.api.fault',
{"url": None, "status": 409, "exception": ""})

111
designate/tests/test_central/test_service.py
View File

@@ -17,11 +17,12 @@
import copy
import random
import mock
import testtools
from testtools.matchers import GreaterThan
from mock import patch
from oslo_log import log as logging
from oslo_db import exception as db_exception
from oslo_messaging.notify import notifier
from designate import exceptions
from designate import objects
@@ -42,7 +43,7 @@ class CentralServiceTest(CentralTestCase):
list = objects.TldList()
list.append(objects.Tld(name='com.'))
with patch.object(self.central_service.storage, 'find_tlds',
with mock.patch.object(self.central_service.storage, 'find_tlds',
return_value=list):
self.central_service.start()
self.assertTrue(self.central_service.check_for_tlds)
@@ -72,12 +73,12 @@ class CentralServiceTest(CentralTestCase):
list.append(objects.Tld(name='biz'))
list.append(objects.Tld(name='z'))
with patch.object(self.central_service.storage, 'find_tlds',
with mock.patch.object(self.central_service.storage, 'find_tlds',
return_value=list):
self.central_service.start()
context = self.get_context()
with patch.object(self.central_service.storage, 'find_tld',
with mock.patch.object(self.central_service.storage, 'find_tld',
return_value=objects.Tld(name='biz')):
with testtools.ExpectedException(exceptions.InvalidDomainName):
self.central_service._is_valid_domain_name(context, 'biz.')
@@ -405,12 +406,13 @@ class CentralServiceTest(CentralTestCase):
self.central_service.count_tenants(self.get_context())
# Domain Tests
def _test_create_domain(self, values):
@mock.patch.object(notifier.Notifier, "info")
def _test_create_domain(self, values, mock_notifier):
# Create a server
self.create_nameserver()
# Reset the list of notifications
self.reset_notifications()
# Reset the mock to avoid the calls from the create_nameserver() call
mock_notifier.reset_mock()
# Create a domain
domain = self.central_service.create_domain(
@@ -422,23 +424,10 @@ class CentralServiceTest(CentralTestCase):
self.assertEqual(domain['email'], values['email'])
self.assertIn('status', domain)
# Ensure we sent exactly 1 notification
notifications = self.get_notifications()
self.assertEqual(len(notifications), 1)
self.assertEqual(mock_notifier.call_count, 1)
# Ensure the notification wrapper contains the correct info
ctxt, message, priority, retry = notifications[0]
self.assertEqual(message['event_type'], 'dns.domain.create')
self.assertEqual(message['priority'], 'INFO')
self.assertIsNotNone(message['timestamp'])
self.assertIsNotNone(message['message_id'])
# Ensure the notification payload contains the correct info
payload = message['payload']
self.assertEqual(payload['id'], domain['id'])
self.assertEqual(payload['name'], domain['name'])
self.assertEqual(payload['tenant_id'], domain['tenant_id'])
mock_notifier.assert_called_once_with(
self.admin_context, 'dns.domain.create', domain)
def test_create_domain_over_tld(self):
values = dict(
@@ -777,52 +766,44 @@ class CentralServiceTest(CentralTestCase):
self.assertEqual(domain['email'], expected_domain['email'])
self.assertIn('status', domain)
def test_update_domain(self):
@mock.patch.object(notifier.Notifier, "info")
def test_update_domain(self, mock_notifier):
# Create a domain
domain = self.create_domain(email='info@example.org')
original_serial = domain.serial
# Reset the list of notifications
self.reset_notifications()
# Update the object
domain.email = 'info@example.net'
# Reset the mock to avoid the calls from the create_domain() call
mock_notifier.reset_mock()
# Perform the update
self.central_service.update_domain(self.admin_context, domain)
# Fetch the domain again
domain = self.central_service.get_domain(self.admin_context, domain.id)
domain = self.central_service.get_domain(
self.admin_context, domain.id)
# Ensure the domain was updated correctly
self.assertTrue(domain.serial > original_serial)
self.assertEqual('info@example.net', domain.email)
# Ensure we sent exactly 1 notification
notifications = self.get_notifications()
self.assertEqual(len(notifications), 1)
self.assertEqual(mock_notifier.call_count, 1)
# Ensure the notification wrapper contains the correct info
ctxt, message, priority, retry = notifications[0]
# Check that the object used in the notify is a Domain and the id
# matches up
notified_domain = mock_notifier.call_args[0][-1]
self.assertIsInstance(notified_domain, objects.Domain)
self.assertEqual(domain.id, notified_domain.id)
self.assertEqual(message['event_type'], 'dns.domain.update')
self.assertEqual(message['priority'], 'INFO')
self.assertIsNotNone(message['timestamp'])
self.assertIsNotNone(message['message_id'])
# Ensure the notification payload contains the correct info
payload = message['payload']
self.assertEqual(domain.id, payload['id'])
self.assertEqual(domain.name, payload['name'])
self.assertEqual(domain.tenant_id, payload['tenant_id'])
mock_notifier.assert_called_once_with(
self.admin_context, 'dns.domain.update', mock.ANY)
def test_update_domain_without_id(self):
# Create a domain
domain = self.create_domain(email='info@example.org')
# Reset the list of notifications
self.reset_notifications()
# Update the object
domain.email = 'info@example.net'
domain.id = None
@@ -835,9 +816,6 @@ class CentralServiceTest(CentralTestCase):
domain = self.create_domain(email='info@example.org')
original_serial = domain.serial
# Reset the list of notifications
self.reset_notifications()
# Update the object
domain.email = 'info@example.net'
@@ -882,7 +860,7 @@ class CentralServiceTest(CentralTestCase):
i[0] = True
raise db_exception.DBDeadlock()
with patch.object(self.central_service.storage, 'commit',
with mock.patch.object(self.central_service.storage, 'commit',
side_effect=fail_once_then_pass):
# Perform the update
domain = self.central_service.update_domain(
@@ -896,12 +874,12 @@ class CentralServiceTest(CentralTestCase):
self.assertTrue(domain.serial > original_serial)
self.assertEqual('info@example.net', domain.email)
def test_delete_domain(self):
@mock.patch.object(notifier.Notifier, "info")
def test_delete_domain(self, mock_notifier):
# Create a domain
domain = self.create_domain()
# Reset the list of notifications
self.reset_notifications()
mock_notifier.reset_mock()
# Delete the domain
self.central_service.delete_domain(self.admin_context, domain['id'])
@@ -922,22 +900,16 @@ class CentralServiceTest(CentralTestCase):
self.assertEqual(deleted_domain.serial, domain.serial)
self.assertEqual(deleted_domain.pool_id, domain.pool_id)
# Ensure we sent exactly 1 notification
notifications = self.get_notifications()
self.assertEqual(len(notifications), 1)
self.assertEqual(mock_notifier.call_count, 1)
# Ensure the notification wrapper contains the correct info
ctxt, message, priority, retry = notifications.pop()
self.assertEqual(message['event_type'], 'dns.domain.delete')
self.assertEqual(message['priority'], 'INFO')
self.assertIsNotNone(message['timestamp'])
self.assertIsNotNone(message['message_id'])
# Check that the object used in the notify is a Domain and the id
# matches up
notified_domain = mock_notifier.call_args[0][-1]
self.assertIsInstance(notified_domain, objects.Domain)
self.assertEqual(deleted_domain.id, notified_domain.id)
# Ensure the notification payload contains the correct info
payload = message['payload']
self.assertEqual(payload['id'], domain['id'])
self.assertEqual(payload['name'], domain['name'])
self.assertEqual(payload['tenant_id'], domain['tenant_id'])
mock_notifier.assert_called_once_with(
self.admin_context, 'dns.domain.delete', mock.ANY)
def test_delete_parent_domain(self):
# Create the Parent Domain using fixture 0
@@ -1294,7 +1266,7 @@ class CentralServiceTest(CentralTestCase):
i[0] = True
raise db_exception.DBDeadlock()
with patch.object(self.central_service.storage, 'commit',
with mock.patch.object(self.central_service.storage, 'commit',
side_effect=fail_once_then_pass):
# Perform the update
recordset = self.central_service.update_recordset(
@@ -2606,9 +2578,6 @@ class CentralServiceTest(CentralTestCase):
# Create a domain
domain = self.create_domain()
# Reset the list of notifications
self.reset_notifications()
# Delete the domain
self.central_service.delete_domain(self.admin_context, domain['id'])