Hopiu/dj-database-url
1
0
Fork 0
mirror of https://github.com/jazzband/dj-database-url.git synced 2026-03-16 22:20:24 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Re-format code to black, isort, flake8

Browse source
This commit is contained in:
Matt Seymour 2022-05-04 09:43:30 +01:00 committed by Matt Seymour
parent 80770bb243
commit 3e87a497f8
3 changed files with 258 additions and 262 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

118
dj_database_url.py
View file

@ -14,46 +14,46 @@ except ImportError:
# Register database schemes in URLs.
urlparse.uses_netloc.append('postgres')
urlparse.uses_netloc.append('postgresql')
urlparse.uses_netloc.append('pgsql')
urlparse.uses_netloc.append('postgis')
urlparse.uses_netloc.append('mysql')
urlparse.uses_netloc.append('mysql2')
urlparse.uses_netloc.append('mysqlgis')
urlparse.uses_netloc.append('mysql-connector')
urlparse.uses_netloc.append('mssql')
urlparse.uses_netloc.append('spatialite')
urlparse.uses_netloc.append('sqlite')
urlparse.uses_netloc.append('oracle')
urlparse.uses_netloc.append('oraclegis')
urlparse.uses_netloc.append('redshift')
urlparse.uses_netloc.append("postgres")
urlparse.uses_netloc.append("postgresql")
urlparse.uses_netloc.append("pgsql")
urlparse.uses_netloc.append("postgis")
urlparse.uses_netloc.append("mysql")
urlparse.uses_netloc.append("mysql2")
urlparse.uses_netloc.append("mysqlgis")
urlparse.uses_netloc.append("mysql-connector")
urlparse.uses_netloc.append("mssql")
urlparse.uses_netloc.append("spatialite")
urlparse.uses_netloc.append("sqlite")
urlparse.uses_netloc.append("oracle")
urlparse.uses_netloc.append("oraclegis")
urlparse.uses_netloc.append("redshift")
DEFAULT_ENV = 'DATABASE_URL'
DEFAULT_ENV = "DATABASE_URL"
SCHEMES = {
'postgis': 'django.contrib.gis.db.backends.postgis',
'mysql': 'django.db.backends.mysql',
'mysql2': 'django.db.backends.mysql',
'mysqlgis': 'django.contrib.gis.db.backends.mysql',
'mysql-connector': 'mysql.connector.django',
'mssql': 'sql_server.pyodbc',
'spatialite': 'django.contrib.gis.db.backends.spatialite',
'sqlite': 'django.db.backends.sqlite3',
'oracle': 'django.db.backends.oracle',
'oraclegis': 'django.contrib.gis.db.backends.oracle',
'redshift': 'django_redshift_backend',
"postgis": "django.contrib.gis.db.backends.postgis",
"mysql": "django.db.backends.mysql",
"mysql2": "django.db.backends.mysql",
"mysqlgis": "django.contrib.gis.db.backends.mysql",
"mysql-connector": "mysql.connector.django",
"mssql": "sql_server.pyodbc",
"spatialite": "django.contrib.gis.db.backends.spatialite",
"sqlite": "django.db.backends.sqlite3",
"oracle": "django.db.backends.oracle",
"oraclegis": "django.contrib.gis.db.backends.oracle",
"redshift": "django_redshift_backend",
}
# https://docs.djangoproject.com/en/2.0/releases/2.0/#id1
if DJANGO_VERSION and DJANGO_VERSION < (2, 0):
SCHEMES['postgres'] = 'django.db.backends.postgresql_psycopg2'
SCHEMES['postgresql'] = 'django.db.backends.postgresql_psycopg2'
SCHEMES['pgsql'] = 'django.db.backends.postgresql_psycopg2'
SCHEMES["postgres"] = "django.db.backends.postgresql_psycopg2"
SCHEMES["postgresql"] = "django.db.backends.postgresql_psycopg2"
SCHEMES["pgsql"] = "django.db.backends.postgresql_psycopg2"
else:
SCHEMES['postgres'] = 'django.db.backends.postgresql'
SCHEMES['postgresql'] = 'django.db.backends.postgresql'
SCHEMES['pgsql'] = 'django.db.backends.postgresql'
SCHEMES["postgres"] = "django.db.backends.postgresql"
SCHEMES["postgresql"] = "django.db.backends.postgresql"
SCHEMES["pgsql"] = "django.db.backends.postgresql"
def config(
@ -74,11 +74,11 @@ def config(
def parse(url, engine=None, conn_max_age=0, ssl_require=False):
"""Parses a database URL."""
if url == 'sqlite://:memory:':
if url == "sqlite://:memory:":
# this is a special case, because if we pass this URL into
# urlparse, urlparse will choke trying to interpret "memory"
# as a port number
return {'ENGINE': SCHEMES['sqlite'], 'NAME': ':memory:'}
return {"ENGINE": SCHEMES["sqlite"], "NAME": ":memory:"}
# note: no other settings are required for sqlite
# otherwise parse the url as normal
@ -88,74 +88,74 @@ def parse(url, engine=None, conn_max_age=0, ssl_require=False):
# Split query strings from path.
path = url.path[1:]
if '?' in path and not url.query:
path, query = path.split('?', 2)
if "?" in path and not url.query:
path, query = path.split("?", 2)
else:
path, query = path, url.query
query = urlparse.parse_qs(query)
# If we are using sqlite and we have no path, then assume we
# want an in-memory database (this is the behaviour of sqlalchemy)
if url.scheme == 'sqlite' and path == '':
path = ':memory:'
if url.scheme == "sqlite" and path == "":
path = ":memory:"
# Handle postgres percent-encoded paths.
hostname = url.hostname or ''
if '%2f' in hostname.lower():
hostname = url.hostname or ""
if "%2f" in hostname.lower():
# Switch to url.netloc to avoid lower cased paths
hostname = url.netloc
if "@" in hostname:
hostname = hostname.rsplit("@", 1)[1]
if ":" in hostname:
hostname = hostname.split(":", 1)[0]
hostname = hostname.replace('%2f', '/').replace('%2F', '/')
hostname = hostname.replace("%2f", "/").replace("%2F", "/")
# Lookup specified engine.
engine = SCHEMES[url.scheme] if engine is None else engine
port = (
str(url.port)
if url.port and engine in [SCHEMES['oracle'], SCHEMES['mssql']]
if url.port and engine in [SCHEMES["oracle"], SCHEMES["mssql"]]
else url.port
)
# Update with environment configuration.
config.update(
{
'NAME': urlparse.unquote(path or ''),
'USER': urlparse.unquote(url.username or ''),
'PASSWORD': urlparse.unquote(url.password or ''),
'HOST': hostname,
'PORT': port or '',
'CONN_MAX_AGE': conn_max_age,
"NAME": urlparse.unquote(path or ""),
"USER": urlparse.unquote(url.username or ""),
"PASSWORD": urlparse.unquote(url.password or ""),
"HOST": hostname,
"PORT": port or "",
"CONN_MAX_AGE": conn_max_age,
}
)
# Pass the query string into OPTIONS.
options = {}
for key, values in query.items():
if url.scheme == 'mysql' and key == 'ssl-ca':
options['ssl'] = {'ca': values[-1]}
if url.scheme == "mysql" and key == "ssl-ca":
options["ssl"] = {"ca": values[-1]}
continue
options[key] = values[-1]
if ssl_require:
options['sslmode'] = 'require'
options["sslmode"] = "require"
# Support for Postgres Schema URLs
if 'currentSchema' in options and engine in (
'django.contrib.gis.db.backends.postgis',
'django.db.backends.postgresql_psycopg2',
'django.db.backends.postgresql',
'django_redshift_backend',
if "currentSchema" in options and engine in (
"django.contrib.gis.db.backends.postgis",
"django.db.backends.postgresql_psycopg2",
"django.db.backends.postgresql",
"django_redshift_backend",
):
options['options'] = '-c search_path={0}'.format(options.pop('currentSchema'))
options["options"] = "-c search_path={0}".format(options.pop("currentSchema"))
if options:
config['OPTIONS'] = options
config["OPTIONS"] = options
if engine:
config['ENGINE'] = engine
config["ENGINE"] = engine
return config

2
setup.py
View file

@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
from setuptools import setup
with open('README.rst') as readme_rst:
with open("README.rst") as readme_rst:
readme = readme_rst.read()
setup(

400
test_dj_database_url.py
View file

@ -1,6 +1,3 @@
# -*- coding: utf-8 -*-
#!/usr/bin/env python
import os
import unittest
@ -8,343 +5,342 @@ from django import VERSION as DJANGO_VERSION
import dj_database_url
POSTGIS_URL = 'postgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn'
POSTGIS_URL = "postgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn"
# Django deprecated the `django.db.backends.postgresql_psycopg2` in 2.0.
# https://docs.djangoproject.com/en/2.0/releases/2.0/#id1
EXPECTED_POSTGRES_ENGINE = 'django.db.backends.postgresql'
EXPECTED_POSTGRES_ENGINE = "django.db.backends.postgresql"
if DJANGO_VERSION < (2, 0):
EXPECTED_POSTGRES_ENGINE = 'django.db.backends.postgresql_psycopg2'
EXPECTED_POSTGRES_ENGINE = "django.db.backends.postgresql_psycopg2"
class DatabaseTestSuite(unittest.TestCase):
def test_postgres_parsing(self):
url = 'postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn'
url = "postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
def test_postgres_unix_socket_parsing(self):
url = 'postgres://%2Fvar%2Frun%2Fpostgresql/d8r82722r2kuvn'
url = "postgres://%2Fvar%2Frun%2Fpostgresql/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == '/var/run/postgresql'
assert url['USER'] == ''
assert url['PASSWORD'] == ''
assert url['PORT'] == ''
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "/var/run/postgresql"
assert url["USER"] == ""
assert url["PASSWORD"] == ""
assert url["PORT"] == ""
url = 'postgres://%2FUsers%2Fpostgres%2FRuN/d8r82722r2kuvn'
url = "postgres://%2FUsers%2Fpostgres%2FRuN/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['HOST'] == '/Users/postgres/RuN'
assert url['USER'] == ''
assert url['PASSWORD'] == ''
assert url['PORT'] == ''
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["HOST"] == "/Users/postgres/RuN"
assert url["USER"] == ""
assert url["PASSWORD"] == ""
assert url["PORT"] == ""
def test_ipv6_parsing(self):
url = 'postgres://ieRaekei9wilaim7:wegauwhgeuioweg@[2001:db8:1234::1234:5678:90af]:5431/d8r82722r2kuvn'
url = "postgres://ieRaekei9wilaim7:wegauwhgeuioweg@[2001:db8:1234::1234:5678:90af]:5431/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == '2001:db8:1234::1234:5678:90af'
assert url['USER'] == 'ieRaekei9wilaim7'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "2001:db8:1234::1234:5678:90af"
assert url["USER"] == "ieRaekei9wilaim7"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
def test_postgres_search_path_parsing(self):
url = 'postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?currentSchema=otherschema'
url = "postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?currentSchema=otherschema"
url = dj_database_url.parse(url)
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url['OPTIONS']['options'] == '-c search_path=otherschema'
assert 'currentSchema' not in url['OPTIONS']
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
assert url["OPTIONS"]["options"] == "-c search_path=otherschema"
assert "currentSchema" not in url["OPTIONS"]
def test_postgres_parsing_with_special_characters(self):
url = 'postgres://%23user:%23password@ec2-107-21-253-135.compute-1.amazonaws.com:5431/%23database'
url = "postgres://%23user:%23password@ec2-107-21-253-135.compute-1.amazonaws.com:5431/%23database"
url = dj_database_url.parse(url)
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == '#database'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == '#user'
assert url['PASSWORD'] == '#password'
assert url['PORT'] == 5431
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "#database"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "#user"
assert url["PASSWORD"] == "#password"
assert url["PORT"] == 5431
def test_postgis_parsing(self):
url = 'postgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn'
url = "postgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.contrib.gis.db.backends.postgis'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url["ENGINE"] == "django.contrib.gis.db.backends.postgis"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
def test_postgis_search_path_parsing(self):
url = 'postgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?currentSchema=otherschema'
url = "postgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?currentSchema=otherschema"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.contrib.gis.db.backends.postgis'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url['OPTIONS']['options'] == '-c search_path=otherschema'
assert 'currentSchema' not in url['OPTIONS']
assert url["ENGINE"] == "django.contrib.gis.db.backends.postgis"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
assert url["OPTIONS"]["options"] == "-c search_path=otherschema"
assert "currentSchema" not in url["OPTIONS"]
def test_mysql_gis_parsing(self):
url = 'mysqlgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn'
url = "mysqlgis://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.contrib.gis.db.backends.mysql'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url["ENGINE"] == "django.contrib.gis.db.backends.mysql"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
def test_mysql_connector_parsing(self):
url = 'mysql-connector://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn'
url = "mysql-connector://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'mysql.connector.django'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url["ENGINE"] == "mysql.connector.django"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
def test_cleardb_parsing(self):
url = 'mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true'
url = "mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.db.backends.mysql'
assert url['NAME'] == 'heroku_97681db3eff7580'
assert url['HOST'] == 'us-cdbr-east.cleardb.com'
assert url['USER'] == 'bea6eb025ca0d8'
assert url['PASSWORD'] == '69772142'
assert url['PORT'] == ''
assert url["ENGINE"] == "django.db.backends.mysql"
assert url["NAME"] == "heroku_97681db3eff7580"
assert url["HOST"] == "us-cdbr-east.cleardb.com"
assert url["USER"] == "bea6eb025ca0d8"
assert url["PASSWORD"] == "69772142"
assert url["PORT"] == ""
def test_database_url(self):
del os.environ['DATABASE_URL']
del os.environ["DATABASE_URL"]
a = dj_database_url.config()
assert not a
os.environ[
'DATABASE_URL'
] = 'postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn'
"DATABASE_URL"
] = "postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn"
url = dj_database_url.config()
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
def test_empty_sqlite_url(self):
url = 'sqlite://'
url = "sqlite://"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.db.backends.sqlite3'
assert url['NAME'] == ':memory:'
assert url["ENGINE"] == "django.db.backends.sqlite3"
assert url["NAME"] == ":memory:"
def test_memory_sqlite_url(self):
url = 'sqlite://:memory:'
url = "sqlite://:memory:"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.db.backends.sqlite3'
assert url['NAME'] == ':memory:'
assert url["ENGINE"] == "django.db.backends.sqlite3"
assert url["NAME"] == ":memory:"
def test_parse_engine_setting(self):
engine = 'django_mysqlpool.backends.mysqlpool'
url = 'mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true'
engine = "django_mysqlpool.backends.mysqlpool"
url = "mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true"
url = dj_database_url.parse(url, engine)
assert url['ENGINE'] == engine
assert url["ENGINE"] == engine
def test_config_engine_setting(self):
engine = 'django_mysqlpool.backends.mysqlpool'
engine = "django_mysqlpool.backends.mysqlpool"
os.environ[
'DATABASE_URL'
] = 'mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true'
"DATABASE_URL"
] = "mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true"
url = dj_database_url.config(engine=engine)
assert url['ENGINE'] == engine
assert url["ENGINE"] == engine
def test_parse_conn_max_age_setting(self):
conn_max_age = 600
url = 'mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true'
url = "mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true"
url = dj_database_url.parse(url, conn_max_age=conn_max_age)
assert url['CONN_MAX_AGE'] == conn_max_age
assert url["CONN_MAX_AGE"] == conn_max_age
def test_config_conn_max_age_setting(self):
conn_max_age = 600
os.environ[
'DATABASE_URL'
] = 'mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true'
"DATABASE_URL"
] = "mysql://bea6eb025ca0d8:69772142@us-cdbr-east.cleardb.com/heroku_97681db3eff7580?reconnect=true"
url = dj_database_url.config(conn_max_age=conn_max_age)
assert url['CONN_MAX_AGE'] == conn_max_age
assert url["CONN_MAX_AGE"] == conn_max_age
def test_database_url_with_options(self):
# Test full options
os.environ[
'DATABASE_URL'
] = 'postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?sslrootcert=rds-combined-ca-bundle.pem&sslmode=verify-full'
"DATABASE_URL"
] = "postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?sslrootcert=rds-combined-ca-bundle.pem&sslmode=verify-full"
url = dj_database_url.config()
assert url['ENGINE'] == EXPECTED_POSTGRES_ENGINE
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5431
assert url['OPTIONS'] == {
'sslrootcert': 'rds-combined-ca-bundle.pem',
'sslmode': 'verify-full',
assert url["ENGINE"] == EXPECTED_POSTGRES_ENGINE
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5431
assert url["OPTIONS"] == {
"sslrootcert": "rds-combined-ca-bundle.pem",
"sslmode": "verify-full",
}
# Test empty options
os.environ[
'DATABASE_URL'
] = 'postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?'
"DATABASE_URL"
] = "postgres://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5431/d8r82722r2kuvn?"
url = dj_database_url.config()
assert 'OPTIONS' not in url
assert "OPTIONS" not in url
def test_mysql_database_url_with_sslca_options(self):
os.environ[
'DATABASE_URL'
] = 'mysql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:3306/d8r82722r2kuvn?ssl-ca=rds-combined-ca-bundle.pem'
"DATABASE_URL"
] = "mysql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:3306/d8r82722r2kuvn?ssl-ca=rds-combined-ca-bundle.pem"
url = dj_database_url.config()
assert url['ENGINE'] == 'django.db.backends.mysql'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 3306
assert url['OPTIONS'] == {'ssl': {'ca': 'rds-combined-ca-bundle.pem'}}
assert url["ENGINE"] == "django.db.backends.mysql"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 3306
assert url["OPTIONS"] == {"ssl": {"ca": "rds-combined-ca-bundle.pem"}}
# Test empty options
os.environ[
'DATABASE_URL'
] = 'mysql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:3306/d8r82722r2kuvn?'
"DATABASE_URL"
] = "mysql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:3306/d8r82722r2kuvn?"
url = dj_database_url.config()
assert 'OPTIONS' not in url
assert "OPTIONS" not in url
def test_oracle_parsing(self):
url = 'oracle://scott:tiger@oraclehost:1521/hr'
url = "oracle://scott:tiger@oraclehost:1521/hr"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.db.backends.oracle'
assert url['NAME'] == 'hr'
assert url['HOST'] == 'oraclehost'
assert url['USER'] == 'scott'
assert url['PASSWORD'] == 'tiger'
assert url['PORT'] == '1521'
assert url["ENGINE"] == "django.db.backends.oracle"
assert url["NAME"] == "hr"
assert url["HOST"] == "oraclehost"
assert url["USER"] == "scott"
assert url["PASSWORD"] == "tiger"
assert url["PORT"] == "1521"
def test_oracle_gis_parsing(self):
url = 'oraclegis://scott:tiger@oraclehost:1521/hr'
url = "oraclegis://scott:tiger@oraclehost:1521/hr"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.contrib.gis.db.backends.oracle'
assert url['NAME'] == 'hr'
assert url['HOST'] == 'oraclehost'
assert url['USER'] == 'scott'
assert url['PASSWORD'] == 'tiger'
assert url['PORT'] == 1521
assert url["ENGINE"] == "django.contrib.gis.db.backends.oracle"
assert url["NAME"] == "hr"
assert url["HOST"] == "oraclehost"
assert url["USER"] == "scott"
assert url["PASSWORD"] == "tiger"
assert url["PORT"] == 1521
def test_oracle_dsn_parsing(self):
url = (
'oracle://scott:tiger@/'
'(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)'
'(HOST=oraclehost)(PORT=1521)))'
'(CONNECT_DATA=(SID=hr)))'
"oracle://scott:tiger@/"
"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)"
"(HOST=oraclehost)(PORT=1521)))"
"(CONNECT_DATA=(SID=hr)))"
)
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.db.backends.oracle'
assert url['USER'] == 'scott'
assert url['PASSWORD'] == 'tiger'
assert url['HOST'] == ''
assert url['PORT'] == ''
assert url["ENGINE"] == "django.db.backends.oracle"
assert url["USER"] == "scott"
assert url["PASSWORD"] == "tiger"
assert url["HOST"] == ""
assert url["PORT"] == ""
dsn = (
'(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)'
'(HOST=oraclehost)(PORT=1521)))'
'(CONNECT_DATA=(SID=hr)))'
"(DESCRIPTION=(ADDRESS_LIST=(ADDRESS=(PROTOCOL=TCP)"
"(HOST=oraclehost)(PORT=1521)))"
"(CONNECT_DATA=(SID=hr)))"
)
assert url['NAME'] == dsn
assert url["NAME"] == dsn
def test_oracle_tns_parsing(self):
url = 'oracle://scott:tiger@/tnsname'
url = "oracle://scott:tiger@/tnsname"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django.db.backends.oracle'
assert url['USER'] == 'scott'
assert url['PASSWORD'] == 'tiger'
assert url['NAME'] == 'tnsname'
assert url['HOST'] == ''
assert url['PORT'] == ''
assert url["ENGINE"] == "django.db.backends.oracle"
assert url["USER"] == "scott"
assert url["PASSWORD"] == "tiger"
assert url["NAME"] == "tnsname"
assert url["HOST"] == ""
assert url["PORT"] == ""
def test_redshift_parsing(self):
url = 'redshift://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5439/d8r82722r2kuvn?currentSchema=otherschema'
url = "redshift://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com:5439/d8r82722r2kuvn?currentSchema=otherschema"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'django_redshift_backend'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == 5439
assert url['OPTIONS']['options'] == '-c search_path=otherschema'
assert 'currentSchema' not in url['OPTIONS']
assert url["ENGINE"] == "django_redshift_backend"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == 5439
assert url["OPTIONS"]["options"] == "-c search_path=otherschema"
assert "currentSchema" not in url["OPTIONS"]
def test_mssql_parsing(self):
url = 'mssql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com/d8r82722r2kuvn?driver=ODBC Driver 13 for SQL Server'
url = "mssql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com/d8r82722r2kuvn?driver=ODBC Driver 13 for SQL Server"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'sql_server.pyodbc'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == ''
assert url['OPTIONS']['driver'] == 'ODBC Driver 13 for SQL Server'
assert 'currentSchema' not in url['OPTIONS']
assert url["ENGINE"] == "sql_server.pyodbc"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == ""
assert url["OPTIONS"]["driver"] == "ODBC Driver 13 for SQL Server"
assert "currentSchema" not in url["OPTIONS"]
def test_mssql_instance_port_parsing(self):
url = 'mssql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com\\insnsnss:12345/d8r82722r2kuvn?driver=ODBC Driver 13 for SQL Server'
url = "mssql://uf07k1i6d8ia0v:wegauwhgeuioweg@ec2-107-21-253-135.compute-1.amazonaws.com\\insnsnss:12345/d8r82722r2kuvn?driver=ODBC Driver 13 for SQL Server"
url = dj_database_url.parse(url)
assert url['ENGINE'] == 'sql_server.pyodbc'
assert url['NAME'] == 'd8r82722r2kuvn'
assert url['HOST'] == 'ec2-107-21-253-135.compute-1.amazonaws.com\\insnsnss'
assert url['USER'] == 'uf07k1i6d8ia0v'
assert url['PASSWORD'] == 'wegauwhgeuioweg'
assert url['PORT'] == '12345'
assert url['OPTIONS']['driver'] == 'ODBC Driver 13 for SQL Server'
assert 'currentSchema' not in url['OPTIONS']
assert url["ENGINE"] == "sql_server.pyodbc"
assert url["NAME"] == "d8r82722r2kuvn"
assert url["HOST"] == "ec2-107-21-253-135.compute-1.amazonaws.com\\insnsnss"
assert url["USER"] == "uf07k1i6d8ia0v"
assert url["PASSWORD"] == "wegauwhgeuioweg"
assert url["PORT"] == "12345"
assert url["OPTIONS"]["driver"] == "ODBC Driver 13 for SQL Server"
assert "currentSchema" not in url["OPTIONS"]
if __name__ == '__main__':
if __name__ == "__main__":
unittest.main()