Page MenuHomeSoftware Heritage
Paste P920

pytest lister celery timeout
ActivePublic
Actions

Authored by tenma on Jan 12 2021, 8:11 PM.
(swh) ~/dev/swh-env/swh-lister$ pytest swh/lister/bitbucket/tests/test_tasks.py
==================================================================================== test session starts =====================================================================================
platform linux -- Python 3.7.3, pytest-6.1.2, py-1.9.0, pluggy-0.13.1
rootdir: /home/tenma/dev/swh-env/swh-lister, configfile: pytest.ini
plugins: flask-1.1.0, requests-mock-1.8.0, hypothesis-5.42.3, cov-2.10.1, xdist-2.1.0, django-4.1.0, postgresql-2.5.2, forked-1.3.0, asyncio-0.14.0, mock-3.3.1, dash-1.18.0, swh.journal-0.5.2.dev2+g8dd9712, swh.core-0.11.1.dev7+gd33457e
collected 2 items
swh/lister/bitbucket/tests/test_tasks.py EE [100%]
=========================================================================================== ERRORS ===========================================================================================
________________________________________________________________________________ ERROR at setup of test_ping _________________________________________________________________________________
swh_scheduler_celery_app = <Celery celery.tests at 0x7f5241caeda0>
swh_scheduler_celery_includes = ['swh.scheduler.tests.tasks', 'swh.lister.bitbucket.tasks', 'swh.lister.cgit.tasks', 'swh.lister.cran.tasks', 'swh.lister.debian.tasks', 'swh.lister.gitea.tasks', ...]
@pytest.fixture(scope="session")
def swh_scheduler_celery_worker(
swh_scheduler_celery_app, swh_scheduler_celery_includes,
):
"""Spawn a worker"""
for module in swh_scheduler_celery_includes:
swh_scheduler_celery_app.loader.import_task_module(module)
> with worker.start_worker(swh_scheduler_celery_app, pool="solo") as w:
../swh-scheduler/swh/scheduler/pytest_plugin.py:103:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib/python3.7/contextlib.py:112: in __enter__
return next(self.gen)
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/contrib/testing/worker.py:82: in start_worker
assert ping.delay().get(timeout=ping_task_timeout) == 'pong'
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/result.py:230: in get
on_message=on_message,
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/backends/base.py:655: in wait_for_pending
no_ack=no_ack,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <celery.backends.cache.CacheBackend object at 0x7f523b8b26a0>, task_id = '903dbe6a-e48e-4f21-af94-272e49849ea9', timeout = 10.0, interval = 0.5, no_ack = True
on_interval = <promise@0x7f523b8c36d0>
def wait_for(self, task_id,
timeout=None, interval=0.5, no_ack=True, on_interval=None):
"""Wait for task and return its result.
If the task raises an exception, this exception
will be re-raised by :func:`wait_for`.
Raises:
celery.exceptions.TimeoutError:
If `timeout` is not :const:`None`, and the operation
takes longer than `timeout` seconds.
"""
self._ensure_not_eager()
time_elapsed = 0.0
while 1:
meta = self.get_task_meta(task_id)
if meta['status'] in states.READY_STATES:
return meta
if on_interval:
on_interval()
# avoid hammering the CPU checking status.
time.sleep(interval)
time_elapsed += interval
if timeout and time_elapsed >= timeout:
> raise TimeoutError('The operation timed out.')
E celery.exceptions.TimeoutError: The operation timed out.
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/backends/base.py:687: TimeoutError
----------------------------------------------------------------------------------- Captured stderr setup ------------------------------------------------------------------------------------
[2021-01-12 19:31:05,011: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5072//
[2021-01-12 19:31:05,436: INFO/MainProcess] sync with loader@14f9e84ef88d
------------------------------------------------------------------------------------- Captured log setup -------------------------------------------------------------------------------------
INFO celery.worker.consumer.connection:connection.py:22 Connected to amqp://guest:**@127.0.0.1:5072//
INFO celery.worker.control:control.py:310 sync with loader@14f9e84ef88d
_____________________________________________________________________________ ERROR at setup of test_incremental _____________________________________________________________________________
swh_scheduler_celery_app = <Celery celery.tests at 0x7f5241caeda0>
swh_scheduler_celery_includes = ['swh.scheduler.tests.tasks', 'swh.lister.bitbucket.tasks', 'swh.lister.cgit.tasks', 'swh.lister.cran.tasks', 'swh.lister.debian.tasks', 'swh.lister.gitea.tasks', ...]
@pytest.fixture(scope="session")
def swh_scheduler_celery_worker(
swh_scheduler_celery_app, swh_scheduler_celery_includes,
):
"""Spawn a worker"""
for module in swh_scheduler_celery_includes:
swh_scheduler_celery_app.loader.import_task_module(module)
> with worker.start_worker(swh_scheduler_celery_app, pool="solo") as w:
../swh-scheduler/swh/scheduler/pytest_plugin.py:103:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/usr/lib/python3.7/contextlib.py:112: in __enter__
return next(self.gen)
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/contrib/testing/worker.py:82: in start_worker
assert ping.delay().get(timeout=ping_task_timeout) == 'pong'
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/result.py:230: in get
on_message=on_message,
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/backends/base.py:655: in wait_for_pending
no_ack=no_ack,
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
self = <celery.backends.cache.CacheBackend object at 0x7f523b8b26a0>, task_id = '903dbe6a-e48e-4f21-af94-272e49849ea9', timeout = 10.0, interval = 0.5, no_ack = True
on_interval = <promise@0x7f523b8c36d0>
def wait_for(self, task_id,
timeout=None, interval=0.5, no_ack=True, on_interval=None):
"""Wait for task and return its result.
If the task raises an exception, this exception
will be re-raised by :func:`wait_for`.
Raises:
celery.exceptions.TimeoutError:
If `timeout` is not :const:`None`, and the operation
takes longer than `timeout` seconds.
"""
self._ensure_not_eager()
time_elapsed = 0.0
while 1:
meta = self.get_task_meta(task_id)
if meta['status'] in states.READY_STATES:
return meta
if on_interval:
on_interval()
# avoid hammering the CPU checking status.
time.sleep(interval)
time_elapsed += interval
if timeout and time_elapsed >= timeout:
> raise TimeoutError('The operation timed out.')
E celery.exceptions.TimeoutError: The operation timed out.
../../../.virtualenvs/swh/lib/python3.7/site-packages/celery/backends/base.py:687: TimeoutError
================================================================================== short test summary info ===================================================================================
ERROR swh/lister/bitbucket/tests/test_tasks.py::test_ping - celery.exceptions.TimeoutError: The operation timed out.
ERROR swh/lister/bitbucket/tests/test_tasks.py::test_incremental - celery.exceptions.TimeoutError: The operation timed out.
===================================================================================== 2 errors in 11.70s =====================================================================================
--- Logging error ---
Traceback (most recent call last):
File "/usr/lib/python3.7/logging/__init__.py", line 1037, in emit
stream.write(msg + self.terminator)
ValueError: I/O operation on closed file.
Call stack:
File "/usr/lib/python3.7/threading.py", line 885, in _bootstrap
self._bootstrap_inner()
File "/usr/lib/python3.7/threading.py", line 917, in _bootstrap_inner
self.run()
File "/usr/lib/python3.7/threading.py", line 865, in run
self._target(*self._args, **self._kwargs)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/worker/worker.py", line 203, in start
self.blueprint.start(self)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/bootsteps.py", line 116, in start
step.start(parent)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/bootsteps.py", line 365, in start
return self.obj.start()
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 311, in start
blueprint.start(self)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/bootsteps.py", line 116, in start
step.start(parent)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/worker/consumer/consumer.py", line 592, in start
c.loop(*c.loop_args())
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/worker/loops.py", line 81, in asynloop
next(loop)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/asynchronous/hub.py", line 361, in create_loop
cb(*cbargs)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/transport/base.py", line 235, in on_readable
reader(loop)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/transport/base.py", line 217, in _read
drain_events(timeout=0)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/amqp/connection.py", line 522, in drain_events
while not self.blocking_read(timeout):
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/amqp/connection.py", line 528, in blocking_read
return self.on_inbound_frame(frame)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/amqp/method_framing.py", line 77, in on_frame
callback(channel, msg.frame_method, msg.frame_args, msg)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/amqp/connection.py", line 535, in on_inbound_method
method_sig, payload, content,
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/amqp/abstract_channel.py", line 143, in dispatch_method
listener(*args)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/amqp/channel.py", line 1613, in _on_basic_deliver
fun(msg)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/messaging.py", line 620, in _receive_callback
return on_m(message) if on_m else self.receive(decoded, message)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/messaging.py", line 586, in receive
[callback(body, message) for callback in callbacks]
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/messaging.py", line 586, in <listcomp>
[callback(body, message) for callback in callbacks]
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/worker/pidbox.py", line 44, in on_message
self.node.handle_message(body, message)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/pidbox.py", line 143, in handle_message
return self.dispatch(**body)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/pidbox.py", line 102, in dispatch
reply = handle(method, arguments)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/pidbox.py", line 121, in handle_call
return self.handle(method, arguments)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/kombu/pidbox.py", line 118, in handle
return self.handlers[method](self.state, **arguments)
File "/home/tenma/.virtualenvs/swh/lib/python3.7/site-packages/celery/worker/control.py", line 310, in hello
logger.info('sync with %s', from_node)
Message: 'sync with %s'
Arguments: ('listers@d7590aa5e875',)
^CException ignored in: <module 'threading' from '/usr/lib/python3.7/threading.py'>
Traceback (most recent call last):
File "/usr/lib/python3.7/threading.py", line 1281, in _shutdown
t.join()
File "/usr/lib/python3.7/threading.py", line 1032, in join
self._wait_for_tstate_lock()
File "/usr/lib/python3.7/threading.py", line 1048, in _wait_for_tstate_lock
elif lock.acquire(block, timeout):
KeyboardInterrupt
About · Developer information · Legal