Python django.db.connections 模块，databases() 实例源码

def test_backend_db():
    """Ensure that we are always testing sqlite on fast in memory DB"""
    from django.db import connection, connections

    if connection.vendor == "sqlite":
        assert connections.databases["default"]["NAME"] == ":memory:"

def test_broken_database(settings, monkeypatch):
    settings.DATABASES = {
        "default": {
            "ENGINE": "django.db.backends.sqlite3",
            "NAME": ":memory:"},

        "broken": {
            "ENGINE": "django.db.backends.postgresql_psycopg2",
            "NAME": "not-existent-db",
            "HOST": "127.0.0.1",
            "PORT": "21",
            "USER": "",
            "PASSWORD": ""
        },

    }
    monkeypatch.setattr(connections, "_databases", None)
    del connections.databases

    ret = get_databases()
    assert sorted(ret.keys()) == ["broken", "default"]
    # assert ret["default"]["engine"] == "django.db.backends.sqlite3"
    assert ret["broken"]["engine"] == "django.db.backends.postgresql_psycopg2"
    assert ret["broken"]["host"] == "127.0.0.1:21"
    assert ret["broken"]["name"] == "not-existent-db"
    assert ret["broken"]["error"] == """could not connect to server: Connection refused
\tIs the server running on host "127.0.0.1" and accepting
\tTCP/IP connections on port 21?
"""

def __exit__(self, return_type, return_value, traceback):
        if os.path.exists(self.fullpath):
            os.remove(self.fullpath)
        name = db_name_from_filename(self.filename)
        if name in settings.DATABASES:
            settings.DATABASES.pop(name)
        if name in connections.databases:
            connections.databases.pop(name)

def add_db_to_settings(dbname, filename, gemini_path=GEMINI_DB_PATH):
    """Add a new db to settings.DATABASES"""
    connection = {
        'ENGINE': 'django.db.backends.sqlite3',
        'NAME': join(gemini_path, filename)
    }
    settings.DATABASES[dbname] = connection
    connections.databases[dbname] = connection
    logger.debug("(+) Adding connection '{}'".format(dbname))

def remove_db_from_settings(dbname):
    """Remove that connection from settings.DATABASES and connections.databases."""
    settings.DATABASES.pop(dbname, None)
    connections.databases.pop(dbname, None)

def mock_django_connection(disabled_features=None):
    """ Overwrite the Django database configuration with a mocked version.

    This is a helper function that does the actual monkey patching.
    """
    db = connections.databases['default']
    db['PASSWORD'] = '****'
    db['USER'] = '**Database disabled for unit tests**'
    ConnectionHandler.__getitem__ = MagicMock(name='mock_connection')
    # noinspection PyUnresolvedReferences
    mock_connection = ConnectionHandler.__getitem__.return_value
    if disabled_features:
        for feature in disabled_features:
            setattr(mock_connection.features, feature, False)
    mock_ops = mock_connection.ops

    # noinspection PyUnusedLocal
    def compiler(queryset, connection, using, **kwargs):
        result = MagicMock(name='mock_connection.ops.compiler()')
        # noinspection PyProtectedMember
        result.execute_sql.side_effect = NotSupportedError(
            "Mock database tried to execute SQL for {} model.".format(
                queryset.model._meta.object_name))
        result.has_results.side_effect = result.execute_sql.side_effect
        return result

    mock_ops.compiler.return_value.side_effect = compiler
    mock_ops.integer_field_range.return_value = (-sys.maxsize - 1, sys.maxsize)
    mock_ops.max_name_length.return_value = sys.maxsize

    Model.refresh_from_db = Mock()  # Make this into a noop.

def set_compiler(self):
        engine = connections.databases[self._using]['ENGINE'].rsplit('.')[-1]
        self.compile = SMARTSQL_COMPILERS[engine]

