diff options
Diffstat (limited to 'Lib/test/test_xmlrpc.py')
| -rw-r--r-- | Lib/test/test_xmlrpc.py | 147 |
1 files changed, 142 insertions, 5 deletions
diff --git a/Lib/test/test_xmlrpc.py b/Lib/test/test_xmlrpc.py index 7ae0dce..02d9f5c 100644 --- a/Lib/test/test_xmlrpc.py +++ b/Lib/test/test_xmlrpc.py @@ -7,6 +7,7 @@ from unittest import mock import xmlrpc.client as xmlrpclib import xmlrpc.server import http.client +import http, http.server import socket import os import re @@ -183,6 +184,27 @@ class XMLRPCTestCase(unittest.TestCase): xmlrpclib.loads(strg)[0][0]) self.assertRaises(TypeError, xmlrpclib.dumps, (arg1,)) + def test_dump_encoding(self): + value = {'key\u20ac\xa4': + 'value\u20ac\xa4'} + strg = xmlrpclib.dumps((value,), encoding='iso-8859-15') + strg = "<?xml version='1.0' encoding='iso-8859-15'?>" + strg + self.assertEqual(xmlrpclib.loads(strg)[0][0], value) + strg = strg.encode('iso-8859-15', 'xmlcharrefreplace') + self.assertEqual(xmlrpclib.loads(strg)[0][0], value) + + strg = xmlrpclib.dumps((value,), encoding='iso-8859-15', + methodresponse=True) + self.assertEqual(xmlrpclib.loads(strg)[0][0], value) + strg = strg.encode('iso-8859-15', 'xmlcharrefreplace') + self.assertEqual(xmlrpclib.loads(strg)[0][0], value) + + methodname = 'method\u20ac\xa4' + strg = xmlrpclib.dumps((value,), encoding='iso-8859-15', + methodname=methodname) + self.assertEqual(xmlrpclib.loads(strg)[0][0], value) + self.assertEqual(xmlrpclib.loads(strg)[1], methodname) + def test_dump_bytes(self): sample = b"my dog has fleas" self.assertEqual(sample, xmlrpclib.Binary(sample)) @@ -202,6 +224,20 @@ class XMLRPCTestCase(unittest.TestCase): self.assertIs(type(newvalue), xmlrpclib.Binary) self.assertIsNone(m) + def test_loads_unsupported(self): + ResponseError = xmlrpclib.ResponseError + data = '<params><param><value><spam/></value></param></params>' + self.assertRaises(ResponseError, xmlrpclib.loads, data) + data = ('<params><param><value><array>' + '<value><spam/></value>' + '</array></value></param></params>') + self.assertRaises(ResponseError, xmlrpclib.loads, data) + data = ('<params><param><value><struct>' + '<member><name>a</name><value><spam/></value></member>' + '<member><name>b</name><value><spam/></value></member>' + '</struct></value></param></params>') + self.assertRaises(ResponseError, xmlrpclib.loads, data) + def test_get_host_info(self): # see bug #3613, this raised a TypeError transp = xmlrpc.client.Transport() @@ -223,6 +259,42 @@ class XMLRPCTestCase(unittest.TestCase): except OSError: self.assertTrue(has_ssl) + @unittest.skipUnless(threading, "Threading required for this test.") + def test_keepalive_disconnect(self): + class RequestHandler(http.server.BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + handled = False + + def do_POST(self): + length = int(self.headers.get("Content-Length")) + self.rfile.read(length) + if self.handled: + self.close_connection = True + return + response = xmlrpclib.dumps((5,), methodresponse=True) + response = response.encode() + self.send_response(http.HTTPStatus.OK) + self.send_header("Content-Length", len(response)) + self.end_headers() + self.wfile.write(response) + self.handled = True + self.close_connection = False + + def run_server(): + server.socket.settimeout(float(1)) # Don't hang if client fails + server.handle_request() # First request and attempt at second + server.handle_request() # Retried second request + + server = http.server.HTTPServer((support.HOST, 0), RequestHandler) + self.addCleanup(server.server_close) + thread = threading.Thread(target=run_server) + thread.start() + self.addCleanup(thread.join) + url = "http://{}:{}/".format(*server.server_address) + with xmlrpclib.ServerProxy(url) as p: + self.assertEqual(p.method(), 5) + self.assertEqual(p.method(), 5) + class HelperTestCase(unittest.TestCase): def test_escape(self): self.assertEqual(xmlrpclib.escape("a&b"), "a&b") @@ -287,7 +359,7 @@ class DateTimeTestCase(unittest.TestCase): def test_repr(self): d = datetime.datetime(2007,1,2,3,4,5) t = xmlrpclib.DateTime(d) - val ="<DateTime '20070102T03:04:05' at %x>" % id(t) + val ="<DateTime '20070102T03:04:05' at %#x>" % id(t) self.assertEqual(repr(t), val) def test_decode(self): @@ -371,7 +443,7 @@ ADDR = PORT = URL = None # The evt is set twice. First when the server is ready to serve. # Second when the server has been shutdown. The user must clear # the event after it has been set the first time to catch the second set. -def http_server(evt, numrequests, requestHandler=None): +def http_server(evt, numrequests, requestHandler=None, encoding=None): class TestInstanceClass: def div(self, x, y): return x // y @@ -400,6 +472,7 @@ def http_server(evt, numrequests, requestHandler=None): if not requestHandler: requestHandler = xmlrpc.server.SimpleXMLRPCRequestHandler serv = MyXMLRPCServer(("localhost", 0), requestHandler, + encoding=encoding, logRequests=False, bind_and_activate=False) try: serv.server_bind() @@ -415,6 +488,7 @@ def http_server(evt, numrequests, requestHandler=None): serv.register_multicall_functions() serv.register_function(pow) serv.register_function(lambda x,y: x+y, 'add') + serv.register_function(lambda x: x, 'têšt') serv.register_function(my_function) testInstance = TestInstanceClass() serv.register_instance(testInstance, allow_dotted_names=True) @@ -582,6 +656,30 @@ class SimpleServerTestCase(BaseServerTestCase): # protocol error; provide additional information in test output self.fail("%s\n%s" % (e, getattr(e, "headers", ""))) + def test_client_encoding(self): + start_string = '\u20ac' + end_string = '\xa4' + + try: + p = xmlrpclib.ServerProxy(URL, encoding='iso-8859-15') + self.assertEqual(p.add(start_string, end_string), + start_string + end_string) + except (xmlrpclib.ProtocolError, socket.error) as e: + # ignore failures due to non-blocking socket unavailable errors. + if not is_unavailable_exception(e): + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, getattr(e, "headers", ""))) + + def test_nonascii_methodname(self): + try: + p = xmlrpclib.ServerProxy(URL, encoding='ascii') + self.assertEqual(p.têšt(42), 42) + except (xmlrpclib.ProtocolError, socket.error) as e: + # ignore failures due to non-blocking socket unavailable errors. + if not is_unavailable_exception(e): + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, getattr(e, "headers", ""))) + # [ch] The test 404 is causing lots of false alarms. def XXXtest_404(self): # send POST with http.client, it should return 404 header and @@ -595,7 +693,7 @@ class SimpleServerTestCase(BaseServerTestCase): self.assertEqual(response.reason, 'Not Found') def test_introspection1(self): - expected_methods = set(['pow', 'div', 'my_function', 'add', + expected_methods = set(['pow', 'div', 'my_function', 'add', 'têšt', 'system.listMethods', 'system.methodHelp', 'system.methodSignature', 'system.multicall', 'Fixture']) @@ -713,6 +811,43 @@ class SimpleServerTestCase(BaseServerTestCase): conn.request('POST', '/RPC2 HTTP/1.0\r\nContent-Length: 100\r\n\r\nbye') conn.close() + def test_context_manager(self): + with xmlrpclib.ServerProxy(URL) as server: + server.add(2, 3) + self.assertNotEqual(server('transport')._connection, + (None, None)) + self.assertEqual(server('transport')._connection, + (None, None)) + + def test_context_manager_method_error(self): + try: + with xmlrpclib.ServerProxy(URL) as server: + server.add(2, "a") + except xmlrpclib.Fault: + pass + self.assertEqual(server('transport')._connection, + (None, None)) + + +class SimpleServerEncodingTestCase(BaseServerTestCase): + @staticmethod + def threadFunc(evt, numrequests, requestHandler=None, encoding=None): + http_server(evt, numrequests, requestHandler, 'iso-8859-15') + + def test_server_encoding(self): + start_string = '\u20ac' + end_string = '\xa4' + + try: + p = xmlrpclib.ServerProxy(URL) + self.assertEqual(p.add(start_string, end_string), + start_string + end_string) + except (xmlrpclib.ProtocolError, socket.error) as e: + # ignore failures due to non-blocking socket unavailable errors. + if not is_unavailable_exception(e): + # protocol error; provide additional information in test output + self.fail("%s\n%s" % (e, getattr(e, "headers", ""))) + class MultiPathServerTestCase(BaseServerTestCase): threadFunc = staticmethod(http_multi_server) @@ -919,6 +1054,7 @@ class ServerProxyTestCase(unittest.TestCase): p = xmlrpclib.ServerProxy(self.url, transport=t) self.assertEqual(p('transport'), t) + # This is a contrived way to make a failure occur on the server side # in order to test the _send_traceback_header flag on the server class FailingMessageClass(http.client.HTTPMessage): @@ -1125,8 +1261,9 @@ class UseBuiltinTypesTestCase(unittest.TestCase): def test_main(): support.run_unittest(XMLRPCTestCase, HelperTestCase, DateTimeTestCase, BinaryTestCase, FaultTestCase, UseBuiltinTypesTestCase, - SimpleServerTestCase, KeepaliveServerTestCase1, - KeepaliveServerTestCase2, GzipServerTestCase, GzipUtilTestCase, + SimpleServerTestCase, SimpleServerEncodingTestCase, + KeepaliveServerTestCase1, KeepaliveServerTestCase2, + GzipServerTestCase, GzipUtilTestCase, MultiPathServerTestCase, ServerProxyTestCase, FailingServerTestCase, CGIHandlerTestCase) |