put value into a list if it s not already one if not isinstancevalue list value value return value def assertandparsehtmlself html usermsg msg try dom parsehtmlhtml except htmlparseerror as e standardmsg sns msg e self failself formatmessageusermsg standardmsg return dom class assertnumqueriescontextcapturequeriescontext def initself testcase num connection self testcase testcase self num num super initconnection def exitself exctype excvalue traceback super exitexctype excvalue traceback if exctype is not none return executed lenself self testcase assertequal executed self num d queries executed d expectedncaptured queries were ns executed self num n join d s i querysql for i query in enumerateself capturedqueries start1 class asserttemplateusedcontext def initself testcase templatename msgprefix countnone self testcase testcase self templatename templatename self msgprefix msgprefix self count count self renderedtemplates self context contextlist def ontemplaterenderself sender signal template context kwargs self renderedtemplates appendtemplate self context appendcopycontext def testself self testcase asserttemplateused self templatename t name for t in self renderedtemplates if t name is not none self msgprefix self count def enterself templaterendered connectself ontemplaterender return self def exitself exctype excvalue traceback templaterendered disconnectself ontemplaterender if exctype is not none return self test class asserttemplatenotusedcontextasserttemplateusedcontext def testself renderedtemplatenames t name for t in self renderedtemplates if t name is not none self testcase assertfalse self templatename in renderedtemplatenames fself msgprefixtemplate self templatename was used funexpectedly in rendering the response class databaseoperationforbiddenassertionerror pass class databasefailure def initself wrapped message self wrapped wrapped self message message def callself raise databaseoperationforbiddenself message class simpletestcaseunittest testcase the class we ll use for the test client self client can be overridden in derived classes clientclass client asyncclientclass asyncclient overriddensettings none modifiedsettings none databases set disalloweddatabasemsg database operations to aliasr are not allowed in simpletestcase subclasses either subclass testcase or transactiontestcase to ensure proper test isolation or add aliasr to tests databases to silence this failure disallowedconnectionmethods connect connections temporaryconnection connections cursor queries chunkedcursor queries classmethod def setupclasscls super setupclass if cls overriddensettings cls clsoverriddencontext overridesettingscls overriddensettings cls clsoverriddencontext enable cls addclasscleanupcls clsoverriddencontext disable if cls modifiedsettings cls clsmodifiedcontext modifysettingscls modifiedsettings cls clsmodifiedcontext enable cls addclasscleanupcls clsmodifiedcontext disable cls adddatabasesfailures cls addclasscleanupcls removedatabasesfailures classmethod def validatedatabasescls if cls databases all return frozensetconnections for alias in cls databases if alias not in connections message s s databases refers to r which is not defined in settings databases cls module cls qualname alias closematches getclosematchesalias listconnections if closematches message did you mean r closematches0 raise improperlyconfiguredmessage return frozensetcls databases classmethod def adddatabasesfailurescls cls databases cls validatedatabases for alias in connections if alias in cls databases continue connection connectionsalias for name operation in cls disallowedconnectionmethods message cls disalloweddatabasemsg test s s cls module cls qualname alias alias operation operation method getattrconnection name setattrconnection name databasefailuremethod message classmethod def removedatabasesfailurescls for alias in connections if alias in cls databases continue connection connectionsalias for name in cls disallowedconnectionmethods method getattrconnection name setattrconnection name method wrapped def callself resultnone self setupandcallresult def debugself perform the following in order presetup run test postteardown skipping prepost hooks if test is set to be skipped if debugtrue reraise any errors in setup and use super debug instead of call to run the test convert async test methods perform pretest setup create a test client clear the mail test outbox perform posttest things pass def settingsself kwargs return overridesettingskwargs def modifysettingsself kwargs return modifysettingskwargs def assertredirects self response expectedurl statuscode302 targetstatuscode200 msgprefix fetchredirectresponsetrue if msgprefix msgprefix if hasattrresponse redirectchain the request was a followed redirect self asserttrue response redirectchain msgprefix response didn t redirect as expected response code was d expected d response statuscode statuscode self assertequal response redirectchain01 statuscode msgprefix initial response didn t redirect as expected response code was d expected d response redirectchain01 statuscode url statuscode response redirectchain1 self assertequal response statuscode targetstatuscode msgprefix response didn t redirect as expected final response code was d expected d response statuscode targetstatuscode else not a followed redirect self assertequal response statuscode statuscode msgprefix response didn t redirect as expected response code was d expected d response statuscode statuscode url response url scheme netloc path query fragment urlspliturl prepend the request path to handle relative path redirects if not path startswith url urljoinresponse requestpathinfo url path urljoinresponse requestpathinfo path if fetchredirectresponse netloc might be empty or in cases where django tests the http scheme the convention is for netloc to be testserver trust both as internal urls here domain port splitdomainportnetloc if domain and not validatehostdomain settings allowedhosts raise valueerror the test client is unable to fetch remote urls got s if the host is served by django add s to allowedhosts otherwise use assertredirects fetchredirectresponsefalse url domain get the redirection page using the same client that was used to obtain the original response extra response client extra or headers response client headers or redirectresponse response client get path querydictquery securescheme https headersheaders extra self assertequal redirectresponse statuscode targetstatuscode msgprefix couldn t retrieve redirection page s response code was d expected d path redirectresponse statuscode targetstatuscode self asserturlequal url expectedurl msgprefix response redirected to s expected s url expectedurl def asserturlequalself url1 url2 msgprefix def normalizeurl if the response supports deferred rendering and hasn t been rendered yet then ensure that it does get rendered before proceeding further assert that a response indicates that some content was retrieved successfully i e the http status code was as expected and that text occurs count times in the content of the response if count is none the count doesn t matter the assertion is true if the text occurs at least once in the response assert that a response indicates that some content was retrieved successfully i e the http status code was as expected and that text doesn t occur in the content of the response raise a valueerror if the given response doesn t have the required attribute assert that a field named field on the given form object has specific errors errors can be either a single error message or a list of errors messages using errors test that the field has no errors you can pass fieldnone to check the form s nonfield errors similar to assertformerror but for formsets use formindexnone to check the formset s nonform errors in that case you must also use fieldnone otherwise use an integer to check the formset s nth form for errors other parameters are the same as assertformerror use this template with context manager assert that the template with the provided name was used in rendering the response also usable as context manager use asserttemplateused as context manager assert that the template with the provided name was not used in rendering the response also usable as context manager use asserttemplatenotused as context manager assertion used in context manager fashion assertion was passed a callable assert that expectedmessage is found in the message of a raised exception args expectedexception exception class expected to be raised expectedmessage expected error message string value args function to be called and extra positional args kwargs extra kwargs same as assertraisesmessage but for assertwarns instead of assertraises assert that a form field behaves correctly with various inputs args fieldclass the class of the field to be tested valid a dictionary mapping valid inputs to their expected cleaned values invalid a dictionary mapping invalid inputs to one or more raised error messages fieldargs the args passed to instantiate the field fieldkwargs the kwargs passed to instantiate the field emptyvalue the expected clean output for inputs in emptyvalues test valid inputs test invalid inputs test required inputs test that maxlength and minlength are always accepted assert that two html snippets are semantically the same whitespace in most cases is ignored and attribute ordering is not significant the arguments must be valid html assert that two html snippets are not semantically equivalent dom1 assertandparsehtml self html1 msg first argument is not valid html dom2 assertandparsehtml self html2 msg second argument is not valid html if dom1 dom2 standardmsg s s safereprdom1 true safereprdom2 true self failself formatmessagemsg standardmsg def assertinhtmlself needle haystack countnone msgprefix parsedneedle assertandparsehtml self needle none first argument is not valid html parsedhaystack assertandparsehtml self haystack none second argument is not valid html realcount parsedhaystack countparsedneedle if msgprefix msgprefix haystackrepr safereprhaystack if count is not none self assertequal realcount count fmsgprefixfound realcount instances of needle r expected fcount in the following responsenhaystackrepr else self asserttrue realcount 0 fmsgprefixcouldn t find needle r in the following responsen fhaystackrepr def assertjsonequalself raw expecteddata msgnone try data json loadsraw except json jsondecodeerror self failfirst argument is not valid json r raw if isinstanceexpecteddata str try expecteddata json loadsexpecteddata except valueerror self failsecond argument is not valid json r expecteddata self assertequaldata expecteddata msgmsg def assertjsonnotequalself raw expecteddata msgnone try data json loadsraw except json jsondecodeerror self failfirst argument is not valid json r raw if isinstanceexpecteddata str try expecteddata json loadsexpecteddata except json jsondecodeerror self failsecond argument is not valid json r expecteddata self assertnotequaldata expecteddata msgmsg def assertxmlequalself xml1 xml2 msgnone try result comparexmlxml1 xml2 except exception as e standardmsg first or second argument is not valid xmlns e self failself formatmessagemsg standardmsg else if not result standardmsg s s safereprxml1 true safereprxml2 true diff n n join difflib ndiffxml1 splitlines xml2 splitlines standardmsg self truncatemessagestandardmsg diff self failself formatmessagemsg standardmsg def assertxmlnotequalself xml1 xml2 msgnone try result comparexmlxml1 xml2 except exception as e standardmsg first or second argument is not valid xmlns e self failself formatmessagemsg standardmsg else if result standardmsg s s safereprxml1 true safereprxml2 true self failself formatmessagemsg standardmsg class transactiontestcasesimpletestcase subclasses can ask for resetting of auto increment sequence before each test case resetsequences false subclasses can enable only a subset of apps for faster tests availableapps none subclasses can define fixtures which will be automatically installed fixtures none databases defaultdbalias disalloweddatabasemsg database operations to aliasr are not allowed in this test add aliasr to tests databases to ensure proper test isolation and silence this failure if transactions aren t available django will serialize the database contents into a fixture during setup and flush and reload them during teardown as flush does not restore data from migrations this can be slow this flag allows enabling on a percase basis serializedrollback false def presetupself super presetup if self availableapps is not none apps setavailableappsself availableapps settingchanged send sendersettings wrapped class settinginstalledapps valueself availableapps entertrue for dbname in self databasesnamesincludemirrorsfalse emitpostmigratesignalverbosity0 interactivefalse dbdbname try self fixturesetup except exception if self availableapps is not none apps unsetavailableapps settingchanged send sendersettings wrapped class settinginstalledapps valuesettings installedapps enterfalse raise clear the querieslog so that it s less likely to overflow a single test probably won t execute 9k queries if querieslog overflows then assertnumqueries doesn t work for dbname in self databasesnamesincludemirrorsfalse connectionsdbname querieslog clear classmethod def databasesnamescls includemirrorstrue only consider allowed database aliases including mirrors or not return alias for alias in connections if alias in cls databases and includemirrors or not connectionsalias settingsdicttestmirror def resetsequencesself dbname conn connectionsdbname if conn features supportssequencereset sqllist conn ops sequenceresetbynamesql nostyle conn introspection sequencelist if sqllist with transaction atomicusingdbname with conn cursor as cursor for sql in sqllist cursor executesql def fixturesetupself for dbname in self databasesnamesincludemirrorsfalse reset sequences if self resetsequences self resetsequencesdbname provide replica initial data from migrated apps if needed if self serializedrollback and hasattr connectionsdbname testserializedcontents if self availableapps is not none apps unsetavailableapps connectionsdbname creation deserializedbfromstring connectionsdbname testserializedcontents if self availableapps is not none apps setavailableappsself availableapps if self fixtures callcommandloaddata self fixtures verbosity0 databasedbname def shouldreloadconnectionsself return true def postteardownself try self fixtureteardown super postteardown if self shouldreloadconnections some db cursors include sql statements as part of cursor creation if you have a test that does a rollback the effect of these statements is lost which can affect the operation of tests e g losing a timezone setting causing objects to be created with the wrong time to make sure this doesn t happen get a clean connection at the start of every test for conn in connections allinitializedonlytrue conn close finally if self availableapps is not none apps unsetavailableapps settingchanged send sendersettings wrapped class settinginstalledapps valuesettings installedapps enterfalse def fixtureteardownself allow truncate cascade and don t emit the postmigrate signal when flushing only a subset of the apps for dbname in self databasesnamesincludemirrorsfalse flush the database inhibitpostmigrate self availableapps is not none or inhibit the postmigrate signal when using serialized rollback to avoid trying to recreate the serialized data self serializedrollback and hasattrconnectionsdbname testserializedcontents callcommand flush verbosity0 interactivefalse databasedbname resetsequencesfalse allowcascadeself availableapps is not none inhibitpostmigrateinhibitpostmigrate def assertquerysetequalself qs values transformnone orderedtrue msgnone values listvalues items qs if transform is not none items maptransform items if not ordered return self assertdictequalcounteritems countervalues msgmsg for example qs iterator could be passed as qs but it does not have ordered attribute if lenvalues 1 and hasattrqs ordered and not qs ordered raise valueerror trying to compare nonordered queryset against more than one ordered value return self assertequallistitems values msgmsg def assertnumqueriesself num funcnone args usingdefaultdbalias kwargs conn connectionsusing context assertnumqueriescontextself num conn if func is none return context with context funcargs kwargs def connectionssupporttransactionsaliasesnone conns connections all if aliases is none else connectionsalias for alias in aliases return allconn features supportstransactions for conn in conns class testdata memoattr testdatamemo def initself name data self name name self data data def getmemoself testcase try memo getattrtestcase self memoattr except attributeerror memo setattrtestcase self memoattr memo return memo def getself instance owner if instance is none return self data memo self getmemoinstance data deepcopyself data memo setattrinstance self name data return data def reprself return testdata namer datar self name self data class testcasetransactiontestcase classmethod def enteratomicscls rollback atomic blocks opened by the previous method for dbname in reversedcls databasesnames transaction setrollbacktrue usingdbname atomicsdbname exitnone none none classmethod def databasessupporttransactionscls return connectionssupporttransactionscls databases classmethod def setupclasscls super setupclass if not cls databasessupporttransactions return cls clsatomics cls enteratomics if cls fixtures for dbname in cls databasesnamesincludemirrorsfalse try callcommand loaddata cls fixtures verbosity0 databasedbname except exception cls rollbackatomicscls clsatomics raise preattrs cls dict copy try cls setuptestdata except exception cls rollbackatomicscls clsatomics raise for name value in cls dict items if value is not preattrs getname setattrcls name testdataname value classmethod def teardownclasscls if cls databasessupporttransactions cls rollbackatomicscls clsatomics for conn in connections allinitializedonlytrue conn close super teardownclass classmethod def setuptestdatacls if the backend does not support transactions we should reload class data before each test context manager to capture transaction oncommit callbacks callbacks startcount lenconnectionsusing runoncommit try yield callbacks finally while true callbackcount lenconnectionsusing runoncommit for callback robust in connectionsusing runoncommit startcount callbacks appendcallback if execute if robust try callback except exception as e logger error ferror calling callback qualname in foncommit s e excinfotrue else callback if callbackcount lenconnectionsusing runoncommit break startcount callbackcount class checkcondition trigger access for all bases override this descriptor s value and set the skip reason assume a class is decorated defer raising to allow importing test class s module retrieve the possibly existing value from the class s dict to avoid triggering the descriptor skip a test if a database has at least one of the named features return deferredskip lambda any getattrconnection features feature false for feature in features database has features s joinfeatures skipifdbfeature def skipunlessdbfeaturefeatures skip a test unless a database has any of the named features return deferredskip lambda not any getattrconnection features feature false for feature in features database doesn t support any of the features s joinfeatures skipunlessanydbfeature class quietwsgirequesthandlerwsgirequesthandler def logmessageargs pass class fsfileshandlerwsgihandler def initself application self application application self baseurl urlparseself getbaseurl super init def shouldhandleself path return path startswithself baseurl2 and not self baseurl1 def filepathself url emulate behavior of django contrib staticfiles views serve when it invokes staticfiles finders functionality todo modify ifwhen that internal api is refactored handler for serving static files a private class that is meant to be used solely as a convenience by liveserverthread handler for serving the media files a private class that is meant to be used solely as a convenience by liveserverthread thread for running a live http server while the tests are running serverclass threadedwsgiserver def initself host statichandler connectionsoverridenone port0 self host host self port port self isready threading event self error none self statichandler statichandler self connectionsoverride connectionsoverride super init def runself if self connectionsoverride override this thread s database connections with the ones provided by the main thread for alias conn in self connectionsoverride items connectionsalias conn try create the handler for serving static and media files handler self statichandlermediafileshandlerwsgihandler self httpd self createserver connectionsoverrideself connectionsoverride if binding to port zero assign the port allocated by the os if self port 0 self port self httpd serveraddress1 self httpd setapphandler self isready set self httpd serveforever except exception as e self error e self isready set finally connections closeall def createserverself connectionsoverridenone return self serverclass self host self port quietwsgirequesthandler allowreuseaddressfalse connectionsoverrideconnectionsoverride def terminateself if hasattrself httpd stop the wsgi server self httpd shutdown self httpd serverclose self join class liveservertestcasetransactiontestcase host localhost port 0 serverthreadclass liveserverthread statichandler staticfileshandler classproperty def liveserverurlcls return http s s cls host cls serverthread port classproperty def allowedhostcls return cls host classmethod def makeconnectionsoverridecls connectionsoverride for conn in connections all if using inmemory sqlite databases pass the connections to the server thread if conn vendor sqlite and conn isinmemorydb connectionsoverrideconn alias conn return connectionsoverride classmethod def setupclasscls super setupclass cls liveservermodifiedsettings modifysettings allowedhostsappend cls allowedhost cls liveservermodifiedsettings enable cls addclasscleanupcls liveservermodifiedsettings disable cls startserverthread classmethod def startserverthreadcls connectionsoverride cls makeconnectionsoverride for conn in connectionsoverride values explicitly enable threadshareability for this connection conn incthreadsharing cls serverthread cls createserverthreadconnectionsoverride cls serverthread daemon true cls serverthread start cls addclasscleanupcls terminatethread wait for the live server to be ready cls serverthread isready wait if cls serverthread error raise cls serverthread error classmethod def createserverthreadcls connectionsoverride return cls serverthreadclass cls host cls statichandler connectionsoverrideconnectionsoverride portcls port classmethod def terminatethreadcls terminate the live server s thread cls serverthread terminate restore shared connections nonshareability for conn in cls serverthread connectionsoverride values conn decthreadsharing class serializemixin lockfile none def initsubclasscls kwargs super initsubclasskwargs if cls lockfile is none raise valueerror lockfile isn t set set it to a unique value in the base class formatcls name classmethod def setupclasscls cls lockfile opencls lockfile cls addclasscleanupcls lockfile close locks lockcls lockfile locks lockex super setupclass put value into a list if it s not already one the class we ll use for the test client self client can be overridden in derived classes wrapper around default __call__ method to perform common django test set up this means that user defined testcases aren t required to include a call to super setup perform the same as __call__ without catching the exception perform the following in order pre setup run test post teardown skipping pre post hooks if test is set to be skipped if debug true reraise any errors in setup and use super debug instead of __call__ to run the test convert async test methods perform pre test setup create a test client clear the mail test outbox perform post test things a context manager that temporarily sets a setting and reverts to the original value when exiting the context a context manager that temporarily applies changes a list setting and reverts back to the original value when exiting the context assert that a response redirected to a specific url and that the redirect url can be loaded won t work for external links since it uses the test client to do a request use fetch_redirect_response false to check such links without fetching them the request was a followed redirect not a followed redirect prepend the request path to handle relative path redirects netloc might be empty or in cases where django tests the http scheme the convention is for netloc to be testserver trust both as internal urls here get the redirection page using the same client that was used to obtain the original response assert that two urls are the same ignoring the order of query string parameters except for parameters with the same name for example path x 1 y 2 is equal to path y 2 x 1 but path a 1 a 2 isn t equal to path a 2 a 1 sort the url s query string parameters coerce reverse_lazy urls if the response supports deferred rendering and hasn t been rendered yet then ensure that it does get rendered before proceeding further assert that a response indicates that some content was retrieved successfully i e the http status code was as expected and that text occurs count times in the content of the response if count is none the count doesn t matter the assertion is true if the text occurs at least once in the response assert that a response indicates that some content was retrieved successfully i e the http status code was as expected and that text doesn t occur in the content of the response raise a valueerror if the given response doesn t have the required attribute assert that a field named field on the given form object has specific errors errors can be either a single error message or a list of errors messages using errors test that the field has no errors you can pass field none to check the form s non field errors similar to assertformerror but for formsets use form_index none to check the formset s non form errors in that case you must also use field none otherwise use an integer to check the formset s n th form for errors other parameters are the same as assertformerror use this template with context manager assert that the template with the provided name was used in rendering the response also usable as context manager use asserttemplateused as context manager assert that the template with the provided name was not used in rendering the response also usable as context manager use asserttemplatenotused as context manager assertion used in context manager fashion assertion was passed a callable assert that expected_message is found in the message of a raised exception args expected_exception exception class expected to be raised expected_message expected error message string value args function to be called and extra positional args kwargs extra kwargs same as assertraisesmessage but for assertwarns instead of assertraises assert that a form field behaves correctly with various inputs args fieldclass the class of the field to be tested valid a dictionary mapping valid inputs to their expected cleaned values invalid a dictionary mapping invalid inputs to one or more raised error messages field_args the args passed to instantiate the field field_kwargs the kwargs passed to instantiate the field empty_value the expected clean output for inputs in empty_values test valid inputs test invalid inputs test required inputs test that max_length and min_length are always accepted assert that two html snippets are semantically the same whitespace in most cases is ignored and attribute ordering is not significant the arguments must be valid html assert that two html snippets are not semantically equivalent assert that the json fragments raw and expected_data are equal usual json non significant whitespace rules apply as the heavyweight is delegated to the json library assert that the json fragments raw and expected_data are not equal usual json non significant whitespace rules apply as the heavyweight is delegated to the json library assert that two xml snippets are semantically the same whitespace in most cases is ignored and attribute ordering is not significant the arguments must be valid xml assert that two xml snippets are not semantically equivalent whitespace in most cases is ignored and attribute ordering is not significant the arguments must be valid xml subclasses can ask for resetting of auto increment sequence before each test case subclasses can enable only a subset of apps for faster tests subclasses can define fixtures which will be automatically installed if transactions aren t available django will serialize the database contents into a fixture during setup and flush and reload them during teardown as flush does not restore data from migrations this can be slow this flag allows enabling on a per case basis perform pre test setup if the class has an available_apps attribute restrict the app registry to these applications then fire the post_migrate signal it must run with the correct set of applications for the test case if the class has a fixtures attribute install those fixtures clear the queries_log so that it s less likely to overflow a single test probably won t execute 9k queries if queries_log overflows then assertnumqueries doesn t work only consider allowed database aliases including mirrors or not reset sequences provide replica initial data from migrated apps if needed perform post test things flush the contents of the database to leave a clean slate if the class has an available_apps attribute don t fire post_migrate force close the connection so the next test gets a clean cursor some db cursors include sql statements as part of cursor creation if you have a test that does a rollback the effect of these statements is lost which can affect the operation of tests e g losing a timezone setting causing objects to be created with the wrong time to make sure this doesn t happen get a clean connection at the start of every test allow truncate cascade and don t emit the post_migrate signal when flushing only a subset of the apps flush the database inhibit the post_migrate signal when using serialized rollback to avoid trying to recreate the serialized data for example qs iterator could be passed as qs but it does not have ordered attribute return whether or not all or specified connections support transactions descriptor to provide testcase instance isolation for attributes assigned during the setuptestdata phase allow safe alteration of objects assigned in setuptestdata by test methods by exposing deep copies instead of the original objects objects are deep copied using a memo kept on the test case instance in order to maintain their original relationships similar to transactiontestcase but use transaction atomic to achieve test isolation in most situations testcase should be preferred to transactiontestcase as it allows faster execution however there are some situations where using transactiontestcase might be necessary e g testing some transactional behavior on database backends with no transaction support testcase behaves as transactiontestcase open atomic blocks for multiple databases rollback atomic blocks opened by the previous method load initial data for the testcase if the backend does not support transactions we should reload class data before each test context manager to capture transaction on_commit callbacks descriptor class for deferred condition checking trigger access for all bases override this descriptor s value and set the skip reason assume a class is decorated defer raising to allow importing test class s module retrieve the possibly existing value from the class s dict to avoid triggering the descriptor skip a test if a database has at least one of the named features skip a test unless a database has all the named features skip a test unless a database has any of the named features a wsgirequesthandler that doesn t log to standard output any of the requests received so as to not clutter the test result output wsgi middleware that intercepts calls to a directory as defined by one of the _root settings and serves those files publishing them under _url check if the path should be handled ignore the path if the host is provided as part of the base_url the request s path isn t under the media path or equal return the relative path to the file on disk for the given url emulate behavior of django contrib staticfiles views serve when it invokes staticfiles finders functionality todo modify if when that internal api is refactored handler for serving static files a private class that is meant to be used solely as a convenience by liveserverthread handler for serving the media files a private class that is meant to be used solely as a convenience by liveserverthread thread for running a live http server while the tests are running set up the live server and databases and then loop over handling http requests override this thread s database connections with the ones provided by the main thread create the handler for serving static and media files if binding to port zero assign the port allocated by the os stop the wsgi server do basically the same as transactiontestcase but also launch a live http server in a separate thread so that the tests may use another testing framework such as selenium for example instead of the built in dummy client it inherits from transactiontestcase instead of testcase because the threads don t share the same transactions unless if using in memory sqlite and each thread needs to commit all their transactions so that the other thread can see the changes if using in memory sqlite databases pass the connections to the server thread explicitly enable thread shareability for this connection wait for the live server to be ready terminate the live server s thread restore shared connections non shareability enforce serialization of testcases that share a common resource define a common lockfile for each set of testcases to serialize this file must exist on the filesystem place it early in the mro in order to isolate setupclass teardownclass
import difflib
import json
import logging
import posixpath
import sys
import threading
import unittest
from collections import Counter
from contextlib import contextmanager
from copy import copy, deepcopy
from difflib import get_close_matches
from functools import wraps
from unittest.suite import _DebugResult
from unittest.util import safe_repr
from urllib.parse import (
    parse_qsl,
    unquote,
    urlencode,
    urljoin,
    urlparse,
    urlsplit,
    urlunparse,
)
from urllib.request import url2pathname

