Merge pull request RustPython#3926 from youknowone/unittest

update signal and venv

Author: youknowone

Lib/signal.py

@@ -1,6 +1,5 @@
 import _signal
 from _signal import *
-from functools import wraps as _wraps
 from enum import IntEnum as _IntEnum
 
 _globals = globals()
@@ -42,6 +41,16 @@ def _enum_to_int(value):
         return value
 
 
+# Similar to functools.wraps(), but only assign __doc__.
+# __module__ should be preserved,
+# __name__ and __qualname__ are already fine,
+# __annotations__ is not set.
+def _wraps(wrapped):
+    def decorator(wrapper):
+        wrapper.__doc__ = wrapped.__doc__
+        return wrapper
+    return decorator
+
 @_wraps(_signal.signal)
 def signal(signalnum, handler):
     handler = _signal.signal(_enum_to_int(signalnum), _enum_to_int(handler))
@@ -59,7 +68,6 @@ def getsignal(signalnum):
     def pthread_sigmask(how, mask):
         sigs_set = _signal.pthread_sigmask(how, mask)
         return set(_int_to_enum(x, Signals) for x in sigs_set)
-    pthread_sigmask.__doc__ = _signal.pthread_sigmask.__doc__
 
 
 if 'sigpending' in _globals:
@@ -73,7 +81,6 @@ def sigpending():
     def sigwait(sigset):
         retsig = _signal.sigwait(sigset)
         return _int_to_enum(retsig, Signals)
-    sigwait.__doc__ = _signal.sigwait
 
 
 if 'valid_signals' in _globals:

Lib/test/test_signal.py

@@ -1,4 +1,5 @@
 import errno
+import inspect
 import os
 import random
 import signal
@@ -33,6 +34,14 @@ def test_enums(self):
                 self.assertIsInstance(sig, signal.Signals)
                 self.assertEqual(sys.platform, "win32")
 
+    def test_functions_module_attr(self):
+        # Issue #27718: If __all__ is not defined all non-builtin functions
+        # should have correct __module__ to be displayed by pydoc.
+        for name in dir(signal):
+            value = getattr(signal, name)
+            if inspect.isroutine(value) and not inspect.isbuiltin(value):
+                self.assertEqual(value.__module__, 'signal')
+
 
 @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
 class PosixTests(unittest.TestCase):
@@ -552,16 +561,20 @@ def handler(signum, frame):
         else:
             write.setblocking(False)
 
-        # Start with large chunk size to reduce the
-        # number of send needed to fill the buffer.
         written = 0
-        for chunk_size in (2 ** 16, 2 ** 8, 1):
+        if sys.platform == "vxworks":
+            CHUNK_SIZES = (1,)
+        else:
+            # Start with large chunk size to reduce the
+            # number of send needed to fill the buffer.
+            CHUNK_SIZES = (2 ** 16, 2 ** 8, 1)
+        for chunk_size in CHUNK_SIZES:
             chunk = b"x" * chunk_size
             try:
                 while True:
                     write.send(chunk)
                     written += chunk_size
-            except (BlockingIOError, socket.timeout):
+            except (BlockingIOError, TimeoutError):
                 pass
 
         print(f"%s bytes written into the socketpair" % written, flush=True)
@@ -628,6 +641,7 @@ def handler(signum, frame):
 
 
 @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
+@unittest.skipUnless(hasattr(signal, 'siginterrupt'), "needs signal.siginterrupt()")
 class SiginterruptTest(unittest.TestCase):
 
     def readpipe_interrupted(self, interrupt):
@@ -678,7 +692,7 @@ def handler(signum, frame):
                 # wait until the child process is loaded and has started
                 first_line = process.stdout.readline()
 
-                stdout, stderr = process.communicate(timeout=5.0)
+                stdout, stderr = process.communicate(timeout=support.SHORT_TIMEOUT)
             except subprocess.TimeoutExpired:
                 process.kill()
                 return False
@@ -717,6 +731,8 @@ def test_siginterrupt_off(self):
 
 
 @unittest.skipIf(sys.platform == "win32", "Not valid on Windows")
