[Zodb-checkins] SVN: ZODB/trunk/src/ZEO/ Plumbed the
record_iternext API through ZEO.
Jim Fulton
jim at zope.com
Thu May 4 14:30:21 EDT 2006
Log message for revision 67970:
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.
Changed:
U ZODB/trunk/src/ZEO/ClientStorage.py
U ZODB/trunk/src/ZEO/ServerStub.py
U ZODB/trunk/src/ZEO/StorageServer.py
A ZODB/trunk/src/ZEO/tests/testConversionSupport.py
-=-
Modified: ZODB/trunk/src/ZEO/ClientStorage.py
===================================================================
--- ZODB/trunk/src/ZEO/ClientStorage.py 2006-05-04 18:25:03 UTC (rev 67969)
+++ ZODB/trunk/src/ZEO/ClientStorage.py 2006-05-04 18:30:20 UTC (rev 67970)
@@ -728,6 +728,13 @@
"""
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):
"""Storage API: return current serial number for oid."""
return self._server.getSerial(oid)
Modified: ZODB/trunk/src/ZEO/ServerStub.py
===================================================================
--- ZODB/trunk/src/ZEO/ServerStub.py 2006-05-04 18:25:03 UTC (rev 67969)
+++ ZODB/trunk/src/ZEO/ServerStub.py 2006-05-04 18:30:20 UTC (rev 67970)
@@ -252,6 +252,9 @@
else:
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):
return self.rpc.call('load', oid, version)
Modified: ZODB/trunk/src/ZEO/StorageServer.py
===================================================================
--- ZODB/trunk/src/ZEO/StorageServer.py 2006-05-04 18:25:03 UTC (rev 67969)
+++ ZODB/trunk/src/ZEO/StorageServer.py 2006-05-04 18:30:20 UTC (rev 67970)
@@ -151,6 +151,11 @@
self.load = self.storage.load
self.loadSerial = self.storage.loadSerial
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:
fn = self.storage.getExtensionMethods
except AttributeError:
@@ -233,6 +238,7 @@
'supportsUndo': self.storage.supportsUndo(),
'supportsVersions': self.storage.supportsVersions(),
'extensionMethods': self.getExtensionMethods(),
+ 'supports_record_iternext': hasattr(self, 'record_iternext'),
}
def get_size_info(self):
@@ -628,6 +634,7 @@
else:
return 1
+
class StorageServer:
"""The server side implementation of ZEO.
Added: ZODB/trunk/src/ZEO/tests/testConversionSupport.py
===================================================================
--- ZODB/trunk/src/ZEO/tests/testConversionSupport.py 2006-05-04 18:25:03 UTC (rev 67969)
+++ ZODB/trunk/src/ZEO/tests/testConversionSupport.py 2006-05-04 18:30:20 UTC (rev 67970)
@@ -0,0 +1,162 @@
+##############################################################################
+#
+# 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')
+
Property changes on: ZODB/trunk/src/ZEO/tests/testConversionSupport.py
___________________________________________________________________
Name: svn:keywords
+ Id
Name: svn:eol-style
+ native
More information about the Zodb-checkins
mailing list