from asgiref.sync import async_to_sync, iscoroutinefunction

from django.apps import apps
from django.conf import settings
from django.core import mail
from django.core.exceptions import ImproperlyConfigured, ValidationError
from django.core.files import locks
from django.core.handlers.wsgi import WSGIHandler, get_path_info
from django.core.management import call_command
from django.core.management.color import no_style
from django.core.management.sql import emit_post_migrate_signal
from django.core.servers.basehttp import ThreadedWSGIServer, WSGIRequestHandler
from django.core.signals import setting_changed
from django.db import DEFAULT_DB_ALIAS, connection, connections, transaction
from django.forms.fields import CharField
from django.http import QueryDict
from django.http.request import split_domain_port, validate_host
from django.test.client import AsyncClient, Client
from django.test.html import HTMLParseError, parse_html
from django.test.signals import template_rendered
from django.test.utils import (
    CaptureQueriesContext,
    ContextList,
    compare_xml,
    modify_settings,
    override_settings,
)
from django.utils.functional import classproperty
from django.views.static import serve

logger = logging.getLogger("django.test")

__all__ = (
    "TestCase",
    "TransactionTestCase",
    "SimpleTestCase",
    "skipIfDBFeature",
    "skipUnlessDBFeature",
)


def to_list(value):
    
    if not isinstance(value, list):
        value = [value]
    return value