+@unittest.skipUnless(hasattr(signal, 'getitimer') and hasattr(signal, 'setitimer'),
+                         "needs signal.getitimer() and signal.setitimer()")
 class ItimerTest(unittest.TestCase):
     def setUp(self):
         self.hndl_called = False
@@ -1240,7 +1256,7 @@ def second_handler(signum=None, frame=None):
         self.setsig(signal.SIGALRM, second_handler)  # for ITIMER_REAL
 
         expected_sigs = 0
-        deadline = time.monotonic() + 15.0
+        deadline = time.monotonic() + support.SHORT_TIMEOUT
 
         while expected_sigs < N:
             os.kill(os.getpid(), signal.SIGPROF)
@@ -1274,7 +1290,7 @@ def handler(signum, frame):
         self.setsig(signal.SIGALRM, handler)  # for ITIMER_REAL
 
         expected_sigs = 0
-        deadline = time.monotonic() + 15.0
+        deadline = time.monotonic() + support.SHORT_TIMEOUT
 
         while expected_sigs < N:
             # Hopefully the SIGALRM will be received somewhere during
@@ -1336,7 +1352,7 @@ def cycle_handlers():
                     # race condition, check it.
                     self.assertIsInstance(cm.unraisable.exc_value, OSError)
                     self.assertIn(
-                        f"Signal {signum} ignored due to race condition",
+                        f"Signal {signum:d} ignored due to race condition",
                         str(cm.unraisable.exc_value))
                     ignored = True
 
@@ -1387,6 +1403,27 @@ def handler(a, b):
         self.assertTrue(is_ok)
 
 
+class PidfdSignalTest(unittest.TestCase):
+
+    @unittest.skipUnless(
+        hasattr(signal, "pidfd_send_signal"),
+        "pidfd support not built in",
+    )
+    def test_pidfd_send_signal(self):
+        with self.assertRaises(OSError) as cm:
+            signal.pidfd_send_signal(0, signal.SIGINT)
+        if cm.exception.errno == errno.ENOSYS:
+            self.skipTest("kernel does not support pidfds")
+        elif cm.exception.errno == errno.EPERM:
+            self.skipTest("Not enough privileges to use pidfs")
+        self.assertEqual(cm.exception.errno, errno.EBADF)
+        my_pidfd = os.open(f'/proc/{os.getpid()}', os.O_DIRECTORY)
+        self.addCleanup(os.close, my_pidfd)
+        with self.assertRaisesRegex(TypeError, "^siginfo must be None$"):
+            signal.pidfd_send_signal(my_pidfd, signal.SIGINT, object(), 0)
+        with self.assertRaises(KeyboardInterrupt):
+            signal.pidfd_send_signal(my_pidfd, signal.SIGINT)
+
 def tearDownModule():
     support.reap_children()
 

Lib/test/test_venv.py

