/usr/share/pyshared/ZEO/tests/servertesting.py is in python-zodb 1:3.10.5-0ubuntu3.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 | ##############################################################################
#
# Copyright Zope Foundation and Contributors.
# All Rights Reserved.
#
# This software is subject to the provisions of the Zope Public License,
# Version 2.1 (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.
#
##############################################################################
# Testing the current ZEO implementation is rather hard due to the
# architecture, which mixes concerns, especially between application
# and networking. Still, it's not as bad as it could be.
# The 2 most important classes in the architecture are ZEOStorage and
# StorageServer. A ZEOStorage is created for each client connection.
# The StorageServer maintains data shared or needed for coordination
# among clients.
# The other important part of the architecture is connections.
# Connections are used by ZEOStorages to send messages or return data
# to clients.
# Here, we'll try to provide some testing infrastructure to isolate
# servers from the network.
import ZEO.StorageServer
import ZEO.zrpc.connection
import ZEO.zrpc.error
import ZODB.MappingStorage
class StorageServer(ZEO.StorageServer.StorageServer):
def __init__(self, addr='test_addr', storages=None, **kw):
if storages is None:
storages = {'1': ZODB.MappingStorage.MappingStorage()}
ZEO.StorageServer.StorageServer.__init__(self, addr, storages, **kw)
def DispatcherClass(*args, **kw):
pass
class Connection:
peer_protocol_version = ZEO.zrpc.connection.Connection.current_protocol
connected = True
def __init__(self, name='connection', addr=''):
name = str(name)
self.name = name
self.addr = addr or 'test-addr-'+name
def close(self):
print self.name, 'closed'
self.connected = False
def poll(self):
if not self.connected:
raise ZEO.zrpc.error.DisconnectedError()
def callAsync(self, meth, *args):
print self.name, 'callAsync', meth, repr(args)
callAsyncNoPoll = callAsync
def call_from_thread(self, *args):
if args:
args[0](*args[1:])
def send_reply(self, *args):
pass
def client(server, name='client', addr=''):
zs = ZEO.StorageServer.ZEOStorage(server)
zs.notifyConnected(Connection(name, addr))
zs.register('1', 0)
return zs
|