RustPython · kingiler · Mar 1, 2024 · Mar 1, 2024 · Mar 1, 2024 · Feb 20, 2024
diff --git a/Lib/cgitb.py b/Lib/cgitb.py
@@ -74,7 +74,7 @@ def lookup(name, frame, locals):
         return 'global', frame.f_globals[name]
     if '__builtins__' in frame.f_globals:
         builtins = frame.f_globals['__builtins__']
-        if type(builtins) is type({}):
+        if isinstance(builtins, dict):
             if name in builtins:
                 return 'builtin', builtins[name]
         else:

diff --git a/Lib/colorsys.py b/Lib/colorsys.py
@@ -83,7 +83,7 @@ def rgb_to_hls(r, g, b):
     if l <= 0.5:
         s = rangec / sumc
     else:
-        s = rangec / (2.0-sumc)
+        s = rangec / (2.0-maxc-minc)  # Not always 2.0-sumc: gh-106498.
     rc = (maxc-r) / rangec
     gc = (maxc-g) / rangec
     bc = (maxc-b) / rangec

diff --git a/Lib/csv.py b/Lib/csv.py
@@ -4,17 +4,22 @@
 """
 
 import re
-from _csv import Error, writer, reader, \
+import types
+from _csv import Error, __version__, writer, reader, register_dialect, \
+                 unregister_dialect, get_dialect, list_dialects, \
+                 field_size_limit, \
                  QUOTE_MINIMAL, QUOTE_ALL, QUOTE_NONNUMERIC, QUOTE_NONE, \
+                 QUOTE_STRINGS, QUOTE_NOTNULL, \
                  __doc__
+from _csv import Dialect as _Dialect
 
-from collections import OrderedDict
 from io import StringIO
 
 __all__ = ["QUOTE_MINIMAL", "QUOTE_ALL", "QUOTE_NONNUMERIC", "QUOTE_NONE",
+           "QUOTE_STRINGS", "QUOTE_NOTNULL",
            "Error", "Dialect", "__doc__", "excel", "excel_tab",
            "field_size_limit", "reader", "writer",
-           "Sniffer",
+           "register_dialect", "get_dialect", "list_dialects", "Sniffer",
            "unregister_dialect", "__version__", "DictReader", "DictWriter",
            "unix_dialect"]
 
@@ -57,10 +62,12 @@ class excel(Dialect):
     skipinitialspace = False
     lineterminator = '\r\n'
     quoting = QUOTE_MINIMAL
+register_dialect("excel", excel)
 
 class excel_tab(excel):
     """Describe the usual properties of Excel-generated TAB-delimited files."""
     delimiter = '\t'
+register_dialect("excel-tab", excel_tab)
 
 class unix_dialect(Dialect):
     """Describe the usual properties of Unix-generated CSV files."""
@@ -70,11 +77,14 @@ class unix_dialect(Dialect):
     skipinitialspace = False
     lineterminator = '\n'
     quoting = QUOTE_ALL
+register_dialect("unix", unix_dialect)
 
 
 class DictReader:
     def __init__(self, f, fieldnames=None, restkey=None, restval=None,
                  dialect="excel", *args, **kwds):
+        if fieldnames is not None and iter(fieldnames) is fieldnames:
+            fieldnames = list(fieldnames)
         self._fieldnames = fieldnames   # list of keys for the dict
         self.restkey = restkey          # key to catch long rows
         self.restval = restval          # default value for short rows
@@ -111,7 +121,7 @@ def __next__(self):
         # values
         while row == []:
             row = next(self.reader)
-        d = OrderedDict(zip(self.fieldnames, row))
+        d = dict(zip(self.fieldnames, row))
         lf = len(self.fieldnames)
         lr = len(row)
         if lf < lr:
@@ -121,21 +131,26 @@ def __next__(self):
                 d[key] = self.restval
         return d
 
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 class DictWriter:
     def __init__(self, f, fieldnames, restval="", extrasaction="raise",
                  dialect="excel", *args, **kwds):
+        if fieldnames is not None and iter(fieldnames) is fieldnames:
+            fieldnames = list(fieldnames)
         self.fieldnames = fieldnames    # list of keys for the dict
         self.restval = restval          # for writing short dicts
-        if extrasaction.lower() not in ("raise", "ignore"):
+        extrasaction = extrasaction.lower()
+        if extrasaction not in ("raise", "ignore"):
             raise ValueError("extrasaction (%s) must be 'raise' or 'ignore'"
                              % extrasaction)
         self.extrasaction = extrasaction
         self.writer = writer(f, dialect, *args, **kwds)
 
     def writeheader(self):
         header = dict(zip(self.fieldnames, self.fieldnames))
-        self.writerow(header)
+        return self.writerow(header)
 
     def _dict_to_list(self, rowdict):
         if self.extrasaction == "raise":
@@ -151,11 +166,8 @@ def writerow(self, rowdict):
     def writerows(self, rowdicts):
         return self.writer.writerows(map(self._dict_to_list, rowdicts))
 
-# Guard Sniffer's type checking against builds that exclude complex()
-try:
-    complex
-except NameError:
-    complex = float
+    __class_getitem__ = classmethod(types.GenericAlias)
+
 
 class Sniffer:
     '''
@@ -404,14 +416,10 @@ def has_header(self, sample):
                 continue # skip rows that have irregular number of columns
 
             for col in list(columnTypes.keys()):
-
-                for thisType in [int, float, complex]:
-                    try:
-                        thisType(row[col])
-                        break
-                    except (ValueError, OverflowError):
-                        pass
-                else:
+                thisType = complex
+                try:
+                    thisType(row[col])
+                except (ValueError, OverflowError):
                     # fallback to length of string
                     thisType = len(row[col])
 
@@ -427,7 +435,7 @@ def has_header(self, sample):
         # on whether it's a header
         hasHeader = 0
         for col, colType in columnTypes.items():
-            if type(colType) == type(0): # it's a length
+            if isinstance(colType, int): # it's a length
                 if len(header[col]) != colType:
                     hasHeader += 1
                 else:

diff --git a/Lib/ftplib.py b/Lib/ftplib.py
@@ -434,10 +434,7 @@ def retrbinary(self, cmd, callback, blocksize=8192, rest=None):
         """
         self.voidcmd('TYPE I')
         with self.transfercmd(cmd, rest) as conn:
