/usr/share/pyshared/ZEO/tests/servertesting.py is in python-zodb 1:3.9.7-2.
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 | ##############################################################################
#
# 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
class StorageServer(ZEO.StorageServer.StorageServer):
def DispatcherClass(*args, **kw):
pass
class Conection:
peer_protocol_version = ZEO.zrpc.connection.Connection.current_protocol
connected = True
def __init__(self, name='connection', addr='test-addr'):
self.name = name
self.addr = addr
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)
|