Skip to content

Z
zodburi
Commit 5014cc8d authored by Tres Seaver's avatar Tres Seaver
Browse files
Options

Return to 100% coverage. 

Also, fix bugs uncovered thereby in processing float / tuple arguments.
parent 9e187359
Hide whitespace changes
Inline Side-by-side
Showing with 179 additions and 159 deletions
zodburi/datatypes.py
View file @ 5014cc8d
......@@ -36,3 +36,6 @@ def convert_int(value):
if value in TRUETYPES:
return 1
return int(value)
def convert_tuple(value):
return tuple(value.split(','))
zodburi/resolvers.py
View file @ 5014cc8d
......@@ -13,6 +13,7 @@ import ZConfig
from zodburi.datatypes import convert_bytesize
from zodburi.datatypes import convert_int
from zodburi.datatypes import convert_tuple
class Resolver(object):
......@@ -26,15 +27,14 @@ class Resolver(object):
unused = kw.copy()
new = {}
convert_string = lambda s: s
converters = (
convert_int,
convert_string,
convert_bytesize)
args = (
self._int_args,
self._string_args,
self._bytesize_args)
for convert, arg_names in zip(converters, args):
converters = [
(convert_int, self._int_args),
(convert_string, self._string_args),
(convert_bytesize, self._bytesize_args),
(float, self._float_args),
(convert_tuple, self._tuple_args),
]
for convert, arg_names in converters:
for arg_name in arg_names:
value = unused.pop(arg_name, None)
if value is not None:
......
zodburi/resolvers_relstorage.py
View file @ 5014cc8d
import cgi
import urlparse
from ZODB.DemoStorage import DemoStorage
from relstorage.adapters.postgresql import PostgreSQLAdapter
from relstorage.options import Options
from relstorage.storage import RelStorage
from .resolvers import Resolver
try:
from relstorage.adapters.postgresql import PostgreSQLAdapter
from relstorage.options import Options
from relstorage.storage import RelStorage
except ImportError: #pragma NO COVER
pass
else:
# Not a real resolver, but we use interpret_kwargs
class PostgreSQLAdapterHelper(Resolver):
_int_args = ('connect_timeout', )
_string_args = ('ssl_mode', )
# Not a real resolver, but we use interpret_kwargs
class PostgreSQLAdapterHelper(Resolver):
_int_args = ('connect_timeout', )
_string_args = ('ssl_mode', )
def __call__(self, parsed_uri, kw):
dsn_args = [
('dbname', parsed_uri.path[1:]),
('user', parsed_uri.username),
('password', parsed_uri.password),
('host', parsed_uri.hostname),
('port', parsed_uri.port)
]
def __call__(self, parsed_uri, kw):
dsn_args = [
('dbname', parsed_uri.path[1:]),
('user', parsed_uri.username),
('password', parsed_uri.password),
('host', parsed_uri.hostname),
('port', parsed_uri.port)
]
kw, unused = self.interpret_kwargs(kw)
dsn_args.extend(kw.items())
kw, unused = self.interpret_kwargs(kw)
dsn_args.extend(kw.items())
dsn = ' '.join("%s='%s'"%arg for arg in dsn_args)
dsn = ' '.join("%s='%s'"%arg for arg in dsn_args)
def factory(options):
return PostgreSQLAdapter(dsn=dsn, options=options)
return factory, unused
def factory(options):
return PostgreSQLAdapter(dsn=dsn, options=options)
return factory, unused
# The relstorage support is inspired by django-zodb.
# Oracle and mysql should be easily implementable from here
class RelStorageURIResolver(Resolver):
_int_args = ('poll_interval', 'cache_local_mb', 'commit_lock_timeout',
'commit_lock_id', 'read_only', 'shared_blob_dir',
'keep_history', 'pack_gc', 'pack_dry_run', 'strict_tpc',
'create', 'demostorage',)
_string_args = ('name', 'blob_dir', 'replica_conf',
'cache_module_name', 'cache_prefix',
'cache_delta_size_limit')
_bytesize_args = ('blob_cache_size', 'blob_cache_size_check',
'blob_cache_chunk_size')
_float_args = ('replica_timeout', 'pack_batch_timeout',
'pack_duty_cycle', 'pack_max_delay')
_tuple_args = ('cache_servers',)
# The relstorage support is inspired by django-zodb.
# Oracle and mysql should be easily implementable from here
class RelStorageURIResolver(Resolver):
_int_args = ('poll_interval', 'cache_local_mb', 'commit_lock_timeout',
'commit_lock_id', 'read_only', 'shared_blob_dir',
'keep_history', 'pack_gc', 'pack_dry_run', 'strict_tpc',
'create', 'demostorage',)
_string_args = ('name', 'blob_dir', 'replica_conf', 'cache_module_name',
'cache_prefix', 'cache_delta_size_limit')
_bytesize_args = ('blob_cache_size', 'blob_cache_size_check',
'blob_cache_chunk_size')
_float_args = ('replica_timeout', 'pack_batch_timeout', 'pack_duty_cycle',
'pack_max_delay')
_tuple_args = ('cache_servers',)
def __init__(self, adapter_helper):
self.adapter_helper = adapter_helper
def __init__(self, adapter_helper):
self.adapter_helper = adapter_helper
def __call__(self, uri):
uri = uri.replace('postgres://', 'http://', 1)
parsed_uri = urlparse.urlsplit(uri)
kw = dict(cgi.parse_qsl(parsed_uri.query))
def __call__(self, uri):
uri = uri.replace('postgres://', 'http://', 1)
parsed_uri = urlparse.urlsplit(uri)
kw = dict(cgi.parse_qsl(parsed_uri.query))
adapter_factory, kw = self.adapter_helper(parsed_uri, kw)
kw, unused = self.interpret_kwargs(kw)
adapter_factory, kw = self.adapter_helper(parsed_uri, kw)
kw, unused = self.interpret_kwargs(kw)
demostorage = kw.pop('demostorage', False)
options = Options(**kw)
demostorage = kw.pop('demostorage', False)
options = Options(**kw)
def factory():
adapter = adapter_factory(options)
storage = RelStorage(adapter=adapter, options=options)
if demostorage:
storage = DemoStorage(base=storage)
return storage
return factory, unused
def factory():
adapter = adapter_factory(options)
storage = RelStorage(adapter=adapter, options=options)
if demostorage:
storage = DemoStorage(base=storage)
return storage
return factory, unused
postgresql_resolver = RelStorageURIResolver(PostgreSQLAdapterHelper())
\ No newline at end of file
postgresql_resolver = RelStorageURIResolver(PostgreSQLAdapterHelper())
zodburi/tests/test_resolvers.py
View file @ 5014cc8d
......@@ -10,20 +10,19 @@ class Base:
def test_bytesize_args(self):
resolver = self._makeOne()
names = list(resolver._bytesize_args)
names = sorted(resolver._bytesize_args)
kwargs = {}
for name in names:
kwargs[name] = '10MB'
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(keys, names)
self.assertEqual(sorted(keys), names)
for name, value in args.items():
self.assertEqual(value, 10*1024*1024)
def test_int_args(self):
resolver = self._makeOne()
names = list(resolver._int_args)
names = sorted(resolver._int_args)
kwargs = {}
for name in names:
kwargs[name] = '10'
......@@ -36,40 +35,37 @@ class Base:
def test_string_args(self):
resolver = self._makeOne()
names = list(resolver._string_args)
names = sorted(resolver._string_args)
kwargs = {}
for name in names:
kwargs[name] = 'string'
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(keys, names)
self.assertEqual(sorted(keys), names)
for name, value in args.items():
self.assertEqual(value, 'string')
def test_float_args(self):
resolver = self._makeOne()
names = list(resolver._float_args)
names = sorted(resolver._float_args)
kwargs = {}
for name in names:
kwargs[name] = '3.14'
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(keys, names)
self.assertEqual(sorted(keys), names)
for name, value in args.items():
self.assertEqual(value, 3.14)
def test_tuple_args(self):
resolver = self._makeOne()
names = list(resolver._tuple_args)
names = sorted(resolver._tuple_args)
kwargs = {}
for name in names:
kwargs[name] = 'first,second,third'
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(keys, names)
self.assertEqual(sorted(keys), names)
for name, value in args.items():
self.assertEqual(value, ('first', 'second', 'third'))
......@@ -412,101 +408,118 @@ class TestMappingStorageURIResolver(Base, unittest.TestCase):
self.assertTrue(isinstance(storage, MappingStorage))
self.assertEqual(storage.__name__, 'storagename')
class TestPostgreSQLURIResolver(unittest.TestCase):
def _getTargetClass(self):
from zodburi.resolvers_relstorage import RelStorageURIResolver
return RelStorageURIResolver
def _makeOne(self):
from zodburi.resolvers_relstorage import PostgreSQLAdapterHelper
klass = self._getTargetClass()
return klass(PostgreSQLAdapterHelper())
def setUp(self):
# relstorage.options.Options is little more than a dict.
# We make it comparable to simplify the tests.
from relstorage.options import Options
Options.__eq__ = lambda s, o: vars(s) == vars(o)
def test_bool_args(self):
resolver = self._makeOne()
f = resolver.interpret_kwargs
kwargs = f({'read_only':'1'})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'true'})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'on'})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'off'})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'no'})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'false'})
self.assertEqual(kwargs[0], {'read_only':0})
@mock.patch('zodburi.resolvers_relstorage.PostgreSQLAdapter')
@mock.patch('zodburi.resolvers_relstorage.RelStorage')
def test_call(self, RelStorage, PostgreSQLAdapter):
from relstorage.options import Options
resolver = self._makeOne()
factory, dbkw = resolver('postgres://someuser:somepass@somehost:5432/somedb?read_only=1')
factory()
expected_options = Options(read_only=1)
PostgreSQLAdapter.assert_called_once_with(dsn="dbname='somedb' user='someuser' password='somepass' "
"host='somehost' port='5432'",
options=expected_options)
RelStorage.assert_called_once_with(adapter=PostgreSQLAdapter(), options=expected_options)
@mock.patch('zodburi.resolvers_relstorage.PostgreSQLAdapter')
@mock.patch('zodburi.resolvers_relstorage.RelStorage')
def test_call_adapter_options(self, RelStorage, PostgreSQLAdapter):
from relstorage.options import Options
resolver = self._makeOne()
factory, dbkw = resolver('postgres://someuser:somepass@somehost:5432/somedb?read_only=1'
'&connect_timeout=10')
factory()
expected_options = Options(read_only=1)
PostgreSQLAdapter.assert_called_once_with(dsn="dbname='somedb' user='someuser' password='somepass' "
"host='somehost' port='5432' connect_timeout='10'",
options=expected_options)
RelStorage.assert_called_once_with(adapter=PostgreSQLAdapter(), options=expected_options)
@mock.patch('zodburi.resolvers_relstorage.PostgreSQLAdapter')
@mock.patch('zodburi.resolvers_relstorage.RelStorage')
def test_invoke_factory_demostorage(self, RelStorage, PostgreSQLAdapter):
from ZODB.DemoStorage import DemoStorage
resolver = self._makeOne()
factory, dbkw = resolver('postgres://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?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'})
try:
from zodburi.resolvers_relstorage import RelStorageURIResolver
except ImportError: #pragma NO COVER
pass
else:
class TestPostgreSQLURIResolver(unittest.TestCase, Base):
def _getTargetClass(self):
from zodburi.resolvers_relstorage import RelStorageURIResolver
return RelStorageURIResolver
def _makeOne(self):
from zodburi.resolvers_relstorage import PostgreSQLAdapterHelper
klass = self._getTargetClass()
return klass(PostgreSQLAdapterHelper())
def setUp(self):
# relstorage.options.Options is little more than a dict.
# We make it comparable to simplify the tests.
from relstorage.options import Options
Options.__eq__ = lambda s, o: vars(s) == vars(o)
def test_bool_args(self):
resolver = self._makeOne()
f = resolver.interpret_kwargs
kwargs = f({'read_only':'1'})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'true'})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'on'})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'off'})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'no'})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'false'})
self.assertEqual(kwargs[0], {'read_only':0})
@mock.patch('zodburi.resolvers_relstorage.PostgreSQLAdapter')
@mock.patch('zodburi.resolvers_relstorage.RelStorage')
def test_call(self, RelStorage, PostgreSQLAdapter):
from relstorage.options import Options
resolver = self._makeOne()
factory, dbkw = resolver(
'postgres://someuser:somepass@somehost:5432/somedb'
'?read_only=1')
factory()
expected_options = Options(read_only=1)
PostgreSQLAdapter.assert_called_once_with(
dsn="dbname='somedb' user='someuser' password='somepass' "
"host='somehost' port='5432'", options=expected_options)
RelStorage.assert_called_once_with(
adapter=PostgreSQLAdapter(), options=expected_options)
@mock.patch('zodburi.resolvers_relstorage.PostgreSQLAdapter')
@mock.patch('zodburi.resolvers_relstorage.RelStorage')
def test_call_adapter_options(self, RelStorage, PostgreSQLAdapter):
from relstorage.options import Options
resolver = self._makeOne()
factory, dbkw = resolver(
'postgres://someuser:somepass@somehost:5432/somedb'
'?read_only=1&connect_timeout=10')
factory()
expected_options = Options(read_only=1)
PostgreSQLAdapter.assert_called_once_with(
dsn="dbname='somedb' user='someuser' password='somepass' "
"host='somehost' port='5432' connect_timeout='10'",
options=expected_options)
RelStorage.assert_called_once_with(
adapter=PostgreSQLAdapter(), options=expected_options)
@mock.patch('zodburi.resolvers_relstorage.PostgreSQLAdapter')
@mock.patch('zodburi.resolvers_relstorage.RelStorage')
def test_invoke_factory_demostorage(self, RelStorage, PostgreSQLAdapter):
from ZODB.DemoStorage import DemoStorage
resolver = self._makeOne()
factory, dbkw = resolver(
'postgres://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'
'?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 TestEntryPoints(unittest.TestCase):
def test_it(self):
from pkg_resources import load_entry_point
from zodburi import resolvers, resolvers_relstorage
from zodburi import resolvers
expected = [
('memory', resolvers.MappingStorageURIResolver),
('zeo', resolvers.ClientStorageURIResolver),
('file', resolvers.FileStorageURIResolver),
('zconfig', resolvers.ZConfigURIResolver),
('postgres', resolvers_relstorage.RelStorageURIResolver),
]
try:
from zodburi.resolvers_relstorage import RelStorageURIResolver
except ImportError: #pragma NO COVER
pass
else:
expected.append(
('postgres', RelStorageURIResolver))
for name, cls in expected:
target = load_entry_point('zodburi', 'zodburi.resolvers', name)
self.assertTrue(isinstance(target, cls))
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7