You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何测试Django 1.10+新风格ThreadLocal中间件的线程本地存储?

Testing Django 1.10+ Style ThreadLocal Middleware

Great question—testing the new-style middleware does require a slight shift from the old process_request approach, since the __call__ method wraps the entire request/response cycle and cleans up after itself. Here are a few solid approaches to verify your middleware is correctly setting the request in thread-local storage:

The cleanest way to make your middleware testable is to split the request setup/cleanup logic into separate helper methods. This keeps your code modular and lets you test individual behaviors directly, without relying on the full __call__ flow.

First, refactor your middleware:

from threading import local

class ThreadLocalMiddleware:
    _thread_locals = local()

    def __init__(self, get_response=None):
        self.get_response = get_response

    def _set_thread_local_request(self, request):
        """Helper to set the request in thread-local storage"""
        self._thread_locals.request = request

    def _clear_thread_local_request(self):
        """Helper to clean up the request from thread-local storage"""
        if hasattr(self._thread_locals, 'request'):
            del self._thread_locals.request

    def __call__(self, request):
        self._set_thread_local_request(request)
        try:
            response = self.get_response(request)
        finally:
            # Ensure cleanup happens even if an exception is raised
            self._clear_thread_local_request()
        return response

    def process_exception(self, request, exception):
        self._clear_thread_local_request()

Now you can write straightforward unit tests for each helper method, plus tests for the full cycle:

from django.test import TestCase, RequestFactory
from unittest.mock import Mock
from yourapp.middleware import ThreadLocalMiddleware

class ThreadLocalMiddlewareTests(TestCase):
    def setUp(self):
        self.factory = RequestFactory()
        self.middleware = ThreadLocalMiddleware()

    def test_set_thread_local_request(self):
        request = self.factory.get('/')
        self.middleware._set_thread_local_request(request)
        
        # Verify the request is stored
        self.assertEqual(request, ThreadLocalMiddleware._thread_locals.request)

    def test_clear_thread_local_request(self):
        request = self.factory.get('/')
        ThreadLocalMiddleware._thread_locals.request = request
        
        self.middleware._clear_thread_local_request()
        # Verify the request is removed
        self.assertFalse(hasattr(ThreadLocalMiddleware._thread_locals, 'request'))

    def test_full_request_cycle_stores_and_clears_request(self):
        request = self.factory.get('/')
        mock_response = Mock()
        
        # Create a mock get_response that checks the thread-local during execution
        def mock_get_response(req):
            self.assertEqual(req, ThreadLocalMiddleware._thread_locals.request)
            return mock_response

        self.middleware.get_response = mock_get_response
        response = self.middleware(request)
        
        # Verify the response is passed through
        self.assertEqual(response, mock_response)
        # Verify cleanup happened after the cycle
        self.assertFalse(hasattr(ThreadLocalMiddleware._thread_locals, 'request'))

    def test_exception_triggers_cleanup(self):
        request = self.factory.get('/')
        
        # Make get_response raise an exception
        def failing_get_response(req):
            raise ValueError("Test exception")

        self.middleware.get_response = failing_get_response
        
        with self.assertRaises(ValueError):
            self.middleware(request)
        
        # Even after exception, cleanup should occur
        self.assertFalse(hasattr(ThreadLocalMiddleware._thread_locals, 'request'))

2. Test with a Mock get_response (No Refactor Needed)

If you don't want to refactor the middleware immediately, you can use a mock get_response function to inspect the thread-local storage while the middleware is still in the middle of processing the request.

from django.test import TestCase, RequestFactory
from unittest.mock import Mock
from yourapp.middleware import ThreadLocalMiddleware

class ThreadLocalMiddlewareTests(TestCase):
    def test_request_is_set_during_processing(self):
        factory = RequestFactory()
        request = factory.get('/')
        middleware = ThreadLocalMiddleware()

        # Capture the thread-local state inside the mock get_response
        captured_request = None
        def mock_get_response(req):
            nonlocal captured_request
            captured_request = ThreadLocalMiddleware._thread_locals.request
            return Mock()

        middleware.get_response = mock_get_response
        middleware(request)
        
        # Verify the stored request matches the original
        self.assertEqual(captured_request, request)
        # Verify cleanup happened after
        self.assertFalse(hasattr(ThreadLocalMiddleware._thread_locals, 'request'))

3. Test with a Real View and Test Client

You can also use Django's TestClient to simulate a full request, and add a test view that checks the thread-local storage directly:

First, add a test view (you can put this in your test file temporarily):

from django.http import HttpResponse
from yourapp.middleware import ThreadLocalMiddleware

def thread_local_test_view(request):
    # Check if the current request matches the thread-local request
    match = request is ThreadLocalMiddleware._thread_locals.request
    return HttpResponse(str(match))

Then write the test:

from django.test import TestCase, Client

class ThreadLocalMiddlewareTests(TestCase):
    def test_request_in_thread_locals_during_view_execution(self):
        client = Client()
        # Map the test view to a URL in your test setup (or use reverse if you add it to urls.py)
        response = client.get('/test-thread-local/')
        
        # The view returns "True" if the request matches
        self.assertEqual(response.content.decode(), "True")

Key Takeaway

The main difference from the old middleware style is that you can't just call a standalone process_request method and check the state afterward—you need to inspect the thread-local storage while the request/response cycle is still in progress. Extracting helper methods is the most maintainable approach long-term, as it makes your middleware easier to test and modify.

内容的提问来源于stack exchange,提问作者Vladimir Linhart

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.05.15 07:22:52