mirror of
https://github.com/pewdiepie-archdaemon/odysseus.git
synced 2026-09-10 18:22:20 +02:00
87 lines
2.4 KiB
Python
87 lines
2.4 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
|
|
from src import task_scheduler
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clear_shared_cache():
|
|
task_scheduler._shared_cache.clear()
|
|
task_scheduler._shared_cache_pending.clear()
|
|
yield
|
|
task_scheduler._shared_cache.clear()
|
|
task_scheduler._shared_cache_pending.clear()
|
|
|
|
|
|
async def test_cached_owner_cancellation_wakes_waiters_and_allows_retry():
|
|
key = ("cancelled-owner",)
|
|
fetch_started = asyncio.Event()
|
|
|
|
async def blocked_fetch():
|
|
fetch_started.set()
|
|
await asyncio.Event().wait()
|
|
|
|
owner = asyncio.create_task(task_scheduler._cached(key, 60, blocked_fetch))
|
|
await fetch_started.wait()
|
|
|
|
async def unexpected_fetch():
|
|
pytest.fail("a waiter must share the owner's fetch")
|
|
|
|
waiter = asyncio.create_task(task_scheduler._cached(key, 60, unexpected_fetch))
|
|
await asyncio.sleep(0)
|
|
|
|
owner.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await owner
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await asyncio.wait_for(waiter, timeout=1)
|
|
|
|
assert key not in task_scheduler._shared_cache_pending
|
|
|
|
async def retry_fetch():
|
|
return "fresh"
|
|
|
|
result = await asyncio.wait_for(
|
|
task_scheduler._cached(key, 60, retry_fetch),
|
|
timeout=1,
|
|
)
|
|
assert result == "fresh"
|
|
|
|
|
|
async def test_cached_waiter_cancellation_does_not_cancel_shared_fetch():
|
|
key = ("cancelled-waiter",)
|
|
fetch_started = asyncio.Event()
|
|
release_fetch = asyncio.Event()
|
|
|
|
async def blocked_fetch():
|
|
fetch_started.set()
|
|
await release_fetch.wait()
|
|
return "shared"
|
|
|
|
owner = asyncio.create_task(task_scheduler._cached(key, 60, blocked_fetch))
|
|
await fetch_started.wait()
|
|
|
|
async def unexpected_fetch():
|
|
pytest.fail("a waiter must share the owner's fetch")
|
|
|
|
waiter = asyncio.create_task(task_scheduler._cached(key, 60, unexpected_fetch))
|
|
await asyncio.sleep(0)
|
|
waiter.cancel()
|
|
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await waiter
|
|
|
|
pending = task_scheduler._shared_cache_pending[key]
|
|
assert not pending.cancelled()
|
|
assert not owner.done()
|
|
|
|
release_fetch.set()
|
|
assert await asyncio.wait_for(owner, timeout=1) == "shared"
|
|
assert key not in task_scheduler._shared_cache_pending
|
|
|
|
async def cache_miss():
|
|
pytest.fail("the successful owner result should be cached")
|
|
|
|
assert await task_scheduler._cached(key, 60, cache_miss) == "shared"
|