def assert_and_parse_html(self, html, user_msg, msg):
    try:
        dom = parse_html(html)
    except HTMLParseError as e:
        standardMsg = "%s\n%s" % (msg, e)
        self.fail(self._formatMessage(user_msg, standardMsg))
    return dom


class _AssertNumQueriesContext(CaptureQueriesContext):
    def __init__(self, test_case, num, connection):
        self.test_case = test_case
        self.num = num
        super().__init__(connection)

    def __exit__(self, exc_type, exc_value, traceback):
        super().__exit__(exc_type, exc_value, traceback)
        if exc_type is not None:
            return
        executed = len(self)
        self.test_case.assertEqual(
            executed,
            self.num,
            "%d queries executed, %d expected\nCaptured queries were:\n%s"
            % (
                executed,
                self.num,
                "\n".join(
                    "%d. %s" % (i, query["sql"])
                    for i, query in enumerate(self.captured_queries, start=1)
                ),
            ),
        )


class _AssertTemplateUsedContext:
    def __init__(self, test_case, template_name, msg_prefix="", count=None):
        self.test_case = test_case
        self.template_name = template_name
        self.msg_prefix = msg_prefix
        self.count = count

        self.rendered_templates = []
        self.context = ContextList()

    def on_template_render(self, sender, signal, template, context, **kwargs):
        self.rendered_templates.append(template)
        self.context.append(copy(context))

    def test(self):
        self.test_case._assert_template_used(
            self.template_name,
            [t.name for t in self.rendered_templates if t.name is not None],
            self.msg_prefix,
            self.count,
        )

    def __enter__(self):
        template_rendered.connect(self.on_template_render)
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        template_rendered.disconnect(self.on_template_render)
        if exc_type is not None:
            return
        self.test()


