1 import base64

     2 import datetime

     3 import sys

     4 import time

     5 import unittest

     6 import xmlrpclib

     7 import SimpleXMLRPCServer

     8 import threading

     9 import mimetools

    10 import httplib

    11 import socket

    12 import os

    13 from test import test_support

    14 

    15 try:

    16     unicode

    17 except NameError:

    18     have_unicode = False

    19 else:

    20     have_unicode = True

    21 

    22 alist = [{'astring': 'foo@bar.baz.spam',

    23           'afloat': 7283.43,

    24           'anint': 2**20,

    25           'ashortlong': 2L,

    26           'anotherlist': ['.zyx.41'],

    27           'abase64': xmlrpclib.Binary("my dog has fleas"),

    28           'boolean': xmlrpclib.False,

    29           'unicode': u'\u4000\u6000\u8000',

    30           u'ukey\u4000': 'regular value',

    31           'datetime1': xmlrpclib.DateTime('20050210T11:41:23'),

    32           'datetime2': xmlrpclib.DateTime(

    33                         (2005, 02, 10, 11, 41, 23, 0, 1, -1)),

    34           'datetime3': xmlrpclib.DateTime(

    35                         datetime.datetime(2005, 02, 10, 11, 41, 23)),

    36           }]

    37 

    38 class XMLRPCTestCase(unittest.TestCase):

    39 

    40     def test_dump_load(self):

    41         self.assertEquals(alist,

    42                           xmlrpclib.loads(xmlrpclib.dumps((alist,)))[0][0])

    43 

    44     def test_dump_bare_datetime(self):

    45         # This checks that an unwrapped datetime.date object can be handled

    46         # by the marshalling code.  This can't be done via test_dump_load()

    47         # since with use_datetime set to 1 the unmarshaller would create

    48         # datetime objects for the 'datetime[123]' keys as well

    49         dt = datetime.datetime(2005, 02, 10, 11, 41, 23)

    50         s = xmlrpclib.dumps((dt,))

    51         (newdt,), m = xmlrpclib.loads(s, use_datetime=1)

    52         self.assertEquals(newdt, dt)

    53         self.assertEquals(m, None)

    54 

    55         (newdt,), m = xmlrpclib.loads(s, use_datetime=0)

    56         self.assertEquals(newdt, xmlrpclib.DateTime('20050210T11:41:23'))

    57 

    58     def test_datetime_before_1900(self):

    59         # same as before but with a date before 1900

    60         dt = datetime.datetime(1, 02, 10, 11, 41, 23)

    61         s = xmlrpclib.dumps((dt,))

    62         (newdt,), m = xmlrpclib.loads(s, use_datetime=1)

    63         self.assertEquals(newdt, dt)

    64         self.assertEquals(m, None)

    65 

    66         (newdt,), m = xmlrpclib.loads(s, use_datetime=0)

    67         self.assertEquals(newdt, xmlrpclib.DateTime('00010210T11:41:23'))

    68 

    69     def test_cmp_datetime_DateTime(self):

    70         now = datetime.datetime.now()

    71         dt = xmlrpclib.DateTime(now.timetuple())

    72         self.assert_(dt == now)

    73         self.assert_(now == dt)

    74         then = now + datetime.timedelta(seconds=4)

    75         self.assert_(then >= dt)

    76         self.assert_(dt < then)

    77 

    78     def test_bug_1164912 (self):

    79         d = xmlrpclib.DateTime()

    80         ((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,),

    81                                             methodresponse=True))

    82         self.assert_(isinstance(new_d.value, str))

    83 

    84         # Check that the output of dumps() is still an 8-bit string

    85         s = xmlrpclib.dumps((new_d,), methodresponse=True)

    86         self.assert_(isinstance(s, str))

    87 

    88     def test_newstyle_class(self):

    89         class T(object):

    90             pass

    91         t = T()

    92         t.x = 100

    93         t.y = "Hello"

    94         ((t2,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((t,)))

    95         self.assertEquals(t2, t.__dict__)

    96 

    97     def test_dump_big_long(self):

    98         self.assertRaises(OverflowError, xmlrpclib.dumps, (2L**99,))

    99 

   100     def test_dump_bad_dict(self):

   101         self.assertRaises(TypeError, xmlrpclib.dumps, ({(1,2,3): 1},))

   102 

   103     def test_dump_recursive_seq(self):

   104         l = [1,2,3]

   105         t = [3,4,5,l]

   106         l.append(t)

   107         self.assertRaises(TypeError, xmlrpclib.dumps, (l,))

   108 

   109     def test_dump_recursive_dict(self):

   110         d = {'1':1, '2':1}

   111         t = {'3':3, 'd':d}

   112         d['t'] = t

   113         self.assertRaises(TypeError, xmlrpclib.dumps, (d,))

   114 

   115     def test_dump_big_int(self):

   116         if sys.maxint > 2L**31-1:

   117             self.assertRaises(OverflowError, xmlrpclib.dumps,

   118                               (int(2L**34),))

   119 

   120         xmlrpclib.dumps((xmlrpclib.MAXINT, xmlrpclib.MININT))

   121         self.assertRaises(OverflowError, xmlrpclib.dumps, (xmlrpclib.MAXINT+1,))

   122         self.assertRaises(OverflowError, xmlrpclib.dumps, (xmlrpclib.MININT-1,))

   123 

   124         def dummy_write(s):

   125             pass

   126 

   127         m = xmlrpclib.Marshaller()

   128         m.dump_int(xmlrpclib.MAXINT, dummy_write)

   129         m.dump_int(xmlrpclib.MININT, dummy_write)

   130         self.assertRaises(OverflowError, m.dump_int, xmlrpclib.MAXINT+1, dummy_write)

   131         self.assertRaises(OverflowError, m.dump_int, xmlrpclib.MININT-1, dummy_write)

   132 

   133 

   134     def test_dump_none(self):

   135         value = alist + [None]

   136         arg1 = (alist + [None],)

   137         strg = xmlrpclib.dumps(arg1, allow_none=True)

   138         self.assertEquals(value,

   139                           xmlrpclib.loads(strg)[0][0])

   140         self.assertRaises(TypeError, xmlrpclib.dumps, (arg1,))

   141 

   142     def test_default_encoding_issues(self):

   143         # SF bug #1115989: wrong decoding in '_stringify'

   144         utf8 = """<?xml version='1.0' encoding='iso-8859-1'?>

   145                   <params>

   146                     <param><value>

   147                       <string>abc \x95</string>

   148                       </value></param>

   149                     <param><value>

   150                       <struct>

   151                         <member>

   152                           <name>def \x96</name>

   153                           <value><string>ghi \x97</string></value>

   154                           </member>

   155                         </struct>

   156                       </value></param>

   157                   </params>

   158                   """

   159 

   160         # sys.setdefaultencoding() normally doesn't exist after site.py is

   161         # loaded.  reload(sys) is the way to get it back.

   162         old_encoding = sys.getdefaultencoding()

   163         setdefaultencoding_existed = hasattr(sys, "setdefaultencoding")

   164         reload(sys) # ugh!

   165         sys.setdefaultencoding("iso-8859-1")

   166         try:

   167             (s, d), m = xmlrpclib.loads(utf8)

   168         finally:

   169             sys.setdefaultencoding(old_encoding)

   170             if not setdefaultencoding_existed:

   171                 del sys.setdefaultencoding

   172 

   173         items = d.items()

   174         if have_unicode:

   175             self.assertEquals(s, u"abc \x95")

   176             self.assert_(isinstance(s, unicode))

   177             self.assertEquals(items, [(u"def \x96", u"ghi \x97")])

   178             self.assert_(isinstance(items[0][0], unicode))

   179             self.assert_(isinstance(items[0][1], unicode))

   180         else:

   181             self.assertEquals(s, "abc \xc2\x95")

   182             self.assertEquals(items, [("def \xc2\x96", "ghi \xc2\x97")])

   183 

   184 

   185 class HelperTestCase(unittest.TestCase):

   186     def test_escape(self):

   187         self.assertEqual(xmlrpclib.escape("a&b"), "a&b")

   188         self.assertEqual(xmlrpclib.escape("a<b"), "a&lt;b")

   189         self.assertEqual(xmlrpclib.escape("a>b"), "a>b")

   190 

   191 class FaultTestCase(unittest.TestCase):

   192     def test_repr(self):

   193         f = xmlrpclib.Fault(42, 'Test Fault')

   194         self.assertEqual(repr(f), "<Fault 42: 'Test Fault'>")

   195         self.assertEqual(repr(f), str(f))

   196 

   197     def test_dump_fault(self):

   198         f = xmlrpclib.Fault(42, 'Test Fault')

   199         s = xmlrpclib.dumps((f,))

   200         (newf,), m = xmlrpclib.loads(s)

   201         self.assertEquals(newf, {'faultCode': 42, 'faultString': 'Test Fault'})

   202         self.assertEquals(m, None)

   203 

   204         s = xmlrpclib.Marshaller().dumps(f)

   205         self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, s)

   206 

   207 

   208 class DateTimeTestCase(unittest.TestCase):

   209     def test_default(self):

   210         t = xmlrpclib.DateTime()

   211 

   212     def test_time(self):

   213         d = 1181399930.036952

   214         t = xmlrpclib.DateTime(d)

   215         self.assertEqual(str(t), time.strftime("%Y%m%dT%H:%M:%S", time.localtime(d)))

   216 

   217     def test_time_tuple(self):

   218         d = (2007,6,9,10,38,50,5,160,0)

   219         t = xmlrpclib.DateTime(d)

   220         self.assertEqual(str(t), '20070609T10:38:50')

   221 

   222     def test_time_struct(self):

   223         d = time.localtime(1181399930.036952)

   224         t = xmlrpclib.DateTime(d)

   225         self.assertEqual(str(t),  time.strftime("%Y%m%dT%H:%M:%S", d))

   226 

   227     def test_datetime_datetime(self):

   228         d = datetime.datetime(2007,1,2,3,4,5)

   229         t = xmlrpclib.DateTime(d)

   230         self.assertEqual(str(t), '20070102T03:04:05')

   231 

   232     def test_repr(self):

   233         d = datetime.datetime(2007,1,2,3,4,5)

   234         t = xmlrpclib.DateTime(d)

   235         val ="<DateTime '20070102T03:04:05' at %x>" % id(t)

   236         self.assertEqual(repr(t), val)

   237 

   238     def test_decode(self):

   239         d = ' 20070908T07:11:13  '

   240         t1 = xmlrpclib.DateTime()

   241         t1.decode(d)

   242         tref = xmlrpclib.DateTime(datetime.datetime(2007,9,8,7,11,13))

   243         self.assertEqual(t1, tref)

   244 

   245         t2 = xmlrpclib._datetime(d)

   246         self.assertEqual(t1, tref)

   247 

   248 class BinaryTestCase(unittest.TestCase):

   249     def test_default(self):

   250         t = xmlrpclib.Binary()

   251         self.assertEqual(str(t), '')

   252 

   253     def test_string(self):

   254         d = '\x01\x02\x03abc123\xff\xfe'

   255         t = xmlrpclib.Binary(d)

   256         self.assertEqual(str(t), d)

   257 

   258     def test_decode(self):

   259         d = '\x01\x02\x03abc123\xff\xfe'

   260         de = base64.encodestring(d)

   261         t1 = xmlrpclib.Binary()

   262         t1.decode(de)

   263         self.assertEqual(str(t1), d)

   264 

   265         t2 = xmlrpclib._binary(de)

   266         self.assertEqual(str(t2), d)

   267 

   268 

   269 PORT = None

   270 

   271 # The evt is set twice.  First when the server is ready to serve.

   272 # Second when the server has been shutdown.  The user must clear

   273 # the event after it has been set the first time to catch the second set.

   274 def http_server(evt, numrequests):

   275     class TestInstanceClass:

   276         def div(self, x, y):

   277             return x // y

   278 

   279         def _methodHelp(self, name):

   280             if name == 'div':

   281                 return 'This is the div function'

   282 

   283     def my_function():

   284         '''This is my function'''

   285         return True

   286 

   287     class MyXMLRPCServer(SimpleXMLRPCServer.SimpleXMLRPCServer):

   288         def get_request(self):

   289             # Ensure the socket is always non-blocking.  On Linux, socket

   290             # attributes are not inherited like they are on *BSD and Windows.

   291             s, port = self.socket.accept()

   292             s.setblocking(True)

   293             return s, port

   294 

   295     try:

   296         serv = MyXMLRPCServer(("localhost", 0),

   297                               logRequests=False, bind_and_activate=False)

   298         serv.socket.settimeout(3)

   299         serv.server_bind()

   300         global PORT

   301         PORT = serv.socket.getsockname()[1]

   302         serv.server_activate()

   303         serv.register_introspection_functions()

   304         serv.register_multicall_functions()

   305         serv.register_function(pow)

   306         serv.register_function(lambda x,y: x+y, 'add')

   307         serv.register_function(my_function)

   308         serv.register_instance(TestInstanceClass())

   309         evt.set()

   310 

   311         # handle up to 'numrequests' requests

   312         while numrequests > 0:

   313             serv.handle_request()

   314             numrequests -= 1

   315 

   316     except socket.timeout:

   317         pass

   318     finally:

   319         serv.socket.close()

   320         PORT = None

   321         evt.set()

   322 

   323 # This function prevents errors like:

   324 #    <ProtocolError for localhost:57527/RPC2: 500 Internal Server Error>

   325 def is_unavailable_exception(e):

   326     '''Returns True if the given ProtocolError is the product of a server-side

   327        exception caused by the 'temporarily unavailable' response sometimes

   328        given by operations on non-blocking sockets.'''

   329 

   330     # sometimes we get a -1 error code and/or empty headers

   331     try:

   332         if e.errcode == -1 or e.headers is None:

   333             return True

   334         exc_mess = e.headers.get('X-exception')

   335     except AttributeError:

   336         # Ignore socket.errors here.

   337         exc_mess = str(e)

   338 

   339     if exc_mess and 'temporarily unavailable' in exc_mess.lower():

   340         return True

   341 

   342     return False

   343 

   344 # NOTE: The tests in SimpleServerTestCase will ignore failures caused by

   345 # "temporarily unavailable" exceptions raised in SimpleXMLRPCServer.  This

   346 # condition occurs infrequently on some platforms, frequently on others, and

   347 # is apparently caused by using SimpleXMLRPCServer with a non-blocking socket.

   348 # If the server class is updated at some point in the future to handle this

   349 # situation more gracefully, these tests should be modified appropriately.

   350 

   351 class SimpleServerTestCase(unittest.TestCase):

   352     def setUp(self):

   353         # enable traceback reporting

   354         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True

   355 

   356         self.evt = threading.Event()

   357         # start server thread to handle requests

   358         serv_args = (self.evt, 1)

   359         threading.Thread(target=http_server, args=serv_args).start()

   360 

   361         # wait for the server to be ready

   362         self.evt.wait()

   363         self.evt.clear()

   364 

   365     def tearDown(self):

   366         # wait on the server thread to terminate

   367         self.evt.wait()

   368 

   369         # disable traceback reporting

   370         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = False

   371 

   372     def test_simple1(self):

   373         try:

   374             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   375             self.assertEqual(p.pow(6,8), 6**8)

   376         except (xmlrpclib.ProtocolError, socket.error), e:

   377             # ignore failures due to non-blocking socket 'unavailable' errors

   378             if not is_unavailable_exception(e):

   379                 # protocol error; provide additional information in test output

   380                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   381 

   382     # [ch] The test 404 is causing lots of false alarms.

   383     def XXXtest_404(self):

   384         # send POST with httplib, it should return 404 header and

   385         # 'Not Found' message.

   386         conn = httplib.HTTPConnection('localhost', PORT)

   387         conn.request('POST', '/this-is-not-valid')

   388         response = conn.getresponse()

   389         conn.close()

   390 

   391         self.assertEqual(response.status, 404)

   392         self.assertEqual(response.reason, 'Not Found')

   393 

   394     def test_introspection1(self):

   395         try:

   396             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   397             meth = p.system.listMethods()

   398             expected_methods = set(['pow', 'div', 'my_function', 'add',

   399                                     'system.listMethods', 'system.methodHelp',

   400                                     'system.methodSignature', 'system.multicall'])

   401             self.assertEqual(set(meth), expected_methods)

   402         except (xmlrpclib.ProtocolError, socket.error), e:

   403             # ignore failures due to non-blocking socket 'unavailable' errors

   404             if not is_unavailable_exception(e):

   405                 # protocol error; provide additional information in test output

   406                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   407 

   408     def test_introspection2(self):

   409         try:

   410             # test _methodHelp()

   411             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   412             divhelp = p.system.methodHelp('div')

   413             self.assertEqual(divhelp, 'This is the div function')

   414         except (xmlrpclib.ProtocolError, socket.error), e:

   415             # ignore failures due to non-blocking socket 'unavailable' errors

   416             if not is_unavailable_exception(e):

   417                 # protocol error; provide additional information in test output

   418                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   419 

   420     def test_introspection3(self):

   421         try:

   422             # test native doc

   423             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   424             myfunction = p.system.methodHelp('my_function')

   425             self.assertEqual(myfunction, 'This is my function')

   426         except (xmlrpclib.ProtocolError, socket.error), e:

   427             # ignore failures due to non-blocking socket 'unavailable' errors

   428             if not is_unavailable_exception(e):

   429                 # protocol error; provide additional information in test output

   430                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   431 

   432     def test_introspection4(self):

   433         # the SimpleXMLRPCServer doesn't support signatures, but

   434         # at least check that we can try making the call

   435         try:

   436             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   437             divsig = p.system.methodSignature('div')

   438             self.assertEqual(divsig, 'signatures not supported')

   439         except (xmlrpclib.ProtocolError, socket.error), e:

   440             # ignore failures due to non-blocking socket 'unavailable' errors

   441             if not is_unavailable_exception(e):

   442                 # protocol error; provide additional information in test output

   443                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   444 

   445     def test_multicall(self):

   446         try:

   447             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   448             multicall = xmlrpclib.MultiCall(p)

   449             multicall.add(2,3)

   450             multicall.pow(6,8)

   451             multicall.div(127,42)

   452             add_result, pow_result, div_result = multicall()

   453             self.assertEqual(add_result, 2+3)

   454             self.assertEqual(pow_result, 6**8)

   455             self.assertEqual(div_result, 127//42)

   456         except (xmlrpclib.ProtocolError, socket.error), e:

   457             # ignore failures due to non-blocking socket 'unavailable' errors

   458             if not is_unavailable_exception(e):

   459                 # protocol error; provide additional information in test output

   460                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   461 

   462     def test_non_existing_multicall(self):

   463         try:

   464             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   465             multicall = xmlrpclib.MultiCall(p)

   466             multicall.this_is_not_exists()

   467             result = multicall()

   468 

   469             # result.results contains;

   470             # [{'faultCode': 1, 'faultString': '<type \'exceptions.Exception\'>:'

   471             #   'method "this_is_not_exists" is not supported'>}]

   472 

   473             self.assertEqual(result.results[0]['faultCode'], 1)

   474             self.assertEqual(result.results[0]['faultString'],

   475                 '<type \'exceptions.Exception\'>:method "this_is_not_exists" '

   476                 'is not supported')

   477         except (xmlrpclib.ProtocolError, socket.error), e:

   478             # ignore failures due to non-blocking socket 'unavailable' errors

   479             if not is_unavailable_exception(e):

   480                 # protocol error; provide additional information in test output

   481                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   482 

   483     def test_dotted_attribute(self):

   484         # Raises an AttributeError because private methods are not allowed.

   485         self.assertRaises(AttributeError,

   486                           SimpleXMLRPCServer.resolve_dotted_attribute, str, '__add')

   487 

   488         self.assert_(SimpleXMLRPCServer.resolve_dotted_attribute(str, 'title'))

   489         # Get the test to run faster by sending a request with test_simple1.

   490         # This avoids waiting for the socket timeout.

   491         self.test_simple1()

   492 

   493 # This is a contrived way to make a failure occur on the server side

   494 # in order to test the _send_traceback_header flag on the server

   495 class FailingMessageClass(mimetools.Message):

   496     def __getitem__(self, key):

   497         key = key.lower()

   498         if key == 'content-length':

   499             return 'I am broken'

   500         return mimetools.Message.__getitem__(self, key)

   501 

   502 

   503 class FailingServerTestCase(unittest.TestCase):

   504     def setUp(self):

   505         self.evt = threading.Event()

   506         # start server thread to handle requests

   507         serv_args = (self.evt, 1)

   508         threading.Thread(target=http_server, args=serv_args).start()

   509 

   510         # wait for the server to be ready

   511         self.evt.wait()

   512         self.evt.clear()

   513 

   514     def tearDown(self):

   515         # wait on the server thread to terminate

   516         self.evt.wait()

   517         # reset flag

   518         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = False

   519         # reset message class

   520         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = mimetools.Message

   521 

   522     def test_basic(self):

   523         # check that flag is false by default

   524         flagval = SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header

   525         self.assertEqual(flagval, False)

   526 

   527         # enable traceback reporting

   528         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True

   529 

   530         # test a call that shouldn't fail just as a smoke test

   531         try:

   532             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   533             self.assertEqual(p.pow(6,8), 6**8)

   534         except (xmlrpclib.ProtocolError, socket.error), e:

   535             # ignore failures due to non-blocking socket 'unavailable' errors

   536             if not is_unavailable_exception(e):

   537                 # protocol error; provide additional information in test output

   538                 self.fail("%s\n%s" % (e, getattr(e, "headers", "")))

   539 

   540     def test_fail_no_info(self):

   541         # use the broken message class

   542         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass

   543 

   544         try:

   545             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   546             p.pow(6,8)

   547         except (xmlrpclib.ProtocolError, socket.error), e:

   548             # ignore failures due to non-blocking socket 'unavailable' errors

   549             if not is_unavailable_exception(e) and hasattr(e, "headers"):

   550                 # The two server-side error headers shouldn't be sent back in this case

   551                 self.assertTrue(e.headers.get("X-exception") is None)

   552                 self.assertTrue(e.headers.get("X-traceback") is None)

   553         else:

   554             self.fail('ProtocolError not raised')

   555 

   556     def test_fail_with_info(self):

   557         # use the broken message class

   558         SimpleXMLRPCServer.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass

   559 

   560         # Check that errors in the server send back exception/traceback

   561         # info when flag is set

   562         SimpleXMLRPCServer.SimpleXMLRPCServer._send_traceback_header = True

   563 

   564         try:

   565             p = xmlrpclib.ServerProxy('http://localhost:%d' % PORT)

   566             p.pow(6,8)

   567         except (xmlrpclib.ProtocolError, socket.error), e:

   568             # ignore failures due to non-blocking socket 'unavailable' errors

   569             if not is_unavailable_exception(e) and hasattr(e, "headers"):

   570                 # We should get error info in the response

   571                 expected_err = "invalid literal for int() with base 10: 'I am broken'"

   572                 self.assertEqual(e.headers.get("x-exception"), expected_err)

   573                 self.assertTrue(e.headers.get("x-traceback") is not None)

   574         else:

   575             self.fail('ProtocolError not raised')

   576 

   577 class CGIHandlerTestCase(unittest.TestCase):

   578     def setUp(self):

   579         self.cgi = SimpleXMLRPCServer.CGIXMLRPCRequestHandler()

   580 

   581     def tearDown(self):

   582         self.cgi = None

   583 

   584     def test_cgi_get(self):

   585         os.environ['REQUEST_METHOD'] = 'GET'

   586         # if the method is GET and no request_text is given, it runs handle_get

   587         # get sysout output

   588         tmp = sys.stdout

   589         sys.stdout = open(test_support.TESTFN, "w")

   590         self.cgi.handle_request()

   591         sys.stdout.close()

   592         sys.stdout = tmp

   593 

   594         # parse Status header

   595         handle = open(test_support.TESTFN, "r").read()

   596         status = handle.split()[1]

   597         message = ' '.join(handle.split()[2:4])

   598 

   599         self.assertEqual(status, '400')

   600         self.assertEqual(message, 'Bad Request')

   601 

   602         os.remove(test_support.TESTFN)

   603         os.environ['REQUEST_METHOD'] = ''

   604 

   605     def test_cgi_xmlrpc_response(self):

   606         data = """<?xml version='1.0'?>

   607 <methodCall>

   608     <methodName>test_method</methodName>

   609     <params>

   610         <param>

   611             <value><string>foo</string></value>

   612         </param>

   613         <param>

   614             <value><string>bar</string></value>

   615         </param>

   616      </params>

   617 </methodCall>

   618 """

   619         open("xmldata.txt", "w").write(data)

   620         tmp1 = sys.stdin

   621         tmp2 = sys.stdout

   622 

   623         sys.stdin = open("xmldata.txt", "r")

   624         sys.stdout = open(test_support.TESTFN, "w")

   625 

   626         self.cgi.handle_request()

   627 

   628         sys.stdin.close()

   629         sys.stdout.close()

   630         sys.stdin = tmp1

   631         sys.stdout = tmp2

   632 

   633         # will respond exception, if so, our goal is achieved ;)

   634         handle = open(test_support.TESTFN, "r").read()

   635 

   636         # start with 44th char so as not to get http header, we just need only xml

   637         self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, handle[44:])

   638 

   639         os.remove("xmldata.txt")

   640         os.remove(test_support.TESTFN)

   641 

   642 def test_main():

   643     xmlrpc_tests = [XMLRPCTestCase, HelperTestCase, DateTimeTestCase,

   644          BinaryTestCase, FaultTestCase]

   645 

   646     # The test cases against a SimpleXMLRPCServer raise a socket error

   647     # 10035 (WSAEWOULDBLOCK) in the server thread handle_request call when

   648     # run on Windows. This only happens on the first test to run, but it

   649     # fails every time and so these tests are skipped on win32 platforms.

   650     if sys.platform != 'win32':

   651         xmlrpc_tests.append(SimpleServerTestCase)

   652         xmlrpc_tests.append(FailingServerTestCase)

   653         xmlrpc_tests.append(CGIHandlerTestCase)

   654 

   655     test_support.run_unittest(*xmlrpc_tests)

   656 

   657 if __name__ == "__main__":

   658     test_main()