Update test_xmlrpc + xmlrpc

RustPython · terryluan12 · Dec 30, 2025 · Dec 30, 2025 · Dec 30, 2025 · Dec 30, 2025
commit c96e74bca180815ffa4af7acb69451b6c0a8c619
diff --git a/Lib/test/test_xmlrpc.py b/Lib/test/test_xmlrpc.py
@@ -1042,55 +1042,46 @@ def test_path2(self):
         self.assertEqual(p.add(6,8), 6+8)
         self.assertRaises(xmlrpclib.Fault, p.pow, 6, 8)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_path3(self):
         p = xmlrpclib.ServerProxy(URL+"/is/broken")
         self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_invalid_path(self):
         p = xmlrpclib.ServerProxy(URL+"/invalid")
         self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_path_query_fragment(self):
         p = xmlrpclib.ServerProxy(URL+"/foo?k=v#frag")
         self.assertEqual(p.test(), "/foo?k=v#frag")
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_path_fragment(self):
         p = xmlrpclib.ServerProxy(URL+"/foo#frag")
         self.assertEqual(p.test(), "/foo#frag")
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_path_query(self):
         p = xmlrpclib.ServerProxy(URL+"/foo?k=v")
         self.assertEqual(p.test(), "/foo?k=v")
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_empty_path(self):
         p = xmlrpclib.ServerProxy(URL)
         self.assertEqual(p.test(), "/RPC2")
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_root_path(self):
         p = xmlrpclib.ServerProxy(URL + "/")
         self.assertEqual(p.test(), "/")
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_empty_path_query(self):
         p = xmlrpclib.ServerProxy(URL + "?k=v")
         self.assertEqual(p.test(), "?k=v")
 
-    @unittest.expectedFailure # TODO: RUSTPYTHON
     @support.requires_resource('walltime')
     def test_empty_path_fragment(self):
         p = xmlrpclib.ServerProxy(URL + "#frag")
@@ -1142,7 +1133,6 @@ def test_two(self):
 
 #test special attribute access on the serverproxy, through the __call__
 #function.
-@unittest.skip("TODO: RUSTPYTHON, appears to hang")
 class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
     #ask for two keepalive requests to be handled.
     request_count=2

diff --git a/Lib/xmlrpc/client.py b/Lib/xmlrpc/client.py
@@ -245,41 +245,15 @@ def __repr__(self):
 
 ##
 # Backwards compatibility
-
 boolean = Boolean = bool
 
-##
-# Wrapper for XML-RPC DateTime values.  This converts a time value to
-# the format used by XML-RPC.
-# <p>
-# The value can be given as a datetime object, as a string in the
-# format "yyyymmddThh:mm:ss", as a 9-item time tuple (as returned by
-# time.localtime()), or an integer value (as returned by time.time()).
-# The wrapper uses time.localtime() to convert an integer to a time
-# tuple.
-#
-# @param value The time, given as a datetime object, an ISO 8601 string,
-#              a time tuple, or an integer time value.
-
 
-# Issue #13305: different format codes across platforms
-_day0 = datetime(1, 1, 1)
-def _try(fmt):
-    try:
-        return _day0.strftime(fmt) == '0001'
-    except ValueError:
-        return False
-if _try('%Y'):      # Mac OS X
-    def _iso8601_format(value):
-        return value.strftime("%Y%m%dT%H:%M:%S")
-elif _try('%4Y'):   # Linux
-    def _iso8601_format(value):
-        return value.strftime("%4Y%m%dT%H:%M:%S")
-else:
-    def _iso8601_format(value):
-        return value.strftime("%Y%m%dT%H:%M:%S").zfill(17)
-del _day0
-del _try
+def _iso8601_format(value):
+    if value.tzinfo is not None:
+        # XML-RPC only uses the naive portion of the datetime
+        value = value.replace(tzinfo=None)
+    # XML-RPC doesn't use '-' separator in the date part
+    return value.isoformat(timespec='seconds').replace('-', '')
 
 
 def _strftime(value):
