mirror of
https://github.com/django/django.git
synced 2025-11-20 03:30:00 +00:00
Fixed #33143 -- Raised RuntimeWarning when performing import-time queries.
This commit is contained in:
parent
bd2ff65fdd
commit
fbd16438f4
7 changed files with 227 additions and 3 deletions
0
tests/apps/query_performing_app/__init__.py
Normal file
0
tests/apps/query_performing_app/__init__.py
Normal file
92
tests/apps/query_performing_app/apps.py
Normal file
92
tests/apps/query_performing_app/apps.py
Normal file
|
|
@ -0,0 +1,92 @@
|
|||
from django.apps import AppConfig
|
||||
from django.db import connections
|
||||
|
||||
|
||||
class BaseAppConfig(AppConfig):
|
||||
name = "apps.query_performing_app"
|
||||
database = "default"
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
super().__init__(*args, **kwargs)
|
||||
self.query_results = []
|
||||
|
||||
def ready(self):
|
||||
self.query_results = []
|
||||
self._perform_query()
|
||||
|
||||
def _perform_query(self):
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
class ModelQueryAppConfig(BaseAppConfig):
|
||||
def _perform_query(self):
|
||||
from ..models import TotallyNormal
|
||||
|
||||
queryset = TotallyNormal.objects.using(self.database)
|
||||
queryset.update_or_create(name="new name")
|
||||
self.query_results = list(queryset.values_list("name"))
|
||||
|
||||
|
||||
class QueryDefaultDatabaseModelAppConfig(ModelQueryAppConfig):
|
||||
database = "default"
|
||||
|
||||
|
||||
class QueryOtherDatabaseModelAppConfig(ModelQueryAppConfig):
|
||||
database = "other"
|
||||
|
||||
|
||||
class CursorQueryAppConfig(BaseAppConfig):
|
||||
def _perform_query(self):
|
||||
connection = connections[self.database]
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("SELECT 42" + connection.features.bare_select_suffix)
|
||||
self.query_results = cursor.fetchall()
|
||||
|
||||
|
||||
class QueryDefaultDatabaseCursorAppConfig(CursorQueryAppConfig):
|
||||
database = "default"
|
||||
|
||||
|
||||
class QueryOtherDatabaseCursorAppConfig(CursorQueryAppConfig):
|
||||
database = "other"
|
||||
|
||||
|
||||
class CursorQueryManyAppConfig(BaseAppConfig):
|
||||
def _perform_query(self):
|
||||
from ..models import TotallyNormal
|
||||
|
||||
connection = connections[self.database]
|
||||
table_meta = TotallyNormal._meta
|
||||
with connection.cursor() as cursor:
|
||||
cursor.executemany(
|
||||
"INSERT INTO %s (%s) VALUES(%%s)"
|
||||
% (
|
||||
connection.introspection.identifier_converter(table_meta.db_table),
|
||||
connection.ops.quote_name(table_meta.get_field("name").column),
|
||||
),
|
||||
[("test name 1",), ("test name 2",)],
|
||||
)
|
||||
self.query_results = []
|
||||
|
||||
|
||||
class QueryDefaultDatabaseCursorManyAppConfig(CursorQueryManyAppConfig):
|
||||
database = "default"
|
||||
|
||||
|
||||
class QueryOtherDatabaseCursorManyAppConfig(CursorQueryManyAppConfig):
|
||||
database = "other"
|
||||
|
||||
|
||||
class StoredProcedureQueryAppConfig(BaseAppConfig):
|
||||
def _perform_query(self):
|
||||
with connections[self.database].cursor() as cursor:
|
||||
cursor.callproc("test_procedure")
|
||||
self.query_results = []
|
||||
|
||||
|
||||
class QueryDefaultDatabaseStoredProcedureAppConfig(StoredProcedureQueryAppConfig):
|
||||
database = "default"
|
||||
|
||||
|
||||
class QueryOtherDatabaseStoredProcedureAppConfig(StoredProcedureQueryAppConfig):
|
||||
database = "other"
|
||||
|
|
@ -1,11 +1,18 @@
|
|||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import django
|
||||
from django.apps import AppConfig, apps
|
||||
from django.apps.registry import Apps
|
||||
from django.contrib.admin.models import LogEntry
|
||||
from django.core.exceptions import AppRegistryNotReady, ImproperlyConfigured
|
||||
from django.db import models
|
||||
from django.test import SimpleTestCase, override_settings
|
||||
from django.db import connections, models
|
||||
from django.test import (
|
||||
SimpleTestCase,
|
||||
TransactionTestCase,
|
||||
override_settings,
|
||||
skipUnlessDBFeature,
|
||||
)
|
||||
from django.test.utils import extend_sys_path, isolate_apps
|
||||
|
||||
from .models import SoAlternative, TotallyNormal, new_apps
|
||||
|
|
@ -539,3 +546,77 @@ class NamespacePackageAppTests(SimpleTestCase):
|
|||
with self.settings(INSTALLED_APPS=["nsapp.apps.NSAppConfig"]):
|
||||
app_config = apps.get_app_config("nsapp")
|
||||
self.assertEqual(app_config.path, self.app_path)
|
||||
|
||||
|
||||
class QueryPerformingAppTests(TransactionTestCase):
|
||||
available_apps = ["apps"]
|
||||
databases = {"default", "other"}
|
||||
expected_msg = (
|
||||
"Accessing the database during app initialization is discouraged. To fix this "
|
||||
"warning, avoid executing queries in AppConfig.ready() or when your app "
|
||||
"modules are imported."
|
||||
)
|
||||
|
||||
def test_query_default_database_using_model(self):
|
||||
query_results = self.run_setup("QueryDefaultDatabaseModelAppConfig")
|
||||
self.assertSequenceEqual(query_results, [("new name",)])
|
||||
|
||||
def test_query_other_database_using_model(self):
|
||||
query_results = self.run_setup("QueryOtherDatabaseModelAppConfig")
|
||||
self.assertSequenceEqual(query_results, [("new name",)])
|
||||
|
||||
def test_query_default_database_using_cursor(self):
|
||||
query_results = self.run_setup("QueryDefaultDatabaseCursorAppConfig")
|
||||
self.assertSequenceEqual(query_results, [(42,)])
|
||||
|
||||
def test_query_other_database_using_cursor(self):
|
||||
query_results = self.run_setup("QueryOtherDatabaseCursorAppConfig")
|
||||
self.assertSequenceEqual(query_results, [(42,)])
|
||||
|
||||
def test_query_many_default_database_using_cursor(self):
|
||||
self.run_setup("QueryDefaultDatabaseCursorManyAppConfig")
|
||||
|
||||
def test_query_many_other_database_using_cursor(self):
|
||||
self.run_setup("QueryOtherDatabaseCursorManyAppConfig")
|
||||
|
||||
@skipUnlessDBFeature("create_test_procedure_without_params_sql")
|
||||
def test_query_default_database_using_stored_procedure(self):
|
||||
connection = connections["default"]
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(connection.features.create_test_procedure_without_params_sql)
|
||||
|
||||
try:
|
||||
self.run_setup("QueryDefaultDatabaseStoredProcedureAppConfig")
|
||||
finally:
|
||||
with connection.schema_editor() as editor:
|
||||
editor.remove_procedure("test_procedure")
|
||||
|
||||
@skipUnlessDBFeature("create_test_procedure_without_params_sql")
|
||||
def test_query_other_database_using_stored_procedure(self):
|
||||
connection = connections["other"]
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(connection.features.create_test_procedure_without_params_sql)
|
||||
|
||||
try:
|
||||
self.run_setup("QueryOtherDatabaseStoredProcedureAppConfig")
|
||||
finally:
|
||||
with connection.schema_editor() as editor:
|
||||
editor.remove_procedure("test_procedure")
|
||||
|
||||
def run_setup(self, app_config_name):
|
||||
custom_settings = override_settings(
|
||||
INSTALLED_APPS=[f"apps.query_performing_app.apps.{app_config_name}"]
|
||||
)
|
||||
# Ignore the RuntimeWarning, as override_settings.enable() calls
|
||||
# AppConfig.ready() which will trigger the warning.
|
||||
with self.assertWarnsMessage(RuntimeWarning, self.expected_msg):
|
||||
custom_settings.enable()
|
||||
try:
|
||||
with patch.multiple(apps, ready=False, loading=False, app_configs={}):
|
||||
with self.assertWarnsMessage(RuntimeWarning, self.expected_msg):
|
||||
django.setup()
|
||||
|
||||
app_config = apps.get_app_config("query_performing_app")
|
||||
return app_config.query_results
|
||||
finally:
|
||||
custom_settings.disable()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue