Skip to content

Z
zodburi
Commit 3833be7e authored by Chris Rossi's avatar Chris Rossi
Browse files
Options

Convert db resolvers to storage resolvers.

parent 085b1577
Hide whitespace changes
Inline Side-by-side
Showing with 185 additions and 321 deletions
setup.py
View file @ 3833be7e
......@@ -13,6 +13,7 @@ except:
CHANGES = ''
requires = ['ZODB3']
tests_require = requires + ['mock']
setup(name='zodburi',
version='0.1a1',
......@@ -31,7 +32,7 @@ setup(name='zodburi',
include_package_data=True,
zip_safe=False,
tests_require = requires,
install_requires = requires,
install_requires = tests_require,
test_suite="zodburi",
)
zodburi/datatypes.py
View file @ 3833be7e
......@@ -23,6 +23,16 @@ class SuffixMultiplier:
return int(v[:-self._keysz]) * m
return int(v) * self._default
byte_size = SuffixMultiplier({'kb': 1024,
convert_bytesize = SuffixMultiplier({'kb': 1024,
'mb': 1024*1024,
'gb': 1024*1024*1024L,})
def convert_int(value):
# boolean values are also treated as integers
value = value.lower()
if value in FALSETYPES:
return 0
if value in TRUETYPES:
return 1
return int(value)
zodburi/resolvers.py
View file @ 3833be7e
......@@ -3,10 +3,7 @@ import cgi
from cStringIO import StringIO
import urlparse
from zodburi.datatypes import byte_size
from zodburi.datatypes import FALSETYPES
from zodburi.datatypes import TRUETYPES
from ZEO.ClientStorage import ClientStorage
from ZODB.FileStorage.FileStorage import FileStorage
from ZODB.DemoStorage import DemoStorage
from ZODB.MappingStorage import MappingStorage
......@@ -14,58 +11,39 @@ from ZODB.blob import BlobStorage
from ZODB.DB import DB
import ZConfig
def interpret_int_args(argnames, kw):
newkw = {}
# boolean values are also treated as integers
for name in argnames:
value = kw.get(name)
if value is not None:
value = value.lower()
if value in FALSETYPES:
value = 0
if value in TRUETYPES:
value = 1
value = int(value)
newkw[name] = value
return newkw
def interpret_string_args(argnames, kw):
newkw = {}
# strings
for name in argnames:
value = kw.get(name)
if value is not None:
newkw[name] = value
return newkw
def interpret_bytesize_args(argnames, kw):
newkw = {}
# suffix multiplied
for name in argnames:
value = kw.get(name)
if value is not None:
newkw[name] = byte_size(value)
return newkw
from zodburi.datatypes import convert_bytesize
from zodburi.datatypes import convert_int
class Resolver(object):
_int_args = ()
_string_args = ()
_bytesize_args = ()
def interpret_kwargs(self, kw):
unused = kw.copy()
new = {}
newkw = interpret_int_args(self._int_args, kw)
new.update(newkw)
newkw = interpret_string_args(self._string_args, kw)
new.update(newkw)
newkw = interpret_bytesize_args(self._bytesize_args, kw)
new.update(newkw)
return 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):
for arg_name in arg_names:
value = unused.pop(arg_name, None)
if value is not None:
value = convert(value)
new[arg_name] = value
return new, unused
class MappingStorageURIResolver(Resolver):
_int_args = ('connection_cache_size', 'connection_pool_size')
_string_args = ('database_name',)
_bytesize_args = ()
def __call__(self, uri):
prefix, rest = uri.split('memory://', 1)
result = rest.split('?', 1)
......@@ -75,23 +53,19 @@ class MappingStorageURIResolver(Resolver):
else:
name, query = result
kw = dict(cgi.parse_qsl(query))
kw = self.interpret_kwargs(kw)
dbkw = get_dbkw(kw)
kw, unused = self.interpret_kwargs(kw)
args = (name,)
dbitems = dbkw.items()
dbitems.sort()
key = (args, tuple(dbitems))
def factory():
storage = MappingStorage(*args)
return DB(storage, **dbkw)
return key, args, kw, factory
return MappingStorage(*args)
return factory, unused
class FileStorageURIResolver(Resolver):
# XXX missing: blob_dir, packer, pack_keep_old, pack_gc, stop
_int_args = ('create', 'read_only', 'demostorage', 'connection_cache_size',
'connection_pool_size')
_string_args = ('blobstorage_dir', 'blobstorage_layout', 'database_name')
_int_args = ('create', 'read_only', 'demostorage')
_string_args = ('blobstorage_dir', 'blobstorage_layout')
_bytesize_args = ('quota',)
def __call__(self, uri):
# we can't use urlparse.urlsplit here due to Windows filenames
prefix, rest = uri.split('file://', 1)
......@@ -102,15 +76,9 @@ class FileStorageURIResolver(Resolver):
else:
path, query = result
path = os.path.normpath(path)
kw = dict(cgi.parse_qsl(query))
kw = self.interpret_kwargs(kw)
dbkw = get_dbkw(kw)
items = kw.items()
items.sort()
args = (path,)
dbitems = dbkw.items()
dbitems.sort()
key = (args, tuple(items), tuple(dbitems))
kw = dict(cgi.parse_qsl(query))
kw, unused = self.interpret_kwargs(kw)
demostorage = False
if 'demostorage'in kw:
......@@ -129,34 +97,30 @@ class FileStorageURIResolver(Resolver):
filestorage = FileStorage(*args, **kw)
blobstorage = BlobStorage(blobstorage_dir, filestorage,
layout=blobstorage_layout)
demostorage = DemoStorage(base=blobstorage)
return DB(demostorage, **dbkw)
return DemoStorage(base=blobstorage)
elif blobstorage_dir:
def factory():
filestorage = FileStorage(*args, **kw)
blobstorage = BlobStorage(blobstorage_dir, filestorage,
return BlobStorage(blobstorage_dir, filestorage,
layout=blobstorage_layout)
return DB(blobstorage, **dbkw)
elif demostorage:
def factory():
filestorage = FileStorage(*args, **kw)
demostorage = DemoStorage(base=filestorage)
return DB(demostorage, **dbkw)
return DemoStorage(base=filestorage)
else:
def factory():
filestorage = FileStorage(*args, **kw)
return DB(filestorage, **dbkw)
return FileStorage(*args, **kw)
return factory, unused
return key, args, kw, factory
class ClientStorageURIResolver(Resolver):
_int_args = ('debug', 'min_disconnect_poll', 'max_disconnect_poll',
'wait_for_server_on_startup', 'wait', 'wait_timeout',
'read_only', 'read_only_fallback', 'shared_blob_dir',
'demostorage', 'connection_cache_size',
'connection_pool_size')
'demostorage')
_string_args = ('storage', 'name', 'client', 'var', 'username',
'password', 'realm', 'blob_dir', 'database_name')
'password', 'realm', 'blob_dir')
_bytesize_args = ('cache_size', )
def __call__(self, uri):
......@@ -177,28 +141,15 @@ class ClientStorageURIResolver(Resolver):
path = os.path.normpath(path)
args = (path,)
kw = dict(cgi.parse_qsl(query))
kw = self.interpret_kwargs(kw)
dbkw = get_dbkw(kw)
items = kw.items()
items.sort()
dbitems = dbkw.items()
dbitems.sort()
key = (args, tuple(items), tuple(dbitems))
kw, unused = self.interpret_kwargs(kw)
if 'demostorage' in kw:
kw.pop('demostorage')
def factory():
from ZEO.ClientStorage import ClientStorage
from ZODB.DB import DB
from ZODB.DemoStorage import DemoStorage
demostorage = DemoStorage(base=ClientStorage(*args, **kw))
return DB(demostorage, **dbkw) #pragma NO COVERAGE
return DemoStorage(base=ClientStorage(*args, **kw))
else:
def factory():
from ZEO.ClientStorage import ClientStorage
from ZODB.DB import DB
clientstorage = ClientStorage(*args, **kw)
return DB(clientstorage, **dbkw) #pragma NO COVERAGE
return key, args, kw, factory
return ClientStorage(*args, **kw)
return factory, unused
def get_dbkw(kw):
dbkw = {}
......@@ -220,7 +171,7 @@ class ZConfigURIResolver(object):
schema_xml_template = """
<schema>
<import package="ZODB"/>
<multisection type="ZODB.database" attribute="databases" />
<multisection type="ZODB.storage" attribute="storages" />
</schema>
"""
......@@ -232,16 +183,16 @@ class ZConfigURIResolver(object):
schema_xml = self.schema_xml_template
schema = ZConfig.loadSchemaFile(StringIO(schema_xml))
config, handler = ZConfig.loadConfig(schema, path)
for database in config.databases:
for factory in config.storages:
if not frag:
# use the first defined in the file
break
elif frag == database.name:
elif frag == factory.name:
# match found
break
else:
raise KeyError("No database named %s found" % frag)
return (path, frag), (), {}, database.open
raise KeyError("No storage named %s found" % frag)
return factory.open, {}
RESOLVERS = {
......
zodburi/tests/test_resolvers.py
View file @ 3833be7e
import mock
import unittest
class Base:
......@@ -5,7 +6,7 @@ class Base:
def test_interpret_kwargs_noargs(self):
resolver = self._makeOne()
kwargs = resolver.interpret_kwargs({})
self.assertEqual(kwargs, {})
self.assertEqual(kwargs, ({}, {}))
def test_bytesize_args(self):
resolver = self._makeOne()
......@@ -13,7 +14,7 @@ class Base:
kwargs = {}
for name in names:
kwargs[name] = '10MB'
args = resolver.interpret_kwargs(kwargs)
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(keys, names)
......@@ -26,7 +27,7 @@ class Base:
kwargs = {}
for name in names:
kwargs[name] = '10'
args = resolver.interpret_kwargs(kwargs)
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(sorted(keys), sorted(names))
......@@ -39,7 +40,7 @@ class Base:
kwargs = {}
for name in names:
kwargs[name] = 'string'
args = resolver.interpret_kwargs(kwargs)
args = resolver.interpret_kwargs(kwargs)[0]
keys = args.keys()
keys.sort()
self.assertEqual(keys, names)
......@@ -68,96 +69,70 @@ class TestFileStorageURIResolver(Base, unittest.TestCase):
resolver = self._makeOne()
f = resolver.interpret_kwargs
kwargs = f({'read_only':'1'})
self.assertEqual(kwargs, {'read_only':1})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'true'})
self.assertEqual(kwargs, {'read_only':1})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'on'})
self.assertEqual(kwargs, {'read_only':1})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'off'})
self.assertEqual(kwargs, {'read_only':0})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'no'})
self.assertEqual(kwargs, {'read_only':0})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'false'})
self.assertEqual(kwargs, {'read_only':0})
self.assertEqual(kwargs[0], {'read_only':0})
def test_call_no_qs(self):
@mock.patch('zodburi.resolvers.FileStorage')
def test_call_no_qs(self, FileStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('file:///tmp/foo/bar')
self.assertEqual(args, ('/tmp/foo/bar',))
self.assertEqual(kw, {})
factory, dbkw = resolver('file:///tmp/foo/bar')
factory()
FileStorage.assert_called_once_with('/tmp/foo/bar')
def test_call_abspath(self):
@mock.patch('zodburi.resolvers.FileStorage')
def test_call_abspath(self, FileStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('file:///tmp/foo/bar?read_only=true')
self.assertEqual(args, ('/tmp/foo/bar',))
self.assertEqual(kw, {'read_only':1})
factory, dbkw = resolver('file:///tmp/foo/bar?read_only=true')
factory()
FileStorage.assert_called_once_with('/tmp/foo/bar', read_only=1)
def test_call_abspath_windows(self):
@mock.patch('zodburi.resolvers.FileStorage')
def test_call_abspath_windows(self, FileStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver(
factory, dbkw = resolver(
'file://C:\\foo\\bar?read_only=true')
self.assertEqual(args, ('C:\\foo\\bar',))
self.assertEqual(kw, {'read_only':1})
factory()
FileStorage.assert_called_once_with('C:\\foo\\bar', read_only=1)
def test_call_normpath(self):
@mock.patch('zodburi.resolvers.FileStorage')
def test_call_normpath(self, FileStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('file:///tmp/../foo/bar?read_only=true')
self.assertEqual(args, ('/foo/bar',))
self.assertEqual(kw, {'read_only':1})
factory, dbkw = resolver('file:///tmp/../foo/bar?read_only=true')
factory()
FileStorage.assert_called_once_with('/foo/bar', read_only=1)
def test_invoke_factory_filestorage(self):
import os
self.assertFalse(os.path.exists(os.path.join(self.tmpdir, 'db.db')))
resolver = self._makeOne()
k, args, kw, factory = resolver(
'file://%s/db.db?quota=200' % self.tmpdir)
self.assertEqual(k,
(('%s/db.db' % self.tmpdir,), (('quota', 200),),
(('cache_size', 10000), ('database_name', 'unnamed'),
('pool_size', 7)))
)
factory()
factory, dbkw = resolver('file://%s/db.db?quota=200' % self.tmpdir)
storage = factory()
from ZODB.FileStorage import FileStorage
self.assertTrue(isinstance(storage, FileStorage))
self.assertEqual(storage._quota, 200)
self.assertTrue(os.path.exists(os.path.join(self.tmpdir, 'db.db')))
def test_demostorage(self):
resolver = self._makeOne()
k, args, kw, factory = resolver(
'file:///tmp/../foo/bar?demostorage=true')
self.assertEqual(args, ('/foo/bar',))
self.assertEqual(kw, {})
def test_invoke_factory_demostorage(self):
import os
from ZODB.DemoStorage import DemoStorage
from ZODB.FileStorage import FileStorage
DB_FILE = os.path.join(self.tmpdir, 'db.db')
self.assertFalse(os.path.exists(DB_FILE))
resolver = self._makeOne()
k, args, kw, factory = resolver(
'file://%s/db.db?quota=200&demostorage=true' % self.tmpdir)
self.assertEqual(k,
(('%s/db.db' % self.tmpdir,),
(('demostorage', 1),
('quota', 200),
),
(('cache_size', 10000),
('database_name', 'unnamed'),
('pool_size', 7)
),
)
)
db = factory()
self.assertTrue(isinstance(db._storage, DemoStorage))
self.assertTrue(isinstance(db._storage.base, FileStorage))
self.assertTrue(os.path.exists(DB_FILE))
def test_blobstorage(self):
resolver = self._makeOne()
k, args, kw, factory = resolver(
('file:///tmp/../foo/bar'
'?blobstorage_dir=/foo/bar&blobstorage_layout=bushy'))
self.assertEqual(args, ('/foo/bar',))
self.assertEqual(kw, {})
factory, dbkw = resolver(
'file://%s/db.db?demostorage=true' % self.tmpdir)
storage = factory()
self.assertTrue(isinstance(storage, DemoStorage))
self.assertTrue(isinstance(storage.base, FileStorage))
self.assertEqual(dbkw, {})
self.assertTrue(os.path.exists(os.path.join(self.tmpdir, 'db.db')))
def test_invoke_factory_blobstorage(self):
import os
......@@ -167,35 +142,15 @@ class TestFileStorageURIResolver(Base, unittest.TestCase):
BLOB_DIR = os.path.join(self.tmpdir, 'blob')
self.assertFalse(os.path.exists(DB_FILE))
resolver = self._makeOne()
k, args, kw, factory = resolver(
factory, dbkw = resolver(
'file://%s/db.db?quota=200'
'&blobstorage_dir=%s/blob'
'&blobstorage_layout=bushy' % (self.tmpdir, q(self.tmpdir)))
self.assertEqual(k,
(('%s/db.db' % self.tmpdir,),
(('blobstorage_dir', '%s/blob' % self.tmpdir),
('blobstorage_layout', 'bushy'),
('quota', 200),
),
(('cache_size', 10000),
('database_name', 'unnamed'),
('pool_size', 7)
),
)
)
db = factory()
self.assertTrue(isinstance(db._storage, BlobStorage))
storage = factory()
self.assertTrue(isinstance(storage, BlobStorage))
self.assertTrue(os.path.exists(DB_FILE))
self.assertTrue(os.path.exists(BLOB_DIR))
def test_blobstorage_and_demostorage(self):
resolver = self._makeOne()
k, args, kw, factory = resolver(
('file:///tmp/../foo/bar?demostorage=true'
'&blobstorage_dir=/foo/bar&blobstorage_layout=bushy'))
self.assertEqual(args, ('/foo/bar',))
self.assertEqual(kw, {})
def test_invoke_factory_blobstorage_and_demostorage(self):
import os
from urllib import quote as q
......@@ -204,36 +159,23 @@ class TestFileStorageURIResolver(Base, unittest.TestCase):
BLOB_DIR = os.path.join(self.tmpdir, 'blob')
self.assertFalse(os.path.exists(DB_FILE))
resolver = self._makeOne()
k, args, kw, factory = resolver(
factory, dbkw = resolver(
'file://%s/db.db?quota=200&demostorage=true'
'&blobstorage_dir=%s/blob'
'&blobstorage_layout=bushy' % (self.tmpdir, q(self.tmpdir)))
self.assertEqual(k,
(('%s/db.db' % self.tmpdir,),
(('blobstorage_dir', '%s/blob' % self.tmpdir),
('blobstorage_layout', 'bushy'),
('demostorage', 1),
('quota', 200),
),
(('cache_size', 10000),
('database_name', 'unnamed'),
('pool_size', 7)
),
)
)
db = factory()
self.assertTrue(isinstance(db._storage, DemoStorage))
storage = factory()
self.assertTrue(isinstance(storage, DemoStorage))
self.assertTrue(os.path.exists(DB_FILE))
self.assertTrue(os.path.exists(BLOB_DIR))
def test_dbargs(self):
resolver = self._makeOne()
k, args, kw, factory = resolver(
factory, dbkw = resolver(
('file:///tmp/../foo/bar?connection_pool_size=1'
'&connection_cache_size=1&database_name=dbname'))
self.assertEqual(k[2],
(('cache_size', 1), ('database_name', 'dbname'),
('pool_size', 1)))
self.assertEqual(dbkw, {'connection_cache_size': '1',
'connection_pool_size': '1',
'database_name': 'dbname'})
class TestClientStorageURIResolver(unittest.TestCase):
......@@ -249,90 +191,62 @@ class TestClientStorageURIResolver(unittest.TestCase):
resolver = self._makeOne()
f = resolver.interpret_kwargs
kwargs = f({'read_only':'1'})
self.assertEqual(kwargs, {'read_only':1})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'true'})
self.assertEqual(kwargs, {'read_only':1})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'on'})
self.assertEqual(kwargs, {'read_only':1})
self.assertEqual(kwargs[0], {'read_only':1})
kwargs = f({'read_only':'off'})
self.assertEqual(kwargs, {'read_only':0})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'no'})
self.assertEqual(kwargs, {'read_only':0})
self.assertEqual(kwargs[0], {'read_only':0})
kwargs = f({'read_only':'false'})
self.assertEqual(kwargs, {'read_only':0})
self.assertEqual(kwargs[0], {'read_only':0})
def test_call_tcp_no_port(self):
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo://localhost?debug=true')
self.assertEqual(args, (('localhost', 9991),))
self.assertEqual(kw, {'debug':1})
self.assertEqual(k,
((('localhost', 9991),), (('debug', 1),),
(('cache_size', 10000), ('database_name','unnamed'),
('pool_size', 7))))
def test_call_tcp(self):
@mock.patch('zodburi.resolvers.ClientStorage')
def test_call_tcp_no_port(self, ClientStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo://localhost:8080?debug=true')
self.assertEqual(args, (('localhost', 8080),))
self.assertEqual(kw, {'debug':1})
self.assertEqual(k,
((('localhost', 8080),), (('debug', 1),),
(('cache_size', 10000), ('database_name','unnamed'),
('pool_size', 7))))
def test_call_unix(self):
factory, dbkw = resolver('zeo://localhost?debug=true')
factory()
ClientStorage.assert_called_once_with(('localhost', 9991), debug=1)
@mock.patch('zodburi.resolvers.ClientStorage')
def test_call_tcp(self, ClientStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo:///var/sock?debug=true')
self.assertEqual(args, ('/var/sock',))
self.assertEqual(kw, {'debug':1})
self.assertEqual(k,
(('/var/sock',),
(('debug', 1),),
(('cache_size', 10000),
('database_name', 'unnamed'),
('pool_size', 7))))
def test_invoke_factory(self):
factory, dbkw = resolver('zeo://localhost:8080?debug=true')
factory()
ClientStorage.assert_called_once_with(('localhost', 8080), debug=1)
@mock.patch('zodburi.resolvers.ClientStorage')
def test_call_unix(self, ClientStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo:///var/nosuchfile?wait=false')
self.assertEqual(k, (('/var/nosuchfile',),
(('wait', 0),),
(('cache_size', 10000),
('database_name', 'unnamed'), ('pool_size', 7))))
from ZEO.ClientStorage import ClientDisconnected
self.assertRaises(ClientDisconnected, factory)
def test_demostorage(self):
factory, dbkw = resolver('zeo:///var/sock?debug=true')
factory()
ClientStorage.assert_called_once_with('/var/sock', debug=1)
@mock.patch('zodburi.resolvers.ClientStorage')
def test_invoke_factory(self, ClientStorage):
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo:///var/sock?demostorage=true')
self.assertEqual(args, ('/var/sock',))
self.assertEqual(kw, {})
factory, dbkw = resolver('zeo:///var/nosuchfile?wait=false')
factory()
ClientStorage.assert_called_once_with('/var/nosuchfile', wait=0)
def test_invoke_factory_demostorage(self):
from ZODB.DemoStorage import DemoStorage
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo:///var/nosuchfile?wait=false'
factory, dbkw = resolver('zeo:///var/nosuchfile?wait=false'
'&demostorage=true')
self.assertEqual(k, (('/var/nosuchfile',),
(('demostorage', 1),
('wait', 0),),
(('cache_size', 10000),
('database_name', 'unnamed'),
('pool_size', 7),
),
))
from ZEO.ClientStorage import ClientDisconnected
self.assertRaises(ClientDisconnected, factory)
self.assertTrue(isinstance(factory(), DemoStorage))
def test_dbargs(self):
resolver = self._makeOne()
k, args, kw, factory = resolver('zeo://localhost:8080?debug=true&'
factory, dbkw = resolver('zeo://localhost:8080?debug=true&'
'connection_pool_size=1&'
'connection_cache_size=1&'
'database_name=dbname')
self.assertEqual(k[2],
(('cache_size', 1), ('database_name', 'dbname'),
('pool_size', 1)))
self.assertEqual(dbkw, {'connection_pool_size': '1',
'connection_cache_size': '1',
'database_name': 'dbname'})
class TestZConfigURIResolver(unittest.TestCase):
......@@ -353,48 +267,38 @@ class TestZConfigURIResolver(unittest.TestCase):
def test_named_database(self):
self.tmp.write("""
<zodb otherdb>
<demostorage>
</demostorage>
</zodb>
<zodb demodb>
<mappingstorage>
</mappingstorage>
</zodb>
<demostorage foo>
</demostorage>
<mappingstorage bar>
</mappingstorage>
""")
self.tmp.flush()
resolver = self._makeOne()
k, args, kw, factory = resolver('zconfig://%s#demodb' % self.tmp.name)
db = factory()
factory, dbkw = resolver('zconfig://%s#bar' % self.tmp.name)
storage = factory()
from ZODB.MappingStorage import MappingStorage
self.assertTrue(isinstance(db._storage, MappingStorage))
self.assertTrue(isinstance(storage, MappingStorage), storage)
def test_anonymous_database(self):
self.tmp.write("""
<zodb>
<mappingstorage>
</mappingstorage>
</zodb>
<zodb demodb>
<mappingstorage>
</mappingstorage>
</zodb>
<mappingstorage>
</mappingstorage>
<demostorage demo>
</demostorage>
""")
self.tmp.flush()
resolver = self._makeOne()
k, args, kw, factory = resolver('zconfig://%s' % self.tmp.name)
db = factory()
factory, dbkw = resolver('zconfig://%s' % self.tmp.name)
storage = factory()
from ZODB.MappingStorage import MappingStorage
self.assertTrue(isinstance(db._storage, MappingStorage))
self.assertTrue(isinstance(storage, MappingStorage))
def test_database_not_found(self):
self.tmp.write("""
<zodb x>
<mappingstorage>
</mappingstorage>
</zodb>
<mappingstorage x>
</mappingstorage>
""")
self.tmp.flush()
resolver = self._makeOne()
......@@ -412,22 +316,20 @@ class TestMappingStorageURIResolver(Base, unittest.TestCase):
def test_call_no_qs(self):
resolver = self._makeOne()
k, args, kw, factory = resolver('memory://')
self.assertEqual(args, ('',))
self.assertEqual(kw, {})
db = factory()
factory, dbkw = resolver('memory://')
self.assertEqual(dbkw, {})
storage = factory()
from ZODB.MappingStorage import MappingStorage
self.assertTrue(isinstance(db._storage, MappingStorage))
self.assertTrue(isinstance(storage, MappingStorage))
self.assertEqual(storage.__name__, '')
def test_call_with_qs(self):
uri='memory://storagename?connection_cache_size=100&database_name=fleeb'
resolver = self._makeOne()
k, args, kw, factory = resolver(uri)
self.assertEqual(args, ('storagename',))
self.assertEqual(kw, {})
self.assertEqual(k, (('storagename',),
(('cache_size', 100), ('database_name', 'fleeb'),
('pool_size', 7))))
db = factory()
factory, dbkw = resolver(uri)
self.assertEqual(dbkw, {'connection_cache_size': '100',
'database_name': 'fleeb'})
storage = factory()
from ZODB.MappingStorage import MappingStorage
self.assertTrue(isinstance(db._storage, MappingStorage))
\ No newline at end of file
self.assertTrue(isinstance(storage, MappingStorage))
self.assertEqual(storage.__name__, 'storagename')
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