@@ -850,9 +824,9 @@ def __init__(self, results):
 
     def __getitem__(self, i):
         item = self.results[i]
-        if type(item) == type({}):
+        if isinstance(item, dict):
             raise Fault(item['faultCode'], item['faultString'])
-        elif type(item) == type([]):
+        elif isinstance(item, list):
             return item[0]
         else:
             raise ValueError("unexpected type in multicall result")
@@ -1339,10 +1313,7 @@ def parse_response(self, response):
 
         p, u = self.getparser()
 
-        while 1:
-            data = stream.read(1024)
-            if not data:
-                break
+        while data := stream.read(1024):
             if self.verbose:
                 print("body:", repr(data))
             p.feed(data)

diff --git a/Lib/xmlrpc/server.py b/Lib/xmlrpc/server.py
@@ -268,17 +268,11 @@ def _marshaled_dispatch(self, data, dispatch_method = None, path = None):
         except Fault as fault:
             response = dumps(fault, allow_none=self.allow_none,
                              encoding=self.encoding)
-        except:
-            # report exception back to server
-            exc_type, exc_value, exc_tb = sys.exc_info()
-            try:
-                response = dumps(
-                    Fault(1, "%s:%s" % (exc_type, exc_value)),
-                    encoding=self.encoding, allow_none=self.allow_none,
-                    )
-            finally:
-                # Break reference cycle
-                exc_type = exc_value = exc_tb = None
+        except BaseException as exc:
+            response = dumps(
+                Fault(1, "%s:%s" % (type(exc), exc)),
+                encoding=self.encoding, allow_none=self.allow_none,
+                )
 
         return response.encode(self.encoding, 'xmlcharrefreplace')
 
@@ -368,16 +362,11 @@ def system_multicall(self, call_list):
                     {'faultCode' : fault.faultCode,
                      'faultString' : fault.faultString}
                     )
-            except:
-                exc_type, exc_value, exc_tb = sys.exc_info()
-                try:
-                    results.append(
-                        {'faultCode' : 1,
-                         'faultString' : "%s:%s" % (exc_type, exc_value)}
-                        )
-                finally:
-                    # Break reference cycle
-                    exc_type = exc_value = exc_tb = None
+            except BaseException as exc:
+                results.append(
+                    {'faultCode' : 1,
+                     'faultString' : "%s:%s" % (type(exc), exc)}
+                    )
         return results
 
     def _dispatch(self, method, params):
@@ -440,7 +429,7 @@ class SimpleXMLRPCRequestHandler(BaseHTTPRequestHandler):
 
     # Class attribute listing the accessible path components;
     # paths not on this list will result in a 404 error.
-    rpc_paths = ('/', '/RPC2')
+    rpc_paths = ('/', '/RPC2', '/pydoc.css')
 
     #if not None, encode responses larger than this, if possible
     encode_threshold = 1400 #a common MTU
@@ -634,19 +623,14 @@ def _marshaled_dispatch(self, data, dispatch_method = None, path = None):
         try:
             response = self.dispatchers[path]._marshaled_dispatch(
                data, dispatch_method, path)
-        except:
+        except BaseException as exc:
             # report low level exception back to server
             # (each dispatcher should have handled their own
             # exceptions)
-            exc_type, exc_value = sys.exc_info()[:2]
-            try:
-                response = dumps(
-                    Fault(1, "%s:%s" % (exc_type, exc_value)),
-                    encoding=self.encoding, allow_none=self.allow_none)
-                response = response.encode(self.encoding, 'xmlcharrefreplace')
-            finally:
-                # Break reference cycle
-                exc_type = exc_value = None
+            response = dumps(
+                Fault(1, "%s:%s" % (type(exc), exc)),
+                encoding=self.encoding, allow_none=self.allow_none)
+            response = response.encode(self.encoding, 'xmlcharrefreplace')
         return response
 
 class CGIXMLRPCRequestHandler(SimpleXMLRPCDispatcher):