-            while 1:
-                data = conn.recv(blocksize)
-                if not data:
-                    break
+            while data := conn.recv(blocksize):
                 callback(data)
             # shutdown ssl layer
             if _SSLSocket is not None and isinstance(conn, _SSLSocket):
@@ -496,10 +493,7 @@ def storbinary(self, cmd, fp, blocksize=8192, callback=None, rest=None):
         """
         self.voidcmd('TYPE I')
         with self.transfercmd(cmd, rest) as conn:
-            while 1:
-                buf = fp.read(blocksize)
                if not buf:
-                    break
+            while buf := fp.read(blocksize):
                 conn.sendall(buf)
                 if callback:
                     callback(buf)
@@ -561,7 +555,7 @@ def dir(self, *args):
         LIST command.  (This *should* only be used for a pathname.)'''
         cmd = 'LIST'
         func = None
-        if args[-1:] and type(args[-1]) != type(''):
+        if args[-1:] and not isinstance(args[-1], str):
             args, func = args[:-1], args[-1]
         for arg in args:
             if arg:
@@ -713,28 +707,12 @@ class FTP_TLS(FTP):
         '221 Goodbye.'
         >>>
         '''
-        ssl_version = ssl.PROTOCOL_TLS_CLIENT
 
         def __init__(self, host='', user='', passwd='', acct='',
-                     keyfile=None, certfile=None, context=None,
-                     timeout=_GLOBAL_DEFAULT_TIMEOUT, source_address=None, *,
-                     encoding='utf-8'):
-            if context is not None and keyfile is not None:
-                raise ValueError("context and keyfile arguments are mutually "
-                                 "exclusive")
-            if context is not None and certfile is not None:
-                raise ValueError("context and certfile arguments are mutually "
-                                 "exclusive")
-            if keyfile is not None or certfile is not None:
-                import warnings
-                warnings.warn("keyfile and certfile are deprecated, use a "
-                              "custom context instead", DeprecationWarning, 2)
-            self.keyfile = keyfile
-            self.certfile = certfile
+                     *, context=None, timeout=_GLOBAL_DEFAULT_TIMEOUT,
+                     source_address=None, encoding='utf-8'):
             if context is None:
-                context = ssl._create_stdlib_context(self.ssl_version,
-                                                     certfile=certfile,
-                                                     keyfile=keyfile)
+                context = ssl._create_stdlib_context()
             self.context = context
             self._prot_p = False
             super().__init__(host, user, passwd, acct,
@@ -749,7 +727,7 @@ def auth(self):
             '''Set up secure control connection by using TLS/SSL.'''
             if isinstance(self.sock, ssl.SSLSocket):
                 raise ValueError("Already using TLS")
-            if self.ssl_version >= ssl.PROTOCOL_TLS:
+            if self.context.protocol >= ssl.PROTOCOL_TLS:
                 resp = self.voidcmd('AUTH TLS')
             else:
                 resp = self.voidcmd('AUTH SSL')

diff --git a/Lib/pprint.py b/Lib/pprint.py
@@ -637,19 +637,6 @@ def _recursion(object):
             % (type(object).__name__, id(object)))
 
 
-def _perfcheck(object=None):
-    import time
-    if object is None:
-        object = [("string", (1, 2), [3, 4], {5: 6, 7: 8})] * 100000
-    p = PrettyPrinter()
-    t1 = time.perf_counter()
-    p._safe_repr(object, {}, None, 0, True)
-    t2 = time.perf_counter()
-    p.pformat(object)
-    t3 = time.perf_counter()
-    print("_safe_repr:", t2 - t1)
-    print("pformat:", t3 - t2)
-
 def _wrap_bytes_repr(object, width, allowance):
     current = b''
     last = len(object) // 4 * 4
@@ -666,6 +653,3 @@ def _wrap_bytes_repr(object, width, allowance):
             current = candidate
     if current:
         yield repr(current)
-
-if __name__ == "__main__":
-    _perfcheck()
diff --git a/Lib/test/test_cmd_line.py b/Lib/test/test_cmd_line.py
@@ -278,13 +278,11 @@ def test_invalid_utf8_arg(self):
         code = 'import sys, os; s=os.fsencode(sys.argv[1]); prin
AF94
t(ascii(s))'
 
         # TODO: RUSTPYTHON
-        @unittest.expectedFailure
         def run_default(arg):
             cmd = [sys.executable, '-c', code, arg]
             return subprocess.run(cmd, stdout=subprocess.PIPE, text=True)
 
         # TODO: RUSTPYTHON
-        @unittest.expectedFailure
         def run_c_locale(arg):
             cmd = [sys.executable, '-c', code, arg]
             env = dict(os.environ)
@@ -293,7 +291,6 @@ def run_c_locale(arg):
                                   text=True, env=env)
 
         # TODO: RUSTPYTHON
-        @unittest.expectedFailure
         def run_utf8_mode(arg):
             cmd = [sys.executable, '-X', 'utf8', '-c', code, arg]
             return subprocess.run(cmd, stdout=subprocess.PIPE, text=True)
@@ -411,7 +408,8 @@ def test_empty_PYTHONPATH_issue16309(self):
             path = ":".join(sys.path)
             path = path.encode("ascii", "backslashreplace")
             sys.stdout.buffer.write(path)"""
