Fixed postgis test database initialization

Refs #20968. Allow querying template_postgis presence without
existing test database.
Thanks Tim Graham for the review.
This commit is contained in:
Claude Paroz 2014-12-09 15:27:20 +01:00
parent c7a19f4203
commit df30ae07fc
4 changed files with 30 additions and 26 deletions

View File

@ -1,5 +1,5 @@
from django.conf import settings
from django.db.backends.creation import NO_DB_ALIAS
from django.db.backends import NO_DB_ALIAS
from django.db.backends.postgresql_psycopg2.base import (
DatabaseWrapper as Psycopg2DatabaseWrapper,
DatabaseFeatures as Psycopg2DatabaseFeatures
@ -31,7 +31,7 @@ class DatabaseWrapper(Psycopg2DatabaseWrapper):
@cached_property
def template_postgis(self):
template_postgis = getattr(settings, 'POSTGIS_TEMPLATE', 'template_postgis')
with self.cursor() as cursor:
with self._nodb_connection.cursor() as cursor:
cursor.execute('SELECT 1 FROM pg_database WHERE datname = %s LIMIT 1;', (template_postgis,))
if cursor.fetchone():
return template_postgis
@ -43,4 +43,3 @@ class DatabaseWrapper(Psycopg2DatabaseWrapper):
# Check that postgis extension is installed on PostGIS >= 2
with self.cursor() as cursor:
cursor.execute("CREATE EXTENSION IF NOT EXISTS postgis")
cursor.connection.commit()

View File

@ -2,7 +2,6 @@ from __future__ import unicode_literals
import re
from tempfile import NamedTemporaryFile
import unittest
from django.db import connection
from django.contrib.gis import gdal
@ -292,6 +291,11 @@ class GeoLookupTest(TestCase):
# Right: A >> B => true if xmin(A) > xmax(B)
# See: BOX2D_left() and BOX2D_right() in lwgeom_box2dfloat4.c in PostGIS source.
# The left/right lookup tests are known failures on PostGIS 2.0/2.0.1
# http://trac.osgeo.org/postgis/ticket/2035
if postgis_bug_version():
self.skipTest("PostGIS 2.0/2.0.1 left and right lookups are known to be buggy.")
# Getting the borders for Colorado & Kansas
co_border = State.objects.get(name='Colorado').poly
ks_border = State.objects.get(name='Kansas').poly
@ -325,11 +329,6 @@ class GeoLookupTest(TestCase):
for c in qs:
self.assertIn(c.name, cities)
# The left/right lookup tests are known failures on PostGIS 2.0/2.0.1
# http://trac.osgeo.org/postgis/ticket/2035
if postgis_bug_version():
test_left_right_lookups = unittest.expectedFailure(test_left_right_lookups)
def test_equals_lookups(self):
"Testing the 'same_as' and 'equals' lookup types."
pnt = fromstr('POINT (-95.363151 29.763374)', srid=4326)

View File

@ -30,6 +30,8 @@ TableInfo = namedtuple('TableInfo', ['name', 'type'])
FieldInfo = namedtuple('FieldInfo',
'name type_code display_size internal_size precision scale null_ok')
NO_DB_ALIAS = '__no_db__'
class BaseDatabaseWrapper(object):
"""
@ -477,6 +479,24 @@ class BaseDatabaseWrapper(object):
if must_close:
self.close()
@cached_property
def _nodb_connection(self):
"""
Alternative connection to be used when there is no need to access
the main database, specifically for test db creation/deletion.
This also prevents the production database from being exposed to
potential child threads while (or after) the test database is destroyed.
Refs #10868, #17786, #16969.
"""
settings_dict = self.settings_dict.copy()
settings_dict['NAME'] = None
#backend = load_backend(settings_dict['ENGINE'])
nodb_connection = self.__class__(
settings_dict,
alias=NO_DB_ALIAS,
allow_thread_sharing=False)
return nodb_connection
def _start_transaction_under_autocommit(self):
"""
Only required when autocommits_when_autocommit_is_off = True.

View File

@ -3,9 +3,7 @@ import sys
import time
from django.conf import settings
from django.db.utils import load_backend
from django.utils.encoding import force_bytes
from django.utils.functional import cached_property
from django.utils.six.moves import input
from django.utils.six import StringIO
from django.db import router
@ -17,7 +15,6 @@ from .utils import truncate_name
# The prefix to put on the default database name when creating
# the test database.
TEST_DATABASE_PREFIX = 'test_'
NO_DB_ALIAS = '__no_db__'
class BaseDatabaseCreation(object):
@ -34,23 +31,12 @@ class BaseDatabaseCreation(object):
def __init__(self, connection):
self.connection = connection
@cached_property
@property
def _nodb_connection(self):
"""
Alternative connection to be used when there is no need to access
the main database, specifically for test db creation/deletion.
This also prevents the production database from being exposed to
potential child threads while (or after) the test database is destroyed.
Refs #10868, #17786, #16969.
Used to be defined here, now moved to DatabaseWrapper.
"""
settings_dict = self.connection.settings_dict.copy()
settings_dict['NAME'] = None
backend = load_backend(settings_dict['ENGINE'])
nodb_connection = backend.DatabaseWrapper(
settings_dict,
alias=NO_DB_ALIAS,
allow_thread_sharing=False)
return nodb_connection
return self.connection._nodb_connection
@classmethod
def _digest(cls, *args):