bpo-43651: PEP 597: Fix EncodingWarning in some tests (GH-25145)

* test_asyncio
* test_bz2
* test_math
* test_cmath
* test_cmd_line
* test_cmd_line_script
* test_compile
* test_contextlib
* test_profile
* ctypes/test/test_find
* test_multiprocessing
* test_configparser
* test_csv
* test_dbm_dumb
* test_decimal
* test_difflib
* os.fdopen() calls io.text_encoding() to emit EncodingWarning for right place.

Author: methane

Lib/ctypes/test/test_find.py

@@ -90,7 +90,7 @@ def test_find_on_libpath(self):
             srcname = os.path.join(d, 'dummy.c')
             libname = 'py_ctypes_test_dummy'
             dstname = os.path.join(d, 'lib%s.so' % libname)
-            with open(srcname, 'w') as f:
+            with open(srcname, 'wb') as f:
                 pass
             self.assertTrue(os.path.exists(srcname))
             # compile the file to a shared library

Lib/os.py

@@ -1020,11 +1020,13 @@ def __iter__(self):
     __all__.append("popen")
 
 # Supply os.fdopen()
-def fdopen(fd, *args, **kwargs):
+def fdopen(fd, mode="r", buffering=-1, encoding=None, *args, **kwargs):
     if not isinstance(fd, int):
         raise TypeError("invalid fd type (%s, expected integer)" % type(fd))
     import io
-    return io.open(fd, *args, **kwargs)
+    if "b" not in mode:
+        encoding = io.text_encoding(encoding)
+    return io.open(fd, mode, buffering, encoding, *args, **kwargs)
 
 
 # For testing purposes, make sure the function is available when the C

Lib/test/_test_multiprocessing.py

@@ -826,7 +826,7 @@ def test_stderr_flush(self):
         proc = self.Process(target=self._test_stderr_flush, args=(testfn,))
         proc.start()
         proc.join()
-        with open(testfn, 'r') as f:
+        with open(testfn, encoding="utf-8") as f:
             err = f.read()
             # The whole traceback was printed
             self.assertIn("ZeroDivisionError", err)
