-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(test): test_create_task.py (#423)
* feat(test): test_create_task.py * chore: `black .` --------- Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
- Loading branch information
1 parent
41d195f
commit eef418d
Showing
3 changed files
with
64 additions
and
0 deletions.
There are no files selected for viewing
File renamed without changes.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,64 @@ | ||
import logging | ||
import pytest | ||
import asyncio | ||
from a_sync.asyncio.create_task import create_task | ||
|
||
|
||
@pytest.mark.asyncio_cooperative | ||
async def test_create_task_with_coroutine(): | ||
async def sample_coroutine(): | ||
return "Hello, World!" | ||
|
||
task = create_task(sample_coroutine()) | ||
result = await task | ||
assert result == "Hello, World!" | ||
|
||
|
||
@pytest.mark.asyncio_cooperative | ||
async def test_create_task_with_future(): | ||
loop = asyncio.get_event_loop() | ||
future = loop.create_future() | ||
future.set_result("Future Result") | ||
|
||
task = create_task(future) | ||
result = await task | ||
assert result == "Future Result" | ||
|
||
|
||
@pytest.mark.asyncio_cooperative | ||
async def test_create_task_with_name(): | ||
async def sample_coroutine(): | ||
return "Named Task" | ||
|
||
task = create_task(sample_coroutine(), name="TestTask") | ||
assert task.get_name() == "TestTask" | ||
|
||
|
||
@pytest.mark.asyncio_cooperative | ||
async def test_create_task_skip_gc_until_done(): | ||
async def sample_coroutine(): | ||
return "GC Test" | ||
|
||
task = create_task(sample_coroutine(), skip_gc_until_done=True) | ||
result = await task | ||
assert result == "GC Test" | ||
|
||
|
||
@pytest.mark.asyncio_cooperative | ||
async def test_create_task_log_destroy_pending(): | ||
async def sample_coroutine(): | ||
return "Log Test" | ||
|
||
task = create_task(sample_coroutine(), log_destroy_pending=False) | ||
assert not task._log_destroy_pending | ||
|
||
|
||
@pytest.mark.asyncio_cooperative | ||
async def test_create_task_handles_non_coroutine_awaitable(): | ||
class CustomAwaitable: | ||
def __await__(self): | ||
yield | ||
return "Custom Awaitable Result" | ||
|
||
result = await create_task(CustomAwaitable()) | ||
assert result == "Custom Awaitable Result" |
File renamed without changes.