x/quark
Code Issues Proposed changes

Fix with statement deprecation warnings

Browse Source 
Refactor nested statements for multiple context managers to
remove deprecation warnings. Additionally, refactor imports
to use current exception libraries.

JIRA:NCP-2066
Change-Id: Ie3953d109c13d03bf5028d54dc3f53d8d3649315
This commit is contained in:
Kyle Haley 2016-08-31 18:10:12 -07:00
parent 8d49a9ef6b
commit 391fd5bb7c
23 changed files with 641 additions and 748 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
quark/tests/functional/db/test_pagination.py
View File

@ -76,9 +76,8 @@ class QuarkSubnetsPaginationFunctionalTest(BaseFunctionalTest):
class QuarkPortsPaginationFunctionalTest(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_info):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
self.context.is_admin = True
net = network_api.create_network(self.context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}

5
quark/tests/functional/plugin_modules/test_ip_addresses.py
View File

@ -142,9 +142,8 @@ class QuarkTestReserveIPAdminWithPorts(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
self.context.is_admin = True
macrng_api.create_mac_address_range(self.context, mac)

32
quark/tests/functional/plugin_modules/test_ports.py
View File

@ -19,6 +19,7 @@ import netaddr
import contextlib
from neutron.common import exceptions as q_exc
from neutron_lib import exceptions as n_exc
from quark.db import api as db_api
import quark.plugin_modules.mac_address_ranges as macrng_api
@ -53,9 +54,8 @@ class QuarkFindPortsSorted(BaseFunctionalTest):
class QuarkCreatePortSatisfyIpam(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_v4_info, subnet_v6_info=None):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
self.context.is_admin = True
net = network_api.create_network(self.context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
@ -130,16 +130,15 @@ class QuarkCreatePortSatisfyIpam(BaseFunctionalTest):
with self._stubs(network, subnet_v4_info) as (
net, sub_v4, sub_v6):
ip = "192.168.1.50"
with self.assertRaises(q_exc.IpAddressGenerationFailure):
with self.assertRaises(n_exc.IpAddressGenerationFailure):
port_api.create_port(self.context, _make_body_only_v4(ip))
class QuarkCreatePortWithIpNotMandatory(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_v4_infos, subnet_v6_info=None):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
self.context.is_admin = True
net = network_api.create_network(self.context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
@ -238,9 +237,8 @@ class QuarkCreatePortWithIpNotMandatory(BaseFunctionalTest):
class QuarkCreatePortWithForbiddenMacRange(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_v4_infos, subnet_v6_info=None):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
self.context.is_admin = True
net = network_api.create_network(self.context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC", do_not_use=True)}
@ -353,9 +351,8 @@ class QuarkPortFixedIPOperations(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_info):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
self.context.is_admin = True
macrng_api.create_mac_address_range(self.context, mac)
@ -516,9 +513,8 @@ class QuarkAdvancedServiceCreatePort(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_info):
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
self.context.is_admin = True
macrng_api.create_mac_address_range(self.context, mac)
@ -549,7 +545,7 @@ class QuarkAdvancedServiceCreatePort(BaseFunctionalTest):
'tenant_id': 'someoneelse'}}
self.context.is_admin = True
self.context.is_advsvc = False
self.assertRaises(q_exc.NotAuthorized,
self.assertRaises(n_exc.NotAuthorized,
port_api.create_port, self.context, port_info)
def test_cant_create_port_without_admin(self):
@ -562,5 +558,5 @@ class QuarkAdvancedServiceCreatePort(BaseFunctionalTest):
# check, quark will first attempt to retrieve the network but
# since networks are scoped by tenant when it is not an admin,
# it will not be found
self.assertRaises(q_exc.NetworkNotFound,
self.assertRaises(n_exc.NetworkNotFound,
port_api.create_port, self.context, port_info)

2
quark/tests/functional/plugin_modules/test_routes.py
View File

@ -35,7 +35,7 @@ class QuarkCreateRoutes(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network, subnet, ip_policy):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(mock.patch("neutron.common.rpc.get_notifier")):
with mock.patch("neutron.common.rpc.get_notifier"):
net = network_api.create_network(self.context, network)
subnet['subnet']['network_id'] = net['id']
sub1 = subnet_api.create_subnet(self.context, subnet)

10
quark/tests/functional/plugin_modules/test_security_groups.py
View File

@ -77,12 +77,10 @@ class SecurityGroupsAsyncUpdateTests(BaseFunctionalTest):
cls = 'QuarkSGAsyncProcessClient'
mod_path = 'quark.worker_plugins.sg_update_worker.%s' % cls
job_path = 'quark.plugin_modules.jobs'
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check"),
mock.patch("%s.add_job_to_context" % job_path),
mock.patch("%s.start_update" % mod_path)) as \
(notifier, limit_check, addjob, update):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"), \
mock.patch("%s.add_job_to_context" % job_path), \
mock.patch("%s.start_update" % mod_path) as update:
self.context.is_admin = True
net = network_api.create_network(self.context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}

13
quark/tests/functional/plugin_modules/test_shared_ips.py
View File

@ -73,9 +73,8 @@ class QuarkSharedIPs(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_info, ports_info):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
net = network_api.create_network(self.admin_context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
macrng_api.create_mac_address_range(self.admin_context, mac)
@ -481,8 +480,7 @@ class QuarkSharedIPsQuotaCheck(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_info, ports_info):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier")):
with mock.patch("neutron.common.rpc.get_notifier"):
self.context.is_admin = True
net = network_api.create_network(self.context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
@ -667,9 +665,8 @@ class QuarkTestSharedIpAddressesQuota(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network_info, subnet_info, ports_info):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS.limit_check")):
with mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("neutron.quota.QUOTAS.limit_check"):
net = network_api.create_network(self.admin_context, network_info)
mac = {'mac_address_range': dict(cidr="AA:BB:CC")}
macrng_api.create_mac_address_range(self.admin_context, mac)

6
quark/tests/functional/plugin_modules/test_subnets.py
View File

@ -62,7 +62,7 @@ class QuarkGetSubnetsFromPlugin(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network, subnet):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(mock.patch("neutron.common.rpc.get_notifier")):
with mock.patch("neutron.common.rpc.get_notifier"):
net = network_api.create_network(self.context, network)
subnet['subnet']['network_id'] = net['id']
sub1 = subnet_api.create_subnet(self.context, subnet)
@ -93,7 +93,7 @@ class QuarkCreateSubnets(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network, subnet):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(mock.patch("neutron.common.rpc.get_notifier")):
with mock.patch("neutron.common.rpc.get_notifier"):
net = network_api.create_network(self.context, network)
subnet['subnet']['network_id'] = net['id']
sub1 = subnet_api.create_subnet(self.context, subnet)
@ -227,7 +227,7 @@ class QuarkUpdateSubnets(BaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network, subnet):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(mock.patch("neutron.common.rpc.get_notifier")):
with mock.patch("neutron.common.rpc.get_notifier"):
net = network_api.create_network(self.context, network)
subnet['subnet']['network_id'] = net['id']
sub1 = subnet_api.create_subnet(self.context, subnet)

2
quark/tests/functional/test_ipam.py
View File

@ -82,7 +82,7 @@ class TestQuarkIpamAllocateFromV6Subnet(QuarkIpamBaseFunctionalTest):
@contextlib.contextmanager
def _stubs(self, network, subnet, ip_policy):
self.ipam = quark.ipam.QuarkIpamANY()
with contextlib.nested(mock.patch("neutron.common.rpc.get_notifier")):
with mock.patch("neutron.common.rpc.get_notifier"):
net = network_api.create_network(self.context, network)
subnet['subnet']['network_id'] = net['id']
sub = subnet_api.create_subnet(self.context, subnet)

85
quark/tests/plugin_modules/test_floating_ips.py
View File

@ -18,7 +18,7 @@ import contextlib
import datetime
import mock
import netaddr
from neutron.common import exceptions as ex
from neutron_lib import exceptions as n_exc
from quark.billing import IP_ASSOC
from quark.billing import IP_DISASSOC
@ -36,18 +36,15 @@ class TestRemoveFloatingIPs(test_quark_plugin.TestQuarkPlugin):
flip_model = models.IPAddress()
flip_model.update(flip)
with contextlib.nested(
mock.patch("quark.db.api.floating_ip_find"),
mock.patch("quark.db.api.floating_ip_disassociate_fixed_ip"),
mock.patch("quark.db.api.port_disassociate_ip"),
mock.patch("quark.db.api.ip_address_deallocate"),
mock.patch("quark.ipam.QuarkIpam.deallocate_ip_address"),
with mock.patch("quark.db.api.floating_ip_find") as flip_find, \
mock.patch("quark.db.api.floating_ip_disassociate_fixed_ip"), \
mock.patch("quark.db.api.port_disassociate_ip"), \
mock.patch("quark.db.api.ip_address_deallocate"), \
mock.patch("quark.ipam.QuarkIpam.deallocate_ip_address"), \
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".remove_floating_ip"),
mock.patch("quark.billing.notify"),
mock.patch("quark.billing.build_payload")
) as (flip_find, db_fixed_ip_disassoc, db_port_disassoc, db_dealloc,
mock_dealloc, mock_remove_flip, notify, build_payload):
".remove_floating_ip"), \
mock.patch("quark.billing.notify"), \
mock.patch("quark.billing.build_payload") as build_payload:
flip_find.return_value = flip_model
build_payload.return_value = {'respek': '4reelz'}
yield
@ -230,19 +227,18 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
addr.fixed_ips.append(fixed_ip)
return addr
with contextlib.nested(
mock.patch("quark.db.api.floating_ip_find"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.port_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
with mock.patch("quark.db.api.floating_ip_find") as flip_find, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".register_floating_ip"),
mock.patch("quark.db.api.port_associate_ip"),
mock.patch("quark.db.api.floating_ip_associate_fixed_ip"),
mock.patch("quark.billing.notify"),
mock.patch("quark.billing.build_payload")
) as (flip_find, net_find, port_find, alloc_ip, mock_reg_flip,
port_assoc, fixed_ip_assoc, notify, build_payload):
".register_floating_ip"), \
mock.patch("quark.db.api.port_associate_ip") as port_assoc, \
mock.patch("quark.db.api.floating_ip_associate_fixed_ip") as \
fixed_ip_assoc, \
mock.patch("quark.billing.notify"), \
mock.patch("quark.billing.build_payload") as build_payload:
flip_find.return_value = flip_model
net_find.return_value = net_model
port_find.return_value = port_model
@ -348,14 +344,14 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
def test_create_without_network_id_fails(self):
with self._stubs():
with self.assertRaises(ex.BadRequest):
with self.assertRaises(n_exc.BadRequest):
request = dict(port_id=2, floating_ip_address="10.0.0.1")
self.plugin.create_floatingip(self.context,
dict(floatingip=request))
def test_create_with_invalid_network_fails(self):
with self._stubs():
with self.assertRaises(ex.NetworkNotFound):
with self.assertRaises(n_exc.NetworkNotFound):
request = dict(floating_network_id=123,
port_id=2, floating_ip_address="10.0.0.1")
self.plugin.create_floatingip(self.context,
@ -366,7 +362,7 @@ class TestCreateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
ipam_strategy="ANY")
with self._stubs(network=network):
with self.assertRaises(ex.PortNotFound):
with self.assertRaises(n_exc.PortNotFound):
request = dict(floating_network_id=network["id"],
port_id=2, floating_ip_address="10.0.0.1")
self.plugin.create_floatingip(self.context,
@ -565,25 +561,24 @@ class TestUpdateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
"""We don't want to notify from tests"""
pass
with contextlib.nested(
mock.patch("quark.db.api.floating_ip_find"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.port_find"),
with mock.patch("quark.db.api.floating_ip_find") as flip_find, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".register_floating_ip"),
".register_floating_ip"), \
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".update_floating_ip"),
".update_floating_ip"), \
mock.patch("quark.drivers.unicorn_driver.UnicornDriver"
".remove_floating_ip"),
mock.patch("quark.db.api.port_associate_ip"),
mock.patch("quark.db.api.port_disassociate_ip"),
mock.patch("quark.db.api.floating_ip_associate_fixed_ip"),
mock.patch("quark.db.api.floating_ip_disassociate_fixed_ip"),
mock.patch("quark.billing.notify")
) as (flip_find, net_find, subnet_find, port_find, reg_flip,
update_flip, rem_flip, port_assoc, port_dessoc, flip_assoc,
flip_dessoc, notify):
".remove_floating_ip"), \
mock.patch("quark.db.api.port_associate_ip") as port_assoc, \
mock.patch("quark.db.api.port_disassociate_ip") as \
port_dessoc, \
mock.patch("quark.db.api.floating_ip_associate_fixed_ip") as \
flip_assoc, \
mock.patch("quark.db.api.floating_ip_disassociate_fixed_ip") \
as flip_dessoc, \
mock.patch("quark.billing.notify") as notify:
flip_find.return_value = flip_model
net_find.return_value = net_model
subnet_find.return_value = subnet_model
@ -698,7 +693,7 @@ class TestUpdateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
address_readable=str(addr))
with self._stubs(flip=flip):
with self.assertRaises(ex.PortNotFound):
with self.assertRaises(n_exc.PortNotFound):
content = dict(port_id="123")
self.plugin.update_floatingip(self.context, flip["id"],
dict(floatingip=content))
@ -763,7 +758,7 @@ class TestUpdateFloatingIPs(test_quark_plugin.TestQuarkPlugin):
def test_update_with_missing_port_id_param_should_fail(self):
with self._stubs():
with self.assertRaises(ex.BadRequest):
with self.assertRaises(n_exc.BadRequest):
content = {}
self.plugin.update_floatingip(self.context, "123",
dict(floatingip=content))