class _AssertTemplateNotUsedContext(_AssertTemplateUsedContext):
    def test(self):
        rendered_template_names = [
            t.name for t in self.rendered_templates if t.name is not None
        ]
        self.test_case.assertFalse(
            self.template_name in rendered_template_names,
            f"{self.msg_prefix}Template '{self.template_name}' was used "
            f"unexpectedly in rendering the response",
        )


class DatabaseOperationForbidden(AssertionError):
    pass


class _DatabaseFailure:
    def __init__(self, wrapped, message):
        self.wrapped = wrapped
        self.message = message

    def __call__(self):
        raise DatabaseOperationForbidden(self.message)


class SimpleTestCase(unittest.TestCase):
    
    
    client_class = Client
    async_client_class = AsyncClient
    _overridden_settings = None
    _modified_settings = None

    databases = set()
    _disallowed_database_msg = (
        "Database %(operation)s to %(alias)r are not allowed in SimpleTestCase "
        "subclasses. Either subclass TestCase or TransactionTestCase to ensure "
        "proper test isolation or add %(alias)r to %(test)s.databases to silence "
        "this failure."
    )
    _disallowed_connection_methods = [
        ("connect", "connections"),
        ("temporary_connection", "connections"),
        ("cursor", "queries"),
        ("chunked_cursor", "queries"),
    ]

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        if cls._overridden_settings:
            cls._cls_overridden_context = override_settings(**cls._overridden_settings)
            cls._cls_overridden_context.enable()
            cls.addClassCleanup(cls._cls_overridden_context.disable)
        if cls._modified_settings:
            cls._cls_modified_context = modify_settings(cls._modified_settings)
            cls._cls_modified_context.enable()
            cls.addClassCleanup(cls._cls_modified_context.disable)
        cls._add_databases_failures()
        cls.addClassCleanup(cls._remove_databases_failures)

    @classmethod
    def _validate_databases(cls):
        if cls.databases == "__all__":
            return frozenset(connections)
        for alias in cls.databases:
            if alias not in connections:
                message = (
                    "%s.%s.databases refers to %r which is not defined in "
                    "settings.DATABASES."
                    % (
                        cls.__module__,
                        cls.__qualname__,
                        alias,
                    )
                )
                close_matches = get_close_matches(alias, list(connections))
                if close_matches:
                    message += " Did you mean %r?" % close_matches[0]
                raise ImproperlyConfigured(message)
        return frozenset(cls.databases)

    @classmethod
    def _add_databases_failures(cls):
        cls.databases = cls._validate_databases()
        for alias in connections:
            if alias in cls.databases:
                continue
            connection = connections[alias]
            for name, operation in cls._disallowed_connection_methods:
                message = cls._disallowed_database_msg % {
                    "test": "%s.%s" % (cls.__module__, cls.__qualname__),
                    "alias": alias,
                    "operation": operation,
                }
                method = getattr(connection, name)
                setattr(connection, name, _DatabaseFailure(method, message))

    @classmethod
    def _remove_databases_failures(cls):
        for alias in connections:
            if alias in cls.databases:
                continue
            connection = connections[alias]
            for name, _ in cls._disallowed_connection_methods:
                method = getattr(connection, name)
                setattr(connection, name, method.wrapped)

    def __call__(self, result=None):
        
        self._setup_and_call(result)

    def debug(self):
        
        debug_result = _DebugResult()
        self._setup_and_call(debug_result, debug=True)

    def _setup_and_call(self, result, debug=False):
        
        testMethod = getattr(self, self._testMethodName)
        skipped = getattr(self.__class__, "__unittest_skip__", False) or getattr(
            testMethod, "__unittest_skip__", False
        )

        
        if iscoroutinefunction(testMethod):
            setattr(self, self._testMethodName, async_to_sync(testMethod))

        if not skipped:
            try:
                self._pre_setup()
            except Exception:
                if debug:
                    raise
                result.addError(self, sys.exc_info())
                return
        if debug:
            super().debug()
        else:
            super().__call__(result)
        if not skipped:
            try:
                self._post_teardown()
            except Exception:
                if debug:
                    raise
                result.addError(self, sys.exc_info())
                return

    def _pre_setup(self):
        
        self.client = self.client_class()
        self.async_client = self.async_client_class()
        mail.outbox = []

    def _post_teardown(self):
        
        pass

    def settings(self, **kwargs):
        
        return override_settings(**kwargs)

    def modify_settings(self, **kwargs):
        
        return modify_settings(**kwargs)

    def assertRedirects(
        self,
        response,
        expected_url,
        status_code=302,
        target_status_code=200,
        msg_prefix="",
        fetch_redirect_response=True,
    ):
        
        if msg_prefix:
            msg_prefix += ": "

        if hasattr(response, "redirect_chain"):
            
            self.assertTrue(
                response.redirect_chain,
                msg_prefix
                + (
                    "Response didn't redirect as expected: Response code was %d "
                    "(expected %d)"
                )
                % (response.status_code, status_code),
            )

            self.assertEqual(
                response.redirect_chain[0][1],
                status_code,
                msg_prefix
                + (
                    "Initial response didn't redirect as expected: Response code was "
                    "%d (expected %d)"
                )
                % (response.redirect_chain[0][1], status_code),
            )

            url, status_code = response.redirect_chain[-1]

            self.assertEqual(
                response.status_code,
                target_status_code,
                msg_prefix
                + (
                    "Response didn't redirect as expected: Final Response code was %d "
                    "(expected %d)"
                )
                % (response.status_code, target_status_code),
            )

        else:
            
            self.assertEqual(
                response.status_code,
                status_code,
                msg_prefix
                + (
                    "Response didn't redirect as expected: Response code was %d "
                    "(expected %d)"
                )
                % (response.status_code, status_code),
            )

            url = response.url
            scheme, netloc, path, query, fragment = urlsplit(url)

            
            if not path.startswith("/"):
                url = urljoin(response.request["PATH_INFO"], url)
                path = urljoin(response.request["PATH_INFO"], path)

            if fetch_redirect_response:
                
                
                
                domain, port = split_domain_port(netloc)
                if domain and not validate_host(domain, settings.ALLOWED_HOSTS):
                    raise ValueError(
                        "The test client is unable to fetch remote URLs (got %s). "
                        "If the host is served by Django, add '%s' to ALLOWED_HOSTS. "
                        "Otherwise, use "
                        "assertRedirects(..., fetch_redirect_response=False)."
                        % (url, domain)
                    )
                
                
                extra = response.client.extra or {}
                headers = response.client.headers or {}
                redirect_response = response.client.get(
                    path,
                    QueryDict(query),
                    secure=(scheme == "https"),
                    headers=headers,
                    **extra,
                )
                self.assertEqual(
                    redirect_response.status_code,
                    target_status_code,
                    msg_prefix
                    + (
                        "Couldn't retrieve redirection page '%s': response code was %d "
                        "(expected %d)"
                    )
                    % (path, redirect_response.status_code, target_status_code),
                )

        self.assertURLEqual(
            url,
            expected_url,
            msg_prefix
            + "Response redirected to '%s', expected '%s'" % (url, expected_url),
        )

    def assertURLEqual(self, url1, url2, msg_prefix=""):
        

        def normalize(url):
            
            url = str(url)  
            scheme, netloc, path, params, query, fragment = urlparse(url)
            query_parts = sorted(parse_qsl(query))
            return urlunparse(
                (scheme, netloc, path, params, urlencode(query_parts), fragment)
            )

        if msg_prefix:
            msg_prefix += ": "
        self.assertEqual(
            normalize(url1),
            normalize(url2),
            msg_prefix + "Expected '%s' to equal '%s'." % (url1, url2),
        )

    def _assert_contains(self, response, text, status_code, msg_prefix, html):
        
        
        if (
            hasattr(response, "render")
            and callable(response.render)
            and not response.is_rendered
        ):
            response.render()

        if msg_prefix:
            msg_prefix += ": "

        self.assertEqual(
            response.status_code,
            status_code,
            msg_prefix + "Couldn't retrieve content: Response code was %d"
            " (expected %d)" % (response.status_code, status_code),
        )

        if response.streaming:
            content = b"".join(response.streaming_content)
        else:
            content = response.content
        content_repr = safe_repr(content)
        if not isinstance(text, bytes) or html:
            text = str(text)
            content = content.decode(response.charset)
            text_repr = "'%s'" % text
        else:
            text_repr = repr(text)
        if html:
            content = assert_and_parse_html(
                self, content, None, "Response's content is not valid HTML:"
            )
            text = assert_and_parse_html(
                self, text, None, "Second argument is not valid HTML:"
            )
        real_count = content.count(text)
        return text_repr, real_count, msg_prefix, content_repr

    def assertContains(
        self, response, text, count=None, status_code=200, msg_prefix="", html=False
    ):
        
        text_repr, real_count, msg_prefix, content_repr = self._assert_contains(
            response, text, status_code, msg_prefix, html
        )

        if count is not None:
            self.assertEqual(
                real_count,
                count,
                (
                    f"{msg_prefix}Found {real_count} instances of {text_repr} "
                    f"(expected {count}) in the following response\n{content_repr}"
                ),
            )
        else:
            self.assertTrue(
                real_count != 0,
                (
                    f"{msg_prefix}Couldn't find {text_repr} in the following response\n"
                    f"{content_repr}"
                ),
            )

    def assertNotContains(
        self, response, text, status_code=200, msg_prefix="", html=False
    ):
        
        text_repr, real_count, msg_prefix, content_repr = self._assert_contains(
            response, text, status_code, msg_prefix, html
        )

        self.assertEqual(
            real_count,
            0,
            (
                f"{msg_prefix}{text_repr} unexpectedly found in the following response"
                f"\n{content_repr}"
            ),
        )

    def _check_test_client_response(self, response, attribute, method_name):
        
        if not hasattr(response, attribute):
            raise ValueError(
                f"{method_name}() is only usable on responses fetched using "
                "the Django test Client."
            )

    def _assert_form_error(self, form, field, errors, msg_prefix, form_repr):
        if not form.is_bound:
            self.fail(
                f"{msg_prefix}The {form_repr} is not bound, it will never have any "
                f"errors."
            )

        if field is not None and field not in form.fields:
            self.fail(
                f"{msg_prefix}The {form_repr} does not contain the field {field!r}."
            )
        if field is None:
            field_errors = form.non_field_errors()
            failure_message = f"The non-field errors of {form_repr} don't match."
        else:
            field_errors = form.errors.get(field, [])
            failure_message = (
                f"The errors of field {field!r} on {form_repr} don't match."
            )

        self.assertEqual(field_errors, errors, msg_prefix + failure_message)

    def assertFormError(self, form, field, errors, msg_prefix=""):
        
        if msg_prefix:
            msg_prefix += ": "
        errors = to_list(errors)
        self._assert_form_error(form, field, errors, msg_prefix, f"form {form!r}")

    def assertFormSetError(self, formset, form_index, field, errors, msg_prefix=""):
        
        if form_index is None and field is not None:
            raise ValueError("You must use field=None with form_index=None.")

        if msg_prefix:
            msg_prefix += ": "
        errors = to_list(errors)

        if not formset.is_bound:
            self.fail(
                f"{msg_prefix}The formset {formset!r} is not bound, it will never have "
                f"any errors."
            )
        if form_index is not None and form_index >= formset.total_form_count():
            form_count = formset.total_form_count()
            form_or_forms = "forms" if form_count > 1 else "form"
            self.fail(
                f"{msg_prefix}The formset {formset!r} only has {form_count} "
                f"{form_or_forms}."
            )
        if form_index is not None:
            form_repr = f"form {form_index} of formset {formset!r}"
            self._assert_form_error(
                formset.forms[form_index], field, errors, msg_prefix, form_repr
            )
        else:
            failure_message = f"The non-form errors of formset {formset!r} don't match."
            self.assertEqual(
                formset.non_form_errors(), errors, msg_prefix + failure_message
            )

    def _get_template_used(self, response, template_name, msg_prefix, method_name):
        if response is None and template_name is None:
            raise TypeError("response and/or template_name argument must be provided")

        if msg_prefix:
            msg_prefix += ": "

        if template_name is not None and response is not None:
            self._check_test_client_response(response, "templates", method_name)

        if not hasattr(response, "templates") or (response is None and template_name):
            if response:
                template_name = response
                response = None
            
            return template_name, None, msg_prefix

        template_names = [t.name for t in response.templates if t.name is not None]
        return None, template_names, msg_prefix

    def _assert_template_used(self, template_name, template_names, msg_prefix, count):
        if not template_names:
            self.fail(msg_prefix + "No templates used to render the response")
        self.assertTrue(
            template_name in template_names,
            msg_prefix + "Template '%s' was not a template used to render"
            " the response. Actual template(s) used: %s"
            % (template_name, ", ".join(template_names)),
        )

        if count is not None:
            self.assertEqual(
                template_names.count(template_name),
                count,
                msg_prefix + "Template '%s' was expected to be rendered %d "
                "time(s) but was actually rendered %d time(s)."
                % (template_name, count, template_names.count(template_name)),
            )

    def assertTemplateUsed(
        self, response=None, template_name=None, msg_prefix="", count=None
    ):
        
        context_mgr_template, template_names, msg_prefix = self._get_template_used(
            response,
            template_name,
            msg_prefix,
            "assertTemplateUsed",
        )
        if context_mgr_template:
            
            return _AssertTemplateUsedContext(
                self, context_mgr_template, msg_prefix, count
            )

        self._assert_template_used(template_name, template_names, msg_prefix, count)

    def assertTemplateNotUsed(self, response=None, template_name=None, msg_prefix=""):
        
        context_mgr_template, template_names, msg_prefix = self._get_template_used(
            response,
            template_name,
            msg_prefix,
            "assertTemplateNotUsed",
        )
        if context_mgr_template:
            
            return _AssertTemplateNotUsedContext(self, context_mgr_template, msg_prefix)

        self.assertFalse(
            template_name in template_names,
            msg_prefix
            + "Template '%s' was used unexpectedly in rendering the response"
            % template_name,
        )

    @contextmanager
    def _assert_raises_or_warns_cm(
        self, func, cm_attr, expected_exception, expected_message
    ):
        with func(expected_exception) as cm:
            yield cm
        self.assertIn(expected_message, str(getattr(cm, cm_attr)))

    def _assertFooMessage(
        self, func, cm_attr, expected_exception, expected_message, *args, **kwargs
    ):
        callable_obj = None
        if args:
            callable_obj, *args = args
        cm = self._assert_raises_or_warns_cm(
            func, cm_attr, expected_exception, expected_message
        )
        
        if callable_obj is None:
            return cm
        
        with cm:
            callable_obj(*args, **kwargs)

    def assertRaisesMessage(
        self, expected_exception, expected_message, *args, **kwargs
    ):
        
        return self._assertFooMessage(
            self.assertRaises,
            "exception",
            expected_exception,
            expected_message,
            *args,
            **kwargs,
        )

    def assertWarnsMessage(self, expected_warning, expected_message, *args, **kwargs):
        
        return self._assertFooMessage(
            self.assertWarns,
            "warning",
            expected_warning,
            expected_message,
            *args,
            **kwargs,
        )

    def assertFieldOutput(
        self,
        fieldclass,
        valid,
        invalid,
        field_args=None,
        field_kwargs=None,
        empty_value="",
    ):
        
        if field_args is None:
            field_args = []
        if field_kwargs is None:
            field_kwargs = {}
        required = fieldclass(*field_args, **field_kwargs)
        optional = fieldclass(*field_args, **{**field_kwargs, "required": False})
        
        for input, output in valid.items():
            self.assertEqual(required.clean(input), output)
            self.assertEqual(optional.clean(input), output)
        
        for input, errors in invalid.items():
            with self.assertRaises(ValidationError) as context_manager:
                required.clean(input)
            self.assertEqual(context_manager.exception.messages, errors)

            with self.assertRaises(ValidationError) as context_manager:
                optional.clean(input)
            self.assertEqual(context_manager.exception.messages, errors)
        
        error_required = [required.error_messages["required"]]
        for e in required.empty_values:
            with self.assertRaises(ValidationError) as context_manager:
                required.clean(e)
            self.assertEqual(context_manager.exception.messages, error_required)
            self.assertEqual(optional.clean(e), empty_value)
        
        if issubclass(fieldclass, CharField):
            field_kwargs.update({"min_length": 2, "max_length": 20})
            self.assertIsInstance(fieldclass(*field_args, **field_kwargs), fieldclass)

    def assertHTMLEqual(self, html1, html2, msg=None):
        
        dom1 = assert_and_parse_html(
            self, html1, msg, "First argument is not valid HTML:"
        )
        dom2 = assert_and_parse_html(
            self, html2, msg, "Second argument is not valid HTML:"
        )

        if dom1 != dom2:
            standardMsg = "%s != %s" % (safe_repr(dom1, True), safe_repr(dom2, True))
            diff = "\n" + "\n".join(
                difflib.ndiff(
                    str(dom1).splitlines(),
                    str(dom2).splitlines(),
                )
            )
            standardMsg = self._truncateMessage(standardMsg, diff)
            self.fail(self._formatMessage(msg, standardMsg))

    def assertHTMLNotEqual(self, html1, html2, msg=None):
        
        dom1 = assert_and_parse_html(
            self, html1, msg, "First argument is not valid HTML:"
        )
        dom2 = assert_and_parse_html(
            self, html2, msg, "Second argument is not valid HTML:"
        )

        if dom1 == dom2:
            standardMsg = "%s == %s" % (safe_repr(dom1, True), safe_repr(dom2, True))
            self.fail(self._formatMessage(msg, standardMsg))

    def assertInHTML(self, needle, haystack, count=None, msg_prefix=""):
        parsed_needle = assert_and_parse_html(
            self, needle, None, "First argument is not valid HTML:"
        )
        parsed_haystack = assert_and_parse_html(
            self, haystack, None, "Second argument is not valid HTML:"
        )
        real_count = parsed_haystack.count(parsed_needle)
        if msg_prefix:
            msg_prefix += ": "
        haystack_repr = safe_repr(haystack)
        if count is not None:
            self.assertEqual(
                real_count,
                count,
                (
                    f"{msg_prefix}Found {real_count} instances of {needle!r} (expected "
                    f"{count}) in the following response\n{haystack_repr}"
                ),
            )
        else:
            self.assertTrue(
                real_count != 0,
                (
                    f"{msg_prefix}Couldn't find {needle!r} in the following response\n"
                    f"{haystack_repr}"
                ),
            )

    def assertJSONEqual(self, raw, expected_data, msg=None):
        
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            self.fail("First argument is not valid JSON: %r" % raw)
        if isinstance(expected_data, str):
            try:
                expected_data = json.loads(expected_data)
            except ValueError:
                self.fail("Second argument is not valid JSON: %r" % expected_data)
        self.assertEqual(data, expected_data, msg=msg)

    def assertJSONNotEqual(self, raw, expected_data, msg=None):
        
        try:
            data = json.loads(raw)
        except json.JSONDecodeError:
            self.fail("First argument is not valid JSON: %r" % raw)
        if isinstance(expected_data, str):
            try:
                expected_data = json.loads(expected_data)
            except json.JSONDecodeError:
                self.fail("Second argument is not valid JSON: %r" % expected_data)
        self.assertNotEqual(data, expected_data, msg=msg)

    def assertXMLEqual(self, xml1, xml2, msg=None):
        
        try:
            result = compare_xml(xml1, xml2)
        except Exception as e:
            standardMsg = "First or second argument is not valid XML\n%s" % e
            self.fail(self._formatMessage(msg, standardMsg))
        else:
            if not result:
                standardMsg = "%s != %s" % (
                    safe_repr(xml1, True),
                    safe_repr(xml2, True),
                )
                diff = "\n" + "\n".join(
                    difflib.ndiff(xml1.splitlines(), xml2.splitlines())
                )
                standardMsg = self._truncateMessage(standardMsg, diff)
                self.fail(self._formatMessage(msg, standardMsg))

    def assertXMLNotEqual(self, xml1, xml2, msg=None):
        
        try:
            result = compare_xml(xml1, xml2)
        except Exception as e:
            standardMsg = "First or second argument is not valid XML\n%s" % e
            self.fail(self._formatMessage(msg, standardMsg))
        else:
            if result:
                standardMsg = "%s == %s" % (
                    safe_repr(xml1, True),
                    safe_repr(xml2, True),
                )
                self.fail(self._formatMessage(msg, standardMsg))


