Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

Is there a way to fail a test if an explicitly created background task fails after yielding from the fixture that created it?

Aperta
  • Qualcosa non va?
#196 8 commenti 1 reazione 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
5/5
Tempo stimato
Più di una settimana
Idoneità per principianti
30/100
Tipo di issue
Funzionalità
Chiarezza
Abbastanza chiara
Stato di attività
Ferma
Stack tecnologico
python
Ambito
testing-qa

Direzione di ricerca

Inizia con il fixture asyncio e l'esempio di background-task in questa issue, concentrandoti su come pytest-asyncio gestisce i task creati prima che un fixture restituisca un valore. Verifica se un'eccezione di un task può annullare e far fallire il test invece di lasciare queue.join() bloccato. Il lavoro è completato quando l'errore mostrato viene segnalato da pytest senza richiedere un callback ad hoc.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

I'll start with an example to provide some context:

import asyncio
import contextlib
import itertools

import pytest


class Consumer:
    def __init__(self, queue: asyncio.Queue):
        self._queue = queue

    async def run(self):
        for i in itertools.count(0):
            item = await self._queue.get()
            print(item)

            if i == 3:
                raise ValueError('oups')

            self._queue.task_done()


@pytest.fixture
async def queue() -> asyncio.Queue:
    return asyncio.Queue()


@pytest.fixture
async def consumer_service(queue):
    consumer = Consumer(queue)
    background_task = asyncio.create_task(consumer.run())

    yield

    background_task.cancel()
    with contextlib.suppress(asyncio.CancelledError):
        await run_task


@pytest.mark.asyncio
@pytest.mark.usefixtures('consumer_service')
async def test_consumer_consumes(queue: asyncio.Queue):
    queue.put_nowait('I')
    queue.put_nowait('Will')
    queue.put_nowait('Not')
    queue.put_nowait('Hang')

    await queue.join()

Here, the consumer service fixture will create a background task that runs the consumer's loop. When processing the last item in the queue, a value error is raised and the background task finishes with an exception before marking the last item as done. In this state, the test will hang and there will be no indication that something went astray.

Ideally, there would be some way to react to the background task failing by canceling and failing the test but I have not seen a way to do it. The closest I got to a solution was to attach a done callback to the background task and check for an exception there but I haven't come across an API that would allow me to recover from this or a way to signal the error (without it being captured by pytest).

Lingua principale
Python
Stelle
1.7k
Fork
209
Merge medio
4h 52m
PR unite (30g)
14

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di pytest-dev/pytest-asyncio

Tutte le issue di pytest-dev/pytest-asyncio

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.