index 0c2510e161..6c3c43f11d 100644

return fname

return None

from ctypes.macholib.dyld import dyld_find as _dyld_find

def find_library(name):

possible = ['lib%s.dylib' % name,

index 0912ffd15c..959aa90522 100644

from io import BytesIO

from test.support import run_unittest

from distutils.tests import support

class FakePopen(object):

self.stdout = os.popen(cmd, 'r')

class CygwinCCompilerTestCase(support.TempdirManager,

unittest.TestCase):

index d00c48981e..5d12b4779d 100644

static_f = self.library_filename(lib, lib_type='static')

if sys.platform == 'darwin':

#

# Note that, as of Xcode 7, Apple SDKs may contain textual stub

# libraries with .tbd extensions rather than the normal .dylib

cflags = sysconfig.get_config_var('CFLAGS')

m = re.search(r'-isysroot\s*(\S+)', cflags)

if m is None:

else:

sysroot = m.group(1)

for dir in dirs:

shared = os.path.join(dir, shared_f)

dylib = os.path.join(dir, dylib_f)

index 2ce5c5b64d..6e10a0c4a5 100644

if _environ_checked:

return

try:

import pwd

os.environ['HOME'] = pwd.getpwuid(os.getuid())[5]

index f603a89f7f..92af10f21a 100644

# Bootstrap-related code ######################################################

_CASE_INSENSITIVE_PLATFORMS_STR_KEY = 'win',

_CASE_INSENSITIVE_PLATFORMS = (_CASE_INSENSITIVE_PLATFORMS_BYTES_KEY

+ _CASE_INSENSITIVE_PLATFORMS_STR_KEY)

return 'FileFinder({!r})'.format(self.path)

# Import setup ###############################################################

def _fix_up_module(ns, name, pathname, cpathname=None):

supported_loaders = _get_supported_file_loaders()

sys.path_hooks.extend([FileFinder.path_hook(*supported_loaders)])

sys.meta_path.append(PathFinder)

index 9b9d88bf58..764fd6d186 100755

# If that also doesn't work return the default values

return release, versioninfo, machine

def _java_getprop(name, default):

from java.lang import System

default in case the command should fail.

"""

# XXX Others too ?

return default

csid, cpu_number = vms_lib.getsyi('SYI$_CPU', 0)

return 'Alpha' if cpu_number >= 128 else 'VAX'

def from_subprocess():

"""

Fall back to `uname -p`

system = 'Windows'

release = 'Vista'

vals = system, node, release, version, machine

# Replace 'unknown' values with the more portable ''

_uname_cache = uname_result(*map(_unknown_as_blank, vals))

system, release, version = system_alias(system, release, version)

if system == 'Darwin':

if system == 'Windows':

# MS platforms

index 69670d9d7f..8287267e64 100644

if sys.platform == 'darwin' and sys._framework:

return f'{userbase}/lib/python/site-packages'

return f'{userbase}/lib/python{version[0]}.{version[1]}/site-packages'

index 6df5dd551e..597da09643 100644

else:

_mswindows = True

if _mswindows:

import _winapi

if _can_fork_exec:

from _posixsubprocess import fork_exec as _fork_exec

# used in methods that are called by __del__

else:

import select

import selectors

raise child_exception_type(err_msg)

"""All callers to this function MUST hold self._waitpid_lock."""

# This method is called (indirectly) by __del__, so it cannot

# refer to anything outside of its local scope.

else:

"""Check if child process has terminated. Returns returncode

attribute.

try:

if self.returncode is not None:

return self.returncode # Another thread waited.

if pid == self.pid:

self._handle_exitstatus(sts)

except OSError as e:

if _deadstate is not None:

self.returncode = _deadstate

# This happens if SIGCLD is set to be ignored or

# waiting for child processes has otherwise been

# disabled for our process. This child is dead, we

index ebe3711827..5f2456b5db 100644

'scripts': '{base}/Scripts',

'data': '{base}',

},

}

# For the OS-native venv scheme, we essentially provide an alias:

'home': 'posix_home',

'user': 'nt_user',

}

if sys.platform == 'darwin' and sys._framework:

return {

'prefix': 'posix_prefix',

'home': 'posix_home',

'user': 'osx_framework_user',

}

return {

'prefix': 'posix_prefix',

'home': 'posix_home',

if m:

release = m.group()

elif osname[:6] == "darwin":

return f"{osname}-{release}-{machine}"

index 2e6518cf24..cc08015a73 100644

"check__all__", "skip_if_buggy_ucrt_strfptime",

"check_disallow_instantiation", "check_sanitizer", "skip_if_sanitizer",

# sys

"check_impl_detail", "unix_shell", "setswitchinterval",

# network

"open_urlresource",

is_android = hasattr(sys, 'getandroidapilevel')

unix_shell = '/system/bin/sh' if is_android else '/bin/sh'

else:

unix_shell = None

is_emscripten = sys.platform == "emscripten"

is_wasi = sys.platform == "wasi"

def requires_fork():

return unittest.skipUnless(has_fork_support, "requires working os.fork()")

def requires_subprocess():

"""Used for subprocess, os.spawn calls, fd inheritance"""

index d7871d3e53..6966ee77af 100644

from multiprocessing.util import _cleanup_tests as multiprocessing_cleanup_tests

from test.test_asyncio import utils as test_utils

from test import support

from test.support import socket_helper

from test.support import threading_helper

from test.support import ALWAYS_EQ, LARGEST, SMALLEST

self._basetest_create_connection(conn_fut)

@socket_helper.skip_unless_bind_unix_socket

def test_create_unix_connection(self):

# Issue #20682: On Mac OS X Tiger, getsockname() returns a

# zero-length address for UNIX socket.

self.assertEqual(cm.exception.reason, 'CERTIFICATE_VERIFY_FAILED')

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_ssl_connection(self):

with test_utils.run_test_server(use_ssl=True) as httpd:

create_connection = functools.partial(

@socket_helper.skip_unless_bind_unix_socket

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_ssl_unix_connection(self):

# Issue #20682: On Mac OS X Tiger, getsockname() returns a

# zero-length address for UNIX socket.

return server, path

@socket_helper.skip_unless_bind_unix_socket

def test_create_unix_server(self):

proto = MyProto(loop=self.loop)

server, path = self._make_unix_server(lambda: proto)

server.close()

@unittest.skipUnless(hasattr(socket, 'AF_UNIX'), 'No UNIX Sockets')

def test_create_unix_server_path_socket_error(self):

proto = MyProto(loop=self.loop)

sock = socket.socket()

@socket_helper.skip_unless_bind_unix_socket

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_unix_server_ssl(self):

proto = MyProto(loop=self.loop)

server, path = self._make_ssl_unix_server(

server.close()

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_server_ssl_verify_failed(self):

proto = MyProto(loop=self.loop)

server, host, port = self._make_ssl_server(

@socket_helper.skip_unless_bind_unix_socket

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_unix_server_ssl_verify_failed(self):

proto = MyProto(loop=self.loop)

server, path = self._make_ssl_unix_server(

@socket_helper.skip_unless_bind_unix_socket

@unittest.skipIf(ssl is None, 'No ssl module')

def test_create_unix_server_ssl_verified(self):

proto = MyProto(loop=self.loop)

server, path = self._make_ssl_unix_server(

index 95fc7a159b..837046cd86 100644

import asyncio

from test.test_asyncio import utils as test_utils

def tearDownModule():

self._basetest_open_connection(conn_fut)

@socket_helper.skip_unless_bind_unix_socket

def test_open_unix_connection(self):

with test_utils.run_test_unix_server() as httpd:

conn_fut = asyncio.open_unix_connection(httpd.address)

@socket_helper.skip_unless_bind_unix_socket

@unittest.skipIf(ssl is None, 'No ssl module')

def test_open_unix_connection_no_loop_ssl(self):

with test_utils.run_test_unix_server(use_ssl=True) as httpd:

conn_fut = asyncio.open_unix_connection(

self._basetest_open_connection_error(conn_fut)

@socket_helper.skip_unless_bind_unix_socket

def test_open_unix_connection_error(self):

with test_utils.run_test_unix_server() as httpd:

conn_fut = asyncio.open_unix_connection(httpd.address)

self.assertEqual(messages, [])

@socket_helper.skip_unless_bind_unix_socket

def test_start_unix_server(self):

class MyServer:

index 01c1214c7f..e861c1d1e0 100644

import threading

import unittest

from unittest import mock

from test.support import os_helper

from test.support import socket_helper

@unittest.skipUnless(hasattr(socket, 'AF_UNIX'),

'UNIX Sockets are not supported')

class SelectorEventLoopUnixSocketTests(test_utils.TestCase):

def setUp(self):

index 8e98256a62..819c8ef89c 100644

import sys

import unittest

from multiprocessing import Process

from test.support.import_helper import import_module

from test.support.os_helper import TESTFN, unlink

start_len = "qq"

if (sys.platform.startswith(('netbsd', 'freebsd', 'openbsd'))

if struct.calcsize('l') == 8:

off_t = 'l'

pid_t = 'i'

index cfd8a101dc..c350e87eec 100644

import unittest

from test import support

from test.support import os_helper

from test.support import threading_helper

with open(os.path.join(self.tempdir, filename), 'wb') as f:

f.write(os_helper.TESTFN_UNDECODABLE)

response = self.request(self.base_url + '/')

# On Mac OS the HFS+ filesystem replaces bytes that aren't valid

# UTF-8 into a percent-encoded value.

for name in os.listdir(self.tempdir):

index e54a13c2d8..55dac3c3c6 100644

from test.support.script_helper import (

assert_python_ok, assert_python_failure, run_python_until_end)

from test.support import import_helper

from test.support import os_helper

from test.support import threading_helper

from test.support import warnings_helper

# On Windows and Mac OSX this test consumes large resources; It takes

# a long time to build the >2 GiB file and takes >2 GiB of disk space

# therefore the resource must be enabled to run this test.

support.requires(

'largefile',

'test requires %s bytes and a long time to run' % self.LARGE)

index 55c5cd5658..87453f29c2 100644

import tempfile

from test.support.script_helper import assert_python_ok, assert_python_failure

from test import support

from test.support import os_helper

from test.support import socket_helper

from test.support import threading_helper

# just need a name - file can't be present, or we'll get an

# 'address already in use' error.

os.remove(fn)

return fn

@unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required")

class UnixSocketHandlerTest(SocketHandlerTest):

"""Test for SocketHandler with unix sockets."""

self.assertEqual(self.log_output, "spam\neggs\n")

@unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required")

class UnixDatagramHandlerTest(DatagramHandlerTest):

"""Test for DatagramHandler using Unix sockets."""

self.assertEqual(self.log_output, b'<11>sp\xc3\xa4m\x00')

@unittest.skipUnless(hasattr(socket, "AF_UNIX"), "Unix sockets required")

class UnixSysLogHandlerTest(SysLogHandlerTest):

"""Test for SysLogHandler with Unix sockets."""

index a1a91f661b..92fde54f5a 100644

from test import support

from test.support.script_helper import assert_python_ok

import array

import io

elif sys.platform == 'wasi':

MAX_MARSHAL_STACK_DEPTH = 1500

else:

for i in range(MAX_MARSHAL_STACK_DEPTH - 2):

last.append([0])

last = last[-1]

index dfcf303942..5aabfac885 100644

from test.support import (

)

from test.support.import_helper import import_module

from test.support.os_helper import TESTFN, unlink

with open(TESTFN, "r+b") as f:

self.assertRaises(ValueError, mmap.mmap, f.fileno(), mapsize, access=4)

# Try incompatible flags, prot and access parameters.

with open(TESTFN, "r+b") as f:

self.assertRaises(ValueError, mmap.mmap, f.fileno(), mapsize,

unlink(TESTFN)

def _make_test_file(self, num_zeroes, tail):

requires('largefile',

'test requires %s bytes and a long time to run' % str(0x180000000))

f = open(TESTFN, 'w+b')

index c9f27575b5..40a66e8613 100644

from unittest import mock

from test import support

FEDORA_OS_RELEASE = """\

NAME=Fedora

def test_mac_ver(self):

res = platform.mac_ver()

# We are on a macOS system, check that the right version

# information is returned

output = subprocess.check_output(['sw_vers'], text=True)

else:

self.assertEqual(res[2], 'PowerPC')

@unittest.skipUnless(sys.platform == 'darwin', "OSX only test")

def test_mac_ver_with_fork(self):

index e643d8e5a4..65e0e2d2c6 100644

from test import support

from test.support import import_helper

from test.support import os_helper

from test.support import warnings_helper

from test.support.script_helper import assert_python_ok

"getpid", "getpgrp", "getppid", "getuid", "sync",

]

