Django DRF测试中RequestsClient带Token认证请求失败如何解决
问题背景
需要对配置了认证权限控制的多个Django Rest Framework API接口做测试,最初选用DRF自带的RequestsClient发起测试请求,但是出现了原生requests库调用正常、RequestsClient调用报错的问题。
初始测试代码
from rest_framework.test import APITestCase from rest_framework.test import RequestsClient from django.urls import reverse import requests URL_PREFIX="http://testserver" API_TOKEN="tfhgffhgf675h" class APITests(APITestCase): @staticmethod def _get_full_url(url, *args, **kwargs): return URL_PREFIX + reverse(url, *args, **kwargs) def setUp(self): self.client = RequestsClient() def test_stuff(self): url = self._get_full_url("ahs:agena_results-list") # 原生requests库调用 - 可正常运行 # 注:测试发现RequestsClient()存在特殊处理逻辑,可直接解析http://testserver地址; # 原生requests发起GET请求必须指定127.0.0.1及对应端口 response = requests.get("http://127.0.0.1:8000", headers={"Authorization": f"Token {API_TOKEN}"}) # RequestsClient()调用 - 运行失败 response = self.client.get(url, headers={"Authorization": f"Token {API_TOKEN}"})
异常现象
- 用原生
requests库发起带认证头的请求可以正常执行 - 用
RequestsClient发起带Authorization认证头的请求直接抛出异常,报错指向事务管理错误与MySQL连接异常,核心报错栈如下:
Internal Server Error: /ahs/api/agena_results/ Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/django/db/models/sql/compiler.py", line 1361, in execute_sql cursor.execute(sql, params) File "/usr/local/lib/python3.9/site-packages/django/db/backends/utils.py", line 67, in execute return self._execute_with_wrappers( File "/usr/local/lib/python3.9/site-packages/django/db/backends/utils.py", line 80, in _execute_with_wrappers return executor(sql, params, many, context) File "/usr/local/lib/python3.9/site-packages/django/db/backends/utils.py", line 83, in _execute self.db.validate_no_broken_transaction() File "/usr/local/lib/python3.9/site-packages/django/db/backends/base/base.py", line 480, in validate_no_broken_transaction raise TransactionManagementError( django.db.transaction.TransactionManagementError: An error occurred in the current transaction. You can't execute queries until the end of the 'atomic' block. During handling of the above exception, another exception occurred: Traceback (most recent call last): File "/usr/local/lib/python3.9/site-packages/django/core/handlers/exception.py", line 55, in inner response = get_response(request) File "/usr/local/lib/python3.9/site-packages/django/core/handlers/base.py", line 197, in _get_response response = wrapped_callback(request, *callback_args, **callback_kwargs) File "/usr/local/lib/python3.9/site-packages/django/views/decorators/csrf.py", line 54, in wrapped_view return view_func(*args, **kwargs) File "/usr/local/lib/python3.9/site-packages/rest_framework/viewsets.py", line 125, in view return self.dispatch(request, *args, **kwargs) File "/usr/local/lib/python3.9/site-packages/rest_framework/views.py", line 509, in dispatch response = self.handle_exception(exc) File "/usr/local/lib/python3.9/site-packages/rest_framework/views.py", line 469, in handle_exception self.raise_uncaught_exception(exc) File "/usr/local/lib/python3.9/site-packages/rest_framework/views.py", line 480, in raise_uncaught_exception raise exc File "/usr/local/lib/python3.9/site-packages/rest_framework/views.py", line 497, in dispatch self.initial(request, *args, **kwargs) File "/usr/local/lib/python3.9/site-packages/rest_framework/views.py", line 414, in initial self.perform_authentication(request) File "/usr/local/lib/python3.9/site-packages/rest_framework/views.py", line 324, in perform_authentication request.user File "/usr/local/lib/python3.9/site-packages/rest_framework/request.py", line 227, in user self._authenticate() File "/usr/local/lib/python3.9/site-packages/rest_framework/request.py", line 380, in _authenticate user_auth_tuple = authenticator.authenticate(self) File "/usr/local/lib/python3.9/site-packages/rest_framework/authentication.py", line 196, in authenticate return self.authenticate_credentials(token) File "/usr/local/lib/python3.9/site-packages/rest_framework/authentication.py", line 201, in authenticate_credentials token = model.objects.select_related('user').get(key=key) File "/usr/local/lib/python3.9/site-packages/django/db/models/query.py", line 492, in get num = len(clone) File "/usr/local/lib/python3.9/site-packages/django/db/models/query.py", line 302, in __len__ self._fetch_all() File "/usr/local/lib/python3.9/site-packages/django/db/models/query.py", line 1507, in _fetch_all self._result_cache = list(self._iterable_class(self)) File "/usr/local/lib/python3.9/site-packages/django/db/models/query.py", line 57, in __iter__ results = compiler.execute_sql( File "/usr/local/lib/python3.9/site-packages/django/db/models/sql/compiler.py", line 1364, in execute_sql cursor.close() File "/usr/local/lib/python3.9/site-packages/MySQLdb/cursors.py", line 83, in close while self.nextset(): File "/usr/local/lib/python3.9/site-packages/MySQLdb/cursors.py", line 137, in nextset nr = db.next_result() MySQLdb._exceptions.OperationalError: (2006, '') . ---------------------------------------------------------------------- Ran 1 test in 30.774s
- 异常存在明确规律:移除
RequestsClient请求中的Authorization头后,请求不会崩溃,但会返回403无权限状态码,无法满足测试要求。 - 已尝试的修复方案:直接在
RequestsClient实例上更新公共请求头设置认证信息,问题依然存在,代码如下:
self.client.headers.update({'Authorization': f'Token {API_TOKEN}'})
问题原因
核心原因是选错了测试客户端:RequestsClient本质是对原生requests库的封装,它发起请求时会走真实的HTTP网络栈,和Django测试框架的事务隔离机制完全不兼容:
- Django的
APITestCase默认会把每个测试用例包裹在数据库原子块中,测试结束后自动回滚所有数据改动,不会污染测试库 RequestsClient发起的请求运行在独立的执行上下文里,拿不到测试事务里的数据库连接状态,当它尝试查询Token表做认证时,会遇到事务状态异常,最终抛出MySQL连接断开、事务管理错误- 去掉认证头时不会触发数据库查询操作(直接返回403),所以不会触发这个报错;而原生
requests请求是直接打在运行中的Django服务上,走的是服务本身的独立数据库连接,和测试用例的事务完全隔离,所以能正常运行。
代码错误点
- 不应该在继承
APITestCase的测试类里用RequestsClient做接口测试,二者的运行机制天生冲突 - 硬拼接
http://testserver前缀的逻辑是多余的,Django专用测试客户端本身就能处理相对路径
修复方案
直接换成DRF测试专用的APIClient即可,它是Django测试客户端的DRF扩展,完全兼容测试框架的事务机制,不需要启动真实服务、不需要指定IP端口,还内置了认证相关的快捷方法。
修复后的测试代码示例:
from rest_framework.test import APITestCase from rest_framework.test import APIClient from django.urls import reverse from rest_framework.authtoken.models import Token from django.contrib.auth import get_user_model API_TOKEN="tfhgffhgf675h" User = get_user_model() class APITests(APITestCase): def setUp(self): self.client = APIClient() # 方式1:直接设置认证头,和原有逻辑一致 self.client.credentials(HTTP_AUTHORIZATION=f'Token {API_TOKEN}') # 方式2(更推荐):测试时直接创建测试用户和对应Token,避免硬编码无效Token # self.test_user = User.objects.create_user(username='test', password='test123') # self.token = Token.objects.create(user=self.test_user) # self.client.credentials(HTTP_AUTHORIZATION=f'Token {self.token.key}') def test_stuff(self): # 直接传reverse拿到的相对路径即可,不需要拼接testserver前缀 url = reverse("ahs:agena_results-list") response = self.client.get(url) # 后续断言逻辑 self.assertEqual(response.status_code, 200)
如果确实需要用RequestsClient模拟真实HTTP请求(比如测试跨服务调用、WSGI层逻辑),就不要继承APITestCase,换成StaticLiveServerTestCase,它会在测试启动时真实启动Django服务,这时候用RequestsClient请求真实的服务地址(self.live_server_url拼接路径)就不会有事务冲突问题,但这种方式测试运行速度会慢很多,普通接口测试优先用APIClient。
内容的提问来源于stack exchange,提问作者John
相关产品推荐
相关产品推荐

