Back to index

python3.2  3.2.2
Classes | Functions | Variables
test.test_xmlrpc Namespace Reference

Classes

class  XMLRPCTestCase
class  HelperTestCase
class  FaultTestCase
class  DateTimeTestCase
class  BinaryTestCase
class  BaseServerTestCase
class  SimpleServerTestCase
class  MultiPathServerTestCase
class  BaseKeepaliveServerTestCase
class  KeepaliveServerTestCase1
class  KeepaliveServerTestCase2
class  GzipServerTestCase
class  ServerProxyTestCase
class  FailingMessageClass
class  FailingServerTestCase
class  CGIHandlerTestCase

Functions

def http_server
def http_multi_server
def is_unavailable_exception
def make_request_and_skipIf
def captured_stdout
def test_main

Variables

 threading = None
list alist
 ADDR = PORTURLNone

Function Documentation

def test.test_xmlrpc.captured_stdout (   encoding = 'utf-8')
A variation on support.captured_stdout() which gives a text stream
having a `buffer` attribute.

Definition at line 853 of file test_xmlrpc.py.

00853 
00854 def captured_stdout(encoding='utf-8'):
00855     """A variation on support.captured_stdout() which gives a text stream
00856     having a `buffer` attribute.
00857     """
00858     import io
00859     orig_stdout = sys.stdout
00860     sys.stdout = io.TextIOWrapper(io.BytesIO(), encoding=encoding)
00861     try:
00862         yield sys.stdout
00863     finally:
00864         sys.stdout = orig_stdout
00865 

Here is the caller graph for this function:

def test.test_xmlrpc.http_multi_server (   evt,
  numrequests,
  requestHandler = None 
)

Definition at line 323 of file test_xmlrpc.py.

00323 
00324 def http_multi_server(evt, numrequests, requestHandler=None):
00325     class TestInstanceClass:
00326         def div(self, x, y):
00327             return x // y
00328 
00329         def _methodHelp(self, name):
00330             if name == 'div':
00331                 return 'This is the div function'
00332 
00333     def my_function():
00334         '''This is my function'''
00335         return True
00336 
00337     class MyXMLRPCServer(xmlrpc.server.MultiPathXMLRPCServer):
00338         def get_request(self):
00339             # Ensure the socket is always non-blocking.  On Linux, socket
00340             # attributes are not inherited like they are on *BSD and Windows.
00341             s, port = self.socket.accept()
00342             s.setblocking(True)
00343             return s, port
00344 
00345     if not requestHandler:
00346         requestHandler = xmlrpc.server.SimpleXMLRPCRequestHandler
00347     class MyRequestHandler(requestHandler):
00348         rpc_paths = []
00349 
00350     serv = MyXMLRPCServer(("localhost", 0), MyRequestHandler,
00351                           logRequests=False, bind_and_activate=False)
00352     serv.socket.settimeout(3)
00353     serv.server_bind()
00354     try:
00355         global ADDR, PORT, URL
00356         ADDR, PORT = serv.socket.getsockname()
00357         #connect to IP address directly.  This avoids socket.create_connection()
00358         #trying to connect to to "localhost" using all address families, which
00359         #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
00360         #on AF_INET only.
00361         URL = "http://%s:%d"%(ADDR, PORT)
00362         serv.server_activate()
00363         paths = ["/foo", "/foo/bar"]
00364         for path in paths:
00365             d = serv.add_dispatcher(path, xmlrpc.server.SimpleXMLRPCDispatcher())
00366             d.register_introspection_functions()
00367             d.register_multicall_functions()
00368         serv.get_dispatcher(paths[0]).register_function(pow)
00369         serv.get_dispatcher(paths[1]).register_function(lambda x,y: x+y, 'add')
00370         evt.set()
00371 
00372         # handle up to 'numrequests' requests
00373         while numrequests > 0:
00374             serv.handle_request()
00375             numrequests -= 1
00376 
00377     except socket.timeout:
00378         pass
00379     finally:
00380         serv.socket.close()
00381         PORT = None
00382         evt.set()
00383 
00384 # This function prevents errors like:
#    <ProtocolError for localhost:57527/RPC2: 500 Internal Server Error>
def test.test_xmlrpc.http_server (   evt,
  numrequests,
  requestHandler = None 
)

Definition at line 268 of file test_xmlrpc.py.