@@ -5,8 +5,7 @@
 Licensed to the PSF under a contributor agreement.
 """
 
-# pip isn't working yet
-# import ensurepip
+import ensurepip
 import os
 import os.path
 import re
@@ -15,12 +14,12 @@
 import subprocess
 import sys
 import tempfile
-from test.support import captured_stdout, captured_stderr, requires_zlib
+from test.support import (captured_stdout, captured_stderr, requires_zlib,
+                          skip_if_broken_multiprocessing_synchronize)
 from test.support.os_helper import (can_symlink, EnvironmentVarGuard, rmtree)
-from test.support.import_helper import import_module
-import threading
 import unittest
 import venv
+from unittest.mock import patch
 
 try:
     import ctypes
@@ -80,8 +79,8 @@ def run_with_capture(self, func, *args, **kwargs):
     def get_env_file(self, *args):
         return os.path.join(self.env_dir, *args)
 
-    def get_text_file_contents(self, *args):
-        with open(self.get_env_file(*args), 'r') as f:
+    def get_text_file_contents(self, *args, encoding='utf-8'):
+        with open(self.get_env_file(*args), 'r', encoding=encoding) as f:
             result = f.read()
         return result
 
@@ -139,6 +138,45 @@ def test_prompt(self):
         self.assertEqual(context.prompt, '(My prompt) ')
         self.assertIn("prompt = 'My prompt'\n", data)
 
+        rmtree(self.env_dir)
+        builder = venv.EnvBuilder(prompt='.')
+        cwd = os.path.basename(os.getcwd())
+        self.run_with_capture(builder.create, self.env_dir)
+        context = builder.ensure_directories(self.env_dir)
+        data = self.get_text_file_contents('pyvenv.cfg')
+        self.assertEqual(context.prompt, '(%s) ' % cwd)
+        self.assertIn("prompt = '%s'\n" % cwd, data)
+
+    def test_upgrade_dependencies(self):
+        builder = venv.EnvBuilder()
+        bin_path = 'Scripts' if sys.platform == 'win32' else 'bin'
+        python_exe = os.path.split(sys.executable)[1]
+        with tempfile.TemporaryDirectory() as fake_env_dir:
+            expect_exe = os.path.normcase(
+                os.path.join(fake_env_dir, bin_path, python_exe)
+            )
+            if sys.platform == 'win32':
+                expect_exe = os.path.normcase(os.path.realpath(expect_exe))
+
+            def pip_cmd_checker(cmd):
+                cmd[0] = os.path.normcase(cmd[0])
+                self.assertEqual(
+                    cmd,
+                    [
+                        expect_exe,
+                        '-m',
+                        'pip',
+                        'install',
+                        '--upgrade',
+                        'pip',
+                        'setuptools'
+                    ]
+                )
+
+            fake_context = builder.ensure_directories(fake_env_dir)
+            with patch('venv.subprocess.check_call', pip_cmd_checker):
+                builder.upgrade_dependencies(fake_context)
+
     @requireVenvCreate
     def test_prefixes(self):
         """
@@ -325,10 +363,11 @@ def test_multiprocessing(self):
         """
         Test that the multiprocessing is able to spawn.
         """
-        # Issue bpo-36342: Instanciation of a Pool object imports the
+        # bpo-36342: Instantiation of a Pool object imports the
         # multiprocessing.synchronize module. Skip the test if this module
         # cannot be imported.
