1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136
|
import contextvars
from typing import Optional
import html5lib
from asgiref.local import Local
from django.http import HttpResponse
from django.test import (
AsyncClient,
AsyncRequestFactory,
Client,
RequestFactory,
TestCase,
TransactionTestCase,
)
from debug_toolbar.panels import Panel
from debug_toolbar.store import get_store
from debug_toolbar.toolbar import DebugToolbar
data_contextvar = contextvars.ContextVar("djdt_toolbar_test_client")
class ToolbarTestClient(Client):
def request(self, **request):
# Use a thread/async task context-local variable to guard against a
# concurrent _created signal from a different thread/task.
data = Local()
data.toolbar = None
def handle_toolbar_created(sender, toolbar=None, **kwargs):
data.toolbar = toolbar
DebugToolbar._created.connect(handle_toolbar_created)
try:
response = super().request(**request)
finally:
DebugToolbar._created.disconnect(handle_toolbar_created)
response.toolbar = data.toolbar
return response
class AsyncToolbarTestClient(AsyncClient):
async def request(self, **request):
# Use a thread/async task context-local variable to guard against a
# concurrent _created signal from a different thread/task.
# In cases testsuite will have both regular and async tests or
# multiple async tests running in an eventloop making async_client calls.
data_contextvar.set(None)
def handle_toolbar_created(sender, toolbar=None, **kwargs):
data_contextvar.set(toolbar)
DebugToolbar._created.connect(handle_toolbar_created)
try:
response = await super().request(**request)
finally:
DebugToolbar._created.disconnect(handle_toolbar_created)
response.toolbar = data_contextvar.get()
return response
rf = RequestFactory()
arf = AsyncRequestFactory()
class BaseMixin:
_is_async = False
client_class = ToolbarTestClient
async_client_class = AsyncToolbarTestClient
panel: Optional[Panel] = None
panel_id = None
def setUp(self):
super().setUp()
self._get_response = lambda request: HttpResponse()
self.request = rf.get("/")
if self._is_async:
self.request = arf.get("/")
self.toolbar = DebugToolbar(self.request, self.get_response_async)
else:
self.toolbar = DebugToolbar(self.request, self.get_response)
self.toolbar.stats = {}
if self.panel_id:
self.panel = self.toolbar.get_panel_by_id(self.panel_id)
self.panel.enable_instrumentation()
else:
self.panel = None
def tearDown(self):
if self.panel:
self.panel.disable_instrumentation()
super().tearDown()
def get_response(self, request):
return self._get_response(request)
async def get_response_async(self, request):
return self._get_response(request)
def assertValidHTML(self, content):
parser = html5lib.HTMLParser()
parser.parseFragment(content)
if parser.errors:
msg_parts = ["Invalid HTML:"]
lines = content.split("\n")
for position, errorcode, datavars in parser.errors:
msg_parts.append(f" {html5lib.constants.E[errorcode]}" % datavars)
msg_parts.append(f" {lines[position[0] - 1]}")
raise self.failureException("\n".join(msg_parts))
def reload_stats(self):
data = self.toolbar.store.panel(self.toolbar.request_id, self.panel_id)
self.panel.load_stats_from_store(data)
class BaseTestCase(BaseMixin, TestCase):
pass
class BaseMultiDBTestCase(BaseMixin, TransactionTestCase):
databases = {"default", "replica"}
class IntegrationTestCase(TestCase):
"""Base TestCase for tests involving clients making requests."""
def setUp(self):
# The HistoryPanel keeps track of previous stores in memory.
# This bleeds into other tests and violates their idempotency.
# Clear the store before each test.
get_store().clear()
super().setUp()
|