django/tests/backends/base/test_base_async.py
2025-09-26 09:47:21 -05:00

21 lines
790 B
Python

import unittest
from django.db import connection, new_connection
from django.test import SimpleTestCase
@unittest.skipUnless(connection.supports_async is True, "Async DB test")
class AsyncDatabaseWrapperTests(SimpleTestCase):
async def test_async_cursor(self):
async with new_connection() as conn:
async with conn.acursor() as cursor:
await cursor.execute("SELECT 1")
result = (await cursor.fetchone())[0]
self.assertEqual(result, 1)
async def test_async_cursor_alias(self):
async with new_connection() as conn:
async with conn.acursor() as cursor:
await cursor.aexecute("SELECT 1")
result = (await cursor.afetchone())[0]
self.assertEqual(result, 1)