Commit 0d35890d authored by Jim Fulton's avatar Jim Fulton

Plumbed the record_iternext API through ZEO.

I'm not particularly proud of the tests I wrote.  It's way to hard to
write tests for ZEO.  Without a lot of white-box trickery, nothing can
be tested without making network connections, which is insane. ZEO's 
architecture needs to change.
parent 626961a8
...@@ -728,6 +728,13 @@ class ClientStorage(object): ...@@ -728,6 +728,13 @@ class ClientStorage(object):
""" """
return self._server.history(oid, version, length) return self._server.history(oid, version, length)
def record_iternext(self, next=None):
"""Storage API: get the mext database record.
This is part of the conversion-support API.
"""
return self._server.record_iternext(next)
def getSerial(self, oid): def getSerial(self, oid):
"""Storage API: return current serial number for oid.""" """Storage API: return current serial number for oid."""
return self._server.getSerial(oid) return self._server.getSerial(oid)
......
...@@ -252,6 +252,9 @@ class StorageServer: ...@@ -252,6 +252,9 @@ class StorageServer:
else: else:
return self.rpc.call('history', oid, version, length) return self.rpc.call('history', oid, version, length)
def record_iternext(self, next):
return self.rpc.call('record_iternext', next)
def load(self, oid, version): def load(self, oid, version):
return self.rpc.call('load', oid, version) return self.rpc.call('load', oid, version)
......
...@@ -151,6 +151,11 @@ class ZEOStorage: ...@@ -151,6 +151,11 @@ class ZEOStorage:
self.load = self.storage.load self.load = self.storage.load
self.loadSerial = self.storage.loadSerial self.loadSerial = self.storage.loadSerial
self.modifiedInVersion = self.storage.modifiedInVersion self.modifiedInVersion = self.storage.modifiedInVersion
record_iternext = getattr(self.storage, 'record_iternext', None)
if record_iternext is not None:
self.record_iternext = record_iternext
try: try:
fn = self.storage.getExtensionMethods fn = self.storage.getExtensionMethods
except AttributeError: except AttributeError:
...@@ -233,6 +238,7 @@ class ZEOStorage: ...@@ -233,6 +238,7 @@ class ZEOStorage:
'supportsUndo': self.storage.supportsUndo(), 'supportsUndo': self.storage.supportsUndo(),
'supportsVersions': self.storage.supportsVersions(), 'supportsVersions': self.storage.supportsVersions(),
'extensionMethods': self.getExtensionMethods(), 'extensionMethods': self.getExtensionMethods(),
'supports_record_iternext': hasattr(self, 'record_iternext'),
} }
def get_size_info(self): def get_size_info(self):
...@@ -628,6 +634,7 @@ class ZEOStorage: ...@@ -628,6 +634,7 @@ class ZEOStorage:
else: else:
return 1 return 1
class StorageServer: class StorageServer:
"""The server side implementation of ZEO. """The server side implementation of ZEO.
......
##############################################################################
#
# Copyright (c) 2006 Zope Corporation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.0 (ZPL). A copy of the ZPL should accompany this distribution.
# THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
# WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
# WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
# FOR A PARTICULAR PURPOSE.
#
##############################################################################
import unittest
from zope.testing import doctest
class FakeStorageBase:
def __getattr__(self, name):
if name in ('versionEmpty', 'versions', 'getSerial',
'history', 'load', 'loadSerial', 'modifiedInVersion',
'lastTransaction', 'getSize', 'getName', 'supportsUndo',
'supportsVersions'):
return lambda *a, **k: None
raise AttributeError(name)
def isReadOnly(self):
return False
def __len__(self):
return 4
class FakeStorage(FakeStorageBase):
def record_iternext(self, next=None):
if next == None:
next = '0'
next = str(int(next) + 1)
oid = next
if next == '4':
next = None
return oid, oid*8, 'data ' + oid, next
class FakeServer:
storages = {
'1': FakeStorage(),
'2': FakeStorageBase(),
}
def register_connection(*args):
return None, None
def test_server_record_iternext():
"""
On the server, record_iternext calls are simply delegated to the
underlying storage.
>>> import ZEO.StorageServer
>>> zeo = ZEO.StorageServer.ZEOStorage(FakeServer(), False)
>>> zeo.register('1', False)
>>> next = None
>>> while 1:
... oid, serial, data, next = zeo.record_iternext(next)
... print oid
... if next is None:
... break
1
2
3
4
The storage info also reflects the fact that record_iternext is supported.
>>> zeo.get_info()['supports_record_iternext']
True
>>> zeo = ZEO.StorageServer.ZEOStorage(FakeServer(), False)
>>> zeo.register('2', False)
>>> zeo.get_info()['supports_record_iternext']
False
"""
def test_client_record_iternext():
"""\
The client simply delegates record_iternext calls to it's server stub.
There's really no decent way to test ZEO without running to muc crazy
stuff. I'd rather to a lame test than a really lame test, so here goes.
First, fake out the connection manager so we can make a connection:
>>> import ZEO.ClientStorage
>>> from ZEO.ClientStorage import ClientStorage
>>> oldConnectionManagerClass = ClientStorage.ConnectionManagerClass
>>> class FauxConnectionManagerClass:
... def __init__(*a, **k):
... pass
... def attempt_connect(self):
... return True
>>> ClientStorage.ConnectionManagerClass = FauxConnectionManagerClass
>>> client = ClientStorage('', wait=False)
>>> ClientStorage.ConnectionManagerClass = oldConnectionManagerClass
Now we'll have our way with it's provate _server attr:
>>> client._server = FakeStorage()
>>> next = None
>>> while 1:
... oid, serial, data, next = client.record_iternext(next)
... print oid
... if next is None:
... break
1
2
3
4
"""
def test_server_stub_record_iternext():
"""\
The server stub simply delegates record_iternext calls to it's rpc.
There's really no decent way to test ZEO without running to muc crazy
stuff. I'd rather to a lame test than a really lame test, so here goes.
>>> class FauxRPC:
... storage = FakeStorage()
... def call(self, meth, *args):
... return getattr(self.storage, meth)(*args)
... peer_protocol_version = 1
>>> import ZEO.ServerStub
>>> stub = ZEO.ServerStub.StorageServer(FauxRPC())
>>> next = None
>>> while 1:
... oid, serial, data, next = stub.record_iternext(next)
... print oid
... if next is None:
... break
1
2
3
4
"""
def test_suite():
return doctest.DocTestSuite()
if __name__ == '__main__':
unittest.main(defaultTest='test_suite')
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