class TransactionTestCase(SimpleTestCase):
    
    
    reset_sequences = False

    
    available_apps = None

    
    fixtures = None

    databases = {DEFAULT_DB_ALIAS}
    _disallowed_database_msg = (
        "Database %(operation)s to %(alias)r are not allowed in this test. "
        "Add %(alias)r to %(test)s.databases to ensure proper test isolation "
        "and silence this failure."
    )

    
    
    
    
    serialized_rollback = False

    def _pre_setup(self):
        
        super()._pre_setup()
        if self.available_apps is not None:
            apps.set_available_apps(self.available_apps)
            setting_changed.send(
                sender=settings._wrapped.__class__,
                setting="INSTALLED_APPS",
                value=self.available_apps,
                enter=True,
            )
            for db_name in self._databases_names(include_mirrors=False):
                emit_post_migrate_signal(verbosity=0, interactive=False, db=db_name)
        try:
            self._fixture_setup()
        except Exception:
            if self.available_apps is not None:
                apps.unset_available_apps()
                setting_changed.send(
                    sender=settings._wrapped.__class__,
                    setting="INSTALLED_APPS",
                    value=settings.INSTALLED_APPS,
                    enter=False,
                )
            raise
        
        
        
        for db_name in self._databases_names(include_mirrors=False):
            connections[db_name].queries_log.clear()

    @classmethod
    def _databases_names(cls, include_mirrors=True):
        
        return [
            alias
            for alias in connections
            if alias in cls.databases
            and (
                include_mirrors
                or not connections[alias].settings_dict["TEST"]["MIRROR"]
            )
        ]

    def _reset_sequences(self, db_name):
        conn = connections[db_name]
        if conn.features.supports_sequence_reset:
            sql_list = conn.ops.sequence_reset_by_name_sql(
                no_style(), conn.introspection.sequence_list()
            )
            if sql_list:
                with transaction.atomic(using=db_name):
                    with conn.cursor() as cursor:
                        for sql in sql_list:
                            cursor.execute(sql)

    def _fixture_setup(self):
        for db_name in self._databases_names(include_mirrors=False):
            
            if self.reset_sequences:
                self._reset_sequences(db_name)

            
            if self.serialized_rollback and hasattr(
                connections[db_name], "_test_serialized_contents"
            ):
                if self.available_apps is not None:
                    apps.unset_available_apps()
                connections[db_name].creation.deserialize_db_from_string(
                    connections[db_name]._test_serialized_contents
                )
                if self.available_apps is not None:
                    apps.set_available_apps(self.available_apps)

            if self.fixtures:
                call_command("loaddata", *self.fixtures, verbosity=0, database=db_name)

    def _should_reload_connections(self):
        return True

    def _post_teardown(self):
        
        try:
            self._fixture_teardown()
            super()._post_teardown()
            if self._should_reload_connections():
                
                
                
                
                
                
                for conn in connections.all(initialized_only=True):
                    conn.close()
        finally:
            if self.available_apps is not None:
                apps.unset_available_apps()
                setting_changed.send(
                    sender=settings._wrapped.__class__,
                    setting="INSTALLED_APPS",
                    value=settings.INSTALLED_APPS,
                    enter=False,
                )

    def _fixture_teardown(self):
        
        
        for db_name in self._databases_names(include_mirrors=False):
            
            inhibit_post_migrate = (
                self.available_apps is not None
                or (  
                    
                    self.serialized_rollback
                    and hasattr(connections[db_name], "_test_serialized_contents")
                )
            )
            call_command(
                "flush",
                verbosity=0,
                interactive=False,
                database=db_name,
                reset_sequences=False,
                allow_cascade=self.available_apps is not None,
                inhibit_post_migrate=inhibit_post_migrate,
            )

    def assertQuerySetEqual(self, qs, values, transform=None, ordered=True, msg=None):
        values = list(values)
        items = qs
        if transform is not None:
            items = map(transform, items)
        if not ordered:
            return self.assertDictEqual(Counter(items), Counter(values), msg=msg)
        
        
        if len(values) > 1 and hasattr(qs, "ordered") and not qs.ordered:
            raise ValueError(
                "Trying to compare non-ordered queryset against more than one "
                "ordered value."
            )
        return self.assertEqual(list(items), values, msg=msg)

    def assertNumQueries(self, num, func=None, *args, using=DEFAULT_DB_ALIAS, **kwargs):
        conn = connections[using]

        context = _AssertNumQueriesContext(self, num, conn)
        if func is None:
            return context

        with context:
            func(*args, **kwargs)


