x/quark
Code Issues Proposed changes

Implemented unit tests for Plugin.update_port

Browse Source
This commit is contained in:
Amir Sadoughi
2013-04-02 00:03:24 -05:00
parent c6f71174be
commit 0ff1a5c613
2 changed files with 91 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
quark/plugin.py
View File

@@ -522,24 +522,23 @@ class Plugin(quantum_plugin_base_v2.QuantumPluginBaseV2):
net_address = netaddr.IPAddress(ip_address)
address = db_api.ip_address_find(
context,
address=ip_address,
deallocated=False,
address=net_address,
tenant_id=context.tenant_id,
scope=db_api.ONE)
if not address:
address = self.ipam_driver.allocate_ip_address(
context,
net_id,
port_id,
port_db["network_id"],
id,
self.ipam_reuse_after,
ip_address=ip_address)
else:
address = self.ipam_driver.allocate_ip_address(
context,
net_id,
port_id,
port_db["network_id"],
id,
self.ipam_reuse_after)
port_db["ip_addresses"].extend([address])
return self._make_port_dict(port_db)

94
quark/tests/test_quark_plugin.py
View File

@@ -701,17 +701,93 @@ class TestQuarkCreatePort(TestQuarkPlugin):
class TestQuarkUpdatePort(TestQuarkPlugin):
# port: port doesn't exit - > PortNotFound
# port: { "fixed_ips": [{}] } -> creates ip on network
# port: { "fixed_ips": [{"ip_id": uuid}]} -> doesn't exist fails
# -> does exist succeeds and associates
# port : " " "ip": address -> bad ip address fails
# -> bad ip (no subnet) fails
# -> good ip, good subnet, success
def test_update_not_implemented(self):
with self.assertRaises(NotImplementedError):
@contextlib.contextmanager
def _stubs(self, port, addr, addr2=None):
port_model = None
addr_model = None
addr_model2 = None
if port:
port_model = models.Port()
port_model.update(port)
if addr:
addr_model = models.IPAddress()
addr_model.update(addr)
if addr2:
addr_model2 = models.IPAddress()
addr_model2.update(addr2)
with contextlib.nested(
mock.patch("quark.db.api.port_find"),
mock.patch("quark.ipam.QuarkIpam.allocate_ip_address"),
mock.patch("quark.db.api.ip_address_find")
) as (port_find, alloc_ip, ip_find):
port_find.return_value = port_model
alloc_ip.return_value = addr_model2 if addr_model2 else addr_model
ip_find.return_value = addr_model
yield port_find, alloc_ip, ip_find
def test_update_port_no_ports(self):
with self.assertRaises(exceptions.PortNotFound):
self.plugin.update_port(self.context, 1, {})
def test_update_port_fixed_ips_ip(self):
new_port_ip = dict(port=dict(fixed_ips=[dict()]))
port = dict(port=dict(network_id=1, tenant_id=self.context.tenant_id,
device_id=2))
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
subnet_id=1, network_id=2, version=4, deallocated=True)
with self._stubs(port=port, addr=ip) as (port_find, alloc_ip, ip_find):
response = self.plugin.update_port(self.context, 1, new_port_ip)
self.assertEqual(port_find.call_count, 1)
self.assertEqual(alloc_ip.call_count, 1)
self.assertEqual(ip_find.call_count, 0)
self.assertEqual(response["fixed_ips"][0]["ip_address"],
"192.168.1.100")
def test_update_port_fixed_ips_ip_id(self):
new_port_ip = dict(port=dict(fixed_ips=[dict(ip_id=1)]))
port = dict(port=dict(network_id=1, tenant_id=self.context.tenant_id,
device_id=2))
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
subnet_id=1, network_id=2, version=4, deallocated=True)
with self._stubs(port=port, addr=ip) as (port_find, alloc_ip, ip_find):
response = self.plugin.update_port(self.context, 1, new_port_ip)
self.assertEqual(port_find.call_count, 1)
self.assertEqual(alloc_ip.call_count, 0)
self.assertEqual(ip_find.call_count, 1)
self.assertEqual(response["fixed_ips"][0]["ip_address"],
"192.168.1.100")
def test_update_port_fixed_ips_ip_address_exists(self):
new_port_ip = dict(port=dict(fixed_ips=[dict(
ip_address="192.168.1.100")]))
port = dict(port=dict(network_id=1, tenant_id=self.context.tenant_id,
device_id=2))
ip = dict(id=1, address=3232235876, address_readable="192.168.1.100",
subnet_id=1, network_id=2, version=4, deallocated=True)
with self._stubs(port=port, addr=ip) as (port_find, alloc_ip, ip_find):
response = self.plugin.update_port(self.context, 1, new_port_ip)
self.assertEqual(port_find.call_count, 1)
self.assertEqual(alloc_ip.call_count, 0)
self.assertEqual(ip_find.call_count, 1)
self.assertEqual(response["fixed_ips"][0]["ip_address"],
"192.168.1.100")
def test_update_port_fixed_ips_ip_address_doesnt_exist(self):
new_port_ip = dict(port=dict(fixed_ips=[dict(
ip_address="192.168.1.101")]))
port = dict(port=dict(network_id=1, tenant_id=self.context.tenant_id,
device_id=2))
ip = dict(id=1, address=3232235876, address_readable="192.168.1.101",
subnet_id=1, network_id=2, version=4, deallocated=True)
with self._stubs(port=port, addr=None, addr2=ip) as \
(port_find, alloc_ip, ip_find):
response = self.plugin.update_port(self.context, 1, new_port_ip)
self.assertEqual(port_find.call_count, 1)
self.assertEqual(alloc_ip.call_count, 1)
self.assertEqual(ip_find.call_count, 1)
self.assertEqual(response["fixed_ips"][0]["ip_address"],
"192.168.1.101")
class TestQuarkGetPortCount(TestQuarkPlugin):
def test_get_port_count(self):