openstack/deb-designate
Code Issues Proposed changes

Switch to mock fixtures, minor cleanup

Browse Source 
Change-Id: I49349b0c5c244201fd1f2866fa4f76cec642d106
This commit is contained in:
Federico Ceratto 2016-01-27 13:19:45 +00:00
parent 691a5f4c7b
commit 6aa235ee09
5 changed files with 51 additions and 44 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
designate/tests/test_mdns/test_handler.py
View File

@ -439,8 +439,6 @@ class MdnsRequestHandlerTest(MdnsTestCase):
request = dns.message.from_wire(binascii.a2b_hex(payload))
request.environ = {'addr': self.addr, 'context': self.context}
response = next(self.handler(request)).to_wire()
print("response:", dns.message.from_wire(response))
print(''.join("%02x" % ord(i) for i in response))
self.assertEqual(expected_response, binascii.b2a_hex(response))
def test_dispatch_opcode_query_TXT_quoted_strings(self):
@ -472,8 +470,6 @@ class MdnsRequestHandlerTest(MdnsTestCase):
request = dns.message.from_wire(binascii.a2b_hex(payload))
request.environ = {'addr': self.addr, 'context': self.context}
response = next(self.handler(request)).to_wire()
print("response:", dns.message.from_wire(response))
print(''.join("%02x" % ord(i) for i in response))
self.assertEqual(expected_response, binascii.b2a_hex(response))
def test_dispatch_opcode_query_MX(self):

12
designate/tests/unit/test_api/test_middleware.py
View File

