-
-
Notifications
You must be signed in to change notification settings - Fork 7
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Fix code for Python3 #11
base: main
Are you sure you want to change the base?
Conversation
Could you explain what it is that you're trying to fix? All I see is some style and default args changes. |
Sure. I get lots of |
@szymonlipinski sounds like you pass the wrong data types when calling magicbus APIs. |
@webknjaz I think that the issue is in
The issue can be replicated with a very short snippet: >>> from magicbus.plugins.loggers import StdoutLogger
>>> StdoutLogger(bus=mock.Mock(id=5)).log("hello", 10)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/Users/amol/wrk/crunch/venv/lib/python3.7/site-packages/magicbus/plugins/loggers.py", line 34, in log
self.stream.write(complete_msg)
TypeError: write() argument must be str, not bytes I think that the PR is wrong as it changes too many things (it changes also |
Fair enough |
It looks like the tests hang on python 2.7 even if I reverted all changes from this PR: https://travis-ci.org/github/cherrypy/magicbus/jobs/757368941 |
|
||
else: | ||
if not isinstance(complete_msg, unicodestr): | ||
complete_msg = complete_msg.decode("utf-8") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I wonder if in this case we might want to add errors="backslashreplace"
given that we are mostly guessing utf-8
might be a good choice.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
PPS: This can probably just become
if isinstance(complete_msg, unicodestr):
encoding = self.encoding or "utf-8"
complete_msg = complete_msg.encode(encoding, errors="backslashreplace")
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Sorry, I misread the two if
branches o_O
You are encoding
in one and decoding
in the other, my suggestion of squashing the two branches was totally wrong :D
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yea, you're right. I haven't noticed that too :) I have reverted the changes. I think it's fine now.
magicbus/plugins/loggers.py
Outdated
complete_msg = complete_msg.decode("utf-8") | ||
if isinstance(complete_msg, unicodestr): | ||
encoding = self.encoding or "utf-8" | ||
complete_msg = complete_msg.encode(encoding, errors="backslashreplace") |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
no, I meant, both ifs can be removed :D
if self.encoding is not None:
if isinstance(complete_msg, unicodestr):
complete_msg = complete_msg.encode(self.encoding)
else:
if isinstance(complete_msg, unicodestr):
encoding = self.encoding or "utf-8"
complete_msg = complete_msg.encode(encoding, errors="backslashreplace")
VS
if isinstance(complete_msg, unicodestr):
encoding = self.encoding or "utf-8"
complete_msg = complete_msg.encode(encoding, errors="backslashreplace")
@webknjaz Hey, could you take a look at the failing test? It failed also when I reverted all the changes made by this PR. It looks like there is some bigger problem somewhere. All the tests pass on my computer for python2 and python3. |
Since it is only happening in CIs and hang on the tests that have TTY in their name, my best guess is that the CIs don't allocate a TTY when running these tests and that's what's causing the problem. I wonder if we could allocate a TTY on our own in such envs... |
|
||
self.stream.write(complete_msg) | ||
self.stream.flush() | ||
|
||
|
||
class StdoutLogger(StreamLogger): | ||
|
||
def __init__(self, bus, level=None, format=None, encoding='utf-8'): | ||
def __init__(self, bus, level=None, format=None, encoding=None): |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't see how this is related to the problem in question. Could you drop changing the defaults too? It is a behavior change at best (and a public interface change at worst. This does not seem to be fixing bugs.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think that the default change is actually the fix, otherwise you would have to explicitly pass encoding=None
everywhere to make sure this works at all on PY3, as there is no way you can have StdoutLogger
work on PY3 if it's encoding.
See
>>> StdoutLogger(mock.Mock(id=5), encoding="utf8").log("HI", level=20)
Traceback (most recent call last):
File "<stdin>", line 1, in <module>
File "/Users/amol/wrk/crunch/venv/lib/python3.7/site-packages/magicbus/plugins/loggers.py", line 34, in log
self.stream.write(complete_msg)
TypeError: write() argument must be str, not bytes
VS
>>> StdoutLogger(mock.Mock(id=5), encoding=None).log("HI", level=20)
[b'2021-02-11T15:51:15.308588'] (Bus 5) HI
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This looks like a bytes vs unicode py2/3 issue. It's usually better to always have this normalized internally.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I don't personally consider this a string encoding/decoding issue.
The problem is related to the fact that sys.stdout
and sys.stderr
are text mode files, so you can't send bytes to them at all in PY3. The whole "encoding" option doesn't make much sense for sys.stdout/err
in my opinion.
If you want to change the encoding of them you are meant to do sys.stdout.reconfigure(encoding='utf-8')
In PY2 sys.stdout
/sys.stderr
like most io, were mostly agnostic to what you throw at them and thus tried to do their best with whatever they received, but you were still in theory meant to send unicode to them and get it printed according to sys.stdout.encoding
like on PY3.
I think that it was mostly an hack the fact that magicbus allowed you to send bytes to sys.stdout
at all totally ignoring what was sys.stdout.encoding
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yeah. Honestly, it'd be great to integrate stdlib logging
across all CherryPy projects instead of reinventing the wheel.
This all looks like a lack of proper text compat functions. See some examples:
A common technique is to either convert most of the stuff to "native" strings or unicode/text with exceptions for passing them as args to certain stdlib interfaces (that require "native" most of the time). In general, it's a good idea to always accept text and "native" from external calls because with bytes we cannot (shouldn't) automatically assume the encoding. |
This is reproducible (gets stuck) for me locally with
But it passes if I tell pytest not to capture stdin/stdout:
(same with |
Looking further, I now remember that it's #7 that I attempted to fix with #8. When I add $ tox -e py36 -- magicbus/test/test_signals.py::TestSignalHandling::test_SIGHUP_tty --full-trace
py36 develop-inst-noop: ~/src/github/cherrypy/magicbus
py36 installed: attrs==20.3.0,collective.checkdocs==0.2,docutils==0.16,importlib-metadata==3.4.0,iniconfig==1.1.1,-e git+ssh://git@github.com/cherrypy/magicbus.git@f73e0a5d5649c6464de1d4a98c83f5a6067f94b5#egg=MagicBus,packaging==20.9,pluggy==0.13.1,py==1.10.0,pyparsing==2.4.7,pytest==6.2.2,toml==0.10.2,typing-extensions==3.7.4.3,zipp==3.4.0
py36 run-test-pre: PYTHONHASHSEED='51905874'
py36 run-test: commands[0] | pytest magicbus/test/test_signals.py::TestSignalHandling::test_SIGHUP_tty --full-trace
==================================================================== test session starts =====================================================================
platform linux -- Python 3.6.12, pytest-6.2.2, py-1.10.0, pluggy-0.13.1 -- ~/src/github/cherrypy/magicbus/.tox/py36/bin/python
cachedir: .tox/py36/.pytest_cache
rootdir: ~/src/github/cherrypy/magicbus, configfile: pytest.ini
collected 1 item
magicbus/test/test_signals.py::TestSignalHandling::test_SIGHUP_tty ^C
-------------------------------------- generated xml file: ~/src/github/cherrypy/magicbus/junit-test-results.xml --------------------------------------
==================================================================== slowest 10 durations ====================================================================
(1 durations < 0.005s hidden. Use -vv to show these durations.)
!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!! KeyboardInterrupt !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
config = <_pytest.config.Config object at 0x7fcb1a157908>, doit = <function _main at 0x7fcb1a3be510>
def wrap_session(
config: Config, doit: Callable[[Config, "Session"], Optional[Union[int, ExitCode]]]
) -> Union[int, ExitCode]:
"""Skeleton command line program."""
session = Session.from_config(config)
session.exitstatus = ExitCode.OK
initstate = 0
try:
try:
config._do_configure()
initstate = 1
config.hook.pytest_sessionstart(session=session)
initstate = 2
> session.exitstatus = doit(config, session) or 0
.tox/py36/lib/python3.6/site-packages/_pytest/main.py:269:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
config = <_pytest.config.Config object at 0x7fcb1a157908>, session = <Session magicbus exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=1>
def _main(config: Config, session: "Session") -> Optional[Union[int, ExitCode]]:
"""Default command line protocol for initialization, session,
running tests and reporting."""
config.hook.pytest_collection(session=session)
> config.hook.pytest_runtestloop(session=session)
.tox/py36/lib/python3.6/site-packages/_pytest/main.py:323:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_HookCaller 'pytest_runtestloop'>, args = (), kwargs = {'session': <Session magicbus exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=1>}
notincall = set()
def __call__(self, *args, **kwargs):
if args:
raise TypeError("hook calling supports only keyword arguments")
assert not self.is_historic()
if self.spec and self.spec.argnames:
notincall = (
set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
)
if notincall:
warnings.warn(
"Argument(s) {} which are declared in the hookspec "
"can not be found in this hook call".format(tuple(notincall)),
stacklevel=2,
)
> return self._hookexec(self, self.get_hookimpls(), kwargs)
.tox/py36/lib/python3.6/site-packages/pluggy/hooks.py:286:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.config.PytestPluginManager object at 0x7fcb1af01f98>, hook = <_HookCaller 'pytest_runtestloop'>
methods = [<HookImpl plugin_name='main', plugin=<module '_pytest.main' from '~/src/github/cherrypy/magicbus/.tox/py36/lib...t/main.py'>>, <HookImpl plugin_name='logging-plugin', plugin=<_pytest.logging.LoggingPlugin object at 0x7fcb1a0c6240>>]
kwargs = {'session': <Session magicbus exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=1>}
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)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:93:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
hook = <_HookCaller 'pytest_runtestloop'>
methods = [<HookImpl plugin_name='main', plugin=<module '_pytest.main' from '~/src/github/cherrypy/magicbus/.tox/py36/lib...t/main.py'>>, <HookImpl plugin_name='logging-plugin', plugin=<_pytest.logging.LoggingPlugin object at 0x7fcb1a0c6240>>]
kwargs = {'session': <Session magicbus exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=1>}
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
methods,
kwargs,
> firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:87:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
session = <Session magicbus exitstatus=<ExitCode.OK: 0> testsfailed=0 testscollected=1>
def pytest_runtestloop(session: "Session") -> bool:
if session.testsfailed and not session.config.option.continue_on_collection_errors:
raise session.Interrupted(
"%d error%s during collection"
% (session.testsfailed, "s" if session.testsfailed != 1 else "")
)
if session.config.option.collectonly:
return True
for i, item in enumerate(session.items):
nextitem = session.items[i + 1] if i + 1 < len(session.items) else None
> item.config.hook.pytest_runtest_protocol(item=item, nextitem=nextitem)
.tox/py36/lib/python3.6/site-packages/_pytest/main.py:348:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_HookCaller 'pytest_runtest_protocol'>, args = (), kwargs = {'item': <Function test_SIGHUP_tty>, 'nextitem': None}, notincall = set()
def __call__(self, *args, **kwargs):
if args:
raise TypeError("hook calling supports only keyword arguments")
assert not self.is_historic()
if self.spec and self.spec.argnames:
notincall = (
set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
)
if notincall:
warnings.warn(
"Argument(s) {} which are declared in the hookspec "
"can not be found in this hook call".format(tuple(notincall)),
stacklevel=2,
)
> return self._hookexec(self, self.get_hookimpls(), kwargs)
.tox/py36/lib/python3.6/site-packages/pluggy/hooks.py:286:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.config.PytestPluginManager object at 0x7fcb1af01f98>, hook = <_HookCaller 'pytest_runtest_protocol'>
methods = [<HookImpl plugin_name='runner', plugin=<module '_pytest.runner' from '~/src/github/cherrypy/magicbus/.tox/py36...est.warnings' from '~/src/github/cherrypy/magicbus/.tox/py36/lib/python3.6/site-packages/_pytest/warnings.py'>>]
kwargs = {'item': <Function test_SIGHUP_tty>, 'nextitem': None}
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)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:93:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
hook = <_HookCaller 'pytest_runtest_protocol'>
methods = [<HookImpl plugin_name='runner', plugin=<module '_pytest.runner' from '~/src/github/cherrypy/magicbus/.tox/py36...est.warnings' from '~/src/github/cherrypy/magicbus/.tox/py36/lib/python3.6/site-packages/_pytest/warnings.py'>>]
kwargs = {'item': <Function test_SIGHUP_tty>, 'nextitem': None}
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
methods,
kwargs,
> firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:87:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <Function test_SIGHUP_tty>, nextitem = None
def pytest_runtest_protocol(item: Item, nextitem: Optional[Item]) -> bool:
ihook = item.ihook
ihook.pytest_runtest_logstart(nodeid=item.nodeid, location=item.location)
> runtestprotocol(item, nextitem=nextitem)
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:109:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <Function test_SIGHUP_tty>, log = True, nextitem = None
def runtestprotocol(
item: Item, log: bool = True, nextitem: Optional[Item] = None
) -> List[TestReport]:
hasrequest = hasattr(item, "_request")
if hasrequest and not item._request: # type: ignore[attr-defined]
item._initrequest() # type: ignore[attr-defined]
rep = call_and_report(item, "setup", log)
reports = [rep]
if rep.passed:
if item.config.getoption("setupshow", False):
show_test_item(item)
if not item.config.getoption("setuponly", False):
> reports.append(call_and_report(item, "call", log))
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:126:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <Function test_SIGHUP_tty>, when = 'call', log = True, kwds = {}
def call_and_report(
item: Item, when: "Literal['setup', 'call', 'teardown']", log: bool = True, **kwds
) -> TestReport:
> call = call_runtest_hook(item, when, **kwds)
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:215:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <Function test_SIGHUP_tty>, when = 'call', kwds = {}, reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>)
def call_runtest_hook(
item: Item, when: "Literal['setup', 'call', 'teardown']", **kwds
) -> "CallInfo[None]":
if when == "setup":
ihook: Callable[..., None] = item.ihook.pytest_runtest_setup
elif when == "call":
ihook = item.ihook.pytest_runtest_call
elif when == "teardown":
ihook = item.ihook.pytest_runtest_teardown
else:
assert False, f"Unhandled runtest hook case: {when}"
reraise: Tuple[Type[BaseException], ...] = (Exit,)
if not item.config.getoption("usepdb", False):
reraise += (KeyboardInterrupt,)
return CallInfo.from_call(
> lambda: ihook(item=item, **kwds), when=when, reraise=reraise
)
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:255:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
cls = _pytest.runner.CallInfo, func = <function call_runtest_hook.<locals>.<lambda> at 0x7fcb1a025620>, when = 'call'
reraise = (<class '_pytest.outcomes.Exit'>, <class 'KeyboardInterrupt'>)
@classmethod
def from_call(
cls,
func: "Callable[[], TResult]",
when: "Literal['collect', 'setup', 'call', 'teardown']",
reraise: Optional[
Union[Type[BaseException], Tuple[Type[BaseException], ...]]
] = None,
) -> "CallInfo[TResult]":
excinfo = None
start = timing.time()
precise_start = timing.perf_counter()
try:
> result: Optional[TResult] = func()
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:311:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> lambda: ihook(item=item, **kwds), when=when, reraise=reraise
)
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:255:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_HookCaller 'pytest_runtest_call'>, args = (), kwargs = {'item': <Function test_SIGHUP_tty>}, notincall = set()
def __call__(self, *args, **kwargs):
if args:
raise TypeError("hook calling supports only keyword arguments")
assert not self.is_historic()
if self.spec and self.spec.argnames:
notincall = (
set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
)
if notincall:
warnings.warn(
"Argument(s) {} which are declared in the hookspec "
"can not be found in this hook call".format(tuple(notincall)),
stacklevel=2,
)
> return self._hookexec(self, self.get_hookimpls(), kwargs)
.tox/py36/lib/python3.6/site-packages/pluggy/hooks.py:286:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.config.PytestPluginManager object at 0x7fcb1af01f98>, hook = <_HookCaller 'pytest_runtest_call'>
methods = [<HookImpl plugin_name='runner', plugin=<module '_pytest.runner' from '~/src/github/cherrypy/magicbus/.tox/py36...xture=None>>, <HookImpl plugin_name='logging-plugin', plugin=<_pytest.logging.LoggingPlugin object at 0x7fcb1a0c6240>>]
kwargs = {'item': <Function test_SIGHUP_tty>}
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)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:93:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
hook = <_HookCaller 'pytest_runtest_call'>
methods = [<HookImpl plugin_name='runner', plugin=<module '_pytest.runner' from '~/src/github/cherrypy/magicbus/.tox/py36...xture=None>>, <HookImpl plugin_name='logging-plugin', plugin=<_pytest.logging.LoggingPlugin object at 0x7fcb1a0c6240>>]
kwargs = {'item': <Function test_SIGHUP_tty>}
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
methods,
kwargs,
> firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:87:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
item = <Function test_SIGHUP_tty>
def pytest_runtest_call(item: Item) -> None:
_update_current_test_var(item, "call")
try:
del sys.last_type
del sys.last_value
del sys.last_traceback
except AttributeError:
pass
try:
> item.runtest()
.tox/py36/lib/python3.6/site-packages/_pytest/runner.py:162:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <Function test_SIGHUP_tty>
def runtest(self) -> None:
"""Execute the underlying test function."""
> self.ihook.pytest_pyfunc_call(pyfuncitem=self)
.tox/py36/lib/python3.6/site-packages/_pytest/python.py:1641:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_HookCaller 'pytest_pyfunc_call'>, args = (), kwargs = {'pyfuncitem': <Function test_SIGHUP_tty>}, notincall = set()
def __call__(self, *args, **kwargs):
if args:
raise TypeError("hook calling supports only keyword arguments")
assert not self.is_historic()
if self.spec and self.spec.argnames:
notincall = (
set(self.spec.argnames) - set(["__multicall__"]) - set(kwargs.keys())
)
if notincall:
warnings.warn(
"Argument(s) {} which are declared in the hookspec "
"can not be found in this hook call".format(tuple(notincall)),
stacklevel=2,
)
> return self._hookexec(self, self.get_hookimpls(), kwargs)
.tox/py36/lib/python3.6/site-packages/pluggy/hooks.py:286:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <_pytest.config.PytestPluginManager object at 0x7fcb1af01f98>, hook = <_HookCaller 'pytest_pyfunc_call'>
methods = [<HookImpl plugin_name='python', plugin=<module '_pytest.python' from '~/src/github/cherrypy/magicbus/.tox/py36/lib/python3.6/site-packages/_pytest/python.py'>>]
kwargs = {'pyfuncitem': <Function test_SIGHUP_tty>}
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)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:93:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
hook = <_HookCaller 'pytest_pyfunc_call'>
methods = [<HookImpl plugin_name='python', plugin=<module '_pytest.python' from '~/src/github/cherrypy/magicbus/.tox/py36/lib/python3.6/site-packages/_pytest/python.py'>>]
kwargs = {'pyfuncitem': <Function test_SIGHUP_tty>}
self._inner_hookexec = lambda hook, methods, kwargs: hook.multicall(
methods,
kwargs,
> firstresult=hook.spec.opts.get("firstresult") if hook.spec else False,
)
.tox/py36/lib/python3.6/site-packages/pluggy/manager.py:87:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
pyfuncitem = <Function test_SIGHUP_tty>
@hookimpl(trylast=True)
def pytest_pyfunc_call(pyfuncitem: "Function") -> Optional[object]:
testfunction = pyfuncitem.obj
if is_async_function(testfunction):
async_warn_and_skip(pyfuncitem.nodeid)
funcargs = pyfuncitem.funcargs
testargs = {arg: funcargs[arg] for arg in pyfuncitem._fixtureinfo.argnames}
> result = testfunction(**testargs)
.tox/py36/lib/python3.6/site-packages/_pytest/python.py:183:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <magicbus.test.test_signals.TestSignalHandling object at 0x7fcb19d65940>
def test_SIGHUP_tty(self):
# When not daemonized, SIGHUP should exit the process.
try:
from signal import SIGHUP
except ImportError:
return 'skipped (no SIGHUP)'
try:
from os import kill
except ImportError:
return 'skipped (no os.kill)'
p = Process([sys.executable, thismodule, 'tty'])
p.start()
pid = pidfile.wait()
kill(pid, SIGHUP)
> pidfile.join()
magicbus/test/test_signals.py:34:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <magicbus.plugins.opsys.PIDFile object at 0x7fcb19f5f630>, timeout = None, poll_interval = 0.1
def join(self, timeout=None, poll_interval=0.1):
"""Return when the PID file does not exist, or the timeout expires."""
starttime = time.time()
while timeout is None or time.time() - starttime <= timeout:
if not os.path.exists(self.pidfile):
return
> time.sleep(poll_interval)
E KeyboardInterrupt
magicbus/plugins/opsys.py:268: KeyboardInterrupt
=================================================================== no tests ran in 1.54s ====================================================================
ERROR: got KeyboardInterrupt signal
__________________________________________________________________________ summary ___________________________________________________________________________
ERROR: py36: keyboardinterrupt |
@szymonlipinski so I hacked that those annoying unrelated failing tests and it should be fine now. Although, I've added an excessive GHA matrix that is flaky at times. Could you rebase this PR so we could see how it goes? |
This reverts commit 28e9064.
368f806
to
ed05583
Compare
No description provided.