def connections_support_transactions(aliases=None):
    
    conns = (
        connections.all()
        if aliases is None
        else (connections[alias] for alias in aliases)
    )
    return all(conn.features.supports_transactions for conn in conns)


class TestData:
    

    memo_attr = "_testdata_memo"

    def __init__(self, name, data):
        self.name = name
        self.data = data

    def get_memo(self, testcase):
        try:
            memo = getattr(testcase, self.memo_attr)
        except AttributeError:
            memo = {}
            setattr(testcase, self.memo_attr, memo)
        return memo

    def __get__(self, instance, owner):
        if instance is None:
            return self.data
        memo = self.get_memo(instance)
        data = deepcopy(self.data, memo)
        setattr(instance, self.name, data)
        return data

    def __repr__(self):
        return "<TestData: name=%r, data=%r>" % (self.name, self.data)


class TestCase(TransactionTestCase):
    

    @classmethod
    def _enter_atomics(cls):
        
        atomics = {}
        for db_name in cls._databases_names():
            atomic = transaction.atomic(using=db_name)
            atomic._from_testcase = True
            atomic.__enter__()
            atomics[db_name] = atomic
        return atomics

    @classmethod
    def _rollback_atomics(cls, atomics):
        
        for db_name in reversed(cls._databases_names()):
            transaction.set_rollback(True, using=db_name)
            atomics[db_name].__exit__(None, None, None)

    @classmethod
    def _databases_support_transactions(cls):
        return connections_support_transactions(cls.databases)

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        if not cls._databases_support_transactions():
            return
        cls.cls_atomics = cls._enter_atomics()

        if cls.fixtures:
            for db_name in cls._databases_names(include_mirrors=False):
                try:
                    call_command(
                        "loaddata",
                        *cls.fixtures,
                        verbosity=0,
                        database=db_name,
                    )
                except Exception:
                    cls._rollback_atomics(cls.cls_atomics)
                    raise
        pre_attrs = cls.__dict__.copy()
        try:
            cls.setUpTestData()
        except Exception:
            cls._rollback_atomics(cls.cls_atomics)
            raise
        for name, value in cls.__dict__.items():
            if value is not pre_attrs.get(name):
                setattr(cls, name, TestData(name, value))

    @classmethod
    def tearDownClass(cls):
        if cls._databases_support_transactions():
            cls._rollback_atomics(cls.cls_atomics)
            for conn in connections.all(initialized_only=True):
                conn.close()
        super().tearDownClass()

    @classmethod
    def setUpTestData(cls):
        
        pass

    def _should_reload_connections(self):
        if self._databases_support_transactions():
            return False
        return super()._should_reload_connections()

    def _fixture_setup(self):
        if not self._databases_support_transactions():
            
            
            self.setUpTestData()
            return super()._fixture_setup()

        if self.reset_sequences:
            raise TypeError("reset_sequences cannot be used on TestCase instances")
        self.atomics = self._enter_atomics()

    def _fixture_teardown(self):
        if not self._databases_support_transactions():
            return super()._fixture_teardown()
        try:
            for db_name in reversed(self._databases_names()):
                if self._should_check_constraints(connections[db_name]):
                    connections[db_name].check_constraints()
        finally:
            self._rollback_atomics(self.atomics)

    def _should_check_constraints(self, connection):
        return (
            connection.features.can_defer_constraint_checks
            and not connection.needs_rollback
            and connection.is_usable()
        )

    @classmethod
    @contextmanager
    def captureOnCommitCallbacks(cls, *, using=DEFAULT_DB_ALIAS, execute=False):
        
        callbacks = []
        start_count = len(connections[using].run_on_commit)
        try:
            yield callbacks
        finally:
            while True:
                callback_count = len(connections[using].run_on_commit)
                for _, callback, robust in connections[using].run_on_commit[
                    start_count:
                ]:
                    callbacks.append(callback)
                    if execute:
                        if robust:
                            try:
                                callback()
                            except Exception as e:
                                logger.error(
                                    f"Error calling {callback.__qualname__} in "
                                    f"on_commit() (%s).",
                                    e,
                                    exc_info=True,
                                )
                        else:
                            callback()

                if callback_count == len(connections[using].run_on_commit):
                    break
                start_count = callback_count


