Made it so shared=false works properly in noauth

RM8126

noauth was making it so shared=false + no tenant filter was returning shared
networks even though shared was supposed to be filtered out. The reason why
a sentinel value (INVERT_DEFAULTS) is used instead of an additional parameter
was due to it being the path of least resistance. There are intentions to
eventually move the shared attribute to the model.

quark/db/api.py

@@ -22,7 +22,7 @@ from neutron.openstack.common import timeutils
 from neutron.openstack.common import uuidutils
 from sqlalchemy import event
 from sqlalchemy import func as sql_func
-from sqlalchemy import and_, asc, orm, or_
+from sqlalchemy import and_, asc, orm, or_, not_
 
 from quark.db import models
 from quark import network_strategy
@@ -349,6 +349,9 @@ def mac_address_create(context, **mac_dict):
     return mac_address
 
 
+INVERT_DEFAULTS = 'invert_defaults'
+
+
 @scoped
 def network_find(context, fields=None, **filters):
     ids = []
@@ -361,13 +364,15 @@ def network_find(context, fields=None, **filters):
             filters.pop("id")
 
     if "shared" in filters:
+        defaults = STRATEGY.get_assignable_networks(context)
         if True in filters["shared"]:
-            defaults = STRATEGY.get_assignable_networks(context)
             if ids:
                 defaults = [net for net in ids if net in defaults]
                 filters.pop("id")
             if not defaults:
                 return []
+        else:
+            defaults.insert(0, INVERT_DEFAULTS)
         filters.pop("shared")
     return _network_find(context, fields, defaults=defaults, **filters)
 
@@ -377,10 +382,18 @@ def _network_find(context, fields, defaults=None, **filters):
     model_filters = _model_query(context, models.Network, filters, query)
 
     if defaults:
-        if filters:
+        invert_defaults = False
+        if INVERT_DEFAULTS in defaults:
+            invert_defaults = True
+            defaults.pop(0)
+        if filters and invert_defaults:
+            query = query.filter(and_(not_(models.Network.id.in_(defaults)),
+                                      and_(*model_filters)))
+        elif filters and not invert_defaults:
             query = query.filter(or_(models.Network.id.in_(defaults),
                                      and_(*model_filters)))
-        else:
+
+        elif not invert_defaults:
             query = query.filter(models.Network.id.in_(defaults))
     else:
         query = query.filter(*model_filters)

quark/tests/plugin_modules/test_networks.py

@@ -116,26 +116,27 @@ class TestQuarkGetNetworksShared(test_quark_plugin.TestQuarkPlugin):
             yield net_find
 
     def test_get_networks_shared(self):
-        net = dict(id=1, tenant_id=self.context.tenant_id, name="mynet",
-                   status="ACTIVE")
-        with self._stubs(nets=[net]) as net_find:
+        net1 = dict(id=1, tenant_id=self.context.tenant_id, name="mynet",
+                    status="ACTIVE")
+        with self._stubs(nets=[net1]) as net_find:
             self.plugin.get_networks(self.context, {"shared": [True]})
             net_find.assert_called_with(self.context, None,
                                         join_subnets=True,
                                         defaults=["public_network"])
 
     def test_get_networks_shared_false(self):
-        net = dict(id=1, tenant_id=self.context.tenant_id, name="mynet",
-                   status="ACTIVE")
-        with self._stubs(nets=[net]) as net_find:
+        net1 = dict(id=1, tenant_id=self.context.tenant_id, name="mynet",
+                    status="ACTIVE")
+        with self._stubs(nets=[net1]) as net_find:
+            invert = db_api.INVERT_DEFAULTS
             self.plugin.get_networks(self.context, {"shared": [False]})
             net_find.assert_called_with(self.context, None, join_subnets=True,
-                                        defaults=[])
+                                        defaults=[invert, "public_network"])
 
     def test_get_networks_no_shared(self):
-        net = dict(id=1, tenant_id=self.context.tenant_id, name="mynet",
-                   status="ACTIVE")
-        with self._stubs(nets=[net]) as net_find:
+        net1 = dict(id=1, tenant_id=self.context.tenant_id, name="mynet",
+                    status="ACTIVE")
+        with self._stubs(nets=[net1]) as net_find:
             self.plugin.get_networks(self.context, {})
             net_find.assert_called_with(self.context, None, join_subnets=True,
                                         defaults=[])

requirements.txt

@@ -1,6 +1,7 @@
 SQLAlchemy>=0.7.8,<=0.9.99
 alembic
 oslo.config>=1.2.0
+oslo.db
 zope.sqlalchemy
 mysql-python
 http://tarballs.openstack.org/neutron/neutron-master.tar.gz#egg=neutron