-        rc1, out1, err1 = assert_python_ok('-c', code, PYTHONPATH="")
+        # TODO: RUSTPYTHON we must unset RUSTPYTHONPATH as well
+        rc1, out1, err1 = assert_python_ok('-c', code, PYTHONPATH="", RUSTPYTHONPATH="")
         rc2, out2, err2 = assert_python_ok('-c', code, __isolated=False)
         # regarding to Posix specification, outputs should be equal
         # for empty and unset PYTHONPATH

diff --git a/Lib/test/test_colorsys.py b/Lib/test/test_colorsys.py
@@ -69,6 +69,16 @@ def test_hls_values(self):
             self.assertTripleEqual(hls, colorsys.rgb_to_hls(*rgb))
             self.assertTripleEqual(rgb, colorsys.hls_to_rgb(*hls))
 
+    def test_hls_nearwhite(self):  # gh-106498
+        values = (
+            # rgb, hls: these do not work in reverse
+            ((0.9999999999999999, 1, 1), (0.5, 1.0, 1.0)),
+            ((1, 0.9999999999999999, 0.9999999999999999), (0.0, 1.0, 1.0)),
+        )
+        for rgb, hls in values:
+            self.assertTripleEqual(hls, colorsys.rgb_to_hls(*rgb))
+            self.assertTripleEqual((1.0, 1.0, 1.0), colorsys.hls_to_rgb(*hls))
+
     def test_yiq_roundtrip(self):
         for r in frange(0.0, 1.0, 0.2):
             for g in frange(0.0, 1.0, 0.2):