class CheckCondition:
    

    def __init__(self, *conditions):
        self.conditions = conditions

    def add_condition(self, condition, reason):
        return self.__class__(*self.conditions, (condition, reason))

    def __get__(self, instance, cls=None):
        
        if any(getattr(base, "__unittest_skip__", False) for base in cls.__bases__):
            return True
        for condition, reason in self.conditions:
            if condition():
                
                cls.__unittest_skip__ = True
                cls.__unittest_skip_why__ = reason
                return True
        return False


def _deferredSkip(condition, reason, name):
    def decorator(test_func):
        nonlocal condition
        if not (
            isinstance(test_func, type) and issubclass(test_func, unittest.TestCase)
        ):

            @wraps(test_func)
            def skip_wrapper(*args, **kwargs):
                if (
                    args
                    and isinstance(args[0], unittest.TestCase)
                    and connection.alias not in getattr(args[0], "databases", {})
                ):
                    raise ValueError(
                        "%s cannot be used on %s as %s doesn't allow queries "
                        "against the %r database."
                        % (
                            name,
                            args[0],
                            args[0].__class__.__qualname__,
                            connection.alias,
                        )
                    )
                if condition():
                    raise unittest.SkipTest(reason)
                return test_func(*args, **kwargs)

            test_item = skip_wrapper
        else:
            
            test_item = test_func
            databases = getattr(test_item, "databases", None)
            if not databases or connection.alias not in databases:
                
                def condition():
                    raise ValueError(
                        "%s cannot be used on %s as it doesn't allow queries "
                        "against the '%s' database."
                        % (
                            name,
                            test_item,
                            connection.alias,
                        )
                    )

            
            
            skip = test_func.__dict__.get("__unittest_skip__")
            if isinstance(skip, CheckCondition):
                test_item.__unittest_skip__ = skip.add_condition(condition, reason)
            elif skip is not True:
                test_item.__unittest_skip__ = CheckCondition((condition, reason))
        return test_item

    return decorator