38
quark/tests/plugin_modules/test_ip_addresses.py
View File

@ -69,18 +69,14 @@ class TestIpAddresses(test_quark_plugin.TestQuarkPlugin):
def _alloc_ip(context, new_addr, net_id, port_m, *args, **kwargs):
new_addr.extend([addr_model])
with contextlib.nested(
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.port_find"),
mock.patch(
"quark.plugin_modules.ip_addresses.ipam_driver"),
mock.patch(
"quark.plugin_modules.ip_addresses.db_api"
".port_associate_ip"),
mock.patch(
"quark.plugin_modules.ip_addresses"
".validate_and_fetch_segment")
) as (net_f, port_find, mock_ipam, mock_port_associate_ip, validate):
with mock.patch("quark.db.api.network_find"), \
mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.plugin_modules.ip_addresses.ipam_driver") \
as mock_ipam, \
mock.patch("quark.plugin_modules.ip_addresses.db_api"
".port_associate_ip") as mock_port_associate_ip, \
mock.patch("quark.plugin_modules.ip_addresses"
".validate_and_fetch_segment"):
port_find.return_value = port_model
mock_ipam.allocate_ip_address.side_effect = _alloc_ip
mock_port_associate_ip.side_effect = _port_associate_stub
@ -433,16 +429,16 @@ class TestQuarkUpdateIPAddress(test_quark_plugin.TestQuarkPlugin):
addr_model.ports = port_models
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.port_find" % db_mod),
mock.patch("%s.ip_address_find" % db_mod),
mock.patch("%s.port_associate_ip" % db_mod),
mock.patch("%s.port_disassociate_ip" % db_mod),
with mock.patch("%s.port_find" % db_mod) as port_find, \
mock.patch("%s.ip_address_find" % db_mod) as ip_find, \
mock.patch("%s.port_associate_ip" % db_mod) as \
port_associate_ip, \
mock.patch("%s.port_disassociate_ip" % db_mod) as \
port_disassociate_ip, \
mock.patch("quark.plugin_modules.ip_addresses"
".validate_and_fetch_segment"),
mock.patch("quark.plugin_modules.ip_addresses.ipam_driver")
) as (port_find, ip_find, port_associate_ip,
port_disassociate_ip, val, mock_ipam):
".validate_and_fetch_segment"), \
mock.patch("quark.plugin_modules.ip_addresses.ipam_driver") \
as mock_ipam:
port_find.return_value = port_models
ip_find.return_value = addr_model
port_associate_ip.side_effect = _port_associate_stub

38
quark/tests/plugin_modules/test_ip_policies.py
View File

@ -81,12 +81,11 @@ class TestQuarkCreateIpPolicies(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, ip_policy, subnets=None, nets=None):
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.subnet_find" % db_mod),
mock.patch("%s.network_find" % db_mod),
mock.patch("%s.ip_policy_create" % db_mod),
mock.patch("%s.route_find" % db_mod)
) as (subnet_find, net_find, ip_policy_create, route_find):
with mock.patch("%s.subnet_find" % db_mod) as subnet_find, \
mock.patch("%s.network_find" % db_mod) as net_find, \
mock.patch("%s.ip_policy_create" % db_mod) as \
ip_policy_create, \
mock.patch("%s.route_find" % db_mod) as route_find:
subnet_find.return_value = subnets if subnets else None
net_find.return_value = nets if nets else None
ip_policy_create.return_value = ip_policy
@ -297,12 +296,11 @@ class TestQuarkUpdateIpPolicies(test_quark_plugin.TestQuarkPlugin):
if not networks:
networks = []
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.ip_policy_find" % db_mod),
mock.patch("%s.subnet_find" % db_mod),
mock.patch("%s.network_find" % db_mod),
mock.patch("%s.ip_policy_update" % db_mod),
) as (ip_policy_find, subnet_find, network_find, ip_policy_update):
with mock.patch("%s.ip_policy_find" % db_mod) as ip_policy_find, \
mock.patch("%s.subnet_find" % db_mod) as subnet_find, \
mock.patch("%s.network_find" % db_mod) as network_find, \
mock.patch("%s.ip_policy_update" % db_mod) as \
ip_policy_update:
ip_policy_find.return_value = ip_policy
subnet_find.return_value = subnets
network_find.return_value = networks
@ -440,10 +438,8 @@ class TestQuarkDeleteIpPolicies(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, ip_policy):
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.ip_policy_find" % db_mod),
mock.patch("%s.ip_policy_delete" % db_mod),
) as (ip_policy_find, ip_policy_delete):
with mock.patch("%s.ip_policy_find" % db_mod) as ip_policy_find, \
mock.patch("%s.ip_policy_delete" % db_mod) as ip_policy_delete:
ip_policy_find.return_value = ip_policy
yield ip_policy_find, ip_policy_delete
@ -473,12 +469,10 @@ class TestQuarkUpdatePolicySubnetWithRoutes(test_quark_plugin.TestQuarkPlugin):
def _stubs(self, ip_policy, subnets=None, routes=None):
subnets = subnets or []
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.ip_policy_find" % db_mod),
mock.patch("%s.subnet_find" % db_mod),
mock.patch("%s.route_find" % db_mod),
mock.patch("%s.ip_policy_update" % db_mod),
) as (ip_policy_find, subnet_find, route_find, ip_policy_update):
with mock.patch("%s.ip_policy_find" % db_mod) as ip_policy_find, \
mock.patch("%s.subnet_find" % db_mod) as subnet_find, \
mock.patch("%s.route_find" % db_mod) as route_find, \
mock.patch("%s.ip_policy_update" % db_mod) as ip_policy_update:
ip_policy_find.return_value = ip_policy
subnet_find.return_value = subnets
route_find.return_value = routes

7
quark/tests/plugin_modules/test_mac_address_ranges.py
View File

@ -144,10 +144,9 @@ class TestQuarkDeleteMacAddressRanges(test_quark_plugin.TestQuarkPlugin):
db_mod = "quark.db.api"
old_context = self.context
self.context = self.context.elevated()
with contextlib.nested(
mock.patch("%s.mac_address_range_find" % db_mod),
mock.patch("%s.mac_address_range_delete" % db_mod),
) as (mar_find, mar_delete):
with mock.patch("%s.mac_address_range_find" % db_mod) as mar_find, \
mock.patch("%s.mac_address_range_delete" % db_mod) as \
mar_delete:
mar_find.return_value = mac_range
yield mar_delete
self.context = old_context

30
quark/tests/plugin_modules/test_networks.py
View File

@ -199,10 +199,8 @@ class TestQuarkUpdateNetwork(test_quark_plugin.TestQuarkPlugin):
net_mod = net.copy()
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.network_find" % db_mod),
mock.patch("%s.network_update" % db_mod)
) as (net_find, net_update):
with mock.patch("%s.network_find" % db_mod) as net_find, \
mock.patch("%s.network_update" % db_mod) as net_update:
net_find.return_value = net_mod
net_update.return_value = net_mod
yield net_update
@ -266,14 +264,12 @@ class TestQuarkDeleteNetwork(test_quark_plugin.TestQuarkPlugin):
db_mod = "quark.db.api"
strategy_prefix = "quark.network_strategy.JSONStrategy"
with contextlib.nested(
mock.patch("%s.network_find" % db_mod),
mock.patch("%s.network_delete" % db_mod),
mock.patch("quark.drivers.base.BaseDriver.delete_network"),
mock.patch("%s.subnet_delete" % db_mod),
mock.patch("%s.is_provider_network" % strategy_prefix)
) as (net_find, net_delete, driver_net_delete, subnet_del,
is_provider_network):
with mock.patch("%s.network_find" % db_mod) as net_find, \
mock.patch("%s.network_delete" % db_mod) as net_delete, \
mock.patch("quark.drivers.base.BaseDriver.delete_network"), \
mock.patch("%s.subnet_delete" % db_mod), \
mock.patch("%s.is_provider_network" % strategy_prefix) as \
is_provider_network:
net_find.return_value = net_mod
is_provider_network.return_value = True
yield net_delete
@ -341,12 +337,10 @@ class TestQuarkCreateNetwork(test_quark_plugin.TestQuarkPlugin):
found_net = models.Network()
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.network_create" % db_mod),
mock.patch("%s.subnet_create" % db_mod),
mock.patch("quark.drivers.base.BaseDriver.create_network"),
mock.patch("%s.network_find" % db_mod)
) as (net_create, sub_create, driver_net_create, net_find):
with mock.patch("%s.network_create" % db_mod) as net_create, \
mock.patch("%s.subnet_create" % db_mod) as sub_create, \
mock.patch("quark.drivers.base.BaseDriver.create_network"), \
mock.patch("%s.network_find" % db_mod) as net_find:
net_create.return_value = net_mod
sub_create.return_value = subnet_mod
net_find.return_value = found_net

