x/quark
Code Issues Proposed changes

Merge pull request #461 from asadoughi/master

Browse Source 
Fixed /ip_address/{ipAddressId}/ports query for service filter
This commit is contained in:
Matt Dietz
2015-10-09 14:54:40 -05:00
parent 7bbe0458f6 76194f3d77
commit 837c19b919
2 changed files with 30 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
quark/db/api.py
View File

@@ -199,6 +199,10 @@ def port_find(context, limit=None, sorts=None, marker_obj=None, fields=None,
if filters.get("device_id"):
model_filters.append(models.Port.device_id.in_(filters["device_id"]))
if filters.get("service"):
model_filters.append(models.Port.associations.any(
models.PortIpAssociation.service == filters["service"]))
if "join_security_groups" in filters:
query = query.options(orm.joinedload(models.Port.security_groups))

26
quark/tests/functional/mysql/test_ports.py
View File

@@ -3,9 +3,11 @@ import netaddr
import contextlib
from quark.db import api as db_api
from quark import exceptions
import quark.ipam
import quark.plugin
import quark.plugin_modules.ip_addresses as ip_api
import quark.plugin_modules.mac_address_ranges as macrng_api
import quark.plugin_modules.networks as network_api
import quark.plugin_modules.ports as port_api
@@ -67,3 +69,27 @@ class QuarkUpdatePorts(MySqlBaseFunctionalTest):
ip = "192.168.1.50"
port = port_api.update_port(self.context, id, _make_body(ip))
self.assertEqual(ip, port['fixed_ips'][0]['ip_address'])
class QuarkFindPorts(MySqlBaseFunctionalTest):
def test_ip_address_port_find_service(self):
net = db_api.network_create(self.context)
port = db_api.port_create(self.context, network_id=net["id"],
backend_key="", device_id="")
ip_address = db_api.ip_address_create(
self.context, address=netaddr.IPAddress("0.0.0.0"))
self.context.session.flush()
ip_address = db_api.port_associate_ip(self.context, [port], ip_address)
ip_address.set_service_for_port(port, "foobar")
self.context.session.flush()
ports = ip_api.get_ports_for_ip_address(
self.context, ip_address["id"],
filters={"service": "not-foobar"})
self.assertEqual(len(ports), 0)
ports = ip_api.get_ports_for_ip_address(
self.context, ip_address["id"],
filters={"service": "foobar"})
self.assertEqual(len(ports), 1)