@@ -736,9 +720,7 @@ def markup(self, text, escape=None, funcs={}, classes={}, methods={}):
                                 r'RFC[- ]?(\d+)|'
                                 r'PEP[- ]?(\d+)|'
                                 r'(self\.)?((?:\w|\.)+))\b')
-        while 1:
-            match = pattern.search(text, here)
-            if not match: break
+        while match := pattern.search(text, here):
             start, end = match.span()
             results.append(escape(text[here:start]))
 
@@ -747,10 +729,10 @@ def markup(self, text, escape=None, funcs={}, classes={}, methods={}):
                 url = escape(all).replace('"', '&quot;')
                 results.append('<a href="%s">%s</a>' % (url, url))
             elif rfc:
-                url = 'http://www.rfc-editor.org/rfc/rfc%d.txt' % int(rfc)
+                url = 'https://www.rfc-editor.org/rfc/rfc%d.txt' % int(rfc)
                 results.append('<a href="%s">%s</a>' % (url, escape(all)))
             elif pep:
-                url = 'https://www.python.org/dev/peps/pep-%04d/' % int(pep)
+                url = 'https://peps.python.org/pep-%04d/' % int(pep)
                 results.append('<a href="%s">%s</a>' % (url, escape(all)))
             elif text[end:end+1] == '(':
                 results.append(self.namelink(name, methods, funcs, classes))
@@ -801,7 +783,7 @@ def docserver(self, server_name, package_documentation, methods):
 
         server_name = self.escape(server_name)
         head = '<big><big><strong>%s</strong></big></big>' % server_name
-        result = self.heading(head, '#ffffff', '#7799ee')
+        result = self.heading(head)
 
         doc = self.markup(package_documentation, self.preformat, fdict)
         doc = doc and '<tt>%s</tt>' % doc
@@ -812,10 +794,25 @@ def docserver(self, server_name, package_documentation, methods):
         for key, value in method_items:
             contents.append(self.docroutine(value, key, funcs=fdict))
         result = result + self.bigsection(
-            'Methods', '#ffffff', '#eeaa77', ''.join(contents))
+            'Methods', 'functions', ''.join(contents))
 
         return result
 
+
+    def page(self, title, contents):
+        """Format an HTML page."""
+        css_path = "/pydoc.css"
+        css_link = (
+            '<link rel="stylesheet" type="text/css" href="%s">' %
+            css_path)
+        return '''\
+<!DOCTYPE>
+<html lang="en">
+<head>
+<meta charset="utf-8">
+<title>Python: %s</title>
+%s</head><body>%s</body></html>''' % (title, css_link, contents)
+
 class XMLRPCDocGenerator:
     """Generates documentation for an XML-RPC server.
 
@@ -907,6 +904,12 @@ class DocXMLRPCRequestHandler(SimpleXMLRPCRequestHandler):
     for documentation.
     """
 
+    def _get_css(self, url):
+        path_here = os.path.dirname(os.path.realpath(__file__))
+        css_path = os.path.join(path_here, "..", "pydoc_data", "_pydoc.css")
+        with open(css_path, mode="rb") as fp:
+            return fp.read()
+
     def do_GET(self):
         """Handles the HTTP GET request.
 
@@ -918,9 +921,15 @@ def do_GET(self):
             self.report_404()
             return
 
-        response = self.server.generate_html_documentation().encode('utf-8')
+        if self.path.endswith('.css'):
+            content_type = 'text/css'
+            response = self._get_css(self.path)
+        else:
+            content_type = 'text/html'
+            response = self.server.generate_html_documentation().encode('utf-8')
+
         self.send_response(200)
-        self.send_header("Content-type", "text/html")
+        self.send_header('Content-Type', '%s; charset=UTF-8' % content_type)
         self.send_header("Content-length", str(len(response)))
         self.end_headers()
         self.wfile.write(response)