230
quark/tests/plugin_modules/test_ports.py
View File

@ -57,9 +57,7 @@ class TestQuarkGetPorts(test_quark_plugin.TestQuarkPlugin):
port_model.ip_addresses = addr_models
port_models = port_model
with contextlib.nested(
mock.patch("quark.db.api.port_find")
) as (port_find,):
with mock.patch("quark.db.api.port_find") as port_find:
port_find.return_value = port_models
yield
@ -237,9 +235,7 @@ class TestQuarkGetPortsProviderSubnetIds(test_quark_plugin.TestQuarkPlugin):
port_model, addr_model)
port_models = port_model
with contextlib.nested(
mock.patch("quark.db.api.port_find")
) as (port_find,):
with mock.patch("quark.db.api.port_find") as port_find:
port_find.return_value = port_models
yield
@ -308,9 +304,8 @@ class TestQuarkGetPortsByIPAddress(test_quark_plugin.TestQuarkPlugin):
ip_mod.ports = [port_model]
addr_models.append(ip_mod)
with contextlib.nested(
mock.patch("quark.db.api.port_find_by_ip_address")
) as (port_find_by_addr,):
with mock.patch("quark.db.api.port_find_by_ip_address") as \
port_find_by_addr:
port_find_by_addr.return_value = addr_models
yield
@ -354,15 +349,14 @@ class TestQuarkCreatePortFailure(test_quark_plugin.TestQuarkPlugin):
port_model.update(port)
port_models = port_model
with contextlib.nested(
mock.patch("quark.db.api.port_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.port_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address"),
mock.patch("quark.db.api.port_count_all"),
) as (port_create, net_find, port_find, alloc_ip, alloc_mac,
port_count):
with mock.patch("quark.db.api.port_create") as port_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address") as \
alloc_mac, \
mock.patch("quark.db.api.port_count_all") as port_count:
port_create.return_value = port_models
net_find.return_value = network
port_find.return_value = models.Port()
@ -412,15 +406,12 @@ class TestQuarkCreatePortRM9305(test_quark_plugin.TestQuarkPlugin):
port_models = port_model
db_mod = "quark.db.api"
ipam = "quark.ipam.QuarkIpam"
with contextlib.nested(
mock.patch("%s.port_create" % db_mod),
mock.patch("%s.network_find" % db_mod),
mock.patch("%s.port_find" % db_mod),
mock.patch("%s.allocate_ip_address" % ipam),
mock.patch("%s.allocate_mac_address" % ipam),
mock.patch("%s.port_count_all" % db_mod),
) as (port_create, net_find, port_find, alloc_ip, alloc_mac,
port_count):
with mock.patch("%s.port_create" % db_mod) as port_create, \
mock.patch("%s.network_find" % db_mod) as net_find, \
mock.patch("%s.port_find" % db_mod) as port_find, \
mock.patch("%s.allocate_ip_address" % ipam) as alloc_ip, \
mock.patch("%s.allocate_mac_address" % ipam) as alloc_mac, \
mock.patch("%s.port_count_all" % db_mod) as port_count:
port_create.return_value = port_models
net_find.return_value = network
port_find.return_value = None
@ -515,16 +506,16 @@ class TestQuarkCreatePortsSameDevBadRequest(test_quark_plugin.TestQuarkPlugin):
new_ips.extend([ip_mod])
return mock.DEFAULT
with contextlib.nested(
mock.patch("quark.db.api.port_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address"),
mock.patch("quark.db.api.port_count_all"),
mock.patch("neutron.quota.QuotaEngine.limit_check"),
mock.patch("quark.db.api.subnet_find"),
) as (port_create, net_find, alloc_ip, alloc_mac, port_count,
limit_check, subnet_find):
with mock.patch("quark.db.api.port_create") as port_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address") as \
alloc_mac, \
mock.patch("quark.db.api.port_count_all") as port_count, \
mock.patch("neutron.quota.QuotaEngine.limit_check") as \
limit_check, \
mock.patch("quark.db.api.subnet_find") as subnet_find:
port_create.side_effect = _create_db_port
net_find.return_value = network
alloc_ip.side_effect = _alloc_ip
@ -741,15 +732,15 @@ class TestQuarkPortCreateQuota(test_quark_plugin.TestQuarkPlugin):
port_model.update(port)
port_models = port_model
with contextlib.nested(
mock.patch("quark.db.api.port_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address"),
mock.patch("quark.db.api.port_count_all"),
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (port_create, net_find, alloc_ip, alloc_mac, port_count,
limit_check):
with mock.patch("quark.db.api.port_create") as port_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address") as \
alloc_mac, \
mock.patch("quark.db.api.port_count_all") as port_count, \
mock.patch("neutron.quota.QuotaEngine.limit_check") as \
limit_check:
port_create.return_value = port_models
net_find.return_value = network
alloc_ip.return_value = addr
@ -804,12 +795,12 @@ class TestQuarkUpdatePort(test_quark_plugin.TestQuarkPlugin):
port_model.network = net_model
port_model.update(port)
with contextlib.nested(
mock.patch("quark.db.api.port_find"),
mock.patch("quark.db.api.port_update"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port"),
) as (port_find, port_update, alloc_ip, dealloc_ip):
with mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.db.api.port_update") as port_update, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port") as \
dealloc_ip:
port_find.return_value = port_model
port_update.return_value = port_model
if new_ips:
@ -931,18 +922,18 @@ class TestQuarkUpdatePortSecurityGroups(test_quark_plugin.TestQuarkPlugin):
port_model.update(port)
port_model["security_groups"].append(sg_mod)
with contextlib.nested(
mock.patch("quark.db.api.port_find"),
mock.patch("quark.db.api.port_update"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port"),
mock.patch("neutron.quota.QuotaEngine.limit_check"),
with mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.db.api.port_update") as port_update, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port") as \
dealloc_ip, \
mock.patch("neutron.quota.QuotaEngine.limit_check"), \
mock.patch("quark.plugin_modules.ports.STRATEGY"
".is_provider_network"),
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.drivers.base.BaseDriver.update_port")
) as (port_find, port_update, alloc_ip, dealloc_ip, limit_check,
net_strat, sg_find, driver_port_update):
".is_provider_network") as net_strat, \
mock.patch("quark.db.api.security_group_find") as sg_find, \
mock.patch("quark.drivers.base.BaseDriver.update_port") as \
driver_port_update:
port_find.return_value = port_model
def _port_update(context, port_db, **kwargs):
@ -1033,12 +1024,11 @@ class TestQuarkUpdatePortSetsIps(test_quark_plugin.TestQuarkPlugin):
port_model = models.Port()
port_model['network'] = net_model
port_model.update(port)
with contextlib.nested(
mock.patch("quark.db.api.port_find"),
mock.patch("quark.db.api.port_update"),
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port"),
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (port_find, port_update, dealloc_ip, limit_check):
with mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.db.api.port_update") as port_update, \
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port") as \
dealloc_ip, \
mock.patch("neutron.quota.QuotaEngine.limit_check"):
port_find.return_value = port_model
port_update.return_value = port_model
alloc_ip = mock.patch("quark.ipam.QuarkIpam.allocate_ip_address",
@ -1080,13 +1070,13 @@ class TestQuarkCreatePortOnSharedNetworks(test_quark_plugin.TestQuarkPlugin):
port_model.update(port)
port_models = port_model
with contextlib.nested(
mock.patch("quark.db.api.port_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address"),
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (port_create, net_find, alloc_ip, alloc_mac, limit_check):
with mock.patch("quark.db.api.port_create") as port_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address") as \
alloc_mac, \
mock.patch("neutron.quota.QuotaEngine.limit_check"):
port_create.return_value = port_models
net_find.return_value = network
alloc_ip.return_value = addr
@ -1145,14 +1135,14 @@ class TestQuarkDeletePort(test_quark_plugin.TestQuarkPlugin):
port_model.network = net_model
port_models = port_model
with contextlib.nested(
mock.patch("quark.db.api.port_find"),
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port"),
mock.patch("quark.ipam.QuarkIpam.deallocate_mac_address"),
mock.patch("quark.db.api.port_delete"),
mock.patch("quark.drivers.base.BaseDriver.delete_port")
) as (port_find, dealloc_ip, dealloc_mac, db_port_del,
driver_port_del):
with mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.ipam.QuarkIpam.deallocate_ips_by_port") as \
dealloc_ip, \
mock.patch("quark.ipam.QuarkIpam.deallocate_mac_address") as \
dealloc_mac, \
mock.patch("quark.db.api.port_delete") as db_port_del, \
mock.patch("quark.drivers.base.BaseDriver.delete_port") as \
driver_port_del:
port_find.return_value = port_models
dealloc_ip.return_value = addr
dealloc_mac.return_value = mac
@ -1318,25 +1308,24 @@ class TestPortDriverSelection(test_quark_plugin.TestQuarkPlugin):
bar_ipam.allocate_mac_address.return_value = mac
ipam = {"FOO": foo_ipam, "BAR": bar_ipam}
with contextlib.nested(
mock.patch("quark.db.api.port_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("oslo_utils.uuidutils.generate_uuid"),
mock.patch("quark.plugin_views._make_port_dict"),
mock.patch("quark.db.api.port_count_all"),
mock.patch("neutron.quota.QuotaEngine.limit_check"),
with mock.patch("quark.db.api.port_create") as port_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("oslo_utils.uuidutils.generate_uuid") as gen_uuid, \
mock.patch("quark.plugin_views._make_port_dict"), \
mock.patch("quark.db.api.port_count_all") as port_count, \
mock.patch("neutron.quota.QuotaEngine.limit_check"), \
mock.patch("quark.plugin_modules.ports.registry."
"DRIVER_REGISTRY.drivers",
new_callable=mock.PropertyMock(return_value=drivers)),
new_callable=mock.PropertyMock(
return_value=drivers)), \
mock.patch("quark.plugin_modules.ports.registry."
"DRIVER_REGISTRY.port_driver_compat_map",
new_callable=mock.PropertyMock(
return_value=compat_map)),
return_value=compat_map)), \
mock.patch("quark.plugin_modules.ports.ipam."
"IPAM_REGISTRY.strategies",
new_callable=mock.PropertyMock(return_value=ipam))
) as (port_create, net_find, gen_uuid, make_port,
port_count, limit_check, _, _, _):
new_callable=mock.PropertyMock(
return_value=ipam)):
net_find.return_value = network
gen_uuid.return_value = 1
port_count.return_value = 0
@ -1770,17 +1759,16 @@ class TestQuarkPortCreateFiltering(test_quark_plugin.TestQuarkPlugin):
network["network_plugin"] = "BASE"
network["ipam_strategy"] = "ANY"
with contextlib.nested(
mock.patch("quark.db.api.port_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address"),
mock.patch("oslo_utils.uuidutils.generate_uuid"),
mock.patch("quark.plugin_views._make_port_dict"),
mock.patch("quark.db.api.port_count_all"),
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (port_create, net_find, alloc_ip, alloc_mac, gen_uuid, make_port,
port_count, limit_check):
with mock.patch("quark.db.api.port_create") as port_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address") as \
alloc_ip, \
mock.patch("quark.ipam.QuarkIpam.allocate_mac_address") as \
alloc_mac, \
mock.patch("oslo_utils.uuidutils.generate_uuid") as gen_uuid, \
mock.patch("quark.plugin_views._make_port_dict"), \
mock.patch("quark.db.api.port_count_all") as port_count, \
mock.patch("neutron.quota.QuotaEngine.limit_check"):
net_find.return_value = network
alloc_ip.return_value = addr
alloc_mac.return_value = mac
@ -1870,13 +1858,12 @@ class TestQuarkPortCreateFiltering(test_quark_plugin.TestQuarkPlugin):
class TestQuarkPortUpdateFiltering(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("quark.db.api.port_find"),
mock.patch("quark.db.api.port_update"),
mock.patch("quark.drivers.registry.DriverRegistry.get_driver"),
mock.patch("quark.plugin_views._make_port_dict"),
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (port_find, port_update, get_driver, make_port, limit_check):
with mock.patch("quark.db.api.port_find") as port_find, \
mock.patch("quark.db.api.port_update") as port_update, \
mock.patch("quark.drivers.registry."
"DriverRegistry.get_driver"), \
mock.patch("quark.plugin_views._make_port_dict"), \
mock.patch("neutron.quota.QuotaEngine.limit_check"):
yield port_find, port_update
def test_update_port_attribute_filtering(self):
@ -1930,15 +1917,12 @@ class TestQuarkPortCreateAsAdvancedService(test_quark_plugin.TestQuarkPlugin):
port_models = port_model
db_mod = "quark.db.api"
ipam = "quark.ipam.QuarkIpam"
with contextlib.nested(
mock.patch("%s.port_create" % db_mod),
mock.patch("%s.network_find" % db_mod),
mock.patch("%s.port_find" % db_mod),
mock.patch("%s.allocate_ip_address" % ipam),
mock.patch("%s.allocate_mac_address" % ipam),
mock.patch("%s.port_count_all" % db_mod),
) as (port_create, net_find, port_find, alloc_ip, alloc_mac,
port_count):
with mock.patch("%s.port_create" % db_mod) as port_create, \
mock.patch("%s.network_find" % db_mod) as net_find, \
mock.patch("%s.port_find" % db_mod) as port_find, \
mock.patch("%s.allocate_ip_address" % ipam) as alloc_ip, \
mock.patch("%s.allocate_mac_address" % ipam) as alloc_mac, \
mock.patch("%s.port_count_all" % db_mod) as port_count:
port_create.return_value = port_models
net_find.return_value = network
port_find.return_value = None

17
quark/tests/plugin_modules/test_routes.py
View File

@ -57,17 +57,16 @@ class TestQuarkCreateRoutes(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, create_route, find_routes, subnet):
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.route_create" % db_mod),
mock.patch("%s.route_find" % db_mod),
mock.patch("%s.subnet_find" % db_mod),
# QuotaEngine.limit_check must be mocked:
# This module can't run independently otherwise, as the Quota
# model isn't defined in this test unit, and no clean way to
# use a model such that it would be defined. In other words,
# running the other tests still has side-effects, which should
# be investigated and cleaned up later.
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (route_create, route_find, subnet_find, quota):
with mock.patch("%s.route_create" % db_mod) as route_create, \
mock.patch("%s.route_find" % db_mod) as route_find, \
mock.patch("%s.subnet_find" % db_mod) as subnet_find, \
mock.patch("neutron.quota.QuotaEngine.limit_check"):
route_create.return_value = create_route
route_find.return_value = find_routes
subnet_find.return_value = subnet
@ -117,10 +116,8 @@ class TestQuarkDeleteRoutes(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, route):
db_mod = "quark.db.api"
with contextlib.nested(
mock.patch("%s.route_delete" % db_mod),
mock.patch("%s.route_find" % db_mod),
) as (route_delete, route_find):
with mock.patch("%s.route_delete" % db_mod) as route_delete, \
mock.patch("%s.route_find" % db_mod) as route_find:
route_find.return_value = route
yield route_delete, route_find

59
quark/tests/plugin_modules/test_security_groups.py
View File

@ -152,10 +152,8 @@ class TestQuarkUpdateSecurityGroup(test_quark_plugin.TestQuarkPlugin):
updated_group = group.copy()
updated_group["name"] = "bar"
with contextlib.nested(
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.db.api.security_group_update"),
) as (db_find, db_update):
with mock.patch("quark.db.api.security_group_find") as db_find, \
mock.patch("quark.db.api.security_group_update") as db_update:
db_find.return_value = group
db_update.return_value = updated_group
update = dict(security_group=dict(name="bar"))
@ -179,10 +177,8 @@ class TestQuarkCreateSecurityGroup(test_quark_plugin.TestQuarkPlugin):
dbgroup = models.SecurityGroup()
dbgroup.update(security_group)
with contextlib.nested(
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.db.api.security_group_create"),
) as (db_find, db_create):
with mock.patch("quark.db.api.security_group_find") as db_find, \
mock.patch("quark.db.api.security_group_create") as db_create:
db_find.return_value.count.return_value = other
db_create.return_value = dbgroup
yield db_create
@ -234,10 +230,9 @@ class TestQuarkDeleteSecurityGroup(test_quark_plugin.TestQuarkPlugin):
dbgroup = models.SecurityGroup()
dbgroup.update(security_group)
with contextlib.nested(
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.db.api.security_group_delete"),
) as (group_find, db_group_delete):
with mock.patch("quark.db.api.security_group_find") as group_find, \
mock.patch("quark.db.api.security_group_delete") as \
db_group_delete:
group_find.return_value = dbgroup
db_group_delete.return_value = dbgroup
yield db_group_delete
@ -303,13 +298,15 @@ class TestQuarkCreateSecurityGroupRule(test_quark_plugin.TestQuarkPlugin):
dbrule["group_id"] = rule['security_group_id']
return dbrule
with contextlib.nested(
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.db.api.security_group_rule_find"),
mock.patch("quark.db.api.security_group_rule_create"),
mock.patch("quark.protocols.human_readable_protocol"),
mock.patch("neutron.quota.QuotaEngine.limit_check")
) as (group_find, rule_find, rule_create, human, limit_check):
with mock.patch("quark.db.api.security_group_find") as group_find, \
mock.patch("quark.db.api.security_group_rule_find") as \
rule_find, \
mock.patch("quark.db.api.security_group_rule_create") as \
rule_create, \
mock.patch("quark.protocols.human_readable_protocol") as \
human, \
mock.patch("neutron.quota.QuotaEngine.limit_check") as \
limit_check:
group_find.return_value = dbgroup
rule_find.return_value.count.return_value = group.get(
'port_rules', None) if group else 0
@ -435,11 +432,11 @@ class TestQuarkDeleteSecurityGroupRule(test_quark_plugin.TestQuarkPlugin):
dbrule = models.SecurityGroupRule()
dbrule.update(dict(rule, group=dbgroup))
with contextlib.nested(
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.db.api.security_group_rule_find"),
mock.patch("quark.db.api.security_group_rule_delete"),
) as (group_find, rule_find, db_group_delete):
with mock.patch("quark.db.api.security_group_find") as group_find, \
mock.patch("quark.db.api.security_group_rule_find") as \
rule_find, \
mock.patch("quark.db.api.security_group_rule_delete") as \
db_group_delete:
group_find.return_value = dbgroup
rule_find.return_value = dbrule
yield db_group_delete
@ -545,16 +542,16 @@ class TestQuarkProtocolHandling(test_quark_plugin.TestQuarkPlugin):
class TestSecurityGroupExceptions(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, finds_rule=True):
with contextlib.nested(
mock.patch("quark.db.api.security_group_find"),
mock.patch("quark.db.api.security_group_rule_find"),
with mock.patch("quark.db.api.security_group_find") as group_find, \
mock.patch("quark.db.api.security_group_rule_find") as \
rule_find, \
mock.patch("neutron.extensions.securitygroup."
"SecurityGroupNotFound.__init__"),
"SecurityGroupNotFound.__init__") as group_exc, \
mock.patch("neutron.extensions.securitygroup."
"SecurityGroupRuleNotFound.__init__"),
"SecurityGroupRuleNotFound.__init__") as \
rule_exc, \
mock.patch("quark.plugin_modules.security_groups."
"_validate_security_group_rule")
) as (group_find, rule_find, group_exc, rule_exc, validate):
"_validate_security_group_rule"):
group_find.return_value = None
rule_find.return_value = None
if finds_rule:

