Merge "Run storage unit tests in pooled context"

marconi/queues/storage/sqlalchemy/pools.py

@@ -135,6 +135,6 @@ def _normalize(pool, detailed=False):
     }
     if detailed:
         opts = pool[3]
-        ret['options'] = utils.json_decode(opts) if opts else None
+        ret['options'] = utils.json_decode(opts) if opts else {}
 
     return ret

marconi/queues/storage/sqlalchemy/tables.py

@@ -68,6 +68,6 @@ Catalogue = sa.Table('Catalogue', metadata,
                      sa.Column('pool', sa.String(64),
                                sa.ForeignKey('Pools.name',
                                              ondelete='CASCADE')),
-                     sa.Column('project', sa.String(64), nullable=False),
+                     sa.Column('project', sa.String(64)),
                      sa.Column('queue', sa.String(64), nullable=False),
                      sa.UniqueConstraint('project', 'queue'))

marconi/tests/queues/storage/base.py

@@ -24,6 +24,7 @@ from testtools import matchers
 
 from marconi.openstack.common.cache import cache as oslo_cache
 from marconi.openstack.common import timeutils
+from marconi.queues import bootstrap
 from marconi.queues import storage
 from marconi.queues.storage import errors
 from marconi import tests as testing
@@ -52,13 +53,33 @@ class ControllerBaseTest(testing.TestBase):
         cache = oslo_cache.get_cache(self.conf.cache_url)
 
         # pylint: disable=not-callable
-        self.driver = self.driver_class(self.conf, cache)
+        self.conf.register_opts(bootstrap._GENERAL_OPTIONS)
+        pooling = 'pooling' in self.conf and self.conf.pooling
+        if pooling and not self.control_driver_class:
+            self.skipTest("Pooling is enabled, "
+                          "but control driver class is not specified")
+
+        if not pooling:
+            # pylint: disable=not-callable
+            self.driver = self.driver_class(self.conf, cache)
+        else:
+            # pylint: disable=not-callable
+            control = self.control_driver_class(self.conf, cache)
+            uri = "sqlite:///:memory:"
+            for i in range(4):
+                control.pools_controller.create(six.text_type(i), 100, uri)
+            self.driver = self.driver_class(self.conf, cache, control)
+
         self._prepare_conf()
 
         self.addCleanup(self._purge_databases)
 
-        # pylint: disable=not-callable
+        if not pooling:
-        self.controller = self.controller_class(self.driver)
+            # pylint: disable=not-callable
+            self.controller = self.controller_class(self.driver)
+        else:
+            # pylint: disable=not-callable
+            self.controller = self.controller_class(self.driver._pool_catalog)
 
     def _prepare_conf(self):
         """Prepare the conf before running tests

tests/unit/queues/storage/test_impl_mongodb.py

@@ -32,6 +32,7 @@ from marconi.queues.storage import mongodb
 from marconi.queues.storage.mongodb import controllers
 from marconi.queues.storage.mongodb import options
 from marconi.queues.storage.mongodb import utils
+from marconi.queues.storage import pooling
 from marconi import tests as testing
 from marconi.tests.queues.storage import base
 
@@ -403,3 +404,30 @@ class MongodbCatalogueTests(base.CatalogueControllerTest):
     def tearDown(self):
         self.controller.drop_all()
         super(MongodbCatalogueTests, self).tearDown()
+
+
+@testing.requires_mongodb
+class PooledMessageTests(base.MessageControllerTest):
+    config_file = 'wsgi_mongodb_pooled.conf'
+    controller_class = pooling.MessageController
+    driver_class = pooling.DataDriver
+    control_driver_class = mongodb.ControlDriver
+    controller_base_class = pooling.RoutingController
+
+
+@testing.requires_mongodb
+class PooledQueueTests(base.QueueControllerTest):
+    config_file = 'wsgi_mongodb_pooled.conf'
+    controller_class = pooling.QueueController
+    driver_class = pooling.DataDriver
+    control_driver_class = mongodb.ControlDriver
+    controller_base_class = pooling.RoutingController
+
+
+@testing.requires_mongodb
+class PooledClaimsTests(base.ClaimControllerTest):
+    config_file = 'wsgi_mongodb_pooled.conf'
+    controller_class = pooling.ClaimController
+    driver_class = pooling.DataDriver
+    control_driver_class = mongodb.ControlDriver
+    controller_base_class = pooling.RoutingController

tests/unit/queues/storage/test_impl_sqlalchemy.py

@@ -17,6 +17,7 @@ import datetime
 import sqlalchemy as sa
 
 from marconi.queues.storage import errors
+from marconi.queues.storage import pooling
 from marconi.queues.storage import sqlalchemy
 from marconi.queues.storage.sqlalchemy import controllers
 from marconi.queues.storage.sqlalchemy import tables
@@ -100,3 +101,27 @@ class SqlalchemyCatalogueTest(base.CatalogueControllerTest):
 
     def tearDown(self):
         super(SqlalchemyCatalogueTest, self).tearDown()
+
+
+class PooledMessageTests(base.MessageControllerTest):
+    config_file = 'wsgi_sqlalchemy_pooled.conf'
+    controller_class = pooling.MessageController
+    driver_class = pooling.DataDriver
+    control_driver_class = sqlalchemy.ControlDriver
+    controller_base_class = pooling.RoutingController
+
+
+class PooledClaimsTests(base.ClaimControllerTest):
+    config_file = 'wsgi_sqlalchemy_pooled.conf'
+    controller_class = pooling.ClaimController
+    driver_class = pooling.DataDriver
+    control_driver_class = sqlalchemy.ControlDriver
+    controller_base_class = pooling.RoutingController
+
+
+class PooledQueueTests(base.QueueControllerTest):
+    config_file = 'wsgi_sqlalchemy_pooled.conf'
+    controller_class = pooling.QueueController
+    driver_class = pooling.DataDriver
+    control_driver_class = sqlalchemy.ControlDriver
+    controller_base_class = pooling.RoutingController