-        import_module('multiprocessing.synchronize')
+        skip_if_broken_multiprocessing_synchronize()
+
         rmtree(self.env_dir)
         self.run_with_capture(venv.create, self.env_dir)
         envpy = os.path.join(os.path.realpath(self.env_dir),
@@ -413,7 +452,7 @@ def do_test_with_pip(self, system_site_packages):
             # pip's cross-version compatibility may trigger deprecation
             # warnings in current versions of Python. Ensure related
             # environment settings don't cause venv to fail.
-            envvars["PYTHONWARNINGS"] = "e"
+            envvars["PYTHONWARNINGS"] = "ignore"
             # ensurepip is different enough from a normal pip invocation
             # that we want to ensure it ignores the normal pip environment
             # variable settings. We set PIP_NO_INSTALL here specifically
@@ -452,7 +491,8 @@ def do_test_with_pip(self, system_site_packages):
         # Ensure pip is available in the virtual environment
         envpy = os.path.join(os.path.realpath(self.env_dir), self.bindir, self.exe)
         # Ignore DeprecationWarning since pip code is not part of Python
-        out, err = check_output([envpy, '-W', 'ignore::DeprecationWarning', '-I',
+        out, err = check_output([envpy, '-W', 'ignore::DeprecationWarning',
+               '-W', 'ignore::ImportWarning', '-I',
                '-m', 'pip', '--version'])
         # We force everything to text, so unittest gives the detailed diff
         # if we get unexpected results
@@ -468,8 +508,12 @@ def do_test_with_pip(self, system_site_packages):
         # Check the private uninstall command provided for the Windows
         # installers works (at least in a virtual environment)
         with EnvironmentVarGuard() as envvars:
+            # It seems ensurepip._uninstall calls subprocesses which do not
+            # inherit the interpreter settings.
+            envvars["PYTHONWARNINGS"] = "ignore"
             out, err = check_output([envpy,
-                '-W', 'ignore::DeprecationWarning', '-I',
+                '-W', 'ignore::DeprecationWarning',
+                '-W', 'ignore::ImportWarning', '-I',
                 '-m', 'ensurepip._uninstall'])
         # We force everything to text, so unittest gives the detailed diff
         # if we get unexpected results
@@ -481,7 +525,7 @@ def do_test_with_pip(self, system_site_packages):
         #    executing pip with sudo, you may want sudo's -H flag."
         # where $HOME is replaced by the HOME environment variable.
         err = re.sub("^(WARNING: )?The directory .* or its parent directory "
-                     "is not owned by the current user .*$", "",
+                     "is not owned or is not writable by the current user.*$", "",
                      err, flags=re.MULTILINE)
         self.assertEqual(err.rstrip(), "")
         # Being fairly specific regarding the expected behaviour for the
@@ -497,10 +541,8 @@ def do_test_with_pip(self, system_site_packages):
             self.assert_pip_not_installed()
 
     # Issue #26610: pip/pep425tags.py requires ctypes
-    # TODO: RUSTPYTHON
     @unittest.skipUnless(ctypes, 'pip requires ctypes')
-    @requires_zlib
-    @unittest.expectedFailure
+    @requires_zlib()
     def test_with_pip(self):
         self.do_test_with_pip(False)
         self.do_test_with_pip(True)

Lib/venv/__init__.py

@@ -142,6 +142,20 @@ def create_if_needed(d):
         context.bin_name = binname
         context.env_exe = os.path.join(binpath, exename)
         create_if_needed(binpath)
+        # Assign and update the command to use when launching the newly created
+        # environment, in case it isn't simply the executable script (e.g. bpo-45337)
+        context.env_exec_cmd = context.env_exe
+        if sys.platform == 'win32':
+            # bpo-45337: Fix up env_exec_cmd to account for file system redirections.
+            # Some redirects only apply to CreateFile and not CreateProcess
+            real_env_exe = os.path.realpath(context.env_exe)
+            if os.path.normcase(real_env_exe) != os.path.normcase(context.env_exe):
+                logger.warning('Actual environment location may have moved due to '
+                               'redirects, links or junctions.\n'
+                               '  Requested location: "%s"\n'
+                               '  Actual location:    "%s"',
+                               context.env_exe, real_env_exe)
+                context.env_exec_cmd = real_env_exe
         return context
 
     def create_configuration(self, context):
@@ -267,8 +281,9 @@ def setup_python(self, context):
                         os.path.normcase(f).startswith(('python', 'vcruntime'))
                     ]
             else:
-                suffixes = ['python.exe', 'python_d.exe', 'pythonw.exe',
-                            'pythonw_d.exe']
+                suffixes = {'python.exe', 'python_d.exe', 'pythonw.exe', 'pythonw_d.exe'}
+                base_exe = os.path.basename(context.env_exe)
+                suffixes.add(base_exe)
 
             for suffix in suffixes:
                 src = os.path.join(dirname, suffix)
@@ -290,15 +305,11 @@ def setup_python(self, context):
 
     def _setup_pip(self, context):
         """Installs or upgrades pip in a virtual environment"""
-        # TODO: RustPython
-        msg = ("Pip isn't supported yet. To create a virtual environment"
-               "without pip, call venv with the --without-pip flag.")
-        raise NotImplementedError(msg)
         # We run ensurepip in isolated mode to avoid side effects from
         # environment vars, the current directory and anything else
         # intended for the global Python environment
-        cmd = [context.env_exe, '-Im', 'ensurepip', '--upgrade',
-                                                    '--default-pip']
+        cmd = [context.env_exec_cmd, '-Im', 'ensurepip', '--upgrade',
+                                                         '--default-pip']
         subprocess.check_output(cmd, stderr=subprocess.STDOUT)
 
     def setup_scripts(self, context):
@@ -398,11 +409,7 @@ def upgrade_dependencies(self, context):
         logger.debug(
             f'Upgrading {CORE_VENV_DEPS} packages in {context.bin_path}'
         )
-        if sys.platform == 'win32':
-            python_exe = os.path.join(context.bin_path, 'python.exe')
-        else:
-            python_exe = os.path.join(context.bin_path, 'python')
-        cmd = [python_exe, '-m', 'pip', 'install', '--upgrade']
+        cmd = [context.env_exec_cmd, '-m', 'pip', 'install', '--upgrade']
         cmd.extend(CORE_VENV_DEPS)
         subprocess.check_call(cmd)