159
quark/tests/plugin_modules/test_subnets.py
View File

@ -181,12 +181,10 @@ class TestQuarkCreateSubnetOverlapping(test_quark_plugin.TestQuarkPlugin):
subnet_models.append(s)
network = models.Network()
network.update(dict(id=1, subnets=subnet_models))
with contextlib.nested(
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_create"),
mock.patch("neutron.common.rpc.get_notifier")
) as (net_find, subnet_find, subnet_create, get_notifier):
with mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.subnet_create") as subnet_create, \
mock.patch("neutron.common.rpc.get_notifier"):
net_find.return_value = network
subnet_find.return_value = subnet_models
subnet_create.return_value = models.Subnet(
@ -239,14 +237,11 @@ class TestQuarkCreateSubnetAllocationPools(test_quark_plugin.TestQuarkPlugin):
return mock.patch.object(models.Subnet, "allocation_pools")
return mock.MagicMock()
with contextlib.nested(
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_create"),
mock.patch("neutron.common.rpc.get_notifier"),
_allocation_pools_mock(),
) as (net_find, subnet_find, subnet_create, get_notifier,
alloc_pools_method):
with mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.subnet_create") as subnet_create, \
mock.patch("neutron.common.rpc.get_notifier"), \
_allocation_pools_mock() as alloc_pools_method:
net_find.return_value = s["network"]
subnet_find.return_value = []
subnet_create.return_value = s
@ -401,19 +396,18 @@ class TestQuarkCreateSubnet(test_quark_plugin.TestQuarkPlugin):
if allocation_pools is not None:
return mock.patch.object(models.Subnet, "allocation_pools")
return mock.MagicMock()
with contextlib.nested(
mock.patch("quark.db.api.subnet_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.dns_create"),
mock.patch("quark.db.api.route_create"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("neutron.common.rpc.get_notifier"),
_allocation_pools_mock(),
mock.patch("sqlalchemy.orm.session.SessionTransaction.commit"),
mock.patch(
"sqlalchemy.orm.unitofwork.UOWTransaction.register_object")
) as (subnet_create, net_find, dns_create, route_create, subnet_find,
get_notifier, alloc_pools_method, commit, register_object):
with mock.patch("quark.db.api.subnet_create") as subnet_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.dns_create") as dns_create, \
mock.patch("quark.db.api.route_create") as route_create, \
mock.patch("quark.db.api.subnet_find"), \
mock.patch("neutron.common.rpc.get_notifier"), \
_allocation_pools_mock() as alloc_pools_method, \
mock.patch("sqlalchemy.orm.session."
"SessionTransaction.commit"), \
mock.patch("sqlalchemy.orm.unitofwork."
"UOWTransaction.register_object") as \
register_object:
subnet_create.return_value = subnet_mod
net_find.return_value = network
route_create.side_effect = route_models
@ -857,16 +851,13 @@ class TestQuarkAllocationPoolCache(test_quark_plugin.TestQuarkPlugin):
gateway=r["nexthop"],
subnet_id=subnet_mod["id"])
for r in host_routes]
with contextlib.nested(
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_update"),
mock.patch("quark.db.api.dns_create"),
mock.patch("quark.db.api.route_find"),
mock.patch("quark.db.api.route_update"),
mock.patch("quark.db.api.route_create"),
mock.patch("sqlalchemy.orm.session.SessionTransaction.commit")
) as (subnet_find, subnet_update, dns_create, route_find,
route_update, route_create, commit):
with mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.subnet_update") as subnet_update, \
mock.patch("quark.db.api.dns_create"), \
mock.patch("quark.db.api.route_find") as route_find, \
mock.patch("quark.db.api.route_update"), \
mock.patch("quark.db.api.route_create"), \
mock.patch("sqlalchemy.orm.session.SessionTransaction.commit"):
subnet_find.return_value = subnet_mod
if has_subnet:
route_find.return_value = (subnet_mod["routes"][0] if
@ -963,19 +954,17 @@ class TestQuarkUpdateSubnet(test_quark_plugin.TestQuarkPlugin):
gateway=r["nexthop"],
subnet_id=subnet_mod["id"])
for r in host_routes]
with contextlib.nested(
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_update"),
mock.patch("quark.db.api.dns_create"),
mock.patch("quark.db.api.route_find"),
mock.patch("quark.db.api.route_update"),
mock.patch("quark.db.api.route_create"),
mock.patch("sqlalchemy.orm.session.SessionTransaction.commit"),
mock.patch(
"sqlalchemy.orm.unitofwork.UOWTransaction.register_object")
) as (subnet_find, subnet_update,
dns_create,
route_find, route_update, route_create, commit, register_object):
with mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.subnet_update") as subnet_update, \
mock.patch("quark.db.api.dns_create") as dns_create, \
mock.patch("quark.db.api.route_find") as route_find, \
mock.patch("quark.db.api.route_update") as route_update, \
mock.patch("quark.db.api.route_create") as route_create, \
mock.patch("sqlalchemy.orm.session."
"SessionTransaction.commit"), \
mock.patch("sqlalchemy.orm.unitofwork."
"UOWTransaction.register_object") as \
register_object:
subnet_find.return_value = subnet_mod
if has_subnet:
route_find.return_value = (subnet_mod["routes"][0] if
@ -1216,12 +1205,11 @@ class TestQuarkDeleteSubnet(test_quark_plugin.TestQuarkPlugin):
ip_mods.append(ip_mod)
strategy_prefix = "quark.network_strategy.JSONStrategy"
with contextlib.nested(
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_delete"),
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("%s.is_provider_network" % strategy_prefix)
) as (sub_find, sub_delete, get_notifier, is_provider_network):
with mock.patch("quark.db.api.subnet_find") as sub_find, \
mock.patch("quark.db.api.subnet_delete") as sub_delete, \
mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("%s.is_provider_network" % strategy_prefix) as \
is_provider_network:
if subnet_mod:
subnet_mod.allocated_ips = ip_mods
sub_find.return_value = subnet_mod
@ -1288,17 +1276,16 @@ class TestSubnetsQuotas(test_quark_plugin.TestQuarkPlugin):
s["_allocation_pool_cache"] = None
subnet = models.Subnet(**s)
subnets.append(subnet)
with contextlib.nested(
mock.patch("quark.plugin_modules.subnets.get_subnets"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.subnet_create"),
mock.patch("quark.db.api.subnet_delete"),
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("oslo_utils.timeutils.utcnow"),
mock.patch("quark.plugin_modules.subnets._validate_subnet_cidr")
) as (get_subnets, sub_find, net_find, sub_create, sub_del, notify,
time_func, sub_validate):
with mock.patch("quark.plugin_modules.subnets.get_subnets") as \
get_subnets, \
mock.patch("quark.db.api.subnet_find") as sub_find, \
mock.patch("quark.db.api.network_find"), \
mock.patch("quark.db.api.subnet_create") as sub_create, \
mock.patch("quark.db.api.subnet_delete"), \
mock.patch("neutron.common.rpc.get_notifier") as notify, \
mock.patch("oslo_utils.timeutils.utcnow") as time_func, \
mock.patch("quark.plugin_modules.subnets."
"_validate_subnet_cidr"):
sub_create.return_value = subnets[0]
sub_find.return_value = subnets[0]
retsubs = []
@ -1473,17 +1460,14 @@ class TestQuarkDiagnoseSubnets(test_quark_plugin.TestQuarkPlugin):
class TestQuarkCreateSubnetAttrFilters(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("quark.db.api.subnet_create"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.dns_create"),
mock.patch("quark.db.api.route_create"),
mock.patch("quark.plugin_views._make_subnet_dict"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("sqlalchemy.orm.session.SessionTransaction.commit")
) as (subnet_create, net_find, dns_create, route_create, sub_dict,
subnet_find, get_notifier, commit):
with mock.patch("quark.db.api.subnet_create") as subnet_create, \
mock.patch("quark.db.api.network_find") as net_find, \
mock.patch("quark.db.api.dns_create"), \
mock.patch("quark.db.api.route_create") as route_create, \
mock.patch("quark.plugin_views._make_subnet_dict"), \
mock.patch("quark.db.api.subnet_find"), \
mock.patch("neutron.common.rpc.get_notifier"), \
mock.patch("sqlalchemy.orm.session.SessionTransaction.commit"):
route_create.return_value = models.Route()
yield subnet_create, net_find
@ -1534,17 +1518,14 @@ class TestQuarkUpdateSubnetAttrFilters(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self):
pool_mod = "quark.allocation_pool.AllocationPools"
with contextlib.nested(
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_update"),
mock.patch("quark.db.api.dns_create"),
mock.patch("quark.db.api.route_find"),
mock.patch("quark.db.api.route_update"),
mock.patch("quark.db.api.route_create"),
mock.patch(pool_mod),
mock.patch("quark.plugin_views._make_subnet_dict")
) as (subnet_find, subnet_update, dns_create, route_find,
route_update, route_create, make_subnet, gateway_exclude):
with mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.subnet_update") as subnet_update, \
mock.patch("quark.db.api.dns_create"), \
mock.patch("quark.db.api.route_find"), \
mock.patch("quark.db.api.route_update"), \
mock.patch("quark.db.api.route_create"), \
mock.patch(pool_mod), \
mock.patch("quark.plugin_views._make_subnet_dict"):
yield subnet_update, subnet_find
def test_update_subnet_attr_filters(self):

275
quark/tests/test_ipam.py
View File

@ -112,20 +112,24 @@ class QuarkMacAddressAllocateDeallocated(QuarkIpamBaseTest):
mac_range = dict(id=1, first_address=0, last_address=255,
next_auto_assign_mac=0, do_not_use=do_not_use,
cidr="AA:BB:CC/24")
with contextlib.nested(
mock.patch("quark.db.api.mac_address_reallocate"),
mock.patch("quark.db.api.mac_address_reallocate_find"),
with mock.patch("quark.db.api.mac_address_reallocate") as \
addr_realloc, \
mock.patch("quark.db.api.mac_address_reallocate_find") as \
addr_realloc_find, \
mock.patch("quark.db.api."
"mac_address_range_find_allocation_counts"),
mock.patch("quark.db.api.mac_address_create"),
mock.patch("quark.db.api.mac_address_range_find"),
mock.patch("quark.db.api.mac_address_delete"),
mock.patch("quark.db.api.mac_range_update_next_auto_assign_mac"),
mock.patch("quark.db.api.mac_range_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (addr_realloc, addr_realloc_find, mac_range_count,
mac_create, range_find, mac_delete, mac_auto_assign, set_full,
refresh):
"mac_address_range_find_allocation_counts") as \
mac_range_count, \
mock.patch("quark.db.api.mac_address_create") as mac_create, \
mock.patch("quark.db.api.mac_address_range_find") as \
range_find, \
mock.patch("quark.db.api.mac_address_delete") as mac_delete, \
mock.patch("quark.db.api."
"mac_range_update_next_auto_assign_mac") as \
mac_auto_assign, \
mock.patch("quark.db.api.mac_range_update_set_full") as \
set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
address_mod = models.MacAddress(**address)
range_mod = models.MacAddressRange(**mac_range)
if mac_find:
@ -174,14 +178,16 @@ class QuarkNewMacAddressAllocation(QuarkIpamBaseTest):
def _stubs(self, addresses=None, ranges=None):
if not addresses:
addresses = [None]
with contextlib.nested(
mock.patch("quark.db.api.mac_address_find"),
with mock.patch("quark.db.api.mac_address_find") as mac_find, \
mock.patch("quark.db.api."
"mac_address_range_find_allocation_counts"),
mock.patch("quark.db.api.mac_range_update_next_auto_assign_mac"),
mock.patch("quark.db.api.mac_range_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (mac_find, mac_range_count, mac_auto, mac_set_full, refresh):
"mac_address_range_find_allocation_counts") as \
mac_range_count, \
mock.patch("quark.db.api."
"mac_range_update_next_auto_assign_mac"), \
mock.patch("quark.db.api.mac_range_update_set_full") as \
mac_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
address_mod = [mac_helper(a) for a in addresses]
range_mod = (range_helper(ranges[0]), ranges[1])
mac_find.side_effect = address_mod
@ -264,12 +270,12 @@ class QuarkNewMacAddressAllocationCreateConflict(QuarkIpamBaseTest):
def _stubs(self, addresses=None, ranges=None):
if not addresses:
addresses = [None]
with contextlib.nested(
mock.patch("quark.db.api.mac_address_find"),
mock.patch("quark.db.api.mac_address_create"),
with mock.patch("quark.db.api.mac_address_find") as mac_find, \
mock.patch("quark.db.api.mac_address_create") as \
mac_create, \
mock.patch("quark.db.api."
"mac_address_range_find_allocation_counts"),
) as (mac_find, mac_create, mac_range_count):
"mac_address_range_find_allocation_counts") as \
mac_range_count:
mac_find.side_effect = [None, None]
address_mod = [mac_helper(a) for a in addresses]
mac_create.side_effect = address_mod
@ -295,12 +301,13 @@ class QuarkNewMacAddressReallocationDeadlocks(QuarkIpamBaseTest):
addresses = [None]
old_override = cfg.CONF.QUARK.mac_address_retry_max
cfg.CONF.set_override('mac_address_retry_max', 1, 'QUARK')
with contextlib.nested(
mock.patch("quark.db.api.mac_address_reallocate"),
mock.patch("quark.db.api.mac_address_create"),
with mock.patch("quark.db.api.mac_address_reallocate") as \
mac_realloc, \
mock.patch("quark.db.api.mac_address_create") as \
mac_create, \
mock.patch("quark.db.api."
"mac_address_range_find_allocation_counts"),
) as (mac_realloc, mac_create, mac_range_count):
"mac_address_range_find_allocation_counts") as \
mac_range_count:
mac_realloc.side_effect = [Exception, None]
mac_create.side_effect = addresses
mac_range_count.return_value = ranges
@ -321,12 +328,11 @@ class QuarkNewMacAddressReallocationDeadlocks(QuarkIpamBaseTest):
class QuarkMacAddressDeallocation(QuarkIpamBaseTest):
@contextlib.contextmanager
def _stubs(self, mac, mac_range):
with contextlib.nested(
mock.patch("quark.db.api.mac_address_find"),
mock.patch("quark.db.api.mac_address_update"),
mock.patch("quark.db.api.mac_address_range_find"),
mock.patch("quark.db.api.mac_address_delete")
) as (mac_find, mac_update, range_find, mac_delete):
with mock.patch("quark.db.api.mac_address_find") as mac_find, \
mock.patch("quark.db.api.mac_address_update") as mac_update, \
mock.patch("quark.db.api.mac_address_range_find") as \
range_find, \
mock.patch("quark.db.api.mac_address_delete") as mac_delete:
mac_update.return_value = mac
mac_find.return_value = mac
range_find.return_value = mac_range
@ -478,18 +484,20 @@ class QuarkIpamTestBothIpAllocation(QuarkIpamBaseTest):
def _stubs(self, addresses=None, subnets=None):
if not addresses:
addresses = [None, None]
with contextlib.nested(
mock.patch("quark.db.api.ip_address_reallocate"),
mock.patch("quark.db.api.ip_address_reallocate_find"),
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (addr_realloc, addr_realloc_find, addr_find,
subnet_alloc_find, subnet_find, subnet_update,
subnet_set_full, refresh):
with mock.patch("quark.db.api.ip_address_reallocate") as \
addr_realloc, \
mock.patch("quark.db.api.ip_address_reallocate_find") as \
addr_realloc_find, \
mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_alloc_find, \
mock.patch("quark.db.api.subnet_find") as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_update, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
addr_mods = []
sub_mods = []
for a in addresses:
@ -887,16 +895,19 @@ class QuarkIpamTestBothRequiredIpAllocation(QuarkIpamBaseTest):
if not addresses:
addresses = [None, None]
self.context.session.add = mock.Mock()
with contextlib.nested(
mock.patch("quark.db.api.ip_address_reallocate"),
mock.patch("quark.db.api.ip_address_reallocate_find"),
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (addr_realloc, addr_realloc_find, addr_find,
subnet_find, subnet_update, subnet_set_full, refresh):
with mock.patch("quark.db.api.ip_address_reallocate") as \
addr_realloc, \
mock.patch("quark.db.api.ip_address_reallocate_find") as \
addr_realloc_find, \
mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_update, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
addrs = [ip_helper(a) for a in addresses]
addr_realloc.side_effect = addrs[:1]
addr_realloc_find.side_effect = addrs[:1]
@ -1129,12 +1140,14 @@ class QuarkIpamAllocateFromV6Subnet(QuarkIpamBaseTest):
ip_mod["allocated_at"] = timeutils.utcnow()
ip_mod["version"] = ip_address.version
with contextlib.nested(
mock.patch("quark.db.models.IPPolicy.get_cidrs_ip_set"),
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.ip_address_create"),
mock.patch("quark.db.api.ip_address_update")
) as (policy_find, ip_address_find, ip_create, ip_update):
with mock.patch("quark.db.models.IPPolicy.get_cidrs_ip_set") as \
policy_find, \
mock.patch("quark.db.api.ip_address_find") as \
ip_address_find, \
mock.patch("quark.db.api.ip_address_create") as \
ip_create, \
mock.patch("quark.db.api.ip_address_update") as \
ip_update:
policy_find.return_value = policies
ip_address_find.return_value = ip_mod
ip_create.return_value = ip_mod
@ -1194,16 +1207,19 @@ class QuarkNewIPAddressAllocation(QuarkIpamBaseTest):
if not addresses:
addresses = [None]
self.context.session.add = mock.Mock()
with contextlib.nested(
mock.patch("quark.db.api.ip_address_reallocate"),
mock.patch("quark.db.api.ip_address_reallocate_find"),
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (addr_realloc, addr_realloc_find, addr_find,
subnet_find, subnet_update, subnet_set_full, refresh):
with mock.patch("quark.db.api.ip_address_reallocate") as \
addr_realloc, \
mock.patch("quark.db.api.ip_address_reallocate_find") as \
addr_realloc_find, \
mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_update, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
addrs = [ip_helper(a) for a in addresses]
addr_realloc.side_effect = addrs[:1]
addr_realloc_find.side_effect = addrs[:1]
@ -1496,15 +1512,16 @@ class QuarkIPAddressAllocationTestRetries(QuarkIpamBaseTest):
@contextlib.contextmanager
def _stubs(self, address=None, subnets=None):
self.context.session.add = mock.Mock()
with contextlib.nested(
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.ip_address_create"),
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (addr_find, addr_create, subnet_find, subnet_update,
subnet_set_full, refresh):
with mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.ip_address_create") as addr_create, \
mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_update, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
addr_find.side_effect = [None, None, None]
addr_mods = []
for a in address:
@ -1635,14 +1652,14 @@ class QuarkIPAddressAllocateDeallocated(QuarkIpamBaseTest):
@contextlib.contextmanager
def _stubs(self, ip_find, subnet, address, addresses_found,
sub_found=True):
with contextlib.nested(
mock.patch("quark.db.api.ip_address_reallocate"),
mock.patch("quark.db.api.ip_address_reallocate_find"),
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.ip_address_update"),
mock.patch("quark.ipam.QuarkIpamANY._choose_available_subnet")
) as (addr_realloc, addr_realloc_find, addr_find, addr_update,
choose_subnet):
with mock.patch("quark.db.api.ip_address_reallocate") as \
addr_realloc, \
mock.patch("quark.db.api.ip_address_reallocate_find") as \
addr_realloc_find, \
mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.ip_address_update") as addr_update, \
mock.patch("quark.ipam.QuarkIpamANY."
"_choose_available_subnet") as choose_subnet:
addr_mod = models.IPAddress(**address)
subnet_mod = models.Subnet(**subnet)
subnet_mod["next_auto_assign_ip"] = subnet["next_auto_assign_ip"]
@ -1703,13 +1720,15 @@ class TestQuarkIpPoliciesIpAllocation(QuarkIpamBaseTest):
if not addresses:
addresses = [None]
self.context.session.add = mock.Mock()
with contextlib.nested(
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh")
) as (addr_find, subnet_find, subnet_update, subnet_set_full, refresh):
with mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_update, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
addr_find.side_effect = [ip_helper(a) for a in addresses]
sub_mods = []
if subnets:
@ -1813,16 +1832,16 @@ class QuarkIPAddressAllocationNotifications(QuarkIpamBaseTest):
def _stubs(self, address, addresses=None, subnets=None, deleted_at=None):
if not addresses:
addresses = [None]
with contextlib.nested(
mock.patch("quark.db.api.ip_address_find"),
mock.patch("quark.db.api.ip_address_create"),
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("sqlalchemy.orm.session.Session.refresh"),
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("oslo_utils.timeutils.utcnow"),
) as (addr_find, addr_create, subnet_find, subnet_update, refresh,
notify, timeutils):
with mock.patch("quark.db.api.ip_address_find") as addr_find, \
mock.patch("quark.db.api.ip_address_create") as addr_create, \
mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_update, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh, \
mock.patch("neutron.common.rpc.get_notifier") as notify, \
mock.patch("oslo_utils.timeutils.utcnow") as timeutils:
addrs_found = []
for a in addresses:
if a:
@ -1952,12 +1971,14 @@ class QuarkIpamTestV6IpGeneration(QuarkIpamBaseTest):
class QuarkIpamTestSelectSubnet(QuarkIpamBaseTest):
@contextlib.contextmanager
def _stubs(self, subnet, count, increments=True, marks_full=True):
with contextlib.nested(
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh"),
) as (subnet_find, subnet_incr, subnet_set_full, refresh):
with mock.patch("quark.db.api.subnet_find_ordered_by_most_full") \
as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_incr, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
sub_mods = []
sub_mods.append((subnet_helper(subnet), count))
@ -2049,12 +2070,13 @@ class QuarkIpamTestSelectSubnet(QuarkIpamBaseTest):
class QuarkIpamTestSelectSubnetLocking(QuarkIpamBaseTest):
@contextlib.contextmanager
def _stubs(self, subnet, count, increments=True, marks_full=True):
with contextlib.nested(
mock.patch("quark.db.api.subnet_find_ordered_by_most_full"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh"),
) as (subnet_find, subnet_incr, subnet_set_full, refresh):
with mock.patch("quark.db.api.subnet_find_ordered_by_most_full") as \
subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_incr, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh"):
sub_mods = []
sub_mods.append((subnet_helper(subnet), count))
@ -2228,12 +2250,13 @@ class IronicIpamTestSelectSubnet(QuarkIpamBaseTest):
@contextlib.contextmanager
def _stubs(self, subnet, count, increments=True, marks_full=True):
with contextlib.nested(
mock.patch("quark.db.api.subnet_find_unused"),
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip"),
mock.patch("quark.db.api.subnet_update_set_full"),
mock.patch("sqlalchemy.orm.session.Session.refresh"),
) as (subnet_find, subnet_incr, subnet_set_full, refresh):
with mock.patch("quark.db.api.subnet_find_unused") as subnet_find, \
mock.patch("quark.db.api.subnet_update_next_auto_assign_ip") \
as subnet_incr, \
mock.patch("quark.db.api.subnet_update_set_full") as \
subnet_set_full, \
mock.patch("sqlalchemy.orm.session.Session.refresh") as \
refresh:
sub_mods = []
sub_mods.append((subnet_helper(subnet), count))

55
quark/tests/test_migrations.py
View File

@ -1,4 +1,3 @@
import contextlib
import datetime
import os
import tempfile
@ -96,10 +95,8 @@ class Test2748e48cee3a(BaseMigrationTest):
dict(id="222", created_at=datetime.date(1970, 1, 1),
ip_policy_id="111", cidr="192.168.10.0/16"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
tu.utcnow.return_value = datetime.datetime(2004, 2, 14)
uuid.generate_uuid.return_value = "foo"
alembic_command.upgrade(self.config, '2748e48cee3a')
@ -121,10 +118,8 @@ class Test2748e48cee3a(BaseMigrationTest):
dict(id="222", created_at=datetime.date(1970, 1, 1),
ip_policy_id="111", cidr="fd00::/7"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
tu.utcnow.return_value = datetime.datetime(2004, 2, 14)
uuid.generate_uuid.return_value = "foo"
alembic_command.upgrade(self.config, '2748e48cee3a')
@ -168,10 +163,8 @@ class Test2748e48cee3a(BaseMigrationTest):
dict(id="223", created_at=dt, ip_policy_id="113",
cidr="0.0.0.0/24"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
tu.utcnow.return_value = datetime.datetime(2004, 2, 14)
uuid.generate_uuid.return_value = "foo"
alembic_command.upgrade(self.config, '2748e48cee3a')
@ -202,10 +195,8 @@ class Test2748e48cee3a(BaseMigrationTest):
dict(id="223", created_at=datetime.date(1970, 1, 1),
ip_policy_id="111", cidr="192.168.10.0/23"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
tu.utcnow.return_value = datetime.datetime(2004, 2, 14)
uuid.generate_uuid.return_value = "foo"
alembic_command.upgrade(self.config, '2748e48cee3a')
@ -276,10 +267,8 @@ class Test45a07fac3d38(BaseMigrationTest):
self.ip_policy_cidrs.insert(),
dict(id="222", created_at=datetime.date(1970, 1, 1),
ip_policy_id="111", cidr="192.168.10.13/32"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
uuid.generate_uuid.side_effect = (1, 2, 3)
tu.utcnow.return_value = datetime.datetime(1970, 1, 1)
alembic_command.upgrade(self.config, '45a07fac3d38')
@ -305,10 +294,8 @@ class Test45a07fac3d38(BaseMigrationTest):
self.ip_policy_cidrs.insert(),
dict(id="222", created_at=datetime.date(1970, 1, 1),
ip_policy_id="111", cidr="fd00::3/128"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
uuid.generate_uuid.side_effect = (1, 2, 3)
tu.utcnow.return_value = datetime.datetime(1970, 1, 1)
alembic_command.upgrade(self.config, '45a07fac3d38')
@ -479,10 +466,8 @@ class Test552b213c2b8c(BaseMigrationTest):
self.subnets.insert(),
dict(id="000", tenant_id="foo", _cidr="192.168.10.0/24",
ip_policy_id=None))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
dt = datetime.datetime(1970, 1, 1)
tu.utcnow.return_value = dt
uuid.generate_uuid.side_effect = ("666", "667", "668")
@ -514,10 +499,8 @@ class Test552b213c2b8c(BaseMigrationTest):
self.subnets.insert(),
dict(id="000", tenant_id="foo", _cidr="fd00::/64",
ip_policy_id=None))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
dt = datetime.datetime(1970, 1, 1)
tu.utcnow.return_value = dt
uuid.generate_uuid.side_effect = ("666", "667", "668")
@ -563,10 +546,8 @@ class Test552b213c2b8c(BaseMigrationTest):
dict(id="221", created_at=dt,
ip_policy_id="111", cidr="192.168.10.13/32"))
with contextlib.nested(
mock.patch("oslo_utils.uuidutils"),
mock.patch("oslo_utils.timeutils")
) as (uuid, tu):
with mock.patch("oslo_utils.uuidutils") as uuid, \
mock.patch("oslo_utils.timeutils") as tu:
tu.utcnow.return_value = dt
uuid.generate_uuid.side_effect = ("5", "6", "7", "8", "9", "10")
alembic_command.upgrade(self.config, '552b213c2b8c')

107
quark/tests/test_nvp_driver.py
View File

@ -123,9 +123,7 @@ class TestNVPDriver(test_base.TestBase):
class TestNVPDriverCreateNetwork(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
) as (conn,):
with mock.patch("%s._connection" % self.d_pkg) as conn:
connection = self._create_connection()
conn.return_value = connection
yield connection
@ -164,12 +162,11 @@ class TestNVPDriverDefaultTransportZoneBindings(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("quark.drivers.nvp_driver.SA_REGISTRY."
"get_strategy"),
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
) as (sa_get_strategy, conn, switch_list):
with mock.patch("quark.drivers.nvp_driver.SA_REGISTRY."
"get_strategy") as sa_get_strategy, \
mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
switch_list:
connection = self._create_connection()
conn.return_value = connection
@ -216,9 +213,7 @@ class TestNVPDriverProviderNetwork(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self, tz):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
) as (conn,):
with mock.patch("%s._connection" % self.d_pkg) as conn:
connection = self._create_connection()
switch = self._create_lswitch(1, False)
switch.transport_zone = mock.Mock()
@ -351,10 +346,9 @@ class TestNVPDriverProviderNetwork(TestNVPDriver):
class TestNVPDriverDeleteNetwork(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self, network_exists=True):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
) as (conn, switch_list):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
switch_list:
connection = self._create_connection()
conn.return_value = connection
if network_exists:
@ -383,11 +377,10 @@ class TestNVPDriverDeleteNetwork(TestNVPDriver):
class TestNVPDriverDeleteNetworkWithExceptions(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self, network_exists=True, exception=None):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
mock.patch("%s._lswitch_delete" % self.d_pkg),
) as (conn, switch_list, switch_delete):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
switch_list, \
mock.patch("%s._lswitch_delete" % self.d_pkg) as switch_delete:
connection = self._create_connection()
conn.return_value = connection
if network_exists:
@ -437,12 +430,12 @@ class TestNVPDriverCreatePort(TestNVPDriver):
'''In all cases an lswitch should be queried.'''
@contextlib.contextmanager
def _stubs(self, has_lswitch=True, maxed_ports=False, net_details=None):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
mock.patch("%s._get_network_details" % self.d_pkg),
) as (conn, next_conn, get_switches, get_net_dets):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg), \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
get_switches, \
mock.patch("%s._get_network_details" % self.d_pkg) as \
get_net_dets:
connection = self._create_connection(has_switches=has_lswitch,
maxed_ports=maxed_ports)
conn.return_value = connection
@ -591,10 +584,8 @@ class TestNVPDriverCreatePort(TestNVPDriver):
class TestNVPDriverUpdatePort(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
) as (conn, next_conn):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg):
connection = self._create_connection()
connection.securityprofile = self._create_security_profile()
conn.return_value = connection
@ -629,9 +620,7 @@ class TestNVPDriverUpdatePort(TestNVPDriver):
class TestNVPDriverLswitchesForNetwork(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self, single_switch=True):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
) as (conn,):
with mock.patch("%s._connection" % self.d_pkg) as conn:
connection = self._create_connection(
has_switches=True, switch_count=1)
conn.return_value = connection
@ -696,10 +685,8 @@ class TestSwitchCopying(TestNVPDriver):
class TestNVPDriverDeletePort(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self, switch_count=1):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
) as (conn, next_conn):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg):
connection = self._create_connection(switch_count=switch_count)
conn.return_value = connection
yield connection
@ -736,10 +723,8 @@ class TestNVPDriverDeletePort(TestNVPDriver):
class TestNVPDriverDeletePortWithExceptions(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self, switch_exception=None, delete_exception=None):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitch_from_port" % self.d_pkg),
) as (conn, switch):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitch_from_port" % self.d_pkg) as switch:
connection = self._create_connection()
conn.return_value = connection
if switch_exception:
@ -820,10 +805,8 @@ class TestNVPDriverDeletePortWithExceptions(TestNVPDriver):
class TestNVPDriverCreateSecurityGroup(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
) as (conn, next_conn):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg):
connection = self._create_connection()
connection.securityprofile = self._create_security_profile()
conn.return_value = connection
@ -875,10 +858,8 @@ class TestNVPDriverCreateSecurityGroup(TestNVPDriver):
class TestNVPDriverDeleteSecurityGroup(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
) as (conn, next_conn):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg):
connection = self._create_connection()
connection.securityprofile = self._create_security_profile()
conn.return_value = connection
@ -905,10 +886,8 @@ class TestNVPDriverDeleteSecurityGroup(TestNVPDriver):
class TestNVPDriverUpdateSecurityGroup(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
) as (conn, next_conn):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg):
connection = self._create_connection()
connection.securityprofile = self._create_security_profile()
conn.return_value = connection
@ -966,10 +945,8 @@ class TestNVPDriverUpdateSecurityGroup(TestNVPDriver):
class TestNVPDriverCreateSecurityGroupRule(TestNVPDriver):
@contextlib.contextmanager
def _stubs(self):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._next_connection" % self.d_pkg),
) as (conn, next_conn):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._next_connection" % self.d_pkg):
connection = self._create_connection()
connection.securityprofile = self._create_security_profile()
connection.securityrule = self._create_security_rule()
@ -1050,9 +1027,7 @@ class TestNVPDriverDeleteSecurityGroupRule(TestNVPDriver):
for rule in rules:
rulelist['logical_port_%s_rules' % rule.pop('direction')].append(
rule)
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
) as (conn,):
with mock.patch("%s._connection" % self.d_pkg) as conn:
connection = self._create_connection()
connection.securityprofile = self._create_security_profile()
connection.securityrule = self._create_security_rule()
@ -1135,10 +1110,8 @@ class TestNVPGetConnection(TestNVPDriver):
retries=1,
backoff=0,
usages=0))
with contextlib.nested(
mock.patch("aiclib.nvp.Connection"),
mock.patch("%s._next_connection" % self.d_pkg)
) as (aiclib_conn, next_conn):
with mock.patch("aiclib.nvp.Connection") as aiclib_conn, \
mock.patch("%s._next_connection" % self.d_pkg) as next_conn:
yield aiclib_conn, next_conn
cfg.CONF.clear_override("controller_connection", "NVP")
@ -1208,9 +1181,7 @@ class TestNVPNextConnection(TestNVPDriver):
conn2["ip_address"] = "192.168.0.2"
self.driver.nvp_connections.extend([conn1, conn2])
with contextlib.nested(
mock.patch("random.randint")
) as (randint,):
with mock.patch("random.randint") as randint:
randint.return_value = 1
yield randint
cfg.CONF.clear_override("controller_connection", "NVP")

