|
| 1 | +import asyncio |
1 | 2 | import datetime
|
2 | 3 | import os
|
3 | 4 | import unittest
|
4 | 5 | from unittest.mock import patch
|
| 6 | +from asgiref.sync import sync_to_async |
5 | 7 |
|
6 | 8 | import django
|
7 | 9 | from django.contrib.auth.models import User
|
@@ -74,6 +76,33 @@ def test_chunked_cursor_wrapper_singleton(self, mock_wrapper):
|
74 | 76 | # ensure that cursor wrapping is applied only once
|
75 | 77 | self.assertEqual(mock_wrapper.call_count, 1)
|
76 | 78 |
|
| 79 | + @patch( |
| 80 | + "debug_toolbar.panels.sql.tracking.NormalCursorWrapper", |
| 81 | + wraps=sql_tracking.NormalCursorWrapper, |
| 82 | + ) |
| 83 | + async def test_cursor_wrapper_async(self, mock_wrapper): |
| 84 | + await sync_to_async(list)(User.objects.all()) |
| 85 | + |
| 86 | + self.assertEqual(mock_wrapper.call_count, 1) |
| 87 | + |
| 88 | + @patch( |
| 89 | + "debug_toolbar.panels.sql.tracking.NormalCursorWrapper", |
| 90 | + wraps=sql_tracking.NormalCursorWrapper, |
| 91 | + ) |
| 92 | + async def test_cursor_wrapper_asyncio_ctx(self, mock_wrapper): |
| 93 | + self.assertTrue(sql_tracking.recording.get()) |
| 94 | + await sync_to_async(list)(User.objects.all()) |
| 95 | + |
| 96 | + async def task(): |
| 97 | + sql_tracking.recording.set(False) |
| 98 | + await sync_to_async(list)(User.objects.all()) |
| 99 | + |
| 100 | + # Ensure this is called in another context |
| 101 | + await asyncio.create_task(task()) |
| 102 | + # Because it was called in another context, it should not have affected ours |
| 103 | + self.assertTrue(sql_tracking.recording.get()) |
| 104 | + self.assertEqual(mock_wrapper.call_count, 1) |
| 105 | + |
77 | 106 | def test_generate_server_timing(self):
|
78 | 107 | self.assertEqual(len(self.panel._queries), 0)
|
79 | 108 |
|
|
0 commit comments