Using appropriate checks in tests

Lib/test/test_xmlrpc.py

 import base64
 import datetime
 import sys
 import time
 import unittest
+from unittest import mock
 import xmlrpc.client as xmlrpclib
 import xmlrpc.server
 import http.client
 import socket
 import os
 import re
 import io
 import contextlib
 from test import support
 
+try:
+    import gzip
+except ImportError:
+    gzip = None
 try:
     import threading
 except ImportError:
     threading = None
 
 alist = [{'astring': 'foo@bar.baz.spam',
           'afloat': 7283.43,
           'anint': 2**20,
           'ashortlong': 2,
           'anotherlist': ['.zyx.41'],
@@ skipping 216 matching lines @@
 
     def test_dotted_attribute(self):
         # this will raise AttributeError because code don't want us to use
         # private methods
         self.assertRaises(AttributeError,
                           xmlrpc.server.resolve_dotted_attribute, str, '__add')
         self.assertTrue(xmlrpc.server.resolve_dotted_attribute(str, 'title'))
 
 class DateTimeTestCase(unittest.TestCase):
     def test_default(self):
-        t = xmlrpclib.DateTime()
+        with mock.patch('time.localtime') as localtime_mock:
+            time_struct = time.struct_time(
+                [2013, 7, 15, 0, 24, 49, 0, 196, 0])
+            localtime_mock.return_value = time_struct
+            localtime = time.localtime()
+            t = xmlrpclib.DateTime()
+            self.assertEqual(str(t),
+                             time.strftime("%Y%m%dT%H:%M:%S", localtime))
 
     def test_time(self):
         d = 1181399930.036952
         t = xmlrpclib.DateTime(d)
         self.assertEqual(str(t),
                          time.strftime("%Y%m%dT%H:%M:%S", time.localtime(d)))
 
     def test_time_tuple(self):
         d = (2007,6,9,10,38,50,5,160,0)
         t = xmlrpclib.DateTime(d)
@@ skipping 16 matching lines @@
         self.assertEqual(repr(t), val)
 
     def test_decode(self):
         d = ' 20070908T07:11:13  '
         t1 = xmlrpclib.DateTime()
         t1.decode(d)
         tref = xmlrpclib.DateTime(datetime.datetime(2007,9,8,7,11,13))
         self.assertEqual(t1, tref)
 
         t2 = xmlrpclib._datetime(d)
-        self.assertEqual(t1, tref)
+        self.assertEqual(t2, tref)
 
     def test_comparison(self):
         now = datetime.datetime.now()
         dtime = xmlrpclib.DateTime(now.timetuple())
 
         # datetime vs. DateTime
         self.assertTrue(dtime == now)
         self.assertTrue(now == dtime)
         then = now + datetime.timedelta(seconds=4)
         self.assertTrue(then >= dtime)
@@ skipping 61 matching lines @@
 # the event after it has been set the first time to catch the second set.
 def http_server(evt, numrequests, requestHandler=None):
     class TestInstanceClass:
         def div(self, x, y):
             return x // y
 
         def _methodHelp(self, name):
             if name == 'div':
                 return 'This is the div function'
 
+        class Fixture:
+            @staticmethod
+            def getData():
+                return '42'
+
     def my_function():
         '''This is my function'''
         return True
 
     class MyXMLRPCServer(xmlrpc.server.SimpleXMLRPCServer):
         def get_request(self):
             # Ensure the socket is always non-blocking.  On Linux, socket
             # attributes are not inherited like they are on *BSD and Windows.
             s, port = self.socket.accept()
             s.setblocking(True)
@@ skipping 11 matching lines @@
         #trying to connect to "localhost" using all address families, which
         #causes slowdown e.g. on vista which supports AF_INET6.  The server listens
         #on AF_INET only.
         URL = "http://%s:%d"%(ADDR, PORT)
         serv.server_activate()
         serv.register_introspection_functions()
         serv.register_multicall_functions()
         serv.register_function(pow)
         serv.register_function(lambda x,y: x+y, 'add')
         serv.register_function(my_function)
-        serv.register_instance(TestInstanceClass())
+        testInstance = TestInstanceClass()
+        serv.register_instance(testInstance, allow_dotted_names=True)
         evt.set()
 
         # handle up to 'numrequests' requests
         while numrequests > 0:
             serv.handle_request()
             numrequests -= 1
 
     except socket.timeout:
         pass
     finally:
@@ skipping 159 matching lines @@
         conn.request('POST', '/this-is-not-valid')
         response = conn.getresponse()
         conn.close()
 
         self.assertEqual(response.status, 404)
         self.assertEqual(response.reason, 'Not Found')
 
     def test_introspection1(self):
         expected_methods = set(['pow', 'div', 'my_function', 'add',
                                 'system.listMethods', 'system.methodHelp',
-                                'system.methodSignature', 'system.multicall'])
+                                'system.methodSignature', 'system.multicall',
+                                'Fixture'])
         try:
             p = xmlrpclib.ServerProxy(URL)
             meth = p.system.listMethods()
             self.assertEqual(set(meth), expected_methods)
         except (xmlrpclib.ProtocolError, OSError) 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", "")))
 