108
quark/tests/test_optimized_nvp_driver.py
View File

@ -51,11 +51,11 @@ class TestOptimizedNVPDriverDeleteNetwork(TestOptimizedNVPDriver):
'''Need to ensure that network of X switches deletes X switches.'''
@contextlib.contextmanager
def _stubs(self, switch_count=1):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
) as (conn, select_switch, get_switches):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg) as \
select_switch, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
get_switches:
connection = self._create_connection()
switch = self._create_lswitch_mock()
conn.return_value = connection
@ -104,12 +104,12 @@ class TestOptimizedNVPDriverDeleteNetworkWithExceptions(
@contextlib.contextmanager
def _stubs(self, switch_count=1, error_code=500):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
mock.patch("%s._lswitch_delete" % self.d_pkg)
) as (conn, select_switch, get_switches, delete_switch):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg) as \
select_switch, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
get_switches, \
mock.patch("%s._lswitch_delete" % self.d_pkg) as delete_switch:
connection = self._create_connection()
switch = self._create_lswitch_mock()
conn.return_value = connection
@ -150,14 +150,14 @@ class TestOptimizedNVPDriverDeletePortMultiSwitch(TestOptimizedNVPDriver):
@contextlib.contextmanager
def _stubs(self, port_count=2, exception=None):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lport_select_by_id" % self.d_pkg),
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
mock.patch("%s._lport_delete" % self.d_pkg),
) as (conn, select_port, select_switch,
two_switch, port_delete):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lport_select_by_id" % self.d_pkg) as \
select_port, \
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg) as \
select_switch, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
two_switch, \
mock.patch("%s._lport_delete" % self.d_pkg) as port_delete:
connection = self._create_connection()
port = self._create_lport_mock(port_count)
switch = self._create_lswitch_mock()
@ -243,12 +243,13 @@ class TestOptimizedNVPDriverDeletePortSingleSwitch(TestOptimizedNVPDriver):
@contextlib.contextmanager
def _stubs(self, port_count=1):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lport_select_by_id" % self.d_pkg),
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
) as (conn, select_port, select_switch, one_switch):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lport_select_by_id" % self.d_pkg) as \
select_port, \
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg) as \
select_switch, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
one_switch:
connection = self._create_connection()
switch = self._create_lswitch_mock()
conn.return_value = connection
@ -271,12 +272,13 @@ class TestOptimizedNVPDriverDeletePortMissing(TestOptimizedNVPDriver):
@contextlib.contextmanager
def _stubs(self, port_count=2):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lport_select_by_id" % self.d_pkg),
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg),
mock.patch("%s._lswitches_for_network" % self.d_pkg),
) as (conn, select_port, select_switch, one_switch):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lport_select_by_id" % self.d_pkg) as \
select_port, \
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg) as \
select_switch, \
mock.patch("%s._lswitches_for_network" % self.d_pkg) as \
one_switch:
connection = self._create_connection()
port = self._create_lport_mock(port_count)
switch = self._create_lswitch_mock()
@ -301,15 +303,17 @@ class TestOptimizedNVPDriverCreatePort(TestOptimizedNVPDriver):
@contextlib.contextmanager
def _stubs(self, has_lswitch=True, maxed_ports=False):
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._lswitch_select_free" % self.d_pkg),
mock.patch("%s._lswitch_select_first" % self.d_pkg),
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg),
mock.patch("%s._lswitch_create_optimized" % self.d_pkg),
mock.patch("%s._get_network_details" % self.d_pkg)
) as (conn, select_free, select_first,
select_by_id, create_opt, get_net_dets):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._lswitch_select_free" % self.d_pkg) as \
select_free, \
mock.patch("%s._lswitch_select_first" % self.d_pkg) as \
select_first, \
mock.patch("%s._lswitch_select_by_nvp_id" % self.d_pkg) as \
select_by_id, \
mock.patch("%s._lswitch_create_optimized" % self.d_pkg) as \
create_opt, \
mock.patch("%s._get_network_details" % self.d_pkg) as \
get_net_dets:
connection = self._create_connection()
conn.return_value = connection
if has_lswitch:
@ -446,10 +450,9 @@ class TestOptimizedNVPDriverUpdatePort(TestOptimizedNVPDriver):
mod_path = "quark.drivers.%s"
op_path = "optimized_nvp_driver.OptimizedNVPDriver"
lport_path = "%s._lport_select_by_id" % op_path
with contextlib.nested(
mock.patch(mod_path % "nvp_driver.NVPDriver.update_port"),
mock.patch(mod_path % lport_path),
) as (update_port, port_find):
with mock.patch(mod_path % "nvp_driver.NVPDriver.update_port") as \
update_port, \
mock.patch(mod_path % lport_path) as port_find:
ret_port = quark.drivers.optimized_nvp_driver.LSwitchPort()
port_find.return_value = ret_port
update_port.return_value = dict(switch_id=2)
@ -467,10 +470,9 @@ class TestCreateSecurityGroups(TestOptimizedNVPDriver):
class TestDeleteSecurityGroups(TestOptimizedNVPDriver):
def test_delete_security_group(self):
mod_path = "quark.drivers.nvp_driver.NVPDriver"
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._query_security_group" % self.d_pkg),
mock.patch("%s.delete_security_group" % mod_path)):
with mock.patch("%s._connection" % self.d_pkg), \
mock.patch("%s._query_security_group" % self.d_pkg), \
mock.patch("%s.delete_security_group" % mod_path):
session_delete = self.context.session.delete
self.context.session.delete = mock.Mock(return_value=None)
@ -483,11 +485,11 @@ class TestSecurityGroupRules(TestOptimizedNVPDriver):
@contextlib.contextmanager
def _stubs(self, rules=None):
rules = rules or []
with contextlib.nested(
mock.patch("%s._connection" % self.d_pkg),
mock.patch("%s._query_security_group" % self.d_pkg),
mock.patch("%s._check_rule_count_per_port" % self.d_pkg),
) as (conn, query_sec_group, rule_count):
with mock.patch("%s._connection" % self.d_pkg) as conn, \
mock.patch("%s._query_security_group" % self.d_pkg) as \
query_sec_group, \
mock.patch("%s._check_rule_count_per_port" % self.d_pkg) \
as rule_count:
query_sec_group.return_value = (quark.drivers.optimized_nvp_driver.
SecurityProfile())
connection = self._create_connection()

