41 lines
1.1 KiB
Python
41 lines
1.1 KiB
Python
import asyncio
|
|
import gc
|
|
import logging
|
|
import unittest
|
|
|
|
from core import tasks
|
|
|
|
|
|
class SpawnTests(unittest.IsolatedAsyncioTestCase):
|
|
async def test_survives_gc_and_set_emptied(self):
|
|
base = len(tasks._BACKGROUND) # other tests may leave tasks from closed loops
|
|
done = asyncio.Event()
|
|
|
|
async def work():
|
|
await asyncio.sleep(0.01)
|
|
done.set()
|
|
|
|
tasks.spawn(work())
|
|
gc.collect()
|
|
self.assertEqual(len(tasks._BACKGROUND), base + 1)
|
|
await asyncio.wait_for(done.wait(), 1)
|
|
await asyncio.sleep(0)
|
|
self.assertEqual(len(tasks._BACKGROUND), base)
|
|
|
|
async def test_exception_logged_without_message(self):
|
|
base = len(tasks._BACKGROUND)
|
|
|
|
async def boom():
|
|
raise ValueError("secret-token")
|
|
|
|
with self.assertLogs("core.tasks", "ERROR") as cm:
|
|
tasks.spawn(boom())
|
|
await asyncio.sleep(0.02)
|
|
self.assertIn("ValueError", cm.output[0])
|
|
self.assertNotIn("secret-token", cm.output[0])
|
|
self.assertEqual(len(tasks._BACKGROUND), base)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|