def skipIfDBFeature(*features):
    
    return _deferredSkip(
        lambda: any(
            getattr(connection.features, feature, False) for feature in features
        ),
        "Database has feature(s) %s" % ", ".join(features),
        "skipIfDBFeature",
    )


def skipUnlessDBFeature(*features):
    
    return _deferredSkip(
        lambda: not all(
            getattr(connection.features, feature, False) for feature in features
        ),
        "Database doesn't support feature(s): %s" % ", ".join(features),
        "skipUnlessDBFeature",
    )


def skipUnlessAnyDBFeature(*features):
    
    return _deferredSkip(
        lambda: not any(
            getattr(connection.features, feature, False) for feature in features
        ),
        "Database doesn't support any of the feature(s): %s" % ", ".join(features),
        "skipUnlessAnyDBFeature",
    )


class QuietWSGIRequestHandler(WSGIRequestHandler):
    

    def log_message(*args):
        pass


class FSFilesHandler(WSGIHandler):
    

    def __init__(self, application):
        self.application = application
        self.base_url = urlparse(self.get_base_url())
        super().__init__()

    def _should_handle(self, path):
        
        return path.startswith(self.base_url[2]) and not self.base_url[1]

    def file_path(self, url):
        
        relative_url = url.removeprefix(self.base_url[2])
        return url2pathname(relative_url)

    def get_response(self, request):
        from django.http import Http404

        if self._should_handle(request.path):
            try:
                return self.serve(request)
            except Http404:
                pass
        return super().get_response(request)

    def serve(self, request):
        os_rel_path = self.file_path(request.path)
        os_rel_path = posixpath.normpath(unquote(os_rel_path))
        
        
        
        final_rel_path = os_rel_path.replace("\\", "/").lstrip("/")
        return serve(request, final_rel_path, document_root=self.get_base_dir())

    def __call__(self, environ, start_response):
        if not self._should_handle(get_path_info(environ)):
            return self.application(environ, start_response)
        return super().__call__(environ, start_response)


class _StaticFilesHandler(FSFilesHandler):
    

    def get_base_dir(self):
        return settings.STATIC_ROOT

    def get_base_url(self):
        return settings.STATIC_URL


class _MediaFilesHandler(FSFilesHandler):
    

    def get_base_dir(self):
        return settings.MEDIA_ROOT

    def get_base_url(self):
        return settings.MEDIA_URL


class LiveServerThread(threading.Thread):
    

    server_class = ThreadedWSGIServer

    def __init__(self, host, static_handler, connections_override=None, port=0):
        self.host = host
        self.port = port
        self.is_ready = threading.Event()
        self.error = None
        self.static_handler = static_handler
        self.connections_override = connections_override
        super().__init__()

    def run(self):
        
        if self.connections_override:
            
            
            for alias, conn in self.connections_override.items():
                connections[alias] = conn
        try:
            
            handler = self.static_handler(_MediaFilesHandler(WSGIHandler()))
            self.httpd = self._create_server(
                connections_override=self.connections_override,
            )
            
            if self.port == 0:
                self.port = self.httpd.server_address[1]
            self.httpd.set_app(handler)
            self.is_ready.set()
            self.httpd.serve_forever()
        except Exception as e:
            self.error = e
            self.is_ready.set()
        finally:
            connections.close_all()

    def _create_server(self, connections_override=None):
        return self.server_class(
            (self.host, self.port),
            QuietWSGIRequestHandler,
            allow_reuse_address=False,
            connections_override=connections_override,
        )

    def terminate(self):
        if hasattr(self, "httpd"):
            
            self.httpd.shutdown()
            self.httpd.server_close()
        self.join()


class LiveServerTestCase(TransactionTestCase):
    

    host = "localhost"
    port = 0
    server_thread_class = LiveServerThread
    static_handler = _StaticFilesHandler

    @classproperty
    def live_server_url(cls):
        return "http://%s:%s" % (cls.host, cls.server_thread.port)

    @classproperty
    def allowed_host(cls):
        return cls.host

    @classmethod
    def _make_connections_override(cls):
        connections_override = {}
        for conn in connections.all():
            
            
            if conn.vendor == "sqlite" and conn.is_in_memory_db():
                connections_override[conn.alias] = conn
        return connections_override

    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls._live_server_modified_settings = modify_settings(
            ALLOWED_HOSTS={"append": cls.allowed_host},
        )
        cls._live_server_modified_settings.enable()
        cls.addClassCleanup(cls._live_server_modified_settings.disable)
        cls._start_server_thread()

    @classmethod
    def _start_server_thread(cls):
        connections_override = cls._make_connections_override()
        for conn in connections_override.values():
            
            conn.inc_thread_sharing()

        cls.server_thread = cls._create_server_thread(connections_override)
        cls.server_thread.daemon = True
        cls.server_thread.start()
        cls.addClassCleanup(cls._terminate_thread)

        
        cls.server_thread.is_ready.wait()
        if cls.server_thread.error:
            raise cls.server_thread.error

    @classmethod
    def _create_server_thread(cls, connections_override):
        return cls.server_thread_class(
            cls.host,
            cls.static_handler,
            connections_override=connections_override,
            port=cls.port,
        )

    @classmethod
    def _terminate_thread(cls):
        
        cls.server_thread.terminate()
        
        for conn in cls.server_thread.connections_override.values():
            conn.dec_thread_sharing()


class SerializeMixin:
    

    lockfile = None

    def __init_subclass__(cls, /, **kwargs):
        super().__init_subclass__(**kwargs)
        if cls.lockfile is None:
            raise ValueError(
                "{}.lockfile isn't set. Set it to a unique value "
                "in the base class.".format(cls.__name__)
            )

    @classmethod
    def setUpClass(cls):
        cls._lockfile = open(cls.lockfile)
        cls.addClassCleanup(cls._lockfile.close)
        locks.lock(cls._lockfile, locks.LOCK_EX)
        super().setUpClass()