for name in NO_ARG_FUNCTIONS:

posix_func = getattr(posix, name, None)

if posix_func is not None:

with self.subTest(name):

@unittest.skipUnless(hasattr(posix, 'getresuid'),

'test needs posix.getresuid()')

check_stat(uid, gid)

self.assertRaises(OSError, chown_func, first_param, 0, -1)

check_stat(uid, gid)

# test illegal types

for t in str, float:

self.assertRaises(TypeError, chown_func, first_param, t(uid), gid)

self.assertIsInstance(hi, int)

self.assertGreaterEqual(hi, lo)

# OSX evidently just returns 15 without checking the argument.

self.assertRaises(OSError, posix.sched_get_priority_min, -23)

self.assertRaises(OSError, posix.sched_get_priority_max, -23)

index 274a8b7377..06cab9848a 100644

check_chown(dirname, uid, gid)

class TestWhich(BaseTest, unittest.TestCase):

def setUp(self):

self.assertGreaterEqual(size.lines, 0)

@unittest.skipUnless(os.isatty(sys.__stdout__.fileno()), "not on tty")

@unittest.skipUnless(hasattr(os, 'get_terminal_size'),

'need os.get_terminal_size()')

def test_stty_match(self):

index af0d2a4e6f..3b96cc476f 100644

import unittest

from test import support

from test.support import os_helper

from test.support import socket_helper

from test.support import threading_helper

# I've ordered this by protocols that have both a tcp and udp

# protocol, at least for modern Linuxes.

if (sys.platform.startswith(('freebsd', 'netbsd', 'gnukfreebsd'))

# avoid the 'echo' service on this platform, as there is an

# assumption breaking non-standard port/protocol entry

services = ('daytime', 'qotd', 'domain')

def _testFDPassCMSG_LEN(self):

self.createAndSendFDs(1)

@unittest.skipIf(AIX, "skipping, see issue #22397")

@requireAttrs(socket, "CMSG_SPACE")

def testFDPassSeparate(self):

maxcmsgs=2)