23
quark/tests/tools/test_redis_sg_tool.py
View File

@ -152,11 +152,11 @@ class QuarkRedisSgToolPurgeOrphans(QuarkRedisSgToolBase):
def _stubs(self):
ports = [{"device_id": 1, "mac_address": 1}]
vifs = ["1.1", "2.2", "3.3"]
with contextlib.nested(
mock.patch("neutron.context.get_admin_context"),
mock.patch("quark.db.api.ports_with_security_groups_find"),
mock.patch("%s._get_connection" % TOOL_MOD)
) as (get_admin_ctxt, db_ports_groups, get_conn):
with mock.patch("neutron.context.get_admin_context") as \
get_admin_ctxt, \
mock.patch("quark.db.api.ports_with_security_groups_find") \
as db_ports_groups, \
mock.patch("%s._get_connection" % TOOL_MOD) as get_conn:
connection_mock = mock.MagicMock()
get_conn.return_value = connection_mock
ports_with_groups_mock = mock.MagicMock()
@ -216,12 +216,13 @@ class QuarkRedisSgToolWriteGroups(QuarkRedisSgToolBase):
vifs = ["1.1", "2.2", "3.3"]
security_groups = [{"id": 1, "name": "test_group"}]
with contextlib.nested(
mock.patch("neutron.context.get_admin_context"),
mock.patch("quark.db.api.security_group_rule_find"),
mock.patch("quark.db.api.ports_with_security_groups_find"),
mock.patch("%s._get_connection" % TOOL_MOD)
) as (get_admin_ctxt, rule_find, db_ports_groups, get_conn):
with mock.patch("neutron.context.get_admin_context") as \
get_admin_ctxt, \
mock.patch("quark.db.api.security_group_rule_find") as \
rule_find, \
mock.patch("quark.db.api.ports_with_security_groups_find") \
as db_ports_groups, \
mock.patch("%s._get_connection" % TOOL_MOD) as get_conn:
connection_mock = mock.MagicMock()
get_conn.return_value = connection_mock
ports_with_groups_mock = mock.MagicMock()

