pypy/module/select/test/test_select.py::AppTestSelectWithSockets::()::test_read_end_closed
self = <CallInfo when='call' exception: Subprocess failed with exit code 1:
Traceback (most recent call last):
File "/tmp/b.../usession-py3.11-1171/src.py", line 126, in test_read_end_closed
assert iwtd == owtd == [writeend]
AssertionError
>
func = <function <lambda> at 0x00007f78c34e11f0>, when = 'call'
def __init__(self, func, when):
#: context of invocation: one of "setup", "call",
#: "teardown", "memocollect"
self.when = when
self.start = time()
try:
> self.result = func()
../_pytest/runner.py:150:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> return CallInfo(lambda: ihook(item=item, **kwds), when=when)
../_pytest/runner.py:138:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_HookCaller 'pytest_runtest_call'>
kwargs = {'__multicall__': <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_read_end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>, 'item': <AppTestMethod 'test_read_end_closed'>}
def __call__(self, **kwargs):
assert not self.is_historic()
> return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
../_pytest/vendored_packages/pluggy.py:724:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.config.PytestPluginManager object at 0x00007f78b0052608>
hook = <_HookCaller 'pytest_runtest_call'>, methods = []
kwargs = {'__multicall__': <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_read_end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>, 'item': <AppTestMethod 'test_read_end_closed'>}
def _hookexec(self, hook, methods, kwargs):
# called from all hookcaller instances.
# enable_tracing will set its own wrapping function at self._inner_hookexec
> return self._inner_hookexec(hook, methods, kwargs)
../_pytest/vendored_packages/pluggy.py:338:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
hook = <_HookCaller 'pytest_runtest_call'>, methods = []
kwargs = {'__multicall__': <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_read_end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>, 'item': <AppTestMethod 'test_read_end_closed'>}
self._inner_hookexec = lambda hook, methods, kwargs: \
> _MultiCall(methods, kwargs, hook.spec_opts).execute()
../_pytest/vendored_packages/pluggy.py:333:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_read_end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>
def execute(self):
all_kwargs = self.kwargs
self.results = results = []
firstresult = self.specopts.get("firstresult")
while self.hook_impls:
hook_impl = self.hook_impls.pop()
args = [all_kwargs[argname] for argname in hook_impl.argnames]
if hook_impl.hookwrapper:
> return _wrapped_call(hook_impl.function(*args), self.execute)
../_pytest/vendored_packages/pluggy.py:595:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
wrap_controller = <generator object pytest_runtest_call at 0x00007f78b0e0c228>
func = <bound method _MultiCall.execute of <_MultiCall 0 results, 0 meths, kwargs={'i...end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>>
def _wrapped_call(wrap_controller, func):
""" Wrap calling to a function with a generator which needs to yield
exactly once. The yield point will trigger calling the wrapped function
and return its _CallOutcome to the yield point. The generator then needs
to finish (raise StopIteration) in order for the wrapped call to complete.
"""
try:
next(wrap_controller) # first yield
except StopIteration:
_raise_wrapfail(wrap_controller, "did not yield")
call_outcome = _CallOutcome(func)
try:
wrap_controller.send(call_outcome)
_raise_wrapfail(wrap_controller, "has second yield")
except StopIteration:
pass
> return call_outcome.get_result()
../_pytest/vendored_packages/pluggy.py:253:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.vendored_packages.pluggy._CallOutcome instance at 0x00007f78b0dfeda0>
def get_result(self):
if self.excinfo is None:
return self.result
else:
ex = self.excinfo
if _py3:
raise ex[1].with_traceback(ex[2])
> _reraise(*ex) # noqa
../_pytest/vendored_packages/pluggy.py:279:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.vendored_packages.pluggy._CallOutcome instance at 0x00007f78b0dfeda0>
func = <bound method _MultiCall.execute of <_MultiCall 0 results, 0 meths, kwargs={'i...end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>>
def __init__(self, func):
try:
> self.result = func()
../_pytest/vendored_packages/pluggy.py:264:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_MultiCall 0 results, 0 meths, kwargs={'item': <AppTestMethod 'test_read_end_closed'>, '__multicall__': <_MultiCall 0 results, 0 meths, kwargs={...}>}>
def execute(self):
all_kwargs = self.kwargs
self.results = results = []
firstresult = self.specopts.get("firstresult")
while self.hook_impls:
hook_impl = self.hook_impls.pop()
args = [all_kwargs[argname] for argname in hook_impl.argnames]
if hook_impl.hookwrapper:
return _wrapped_call(hook_impl.function(*args), self.execute)
> res = hook_impl.function(*args)
../_pytest/vendored_packages/pluggy.py:596:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <AppTestMethod 'test_read_end_closed'>
def pytest_runtest_call(item):
try:
> item.runtest()
../_pytest/runner.py:91:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <AppTestMethod 'test_read_end_closed'>
def runtest(self):
target = self.obj
src = extract_docstring_if_empty_function(target.im_func)
space = target.im_self.space
if self.config.option.runappdirect:
appexec_definitions = self.parent.obj.__dict__
spaceconfig = getattr(self.parent.obj, 'spaceconfig', None)
usemodules = spaceconfig.get('usemodules') if spaceconfig else None
return run_with_python(self.config.option.python, src, usemodules,
> **appexec_definitions)
tool/pytest/apptest.py:327:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
python_ = '/Users/matti/build-worker-x86_64/pypy-c-jit-macos-x86-64/build/pypy/goal/pypy3.11-c'
target_ = <function test_read_end_closed at 0x00007f78c203b358>
usemodules = ['select', '_socket', 'time', '_thread']
definitions = {'getpair': <bound method AppTestSelectWithSockets.w_getpair of <pypy.module.select.test.test_select.AppTestSelectWith...ockets.w_make_server of <pypy.module.select.test.test_select.AppTestSelectWithSockets instance at 0x00007f78b0dfeea0>>}
defs = ['def w_getpair(self):\n """Helper method which returns a pair of connected sockets."""\n import _socket\n im...ission denied"\n print(e)\n else:\n raise(e)', 'type(self).make_server = w_make_server\n']
symbol = 'make_server'
value = <bound method AppTestSelectWithSockets.w_make_server of <pypy.module.select.test.test_select.AppTestSelectWithSockets instance at 0x00007f78b0dfeea0>>
code = <py._code.code.Code object at 0x00007f78b0e0aad8>
def run_with_python(python_, target_, usemodules, **definitions):
if python_ is None:
py.test.skip("Cannot find the default python3 interpreter to run with -A")
defs = []
for symbol, value in sorted(definitions.items()):
if isinstance(value, tuple) and isinstance(value[0], py.code.Source):
code, args = value
defs.append(str(code))
arg_repr = []
for arg in args:
if isinstance(arg, types.FunctionType):
arg_repr.append(arg.__name__)
elif isinstance(arg, types.MethodType):
arg_repr.append(arg.__name__)
else:
arg_repr.append(py3k_repr(arg))
args = ', '.join(arg_repr)
defs.append("self.%s = anonymous(%s)\n" % (symbol, args))
elif isinstance(value, types.MethodType):
# "def w_method(self)"
code = py.code.Code(value)
defs.append(str(code.source()))
defs.append("type(self).%s = %s\n" % (symbol, value.__name__))
elif isinstance(value, types.ModuleType):
name = value.__name__
defs.append("import %s; self.%s = %s\n" % (name, symbol, name))
elif isinstance(value, (str, unicode, int, long, float, list, tuple,
dict)) or value is None:
defs.append("self.%s = %s\n" % (symbol, py3k_repr(value)))
check_usemodules = ''
if usemodules:
usemodules = [_rename_module(name) for name in usemodules]
check_usemodules = """\
missing = set(%r).difference(sys.builtin_module_names)
if missing:
if not hasattr(sys, 'pypy_version_info'):
# They may be extension modules on CPython
name = None
for name in missing.copy():
if name in ['cpyext', '_cffi_backend', '_rawffi']:
missing.remove(name)
continue
try:
__import__(name)
except ImportError:
pass
else:
missing.remove(name)
del name
if missing:
sys.exit(81)
del missing""" % usemodules
source = list(py.code.Source(target_))
while source[0].startswith(('@py.test.mark.', '@pytest.mark.')):
source.pop(0)
source = source[1:]
pyfile = udir.join('src.py')
if isinstance(target_, str):
# Special case of a docstring; the function name is the first word.
target_name = target_.split('(', 1)[0]
else:
target_name = target_.__name__
with pyfile.open('w') as f:
f.write(helpers % (pypyroot, check_usemodules))
f.write('\n'.join(defs))
f.write('def %s():\n' % target_name)
f.write('\n'.join(source))
f.write("\ntry:\n %s()\n" % target_name)
f.write('finally:\n print("===aefwuiheawiu===")')
helper_dir = os.path.join(pypydir, 'tool', 'cpyext')
env = os.environ.copy()
env['PYTHONPATH'] = helper_dir
res, stdout, stderr = run_subprocess(
python_, [str(pyfile)], env=env)
print pyfile.read()
print "from", str(pyfile)
print >> sys.stdout, stdout
print >> sys.stderr, stderr
if res == 81:
py.test.skip('%r was not compiled w/ required usemodules: %r' %
(python_, usemodules))
elif res != 0:
raise AssertionError(
> "Subprocess failed with exit code %s:\n%s" % (res, stderr))
E AssertionError: Subprocess failed with exit code 1:
E Traceback (most recent call last):
E File "/tmp/buildbot-x86_64/usession-py3.11-1171/src.py", line 131, in <module>
E test_read_end_closed()
E File "/tmp/buildbot-x86_64/usession-py3.11-1171/src.py", line 126, in test_read_end_closed
E assert iwtd == owtd == [writeend]
E AssertionError
tool/pytest/apptest.py:239: AssertionError
---------- Captured stdout call ----------
# -*- encoding: utf-8 -*-
if 1:
import sys
sys.path.append('/Users/matti/build-worker-x86_64/pypy-c-jit-macos-x86-64/build')
missing = set(['select', '_socket', 'time', '_thread']).difference(sys.builtin_module_names)
if missing:
if not hasattr(sys, 'pypy_version_info'):
# They may be extension modules on CPython
name = None
for name in missing.copy():
if name in ['cpyext', '_cffi_backend', '_rawffi']:
missing.remove(name)
continue
try:
__import__(name)
except ImportError:
pass
else:
missing.remove(name)
del name
if missing:
sys.exit(81)
del missing
def skip(message):
print(message)
raise SystemExit(0)
__builtins__.skip = skip
__builtins__.py3k_skip = skip
class ExceptionWrapper:
pass
def raises(exc, *args, **kwargs):
if not args:
return RaisesContext(exc)
func = args[0]
args = args[1:]
import os
try:
if isinstance(func, str):
if func.startswith((' ', os.linesep, '\n')):
# it's probably an indented block, so we prefix if True:
# to avoid SyntaxError
func = "if True:\n" + func
frame = sys._getframe(1)
exec(func, frame.f_globals, frame.f_locals)
else:
func(*args, **kwargs)
except exc as e:
res = ExceptionWrapper()
res.value = e
res.traceback = e.__traceback__
return res
else:
raise AssertionError("DID NOT RAISE")
class RaisesContext(object):
def __init__(self, expected_exception):
self.expected_exception = expected_exception
self.excinfo = None
def __enter__(self):
return self
def __exit__(self, *tp):
__tracebackhide__ = True
if tp[0] is None:
raise AssertionError("DID NOT RAISE")
self.value = tp[1]
return issubclass(tp[0], self.expected_exception)
__builtins__.raises = raises
class Test:
pass
self = Test()
def w_getpair(self):
"""Helper method which returns a pair of connected sockets."""
import _socket
import _thread
self.make_server()
self.make_server()
self.sock.listen(1)
s2 = _socket.socket()
_thread.start_new_thread(s2.connect, (self.sockaddress,))
fd, addr2 = self.sock._accept()
s1 = _socket.socket(_socket.AF_INET, _socket.SOCK_STREAM,
proto=0, fileno=fd)
# speed up the tests that want to fill the buffers
s1.setsockopt(_socket.SOL_SOCKET, _socket.SO_RCVBUF, 4096)
s2.setsockopt(_socket.SOL_SOCKET, _socket.SO_SNDBUF, 4096)
return s1, s2
type(self).getpair = w_getpair
def w_make_server(self):
import _socket
if hasattr(self, 'sock'):
return self.sock
self.sock = _socket.socket()
try_ports = [1023] + list(range(20000, 30000, 437))
for port in try_ports:
print('binding to port %d:' % (port,))
self.sockaddress = ('127.0.0.1', port)
try:
self.sock.bind(self.sockaddress)
break
except _socket.error as e: # should get a "Permission denied"
print(e)
else:
raise(e)
type(self).make_server = w_make_server
def test_read_end_closed():
"""
select.select returns elements from the "write list" (the second
parameter) when they are not writable but when the corresponding
read end has been closed. (this test currently doesn't make the
write end non-writable before testing its selectability. -exarkun)
"""
import select
readend, writeend = self.getpair()
readend.close()
try:
iwtd, owtd, ewtd = select.select([writeend], [writeend], [writeend])
assert iwtd == owtd == [writeend]
assert ewtd == []
finally:
writeend.close()
try:
test_read_end_closed()
finally:
print("===aefwuiheawiu===")
from /tmp/buildbot-x86_64/usession-py3.11-1171/src.py
binding to port 1023:
[Errno 13] Permission denied
binding to port 20000:
[Errno 48] Address already in use
binding to port 20437:
[Errno 48] Address already in use
binding to port 20874:
[Errno 48] Address already in use
binding to port 21311:
[Errno 48] Address already in use
binding to port 21748:
[Errno 48] Address already in use
binding to port 22185:
[Errno 48] Address already in use
binding to port 22622:
===aefwuiheawiu===
---------- Captured stderr call ----------
Traceback (most recent call last):
File "/tmp/buildbot-x86_64/usession-py3.11-1171/src.py", line 131, in <module>
test_read_end_closed()
File "/tmp/buildbot-x86_64/usession-py3.11-1171/src.py", line 126, in test_read_end_closed
assert iwtd == owtd == [writeend]
AssertionError
builder: pypy-c-jit-macos-x86-64 build #1315+
test: pypy/module/select/test/test_select.py::AppTestSelectWithSockets::()::test_read_end_closed