a database connection cannot be used in an async context async def testgetasyncconnectionself with self assertraisessynchronousonlyoperation listsimplemodel objects all class asyncunsafetestsimpletestcase asyncunsafe def dangerousmethodself return true async def testasyncunsafeself asyncunsafe decorator catches bad access and returns the right message msg you cannot call this from an async context use a thread or synctoasync with self assertraisesmessagesynchronousonlyoperation msg self dangerousmethod mock patch dictos environ djangoallowasyncunsafe true asynctosync mock patch is not asyncaware async def testasyncunsafesuppressedself decorator doesn t trigger check when the environment variable to suppress it is set try self dangerousmethod except synchronousonlyoperation self failsynchronousonlyoperation should not be raised class syncviewview def getself request args kwargs return httpresponsehello sync world class asyncviewview async def getself request args kwargs return httpresponsehello async world class viewtestssimpletestcase def testviewsarecorrectlymarkedself tests syncview false asyncview true for viewcls isasync in tests with self subtestviewclsviewcls isasyncisasync self assertisviewcls viewisasync isasync callback viewcls asview self assertisiscoroutinefunctioncallback isasync def testmixedviewsraiseerrorself class mixedviewview def getself request args kwargs return httpresponsehello mixed world async def postself request args kwargs return httpresponsehello mixed world msg fmixedview qualname http handlers must either be all sync or all async with self assertraisesmessageimproperlyconfigured msg mixedview asview def testoptionshandlerrespondscorrectlyself tests syncview false asyncview true for viewcls iscoroutine in tests with self subtestviewclsviewcls iscoroutineiscoroutine instance viewcls response instance optionsnone self assertis asyncio iscoroutineresponse iscoroutine if iscoroutine response asyncio runresponse self assertisinstanceresponse httpresponse def testhttpmethodnotallowedrespondscorrectlyself requestfactory requestfactory tests syncview false asyncview true for viewcls iscoroutine in tests with self subtestviewclsviewcls iscoroutineiscoroutine instance viewcls response instance httpmethodnotallowedrequestfactory post self assertis asyncio iscoroutineresponse iscoroutine if iscoroutine response asyncio runresponse self assertisinstanceresponse httpresponsenotallowed def testbaseviewclassissyncself self assertisview viewisasync false a database connection cannot be used in an async context async_unsafe decorator should work correctly and returns the correct message async_unsafe decorator catches bad access and returns the right message mock patch is not async aware decorator doesn t trigger check when the environment variable to suppress it is set view and by extension any subclasses that don t define handlers are sync
import asyncio
import os
from unittest import mock

from asgiref.sync import async_to_sync, iscoroutinefunction

from django.core.cache import DEFAULT_CACHE_ALIAS, caches
from django.core.exceptions import ImproperlyConfigured, SynchronousOnlyOperation
from django.http import HttpResponse, HttpResponseNotAllowed
from django.test import RequestFactory, SimpleTestCase
from django.utils.asyncio import async_unsafe
from django.views.generic.base import View

from .models import SimpleModel


class CacheTest(SimpleTestCase):
    def test_caches_local(self):
        @async_to_sync
        async def async_cache():
            return caches[DEFAULT_CACHE_ALIAS]

        cache_1 = async_cache()
        cache_2 = async_cache()
        self.assertIs(cache_1, cache_2)


class DatabaseConnectionTest(SimpleTestCase):
    

    async def test_get_async_connection(self):
        with self.assertRaises(SynchronousOnlyOperation):
            list(SimpleModel.objects.all())


class AsyncUnsafeTest(SimpleTestCase):
    

    @async_unsafe
    def dangerous_method(self):
        return True

    async def test_async_unsafe(self):
        
        
        msg = (
            "You cannot call this from an async context - use a thread or "
            "sync_to_async."
        )
        with self.assertRaisesMessage(SynchronousOnlyOperation, msg):
            self.dangerous_method()

    @mock.patch.dict(os.environ, {"DJANGO_ALLOW_ASYNC_UNSAFE": "true"})
    @async_to_sync  
    async def test_async_unsafe_suppressed(self):
        
        
        try:
            self.dangerous_method()
        except SynchronousOnlyOperation:
            self.fail("SynchronousOnlyOperation should not be raised.")


class SyncView(View):
    def get(self, request, *args, **kwargs):
        return HttpResponse("Hello (sync) world!")


class AsyncView(View):
    async def get(self, request, *args, **kwargs):
        return HttpResponse("Hello (async) world!")


class ViewTests(SimpleTestCase):
    def test_views_are_correctly_marked(self):
        tests = [
            (SyncView, False),
            (AsyncView, True),
        ]
        for view_cls, is_async in tests:
            with self.subTest(view_cls=view_cls, is_async=is_async):
                self.assertIs(view_cls.view_is_async, is_async)
                callback = view_cls.as_view()
                self.assertIs(iscoroutinefunction(callback), is_async)

    def test_mixed_views_raise_error(self):
        class MixedView(View):
            def get(self, request, *args, **kwargs):
                return HttpResponse("Hello (mixed) world!")

            async def post(self, request, *args, **kwargs):
                return HttpResponse("Hello (mixed) world!")

        msg = (
            f"{MixedView.__qualname__} HTTP handlers must either be all sync or all "
            "async."
        )
        with self.assertRaisesMessage(ImproperlyConfigured, msg):
            MixedView.as_view()

    def test_options_handler_responds_correctly(self):
        tests = [
            (SyncView, False),
            (AsyncView, True),
        ]
        for view_cls, is_coroutine in tests:
            with self.subTest(view_cls=view_cls, is_coroutine=is_coroutine):
                instance = view_cls()
                response = instance.options(None)
                self.assertIs(
                    asyncio.iscoroutine(response),
                    is_coroutine,
                )
                if is_coroutine:
                    response = asyncio.run(response)

                self.assertIsInstance(response, HttpResponse)

    def test_http_method_not_allowed_responds_correctly(self):
        request_factory = RequestFactory()
        tests = [
            (SyncView, False),
            (AsyncView, True),
        ]
        for view_cls, is_coroutine in tests:
            with self.subTest(view_cls=view_cls, is_coroutine=is_coroutine):
                instance = view_cls()
                response = instance.http_method_not_allowed(request_factory.post("/"))
                self.assertIs(
                    asyncio.iscoroutine(response),
                    is_coroutine,
                )
                if is_coroutine:
                    response = asyncio.run(response)

                self.assertIsInstance(response, HttpResponseNotAllowed)

    def test_base_view_class_is_sync(self):
        
        self.assertIs(View.view_is_async, False)