Isoliertes Mocking von Async-Generatoren in Django-Unit-Tests

from unittest.mock import patch
from django.test import SimpleTestCase
 
class AsyncStreamServiceTest(SimpleTestCase):
    async def test_async_generator_stream(self):
        async def mock_stream():
            yield {"status": "processing"}
            yield {"status": "complete"}
 
        with patch("services.external.fetch_stream", side_effect=mock_stream):
            from services.external import fetch_stream
            results = [data async for data in fetch_stream()]
            self.assertEqual(len(results), 2)
            self.assertEqual(results[1]["status"], "complete")