00268 
00269 def http_server(evt, numrequests, requestHandler=None):
00270     class TestInstanceClass:
00271         def div(self, x, y):
00272             return x // y
00273 
00274         def _methodHelp(self, name):
00275             if name == 'div':
00276                 return 'This is the div function'
00277 
00278     def my_function():
00279         '''This is my function'''
00280         return True
00281 
00282     class MyXMLRPCServer(xmlrpc.server.SimpleXMLRPCServer):
00283         def get_request(self):
00284             # Ensure the socket is always non-blocking.  On Linux, socket
00285             # attributes are not inherited like they are on *BSD and Windows.
00286             s, port = self.socket.accept()
00287             s.setblocking(True)
00288             return s, port
00289 
00290     if not requestHandler:
00291         requestHandler = xmlrpc.server.SimpleXMLRPCRequestHandler
00292     serv = MyXMLRPCServer(("localhost", 0), requestHandler,
00293                           logRequests=False, bind_and_activate=False)
00294     try:
00295         serv.server_bind()
00296         global ADDR, PORT, URL
00297         ADDR, PORT = serv.socket.getsockname()
00298         #connect to IP address directly.  This avoids socket.create_connection()
00299         #trying to connect to to "localhost" using all address families, which
00300         #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
00301         #on AF_INET only.
00302         URL = "http://%s:%d"%(ADDR, PORT)
00303         serv.server_activate()
00304         serv.register_introspection_functions()
00305         serv.register_multicall_functions()
00306         serv.register_function(pow)
00307         serv.register_function(lambda x,y: x+y, 'add')
00308         serv.register_function(my_function)
00309         serv.register_instance(TestInstanceClass())
00310         evt.set()
00311 
00312         # handle up to 'numrequests' requests
00313         while numrequests > 0:
00314             serv.handle_request()
00315             numrequests -= 1
00316 
00317     except socket.timeout:
00318         pass
00319     finally:
00320         serv.socket.close()
00321         PORT = None
00322         evt.set()

Returns True if the given ProtocolError is the product of a server-side
   exception caused by the 'temporarily unavailable' response sometimes
   given by operations on non-blocking sockets.

Definition at line 385 of file test_xmlrpc.py.

00385 
00386 def is_unavailable_exception(e):
00387     '''Returns True if the given ProtocolError is the product of a server-side
00388        exception caused by the 'temporarily unavailable' response sometimes
00389        given by operations on non-blocking sockets.'''
00390 
00391     # sometimes we get a -1 error code and/or empty headers
00392     try:
00393         if e.errcode == -1 or e.headers is None:
00394             return True
00395         exc_mess = e.headers.get('X-exception')
00396     except AttributeError:
00397         # Ignore socket.errors here.
00398         exc_mess = str(e)
00399 
00400     if exc_mess and 'temporarily unavailable' in exc_mess.lower():
00401         return True

Here is the caller graph for this function:

def test.test_xmlrpc.make_request_and_skipIf (   condition,
  reason 
)

Definition at line 402 of file test_xmlrpc.py.

00402 
00403 def make_request_and_skipIf(condition, reason):
00404     # If we skip the test, we have to make a request because the
00405     # the server created in setUp blocks expecting one to come in.
00406     if not condition:
00407         return lambda func: func
00408     def decorator(func):
00409         def make_request_and_skip(self):
00410             try:
00411                 xmlrpclib.ServerProxy(URL).my_function()
00412             except (xmlrpclib.ProtocolError, socket.error) as e:
00413                 if not is_unavailable_exception(e):
00414                     raise
00415             raise unittest.SkipTest(reason)
00416         return make_request_and_skip
00417     return decorator
00418 
@unittest.skipUnless(threading, 'Threading required for this test.')

Here is the call graph for this function:

Here is the caller graph for this function:

Definition at line 935 of file test_xmlrpc.py.

00935 
00936 def test_main():
00937     xmlrpc_tests = [XMLRPCTestCase, HelperTestCase, DateTimeTestCase,
00938          BinaryTestCase, FaultTestCase]
00939     xmlrpc_tests.append(SimpleServerTestCase)
00940     xmlrpc_tests.append(KeepaliveServerTestCase1)
00941     xmlrpc_tests.append(KeepaliveServerTestCase2)
00942     try:
00943         import gzip
00944         xmlrpc_tests.append(GzipServerTestCase)
00945     except ImportError:
00946         pass #gzip not supported in this build
00947     xmlrpc_tests.append(MultiPathServerTestCase)
00948     xmlrpc_tests.append(ServerProxyTestCase)
00949     xmlrpc_tests.append(FailingServerTestCase)
00950     xmlrpc_tests.append(CGIHandlerTestCase)
00951 
00952     support.run_unittest(*xmlrpc_tests)


Variable Documentation

test.test_xmlrpc.ADDR = PORTURLNone

Definition at line 263 of file test_xmlrpc.py.

Initial value:
00001 [{'astring': 'foo@bar.baz.spam',
00002           'afloat': 7283.43,
00003           'anint': 2**20,
00004           'ashortlong': 2,
00005           'anotherlist': ['.zyx.41'],
00006           'abase64': xmlrpclib.Binary(b"my dog has fleas"),
00007           'boolean': False,
00008           'unicode': '\u4000\u6000\u8000',
00009           'ukey\u4000': 'regular value',
00010           'datetime1': xmlrpclib.DateTime('20050210T11:41:23'),
00011           'datetime2': xmlrpclib.DateTime(
00012                         (2005, 2, 10, 11, 41, 23, 0, 1, -1)),
00013           'datetime3': xmlrpclib.DateTime(
00014                         datetime.datetime(2005, 2, 10, 11, 41, 23)),
00015           }]

Definition at line 21 of file test_xmlrpc.py.

Definition at line 19 of file test_xmlrpc.py.