@ -14,6 +14,11 @@
# License for the specific language governing permissions and limitations
# under the License.
"""
Test API middleware
"""
import fixtures
import oslotest.base
import mock
@ -46,11 +51,10 @@ class KeystoneContextMiddlewareTest(oslotest.base.BaseTestCase):
# Replace the DesignateContext class..
self.ctxt = mock.Mock()
ctxt_patcher = mock.patch(
self.useFixture(fixtures.MockPatch(
'designate.context.DesignateContext',
return_value=self.ctxt)
self.addCleanup(ctxt_patcher.stop)
ctxt_patcher.start()
return_value=self.ctxt
))
def test_sudo_by_project_id(self):
self.request.headers.update({

15
designate/tests/unit/test_backend/test_designate.py
View File

@ -13,14 +13,18 @@
# 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 uuid
"""
Unit test Backend
"""
import uuid
from designateclient import exceptions
from mock import patch
from mock import NonCallableMagicMock
from mock import Mock
from oslo_log import log as logging
import fixtures
import oslotest.base
import testtools
@ -70,11 +74,10 @@ class DesignateBackendTest(oslotest.base.BaseTestCase):
# Backends blow up when trying to self.admin_context = ... due to
# policy not being initialized
self.admin_context = Mock()
get_context_patcher = patch(
'designate.context.DesignateContext.get_admin_context')
self.addCleanup(get_context_patcher.stop)
get_context = get_context_patcher.start()
get_context.return_value = self.admin_context
self.useFixture(fixtures.MockPatch(
'designate.context.DesignateContext.get_admin_context',
return_value=self.admin_context
))
self.backend = impl_designate.DesignateBackend(self.target)

11
designate/tests/unit/test_central/test_basic.py
View File

@ -20,6 +20,7 @@ from mock import Mock
from mock import patch
from oslo_config import cfg
from oslo_config import fixture as cfg_fixture
from oslo_log import log as logging
from oslotest import base
import fixtures
import mock
@ -29,6 +30,7 @@ from designate import exceptions
from designate.central.service import Service
import designate.central.service
LOG = logging.getLogger(__name__)
# FIXME: create mock, do not use cfg
cfg.CONF.import_opt('storage_driver', 'designate.central',
@ -44,7 +46,8 @@ def unwrap(f):
for _ in range(42):
try:
uf = getattr(f, '__wrapped_function')
print("Unwrapping %s from %s" % (f.func_name, f.__wrapper_name))
LOG.debug("Unwrapping %s from %s" %
(f.func_name, f.__wrapper_name))
f = uf
except AttributeError:
return f
@ -1771,12 +1774,12 @@ class IsSubzoneTestCase(CentralBasic):
super(IsSubzoneTestCase, self).setUp()
def find_zone(ctx, criterion):
print("Calling find_zone on %r" % criterion)
LOG.debug("Calling find_zone on %r" % criterion)
if criterion['name'] == 'example.com.':
print("Returning %r" % criterion['name'])
LOG.debug("Returning %r" % criterion['name'])
return criterion['name']
print("Not found")
LOG.debug("Not found")
raise exceptions.ZoneNotFound
self.service.storage.find_zone = find_zone

53
designate/tests/unit/test_zone_manager/test_tasks.py
View File

@ -13,12 +13,17 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Unit test Zone Manager tasks
"""
import datetime
import uuid
import mock
from oslotest import base as test
from oslo_utils import timeutils
from oslotest import base as test
import fixtures
import mock
import six
import testtools
@ -155,18 +160,17 @@ class PeriodicExistsTest(TaskTest):
# Mock a ctxt...
self.ctxt = mock.Mock()
get_admin_ctxt_patcher = mock.patch.object(context.DesignateContext,
'get_admin_context')
self.addCleanup(get_admin_ctxt_patcher.stop)
get_admin_context = get_admin_ctxt_patcher.start()
get_admin_context.return_value = self.ctxt
self.useFixture(fixtures.MockPatchObject(
context.DesignateContext, 'get_admin_context',
return_value=self.ctxt
))
# Patch get_notifier so that it returns a mock..
get_notifier_patcher = mock.patch.object(rpc, 'get_notifier')
get_notifier = get_notifier_patcher.start()
self.addCleanup(get_notifier_patcher.stop)
self.mock_notifier = mock.Mock()
get_notifier.return_value = self.mock_notifier
self.useFixture(fixtures.MockPatchObject(
rpc, 'get_notifier',
return_value=self.mock_notifier
))
self.task = tasks.PeriodicExistsTask()
self.task.my_partitions = range(0, 10)
@ -178,11 +182,10 @@ class PeriodicExistsTest(TaskTest):
"audit_period_beginning": str(self.period[0]),
"audit_period_ending": str(self.period[1])
}
get_period_patcher = mock.patch.object(
tasks.PeriodicExistsTask, '_get_period')
self.addCleanup(get_period_patcher.stop)
self.get_period = get_period_patcher.start()
self.get_period.return_value = self.period
self.useFixture(fixtures.MockPatchObject(
tasks.PeriodicExistsTask, '_get_period',
return_value=self.period
))
def test_emit_exists(self):
zone = RoObject(
@ -231,19 +234,17 @@ class PeriodicSecondaryRefreshTest(TaskTest):
# Mock a ctxt...
self.ctxt = mock.Mock()
get_admin_ctxt_patcher = mock.patch.object(context.DesignateContext,
'get_admin_context')
self.addCleanup(get_admin_ctxt_patcher.stop)
get_admin_context = get_admin_ctxt_patcher.start()
get_admin_context.return_value = self.ctxt
self.useFixture(fixtures.MockPatchObject(
context.DesignateContext, 'get_admin_context',
return_value=self.ctxt
))
# Mock a central...
self.central = mock.Mock()
get_central_patcher = mock.patch.object(central_api.CentralAPI,
'get_instance')
self.addCleanup(get_central_patcher.stop)
get_central = get_central_patcher.start()
get_central.return_value = self.central
self.useFixture(fixtures.MockPatchObject(
central_api.CentralAPI, 'get_instance',
return_value=self.central
))
self.task = tasks.PeriodicSecondaryRefreshTask()
self.task.my_partitions = 0, 9