Fix #41. Add MySQL URI resolver test and expand tested options.

relstorage/options.py

@@ -78,3 +78,6 @@ def __init__(self, **kwoptions):
                 setattr(self, key, value)
             else:
                 raise TypeError("Unknown parameter: %s" % key)
+
+    def __repr__(self):
+        return 'relstorage.options.Options(**' + repr(self.__dict__) + ')'

relstorage/tests/test_zodburi.py

@@ -1,13 +1,22 @@
+from __future__ import print_function, absolute_import
+
 import mock
 import unittest
 from pkg_resources import DistributionNotFound
+from relstorage.zodburi_resolver import RelStorageURIResolver
+from relstorage.zodburi_resolver import SuffixMultiplier
+
+class AbstractURIResolverTestBase(unittest.TestCase):
 
+    adapter_name = None
+    prefix = ''
 
-class TestPostgreSQLURIResolver(unittest.TestCase):
+    def _get_helper(self):
+        raise NotImplementedError()
 
     def _makeOne(self):
-        from relstorage.zodburi_resolver import RelStorageURIResolver, PostgreSQLAdapterHelper
-        return RelStorageURIResolver(PostgreSQLAdapterHelper())
+        helper = self._get_helper()
+        return RelStorageURIResolver(helper())
 
     def setUp(self):
         # relstorage.options.Options is little more than a dict.
@@ -16,10 +25,10 @@ def setUp(self):
         Options.__eq__ = lambda s, o: vars(s) == vars(o)
 
         self.patcher1 = mock.patch('relstorage.zodburi_resolver.RelStorage')
-        self.patcher2 = mock.patch('relstorage.adapters.postgresql.PostgreSQLAdapter')
+        self.patcher2 = mock.patch(self.adapter_name)
         self.RelStorage = self.patcher1.start()
         try:
-            self.PostgreSQLAdapter = self.patcher2.start()
+            self.DBAdapter = self.patcher2.start()
         except ImportError as e:
             raise unittest.SkipTest(str(e))
 
@@ -43,61 +52,102 @@ def test_bool_args(self):
         kwargs = f({'read_only':'false'})
         self.assertEqual(kwargs[0], {'read_only':0})
 
+    def _format_db(self, dbname='somedb', user='someuser', password='somepass',
+                   host='somehost', port='5432', **kwargs):
+        raise NotImplementedError()
+
     def test_call(self):
         from relstorage.options import Options
         resolver = self._makeOne()
-        factory, dbkw = resolver(
-            'postgres://someuser:somepass@somehost:5432/somedb'
-            '?read_only=1')
+        factory, _dbkw = resolver(
+            self.prefix + '://someuser:somepass@somehost:5432/somedb'
+            '?read_only=1&cache_servers=123,456')
         factory()
 
-        expected_options = Options(read_only=1)
-        self.PostgreSQLAdapter.assert_called_once_with(
-            dsn="dbname='somedb' user='someuser' password='somepass' "
-                "host='somehost' port='5432'", options=expected_options)
+        expected_options = Options(read_only=1, cache_servers=('123', '456'))
+        self.DBAdapter.assert_called_once_with(
+            options=expected_options, **self._format_db())
         self.RelStorage.assert_called_once_with(
-            adapter=self.PostgreSQLAdapter(), options=expected_options)
+            adapter=self.DBAdapter(), options=expected_options)
 
     def test_call_adapter_options(self):
         from relstorage.options import Options
         resolver = self._makeOne()
-        factory, dbkw = resolver(
-            'postgres://someuser:somepass@somehost:5432/somedb'
+        factory, _dbkw = resolver(
+            self.prefix + '://someuser:somepass@somehost:5432/somedb'
             '?read_only=1&connect_timeout=10')
         factory()
 
         expected_options = Options(read_only=1)
-        self.PostgreSQLAdapter.assert_called_once_with(
-            dsn="dbname='somedb' user='someuser' password='somepass' "
-                "host='somehost' port='5432' connect_timeout='10'",
-            options=expected_options)
+        self.DBAdapter.assert_called_once_with(
+            options=expected_options,
+            **self._format_db(connect_timeout=10))
         self.RelStorage.assert_called_once_with(
-            adapter=self.PostgreSQLAdapter(), options=expected_options)
+            adapter=self.DBAdapter(), options=expected_options)
 
     def test_invoke_factory_demostorage(self):
         from ZODB.DemoStorage import DemoStorage
         resolver = self._makeOne()