@@ -836,14 +836,14 @@ def test_stderr_flush(self):
     @classmethod
     def _test_stderr_flush(cls, testfn):
         fd = os.open(testfn, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
-        sys.stderr = open(fd, 'w', closefd=False)
+        sys.stderr = open(fd, 'w', encoding="utf-8", closefd=False)
         1/0 # MARKER
 
 
     @classmethod
     def _test_sys_exit(cls, reason, testfn):
         fd = os.open(testfn, os.O_WRONLY | os.O_CREAT | os.O_EXCL)
-        sys.stderr = open(fd, 'w', closefd=False)
+        sys.stderr = open(fd, 'w', encoding="utf-8", closefd=False)
         sys.exit(reason)
 
     def test_sys_exit(self):
@@ -864,7 +864,7 @@ def test_sys_exit(self):
             join_process(p)
             self.assertEqual(p.exitcode, 1)
 
-            with open(testfn, 'r') as f:
+            with open(testfn, encoding="utf-8") as f:
                 content = f.read()
             self.assertEqual(content.rstrip(), str(reason))
 
@@ -1118,7 +1118,7 @@ def test_task_done(self):
     def test_no_import_lock_contention(self):
         with os_helper.temp_cwd():
             module_name = 'imported_by_an_imported_module'
-            with open(module_name + '.py', 'w') as f:
+            with open(module_name + '.py', 'w', encoding="utf-8") as f:
                 f.write("""if 1:
                     import multiprocessing
 

Lib/test/multibytecodec_support.py

@@ -296,7 +296,7 @@ def setUp(self):
             self.skipTest("Could not retrieve "+self.mapfileurl)
 
     def open_mapping_file(self):
-        return support.open_urlresource(self.mapfileurl)
+        return support.open_urlresource(self.mapfileurl, encoding="utf-8")
 
     def test_mapping_file(self):
         if self.mapfileurl.endswith('.xml'):

Lib/test/test_asyncio/test_base_events.py

@@ -2096,7 +2096,7 @@ def test_blocking_socket(self):
 
     def test_nonbinary_file(self):
         sock = self.make_socket()
-        with open(os_helper.TESTFN, 'r') as f:
+        with open(os_helper.TESTFN, encoding="utf-8") as f:
             with self.assertRaisesRegex(ValueError, "binary mode"):
                 self.run_loop(self.loop.sock_sendfile(sock, f))
 

Lib/test/test_asyncio/test_events.py

@@ -1349,7 +1349,7 @@ def test_unclosed_pipe_transport(self):
 
         rpipe, wpipe = os.pipe()
         rpipeobj = io.open(rpipe, 'rb', 1024)
-        wpipeobj = io.open(wpipe, 'w', 1024)
+        wpipeobj = io.open(wpipe, 'w', 1024, encoding="utf-8")
 
         async def connect():
             read_transport, _ = await loop.connect_read_pipe(

Lib/test/test_bz2.py

@@ -922,14 +922,14 @@ def test_text_modes(self):
         for mode in ("wt", "xt"):
             if mode == "xt":
                 unlink(self.filename)
-            with self.open(self.filename, mode) as f:
+            with self.open(self.filename, mode, encoding="ascii") as f:
                 f.write(text)
             with open(self.filename, "rb") as f:
                 file_data = ext_decompress(f.read()).decode("ascii")
                 self.assertEqual(file_data, text_native_eol)
-            with self.open(self.filename, "rt") as f:
+            with self.open(self.filename, "rt", encoding="ascii") as f:
                 self.assertEqual(f.read(), text)
-            with self.open(self.filename, "at") as f:
+            with self.open(self.filename, "at", encoding="ascii") as f:
                 f.write(text)
             with open(self.filename, "rb") as f:
                 file_data = ext_decompress(f.read()).decode("ascii")
@@ -938,7 +938,8 @@ def test_text_modes(self):
     def test_x_mode(self):
         for mode in ("x", "xb", "xt"):
             unlink(self.filename)
-            with self.open(self.filename, mode) as f:
+            encoding = "utf-8" if "t" in mode else None
+            with self.open(self.filename, mode, encoding=encoding) as f:
                 pass
             with self.assertRaises(FileExistsError):
                 with self.open(self.filename, mode) as f:
@@ -950,7 +951,7 @@ def test_fileobj(self):
         with self.open(BytesIO(self.DATA), "rb") as f:
             self.assertEqual(f.read(), self.TEXT)
         text = self.TEXT.decode("ascii")
-        with self.open(BytesIO(self.DATA), "rt") as f:
+        with self.open(BytesIO(self.DATA), "rt", encoding="utf-8") as f:
             self.assertEqual(f.read(), text)
 
     def test_bad_params(self):
@@ -989,9 +990,9 @@ def test_encoding_error_handler(self):
     def test_newline(self):
         # Test with explicit newline (universal newline mode disabled).
         text = self.TEXT.decode("ascii")
-        with self.open(self.filename, "wt", newline="\n") as f:
+        with self.open(self.filename, "wt", encoding="utf-8", newline="\n") as f:
             f.write(text)
-        with self.open(self.filename, "rt", newline="\r") as f:
+        with self.open(self.filename, "rt", encoding="utf-8", newline="\r") as f:
             self.assertEqual(f.readlines(), [text])
 
 

Lib/test/test_cmath.py

@@ -60,7 +60,7 @@ class CMathTests(unittest.TestCase):
     test_functions.append(lambda x : cmath.log(14.-27j, x))
 
     def setUp(self):
-        self.test_values = open(test_file)
+        self.test_values = open(test_file, encoding="utf-8")
 
     def tearDown(self):
         self.test_values.close()

Lib/test/test_cmd_line.py

@@ -512,7 +512,7 @@ def test_del___main__(self):
         # the dict whereas the module was destroyed
         filename = os_helper.TESTFN
         self.addCleanup(os_helper.unlink, filename)
-        with open(filename, "w") as script:
+        with open(filename, "w", encoding="utf-8") as script:
             print("import sys", file=script)
             print("del sys.modules['__main__']", file=script)
         assert_python_ok(filename)
@@ -549,9 +549,9 @@ def test_isolatedmode(self):
         with os_helper.temp_cwd() as tmpdir:
             fake = os.path.join(tmpdir, "uuid.py")
             main = os.path.join(tmpdir, "main.py")
-            with open(fake, "w") as f:
+            with open(fake, "w", encoding="utf-8") as f:
                 f.write("raise RuntimeError('isolated mode test')\n")
-            with open(main, "w") as f:
+            with open(main, "w", encoding="utf-8") as f:
                 f.write("import uuid\n")
                 f.write("print('ok')\n")
             self.assertRaises(subprocess.CalledProcessError,

Lib/test/test_cmd_line_script.py

@@ -400,7 +400,7 @@ def test_issue8202_dash_c_file_ignored(self):
         # does not alter the value of sys.path[0]
         with os_helper.temp_dir() as script_dir:
             with os_helper.change_cwd(path=script_dir):
-                with open("-c", "w") as f:
+                with open("-c", "w", encoding="utf-8") as f:
                     f.write("data")
                     rc, out, err = assert_python_ok('-c',
                         'import sys; print("sys.path[0]==%r" % sys.path[0])',
@@ -416,7 +416,7 @@ def test_issue8202_dash_m_file_ignored(self):
         with os_helper.temp_dir() as script_dir:
             script_name = _make_test_script(script_dir, 'other')
             with os_helper.change_cwd(path=script_dir):
-                with open("-m", "w") as f:
+                with open("-m", "w", encoding="utf-8") as f:
                     f.write("data")
                     rc, out, err = assert_python_ok('-m', 'other', *example_args,
                                                     __isolated=False)
@@ -429,7 +429,7 @@ def test_issue20884(self):
         # will be failed.
         with os_helper.temp_dir() as script_dir:
             script_name = os.path.join(script_dir, "issue20884.py")
-            with open(script_name, "w", newline='\n') as f:
+            with open(script_name, "w", encoding="latin1", newline='\n') as f:
                 f.write("#coding: iso-8859-1\n")
                 f.write('"""\n')
                 for _ in range(30):