def test_time_field(self):
        # Only possible to test this on PostGIS at the momemnt.  MySQL
        # complains about permissions, and SpatiaLite/Oracle are
        # insanely difficult to get support compiled in for in GDAL.
        if not connections['default'].ops.postgis:
            self.skipTest("This database does not support 'ogrinspect'ion")

        # Getting the database identifier used by OGR, if None returned
        # GDAL does not have the support compiled in.
        ogr_db = get_ogr_db_string()
        if not ogr_db:
            self.skipTest("Your GDAL installation does not support PostGIS databases")

        # Writing shapefiles via GDAL currently does not support writing OGRTime
        # fields, so we need to actually use a database
        model_def = ogrinspect(ogr_db, 'Measurement',
                               layer_key=AllOGRFields._meta.db_table,
                               decimal=['f_decimal'])

        self.assertTrue(model_def.startswith(
            '# This is an auto-generated Django model module created by ogrinspect.\n'
            'from django.contrib.gis.db import models\n'
            '\n'
            'class Measurement(models.Model):\n'
        ))

        # The ordering of model fields might vary depending on several factors (version of GDAL, etc.)
        self.assertIn('    f_decimal = models.DecimalField(max_digits=0, decimal_places=0)', model_def)
        self.assertIn('    f_int = models.IntegerField()', model_def)
        self.assertIn('    f_datetime = models.DateTimeField()', model_def)
        self.assertIn('    f_time = models.TimeField()', model_def)
        self.assertIn('    f_float = models.FloatField()', model_def)
        self.assertIn('    f_char = models.CharField(max_length=10)', model_def)
        self.assertIn('    f_date = models.DateField()', model_def)

        self.assertTrue(model_def.endswith(
            '    geom = models.PolygonField()\n'
            '    objects = models.GeoManager()'
        ))

def createConnection(self):
        """Create new database connection."""
        db = connections.databases[self.alias]
        backend = load_backend(db['ENGINE'])
        return backend.DatabaseWrapper(
            db, self.alias, allow_thread_sharing=True)

def test_atomic_requests_are_enabled(self):
        # ATOMIC_REQUESTS *must* be set for the default connection.
        self.assertThat(
            connections.databases, ContainsDict({
                "default": ContainsDict({
                    "ATOMIC_REQUESTS": Is(True),
                }),
            }),
        )

def test_isolation_level_is_serializable(self):
        # Transactions *must* be SERIALIZABLE for the default connection.
        self.assertThat(
            connections.databases, ContainsDict({
                "default": ContainsDict({
                    "OPTIONS": ContainsDict({
                        "isolation_level": Equals(
                            ISOLATION_LEVEL_REPEATABLE_READ),
                    }),
                }),
            }),
        )

def get_ogr_db_string():
    """
    Construct the DB string that GDAL will use to inspect the database.
    GDAL will create its own connection to the database, so we re-use the
    connection settings from the Django test.
    """
    db = connections.databases['default']

    # Map from the django backend into the OGR driver name and database identifier
    # http://www.gdal.org/ogr/ogr_formats.html
    #
    # TODO: Support Oracle (OCI).
    drivers = {
        'django.contrib.gis.db.backends.postgis': ('PostgreSQL', "PG:dbname='%(db_name)s'", ' '),
        'django.contrib.gis.db.backends.mysql': ('MySQL', 'MYSQL:"%(db_name)s"', ','),
        'django.contrib.gis.db.backends.spatialite': ('SQLite', '%(db_name)s', '')
    }

    drv_name, db_str, param_sep = drivers[db['ENGINE']]

    # Ensure that GDAL library has driver support for the database.
    try:
        Driver(drv_name)
    except:
        return None

    # SQLite/Spatialite in-memory databases
    if db['NAME'] == ":memory:":
        return None

    # Build the params of the OGR database connection string
    params = [db_str % {'db_name': db['NAME']}]
    def add(key, template):
        value = db.get(key, None)
        # Don't add the parameter if it is not in django's settings
        if value:
            params.append(template % value)
    add('HOST', "host='%s'")
    add('PORT', "port='%s'")
    add('USER', "user='%s'")
    add('PASSWORD', "password='%s'")

    return param_sep.join(params)