31
quark/tests/tools/test_resp_async_middleware.py
View File

@ -12,7 +12,6 @@
# implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import contextlib
import json
import mock
@ -57,10 +56,8 @@ class TestRespAsyncIDMiddleware(test_base.TestBase):
def test_mw_none_context(self):
mw = jobmw.filter_factory(self.conf)(self.app)
with contextlib.nested(
mock.patch('%s._get_resp' % mw_mock_path),
mock.patch('%s._get_ctx' % mw_mock_path)) as \
(get_resp, get_ctx):
with mock.patch('%s._get_resp' % mw_mock_path) as get_resp, \
mock.patch('%s._get_ctx' % mw_mock_path) as get_ctx:
get_resp.return_value = self.resp_return
get_ctx.return_value = self.none_ctx
resp = mw.__call__.request('/', method='GET', body=self.body)
@ -71,10 +68,8 @@ class TestRespAsyncIDMiddleware(test_base.TestBase):
def test_mw_empty_context(self):
mw = jobmw.filter_factory(self.conf)(self.app)
with contextlib.nested(
mock.patch('%s._get_resp' % mw_mock_path),
mock.patch('%s._get_ctx' % mw_mock_path)) as \
(get_resp, get_ctx):
with mock.patch('%s._get_resp' % mw_mock_path) as get_resp, \
mock.patch('%s._get_ctx' % mw_mock_path) as get_ctx:
get_resp.return_value = self.resp_return
get_ctx.return_value = self.no_ctx
resp = mw.__call__.request('/', method='GET', body=self.body)
@ -85,10 +80,8 @@ class TestRespAsyncIDMiddleware(test_base.TestBase):
def test_mw_missing_context(self):
mw = jobmw.filter_factory(self.conf)(self.app)
with contextlib.nested(
mock.patch('%s._get_resp' % mw_mock_path),
mock.patch('%s._get_ctx' % mw_mock_path)) as \
(get_resp, get_ctx):
with mock.patch('%s._get_resp' % mw_mock_path) as get_resp, \
mock.patch('%s._get_ctx' % mw_mock_path) as get_ctx:
get_resp.return_value = self.resp_return
get_ctx.return_value = self.random_ctx
resp = mw.__call__.request('/', method='GET', body=self.body)
@ -99,10 +92,8 @@ class TestRespAsyncIDMiddleware(test_base.TestBase):
def test_mw_modify_resp(self):
mw = jobmw.filter_factory(self.conf)(self.app)
with contextlib.nested(
mock.patch('%s._get_resp' % mw_mock_path),
mock.patch('%s._get_ctx' % mw_mock_path)) as \
(get_resp, get_ctx):
with mock.patch('%s._get_resp' % mw_mock_path) as get_resp, \
mock.patch('%s._get_ctx' % mw_mock_path) as get_ctx:
get_resp.return_value = self.resp_return
get_ctx.return_value = self.job_ctx
resp = mw.__call__.request('/', method='GET', body=self.body)
@ -116,10 +107,8 @@ class TestRespAsyncIDMiddleware(test_base.TestBase):
def test_mw_error_resp(self):
mw = jobmw.filter_factory(self.conf)(self.app)
with contextlib.nested(
mock.patch('%s._get_resp' % mw_mock_path),
mock.patch('%s._get_ctx' % mw_mock_path)) as \
(get_resp, get_ctx):
with mock.patch('%s._get_resp' % mw_mock_path) as get_resp, \
mock.patch('%s._get_ctx' % mw_mock_path) as get_ctx:
get_resp.return_value = self.err_resp
get_ctx.return_value = self.job_ctx
resp = mw.__call__.request('/', method='GET', body=self.body)