@@ skipping 77 matching lines @@
 
     def test_dotted_attribute(self):
         # Raises an AttributeError because private methods are not allowed.
         self.assertRaises(AttributeError,
                           xmlrpc.server.resolve_dotted_attribute, str, '__add')
 
         self.assertTrue(xmlrpc.server.resolve_dotted_attribute(str, 'title'))
         # Get the test to run faster by sending a request with test_simple1.
         # This avoids waiting for the socket timeout.
         self.test_simple1()
+
+    def test_allow_dotted_names_true(self):
+        # XXX also need allow_dotted_names_false test.
+        server = xmlrpclib.ServerProxy("http://%s:%d/RPC2" % (ADDR, PORT))
+        data = server.Fixture.getData()
+        self.assertEqual(data, '42')
 
     def test_unicode_host(self):
         server = xmlrpclib.ServerProxy("http://%s:%d/RPC2" % (ADDR, PORT))
         self.assertEqual(server.add("a", "\xe9"), "a\xe9")
 
     def test_partial_post(self):
         # Check that a partial POST doesn't make the server loop: issue #14001.
         conn = http.client.HTTPConnection(ADDR, PORT)
         conn.request('POST', '/RPC2 HTTP/1.0\r\nContent-Length: 100\r\n\r\nbye')
         conn.close()
@@ skipping 88 matching lines @@
         p = xmlrpclib.ServerProxy(URL)
         #do some requests with close.
         self.assertEqual(p.pow(6,8), 6**8)
         p("transport").close() #same as above, really.
         self.assertEqual(p.pow(6,8), 6**8)
         p("close")()
         self.assertEqual(len(self.RequestHandler.myRequests), 2)
 
 #A test case that verifies that gzip encoding works in both directions
 #(for a request and the response)
+@unittest.skipIf(gzip is None, 'requires gzip')
 class GzipServerTestCase(BaseServerTestCase):
     #a request handler that supports keep-alive and logs requests into a
     #class variable
     class RequestHandler(xmlrpc.server.SimpleXMLRPCRequestHandler):
         parentClass = xmlrpc.server.SimpleXMLRPCRequestHandler
         protocol_version = 'HTTP/1.1'
 
         def do_POST(self):
             #store content of last request in class
             self.__class__.content_length = int(self.headers["content-length"])
@@ skipping 271 matching lines @@
 
     def test_xmlrpcserver_has_use_builtin_types_flag(self):
         server = xmlrpc.server.SimpleXMLRPCServer(("localhost", 0),
             use_builtin_types=True)
         server.server_close()
         self.assertTrue(server.use_builtin_types)
 
 
 @support.reap_threads
 def test_main():
-    xmlrpc_tests = [XMLRPCTestCase, HelperTestCase, DateTimeTestCase,
-         BinaryTestCase, FaultTestCase]
-    xmlrpc_tests.append(UseBuiltinTypesTestCase)
-    xmlrpc_tests.append(SimpleServerTestCase)
-    xmlrpc_tests.append(KeepaliveServerTestCase1)
-    xmlrpc_tests.append(KeepaliveServerTestCase2)
-    try:
-        import gzip
-        xmlrpc_tests.append(GzipServerTestCase)
-    except ImportError:
-        pass #gzip not supported in this build
-    xmlrpc_tests.append(MultiPathServerTestCase)
-    xmlrpc_tests.append(ServerProxyTestCase)
-    xmlrpc_tests.append(FailingServerTestCase)
-    xmlrpc_tests.append(CGIHandlerTestCase)
-
-    support.run_unittest(*xmlrpc_tests)
+    support.run_unittest(XMLRPCTestCase, HelperTestCase, DateTimeTestCase,
+            BinaryTestCase, FaultTestCase, UseBuiltinTypesTestCase,
+            SimpleServerTestCase, KeepaliveServerTestCase1,
+            KeepaliveServerTestCase2, GzipServerTestCase,
+            MultiPathServerTestCase, ServerProxyTestCase, FailingServerTestCase,
+            CGIHandlerTestCase)
+
 
 if __name__ == "__main__":
     test_main()