-        factory, dbkw = resolver(
-            'postgres://someuser:somepass@somehost:5432/somedb'
+        factory, _dbkw = resolver(
+            self.prefix + '://someuser:somepass@somehost:5432/somedb'
             '?read_only=1&demostorage=true')
         self.assertTrue(isinstance(factory(), DemoStorage))
 
     def test_dbargs(self):
         resolver = self._makeOne()
-        factory, dbkw = resolver(
-            'postgres://someuser:somepass@somehost:5432/somedb'
+        _factory, dbkw = resolver(
+            self.prefix + '://someuser:somepass@somehost:5432/somedb'
             '?read_only=1&connection_pool_size=1&connection_cache_size=1'
             '&database_name=dbname')
         self.assertEqual(dbkw, {'connection_pool_size': '1',
                                 'connection_cache_size': '1',
                                 'database_name': 'dbname'})
 
+class TestPostgreSQLURIResolver(AbstractURIResolverTestBase):
+    adapter_name = 'relstorage.adapters.postgresql.PostgreSQLAdapter'
+    prefix = 'postgres'
+
+    def _get_helper(self):
+        from relstorage.zodburi_resolver import PostgreSQLAdapterHelper
+        return PostgreSQLAdapterHelper
+
+
+    def _format_db(self, dbname='somedb', user='someuser', password='somepass',
+                   host='somehost', port='5432', **kwargs):
+        dsn = ("dbname='%s' user='%s' password='%s' host='%s' port='%s'"
+               % (dbname, user, password, host, port))
+        if 'connect_timeout' in kwargs:
+            dsn += " connect_timeout='%s'" % kwargs['connect_timeout']
+        else:
+            assert not kwargs
+
+        return {'dsn': dsn}
+
+
+class TestMySQLURIResolver(AbstractURIResolverTestBase):
+    adapter_name = 'relstorage.adapters.mysql.MySQLAdapter'
+    prefix = 'postgres'
+
+    def _get_helper(self):
+        from relstorage.zodburi_resolver import MySQLAdapterHelper
+        return MySQLAdapterHelper
+
+    def _format_db(self, dbname='somedb', user='someuser', password='somepass',
+                   host='somehost', port='5432', **kwargs):
+        args = dict(locals())
+        args.update(kwargs)
+        args['db'] = args['dbname']; del args['dbname']
+        args['passwd'] = args['password']; del args['password']
+        args['port'] = int(args['port'])
+
+        del args['kwargs']
+        del args['self']
+        return args
 
 class TestEntryPoints(unittest.TestCase):
     def test_it(self):
         from pkg_resources import load_entry_point
         from relstorage.zodburi_resolver import (
-            RelStorageURIResolver,
             PostgreSQLAdapterHelper,
             MySQLAdapterHelper,
             OracleAdapterHelper
@@ -119,10 +169,28 @@ def test_it(self):
                 warnings.warn('%s not found, skipping the zodburi test for %s'%
                               (e.args[0], name))
 
+class TestSuffixMultiplier(unittest.TestCase):
+
+    def test_call_bytesize(self):
+        from relstorage.zodburi_resolver import convert_bytesize
+        self.assertEqual(1024, convert_bytesize('1kb'))
+        self.assertEqual(1024, convert_bytesize('1Kb'))
+
+        self.assertEqual(1024*1024, convert_bytesize('1Mb'))
+        self.assertEqual(1024*1024*6, convert_bytesize('6MB'))
+
+        self.assertEqual(42, convert_bytesize('42'))
+
+    def test_bad_size(self):
+        self.assertRaises(ValueError, SuffixMultiplier, {'ab': 1, 'bc': 2, 'def': 3})
+        self.assertRaises(ValueError, SuffixMultiplier, {'ab': 1, 'def': 3})
+
 def test_suite():
     suite = unittest.TestSuite()
     suite.addTest(unittest.makeSuite(TestPostgreSQLURIResolver))
+    suite.addTest(unittest.makeSuite(TestMySQLURIResolver))
     suite.addTest(unittest.makeSuite(TestEntryPoints))
+    suite.addTest(unittest.makeSuite(TestSuffixMultiplier))
     return suite
 
 if __name__ == '__main__':

relstorage/zodburi_resolver.py

@@ -1,11 +1,14 @@
 from __future__ import print_function, absolute_import
 
-import cgi
+
 try:
     from urllib import parse as urlparse
+    from urllib.parse import parse_qsl
+    from functools import reduce
 except ImportError:
     # Py2
     import urlparse
+    from cgi import parse_qsl
 from ZODB.DemoStorage import DemoStorage
 
 from relstorage.options import Options
@@ -25,16 +28,15 @@ class SuffixMultiplier(object):
     # match, default is the multiplier.  Matches are case insensitive.  Return
     # values are in the fundamental unit.
     def __init__(self, d, default=1):
-        self._d = d
+        self._d = dict(d)
         self._default = default
         # all keys must be the same size
         self._keysz = None
-        for k in d.keys():
-            if self._keysz is None:
-                self._keysz = len(k)
-            else:
-                if self._keysz != len(k):
-                    raise ValueError('suffix length missmatch')
+        def check(a, b):
+            if len(a) != len(b):
+                raise ValueError("suffix length mismatch")
+            return a
+        self._keysz = len(reduce(check, d))
 
     def __call__(self, v):
         v = v.lower()
@@ -176,7 +178,7 @@ def __call__(self, uri):
         uri = uri.replace('mysql://', 'http://', 1)
         uri = uri.replace('oracle://', 'http://', 1)
         parsed_uri = urlparse.urlsplit(uri)
-        kw = dict(cgi.parse_qsl(parsed_uri.query))
+        kw = dict(parse_qsl(parsed_uri.query))
 
         adapter_factory, kw = self.adapter_helper(parsed_uri, kw)
         kw, unused = self.interpret_kwargs(kw)