Source: python-watchgod
Version: 0.8.2-2
Severity: serious
Justification: FTBFS
Tags: trixie sid ftbfs
User: lu...@debian.org
Usertags: ftbfs-20240115 ftbfs-trixie
Hi,
During a rebuild of all packages in sid, your package failed to build
on amd64.
Relevant part (hopefully):
> debian/rules binary
> dh binary --with python3 --buildsystem=pybuild
> dh_update_autotools_config -O--buildsystem=pybuild
> dh_autoreconf -O--buildsystem=pybuild
> dh_auto_configure -O--buildsystem=pybuild
> I: pybuild base:305: python3.12 setup.py config
> running config
> I: pybuild base:305: python3.11 setup.py config
> running config
> dh_auto_build -O--buildsystem=pybuild
> I: pybuild base:305: /usr/bin/python3.12 setup.py build
> running build
> running build_py
> creating /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/__init__.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/version.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/main.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/watcher.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/__main__.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/cli.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> copying watchgod/py.typed ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build/watchgod
> I: pybuild base:305: /usr/bin/python3 setup.py build
> running build
> running build_py
> creating /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/__init__.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/version.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/main.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/watcher.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/__main__.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/cli.py ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> copying watchgod/py.typed ->
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build/watchgod
> dh_auto_test -O--buildsystem=pybuild
> I: pybuild base:305: cd
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build; python3.12 -m pytest
> -W ignore::trio.TrioDeprecationWarning
> ============================= test session starts
> ==============================
> platform linux -- Python 3.12.1, pytest-7.4.4, pluggy-1.3.0
> rootdir: /<<PKGBUILDDIR>>
> configfile: setup.cfg
> plugins: anyio-4.1.0, asyncio-0.20.3, toolbox-0.4, sugar-0.9.7, mock-3.12.0
> asyncio: mode=Mode.STRICT
> collected 162 items
>
> tests/test_cli.py ...................................................... [
> 33%]
> ............... [
> 42%]
> tests/test_run_process.py ............F.. [
> 51%]
> tests/test_watch.py .................................................... [
> 83%]
> ..FFFFFFFFFFFFFFFF..F..F..
> [100%]
>
> =================================== FAILURES
> ===================================
> _________________ test_async_alive_terminates[asyncio+uvloop]
> __________________
>
> pyfuncitem = <Function test_async_alive_terminates[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.12/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7f3d2defd0a0>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> __________________________ test_watch[asyncio+uvloop]
> __________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d599160>
>
> def test_watch(mocker):
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {'r1'}
>
> tests/test_watch.py:310:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a31b0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _____________________________ test_watch[asyncio]
> ______________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d5846e0>
>
> def test_watch(mocker):
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {'r1'}
>
> tests/test_watch.py:310:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a4c10>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________________ test_watch[trio]
> _______________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d561af0>
>
> def test_watch(mocker):
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {'r1'}
>
> tests/test_watch.py:310:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a4d60>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> __________________ test_watch_watcher_kwargs[asyncio+uvloop]
> ___________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d5634a0>
>
> def test_watch_watcher_kwargs(mocker):
> class FakeWatcher:
> def __init__(self, path, arg1=None, arg2=None):
> self._results = iter(
> [
> {arg1},
> set(),
> {arg2},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> kwargs = dict(arg1='foo', arg2='bar')
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, watcher_kwargs=kwargs,
> debounce=5, normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {kwargs['arg1']}
>
> tests/test_watch.py:332:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a5380>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ______________________ test_watch_watcher_kwargs[asyncio]
> ______________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d562720>
>
> def test_watch_watcher_kwargs(mocker):
> class FakeWatcher:
> def __init__(self, path, arg1=None, arg2=None):
> self._results = iter(
> [
> {arg1},
> set(),
> {arg2},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> kwargs = dict(arg1='foo', arg2='bar')
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, watcher_kwargs=kwargs,
> debounce=5, normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {kwargs['arg1']}
>
> tests/test_watch.py:332:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a5af0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________ test_watch_watcher_kwargs[trio]
> ________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d584350>
>
> def test_watch_watcher_kwargs(mocker):
> class FakeWatcher:
> def __init__(self, path, arg1=None, arg2=None):
> self._results = iter(
> [
> {arg1},
> set(),
> {arg2},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> kwargs = dict(arg1='foo', arg2='bar')
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, watcher_kwargs=kwargs,
> debounce=5, normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {kwargs['arg1']}
>
> tests/test_watch.py:332:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a61f0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________ test_watch_stop[asyncio+uvloop]
> ________________________
>
> def test_watch_stop():
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> ]
> )
>
> def check(self):
> return next(self._results)
>
> stop_event = threading.Event()
> stop_event.set()
> ans = []
> > for c in watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> > min_sleep=1, stop_event=stop_event):
>
> tests/test_watch.py:353:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a6730>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ___________________________ test_watch_stop[asyncio]
> ___________________________
>
> def test_watch_stop():
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> ]
> )
>
> def check(self):
> return next(self._results)
>
> stop_event = threading.Event()
> stop_event.set()
> ans = []
> > for c in watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> > min_sleep=1, stop_event=stop_event):
>
> tests/test_watch.py:353:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a6c70>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ____________________________ test_watch_stop[trio]
> _____________________________
>
> def test_watch_stop():
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> ]
> )
>
> def check(self):
> return next(self._results)
>
> stop_event = threading.Event()
> stop_event.set()
> ans = []
> > for c in watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> > min_sleep=1, stop_event=stop_event):
>
> tests/test_watch.py:353:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a71b0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> __________________ test_watch_keyboard_error[asyncio+uvloop]
> ___________________
>
> def test_watch_keyboard_error():
> class FakeWatcher:
> def __init__(self, path):
> pass
>
> def check(self):
> raise KeyboardInterrupt()
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=1)
> > assert list(iter) == []
>
> tests/test_watch.py:367:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a7760>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ______________________ test_watch_keyboard_error[asyncio]
> ______________________
>
> def test_watch_keyboard_error():
> class FakeWatcher:
> def __init__(self, path):
> pass
>
> def check(self):
> raise KeyboardInterrupt()
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=1)
> > assert list(iter) == []
>
> tests/test_watch.py:367:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a7920>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________ test_watch_keyboard_error[trio]
> ________________________
>
> def test_watch_keyboard_error():
> class FakeWatcher:
> def __init__(self, path):
> pass
>
> def check(self):
> raise KeyboardInterrupt()
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=1)
> > assert list(iter) == []
>
> tests/test_watch.py:367:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5a7e60>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ________________________ test_watch_log[asyncio+uvloop]
> ________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d532cc0>
> caplog = <_pytest.logging.LogCaptureFixture object at 0x7f3d2d532750>
>
> def test_watch_log(mocker, caplog):
> mock_log_enabled = mocker.patch('watchgod.main.logger.isEnabledFor')
> mock_log_enabled.return_value = True
>
> class FakeWatcher:
> def __init__(self, path):
> self.files = [1, 2, 3]
>
> def check(self):
> return {'r1'}
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=10)
> > assert next(iter) == {'r1'}
>
> tests/test_watch.py:382:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5c42e0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ___________________________ test_watch_log[asyncio]
> ____________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2d561a30>
> caplog = <_pytest.logging.LogCaptureFixture object at 0x7f3d2d5629c0>
>
> def test_watch_log(mocker, caplog):
> mock_log_enabled = mocker.patch('watchgod.main.logger.isEnabledFor')
> mock_log_enabled.return_value = True
>
> class FakeWatcher:
> def __init__(self, path):
> self.files = [1, 2, 3]
>
> def check(self):
> return {'r1'}
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=10)
> > assert next(iter) == {'r1'}
>
> tests/test_watch.py:382:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5c49e0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _____________________________ test_watch_log[trio]
> _____________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7f3d2de97950>
> caplog = <_pytest.logging.LogCaptureFixture object at 0x7f3d2de948f0>
>
> def test_watch_log(mocker, caplog):
> mock_log_enabled = mocker.patch('watchgod.main.logger.isEnabledFor')
> mock_log_enabled.return_value = True
>
> class FakeWatcher:
> def __init__(self, path):
> self.files = [1, 2, 3]
>
> def check(self):
> return {'r1'}
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=10)
> > assert next(iter) == {'r1'}
>
> tests/test_watch.py:382:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7f3d2d5c50e0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _________________________ test_awatch[asyncio+uvloop]
> __________________________
>
> pyfuncitem = <Function test_awatch[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.12/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7f3d2def5d60>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> _______________________ test_awatch_stop[asyncio+uvloop]
> _______________________
>
> pyfuncitem = <Function test_awatch_stop[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.12/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7f3d2def5880>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> _______________________ test_awatch_log[asyncio+uvloop]
> ________________________
>
> pyfuncitem = <Function test_awatch_log[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.12/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7f3d2def6960>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> =========================== short test summary info
> ============================
> FAILED tests/test_run_process.py::test_async_alive_terminates[asyncio+uvloop]
> FAILED tests/test_watch.py::test_watch[asyncio+uvloop] - RuntimeError:
> Alread...
> FAILED tests/test_watch.py::test_watch[asyncio] - RuntimeError: Already
> runni...
> FAILED tests/test_watch.py::test_watch[trio] - RuntimeError: Already running
> ...
> FAILED tests/test_watch.py::test_watch_watcher_kwargs[asyncio+uvloop] -
> Runti...
> FAILED tests/test_watch.py::test_watch_watcher_kwargs[asyncio] -
> RuntimeError...
> FAILED tests/test_watch.py::test_watch_watcher_kwargs[trio] - RuntimeError:
> A...
> FAILED tests/test_watch.py::test_watch_stop[asyncio+uvloop] - RuntimeError:
> A...
> FAILED tests/test_watch.py::test_watch_stop[asyncio] - RuntimeError: Already
> ...
> FAILED tests/test_watch.py::test_watch_stop[trio] - RuntimeError: Already
> run...
> FAILED tests/test_watch.py::test_watch_keyboard_error[asyncio+uvloop] -
> Runti...
> FAILED tests/test_watch.py::test_watch_keyboard_error[asyncio] -
> RuntimeError...
> FAILED tests/test_watch.py::test_watch_keyboard_error[trio] - RuntimeError:
> A...
> FAILED tests/test_watch.py::test_watch_log[asyncio+uvloop] - RuntimeError:
> Al...
> FAILED tests/test_watch.py::test_watch_log[asyncio] - RuntimeError: Already
> r...
> FAILED tests/test_watch.py::test_watch_log[trio] - RuntimeError: Already
> runn...
> FAILED tests/test_watch.py::test_awatch[asyncio+uvloop] -
> ModuleNotFoundError...
> FAILED tests/test_watch.py::test_awatch_stop[asyncio+uvloop] -
> ModuleNotFound...
> FAILED tests/test_watch.py::test_awatch_log[asyncio+uvloop] -
> ModuleNotFoundE...
> ======================== 19 failed, 143 passed in 2.19s
> ========================
> E: pybuild pybuild:391: test: plugin distutils failed with: exit code=1: cd
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.12_watchgod/build; python3.12 -m pytest
> -W ignore::trio.TrioDeprecationWarning
> I: pybuild base:305: cd
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build; python3.11 -m pytest
> -W ignore::trio.TrioDeprecationWarning
> ============================= test session starts
> ==============================
> platform linux -- Python 3.11.7, pytest-7.4.4, pluggy-1.3.0
> rootdir: /<<PKGBUILDDIR>>
> configfile: setup.cfg
> plugins: anyio-4.1.0, asyncio-0.20.3, toolbox-0.4, sugar-0.9.7, mock-3.12.0
> asyncio: mode=Mode.STRICT
> collected 162 items
>
> tests/test_cli.py ...................................................... [
> 33%]
> ............... [
> 42%]
> tests/test_run_process.py ............F.. [
> 51%]
> tests/test_watch.py .................................................... [
> 83%]
> ..FFFFFFFFFFFFFFFF..F..F..
> [100%]
>
> =================================== FAILURES
> ===================================
> _________________ test_async_alive_terminates[asyncio+uvloop]
> __________________
>
> pyfuncitem = <Function test_async_alive_terminates[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.11/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7fda0f86d690>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> __________________________ test_watch[asyncio+uvloop]
> __________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0f707010>
>
> def test_watch(mocker):
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {'r1'}
>
> tests/test_watch.py:310:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f690040>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _____________________________ test_watch[asyncio]
> ______________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0ed68710>
>
> def test_watch(mocker):
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {'r1'}
>
> tests/test_watch.py:310:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a4660>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________________ test_watch[trio]
> _______________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0eb48490>
>
> def test_watch(mocker):
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {'r1'}
>
> tests/test_watch.py:310:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a53f0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> __________________ test_watch_watcher_kwargs[asyncio+uvloop]
> ___________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0f89b3d0>
>
> def test_watch_watcher_kwargs(mocker):
> class FakeWatcher:
> def __init__(self, path, arg1=None, arg2=None):
> self._results = iter(
> [
> {arg1},
> set(),
> {arg2},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> kwargs = dict(arg1='foo', arg2='bar')
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, watcher_kwargs=kwargs,
> debounce=5, normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {kwargs['arg1']}
>
> tests/test_watch.py:332:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a5a80>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ______________________ test_watch_watcher_kwargs[asyncio]
> ______________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0eb64a10>
>
> def test_watch_watcher_kwargs(mocker):
> class FakeWatcher:
> def __init__(self, path, arg1=None, arg2=None):
> self._results = iter(
> [
> {arg1},
> set(),
> {arg2},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> kwargs = dict(arg1='foo', arg2='bar')
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, watcher_kwargs=kwargs,
> debounce=5, normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {kwargs['arg1']}
>
> tests/test_watch.py:332:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a61f0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________ test_watch_watcher_kwargs[trio]
> ________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0f6e83d0>
>
> def test_watch_watcher_kwargs(mocker):
> class FakeWatcher:
> def __init__(self, path, arg1=None, arg2=None):
> self._results = iter(
> [
> {arg1},
> set(),
> {arg2},
> set(),
> ]
> )
>
> def check(self):
> return next(self._results)
>
> kwargs = dict(arg1='foo', arg2='bar')
>
> iter_ = watch('xxx', watcher_cls=FakeWatcher, watcher_kwargs=kwargs,
> debounce=5, normal_sleep=2, min_sleep=1)
> > assert next(iter_) == {kwargs['arg1']}
>
> tests/test_watch.py:332:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a69d0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________ test_watch_stop[asyncio+uvloop]
> ________________________
>
> def test_watch_stop():
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> ]
> )
>
> def check(self):
> return next(self._results)
>
> stop_event = threading.Event()
> stop_event.set()
> ans = []
> > for c in watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> > min_sleep=1, stop_event=stop_event):
>
> tests/test_watch.py:353:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a7140>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ___________________________ test_watch_stop[asyncio]
> ___________________________
>
> def test_watch_stop():
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> ]
> )
>
> def check(self):
> return next(self._results)
>
> stop_event = threading.Event()
> stop_event.set()
> ans = []
> > for c in watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> > min_sleep=1, stop_event=stop_event):
>
> tests/test_watch.py:353:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a7840>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ____________________________ test_watch_stop[trio]
> _____________________________
>
> def test_watch_stop():
> class FakeWatcher:
> def __init__(self, path):
> self._results = iter(
> [
> {'r1'},
> set(),
> {'r2'},
> ]
> )
>
> def check(self):
> return next(self._results)
>
> stop_event = threading.Event()
> stop_event.set()
> ans = []
> > for c in watch('xxx', watcher_cls=FakeWatcher, debounce=5,
> > min_sleep=1, stop_event=stop_event):
>
> tests/test_watch.py:353:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a7e60>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> __________________ test_watch_keyboard_error[asyncio+uvloop]
> ___________________
>
> def test_watch_keyboard_error():
> class FakeWatcher:
> def __init__(self, path):
> pass
>
> def check(self):
> raise KeyboardInterrupt()
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=1)
> > assert list(iter) == []
>
> tests/test_watch.py:367:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a7d80>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ______________________ test_watch_keyboard_error[asyncio]
> ______________________
>
> def test_watch_keyboard_error():
> class FakeWatcher:
> def __init__(self, path):
> pass
>
> def check(self):
> raise KeyboardInterrupt()
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=1)
> > assert list(iter) == []
>
> tests/test_watch.py:367:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0f6a7a00>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _______________________ test_watch_keyboard_error[trio]
> ________________________
>
> def test_watch_keyboard_error():
> class FakeWatcher:
> def __init__(self, path):
> pass
>
> def check(self):
> raise KeyboardInterrupt()
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=1)
> > assert list(iter) == []
>
> tests/test_watch.py:367:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0ed60900>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ________________________ test_watch_log[asyncio+uvloop]
> ________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0f7e8410>
> caplog = <_pytest.logging.LogCaptureFixture object at 0x7fda0f7e9750>
>
> def test_watch_log(mocker, caplog):
> mock_log_enabled = mocker.patch('watchgod.main.logger.isEnabledFor')
> mock_log_enabled.return_value = True
>
> class FakeWatcher:
> def __init__(self, path):
> self.files = [1, 2, 3]
>
> def check(self):
> return {'r1'}
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=10)
> > assert next(iter) == {'r1'}
>
> tests/test_watch.py:382:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0ed603c0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> ___________________________ test_watch_log[asyncio]
> ____________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0f7c3c90>
> caplog = <_pytest.logging.LogCaptureFixture object at 0x7fda0f6ac750>
>
> def test_watch_log(mocker, caplog):
> mock_log_enabled = mocker.patch('watchgod.main.logger.isEnabledFor')
> mock_log_enabled.return_value = True
>
> class FakeWatcher:
> def __init__(self, path):
> self.files = [1, 2, 3]
>
> def check(self):
> return {'r1'}
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=10)
> > assert next(iter) == {'r1'}
>
> tests/test_watch.py:382:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0ed61e70>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _____________________________ test_watch_log[trio]
> _____________________________
>
> mocker = <pytest_mock.plugin.MockerFixture object at 0x7fda0f6ea5d0>
> caplog = <_pytest.logging.LogCaptureFixture object at 0x7fda0f6e8d10>
>
> def test_watch_log(mocker, caplog):
> mock_log_enabled = mocker.patch('watchgod.main.logger.isEnabledFor')
> mock_log_enabled.return_value = True
>
> class FakeWatcher:
> def __init__(self, path):
> self.files = [1, 2, 3]
>
> def check(self):
> return {'r1'}
>
> iter = watch('xxx', watcher_cls=FakeWatcher, debounce=5, min_sleep=10)
> > assert next(iter) == {'r1'}
>
> tests/test_watch.py:382:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> watchgod/main.py:47: in watch
> yield anyio.run(_awatch.__anext__)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> func = <bound method awatch.__anext__ of <watchgod.main.awatch object at
> 0x7fda0ed61af0>>
> backend = 'asyncio', backend_options = None, args = ()
> asynclib_name = 'asyncio'
>
> def run(
> func: Callable[..., Awaitable[T_Retval]],
> *args: object,
> backend: str = "asyncio",
> backend_options: dict[str, Any] | None = None,
> ) -> T_Retval:
> """
> Run the given coroutine function in an asynchronous event loop.
>
> The current thread must not be already running an event loop.
>
> :param func: a coroutine function
> :param args: positional arguments to ``func``
> :param backend: name of the asynchronous event loop implementation –
> currently
> either ``asyncio`` or ``trio``
> :param backend_options: keyword arguments to call the backend
> ``run()``
> implementation with (documented :ref:`here <backend options>`)
> :return: the return value of the coroutine function
> :raises RuntimeError: if an asynchronous event loop is already
> running in this
> thread
> :raises LookupError: if the named backend is not found
>
> """
> try:
> asynclib_name = sniffio.current_async_library()
> except sniffio.AsyncLibraryNotFoundError:
> pass
> else:
> > raise RuntimeError(f"Already running {asynclib_name} in this
> > thread")
> E RuntimeError: Already running asyncio in this thread
>
> /usr/lib/python3/dist-packages/anyio/_core/_eventloop.py:51: RuntimeError
> _________________________ test_awatch[asyncio+uvloop]
> __________________________
>
> pyfuncitem = <Function test_awatch[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.11/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7fda0f8f9750>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> _______________________ test_awatch_stop[asyncio+uvloop]
> _______________________
>
> pyfuncitem = <Function test_awatch_stop[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.11/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7fda0eb496d0>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> _______________________ test_awatch_log[asyncio+uvloop]
> ________________________
>
> pyfuncitem = <Function test_awatch_log[asyncio+uvloop]>
>
> @pytest.hookimpl(tryfirst=True)
> def pytest_pyfunc_call(pyfuncitem: Any) -> bool | None:
> def run_with_hypothesis(**kwargs: Any) -> None:
> with get_runner(backend_name, backend_options) as runner:
> runner.run_test(original_func, kwargs)
>
> backend = pyfuncitem.funcargs.get("anyio_backend")
> if backend:
> backend_name, backend_options =
> extract_backend_and_options(backend)
>
> if hasattr(pyfuncitem.obj, "hypothesis"):
> # Wrap the inner test function unless it's already wrapped
> original_func = pyfuncitem.obj.hypothesis.inner_test
> if original_func.__qualname__ !=
> run_with_hypothesis.__qualname__:
> if iscoroutinefunction(original_func):
> pyfuncitem.obj.hypothesis.inner_test =
> run_with_hypothesis
>
> return None
>
> if iscoroutinefunction(pyfuncitem.obj):
> funcargs = pyfuncitem.funcargs
> testargs = {arg: funcargs[arg] for arg in
> pyfuncitem._fixtureinfo.argnames}
> > with get_runner(backend_name, backend_options) as runner:
>
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:123:
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
> /usr/lib/python3.11/contextlib.py:137: in __enter__
> return next(self.gen)
> /usr/lib/python3/dist-packages/anyio/pytest_plugin.py:45: in get_runner
> asynclib.create_test_runner(backend_options)
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:2431: in
> create_test_runner
> return TestRunner(**options)
> _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
> _
>
> self = <anyio._backends._asyncio.TestRunner object at 0x7fda0eb666d0>
>
> def __init__(
> self,
> *,
> debug: bool | None = None,
> use_uvloop: bool = False,
> loop_factory: Callable[[], AbstractEventLoop] | None = None,
> ) -> None:
> if use_uvloop and loop_factory is None:
> > import uvloop
> E ModuleNotFoundError: No module named 'uvloop'
>
> /usr/lib/python3/dist-packages/anyio/_backends/_asyncio.py:1814:
> ModuleNotFoundError
> =========================== short test summary info
> ============================
> FAILED tests/test_run_process.py::test_async_alive_terminates[asyncio+uvloop]
> FAILED tests/test_watch.py::test_watch[asyncio+uvloop] - RuntimeError:
> Alread...
> FAILED tests/test_watch.py::test_watch[asyncio] - RuntimeError: Already
> runni...
> FAILED tests/test_watch.py::test_watch[trio] - RuntimeError: Already running
> ...
> FAILED tests/test_watch.py::test_watch_watcher_kwargs[asyncio+uvloop] -
> Runti...
> FAILED tests/test_watch.py::test_watch_watcher_kwargs[asyncio] -
> RuntimeError...
> FAILED tests/test_watch.py::test_watch_watcher_kwargs[trio] - RuntimeError:
> A...
> FAILED tests/test_watch.py::test_watch_stop[asyncio+uvloop] - RuntimeError:
> A...
> FAILED tests/test_watch.py::test_watch_stop[asyncio] - RuntimeError: Already
> ...
> FAILED tests/test_watch.py::test_watch_stop[trio] - RuntimeError: Already
> run...
> FAILED tests/test_watch.py::test_watch_keyboard_error[asyncio+uvloop] -
> Runti...
> FAILED tests/test_watch.py::test_watch_keyboard_error[asyncio] -
> RuntimeError...
> FAILED tests/test_watch.py::test_watch_keyboard_error[trio] - RuntimeError:
> A...
> FAILED tests/test_watch.py::test_watch_log[asyncio+uvloop] - RuntimeError:
> Al...
> FAILED tests/test_watch.py::test_watch_log[asyncio] - RuntimeError: Already
> r...
> FAILED tests/test_watch.py::test_watch_log[trio] - RuntimeError: Already
> runn...
> FAILED tests/test_watch.py::test_awatch[asyncio+uvloop] -
> ModuleNotFoundError...
> FAILED tests/test_watch.py::test_awatch_stop[asyncio+uvloop] -
> ModuleNotFound...
> FAILED tests/test_watch.py::test_awatch_log[asyncio+uvloop] -
> ModuleNotFoundE...
> ======================== 19 failed, 143 passed in 2.10s
> ========================
> E: pybuild pybuild:391: test: plugin distutils failed with: exit code=1: cd
> /<<PKGBUILDDIR>>/.pybuild/cpython3_3.11_watchgod/build; python3.11 -m pytest
> -W ignore::trio.TrioDeprecationWarning
> dh_auto_test: error: pybuild --test --test-pytest -i python{version} -p "3.12
> 3.11" returned exit code 13
The full build log is available from:
http://qa-logs.debian.net/2024/01/15/python-watchgod_0.8.2-2_unstable.log
All bugs filed during this archive rebuild are listed at:
https://bugs.debian.org/cgi-bin/pkgreport.cgi?tag=ftbfs-20240115;users=lu...@debian.org
or:
https://udd.debian.org/bugs/?release=na&merged=ign&fnewerval=7&flastmodval=7&fusertag=only&fusertagtag=ftbfs-20240115&fusertaguser=lu...@debian.org&allbugs=1&cseverity=1&ctags=1&caffected=1#results
A list of current common problems and possible solutions is available at
http://wiki.debian.org/qa.debian.org/FTBFS . You're welcome to contribute!
If you reassign this bug to another package, please mark it as 'affects'-ing
this package. See https://www.debian.org/Bugs/server-control#affects
If you fail to reproduce this, please provide a build log and diff it with mine
so that we can identify if something relevant changed in the meantime.
