# This file is part of the django-environ. # # Copyright (c) 2021-2024, Serghei Iakovlev # Copyright (c) 2013-2021, Daniele Faraglia # # For the full copyright and license information, please view # the LICENSE.txt file that was distributed with this source code. import warnings import pytest from environ import Env from environ.compat import DJANGO_POSTGRES @pytest.mark.parametrize( 'url,engine,name,host,user,passwd,port', [ # postgres://user:password@host:port/dbname ('postgres://enigma:secret@example.com:5431/dbname', DJANGO_POSTGRES, 'dbname', 'example.com', 'enigma', 'secret', 5431), # postgres://path/dbname ('postgres:////var/run/postgresql/dbname', DJANGO_POSTGRES, 'dbname', '/var/run/postgresql', '', '', ''), # postgis://user:password@host:port/dbname ('postgis://enigma:secret@example.com:5431/dbname', 'django.contrib.gis.db.backends.postgis', 'dbname', 'example.com', 'enigma', 'secret', 5431), # postgres://host,host,host/dbname ('postgres://node1,node2,node3/db', DJANGO_POSTGRES, 'db', 'node1,node2,node3', '', '', '' ), # cockroachdb://username:secret@test.example.com:26258/dbname ('cockroachdb://username:secret@test.example.com:26258/dbname', 'django_cockroachdb', 'dbname', 'test.example.com', 'username', 'secret', 26258), # mysqlgis://user:password@host:port/dbname ('mysqlgis://enigma:secret@example.com:5431/dbname', 'django.contrib.gis.db.backends.mysql', 'dbname', 'example.com', 'enigma', 'secret', 5431), # mysql://user:password@host/dbname?options ('mysql://enigma:secret@reconnect.com/dbname?reconnect=true', 'django.db.backends.mysql', 'dbname', 'reconnect.com', 'enigma', 'secret', ''), # mysql://user@host/dbname ('mysql://enigma@localhost/dbname', 'django.db.backends.mysql', 'dbname', 'localhost', 'enigma', '', ''), # sqlite:// ('sqlite://', 'django.db.backends.sqlite3', ':memory:', '', '', '', ''), # sqlite:////absolute/path/to/db/file ('sqlite:////full/path/to/your/file.sqlite', 'django.db.backends.sqlite3', '/full/path/to/your/file.sqlite', '', '', '', ''), # sqlite://:memory: ('sqlite://:memory:', 'django.db.backends.sqlite3', ':memory:', '', '', '', ''), # ldap://user:password@host ('ldap://cn=admin,dc=nodomain,dc=org:secret@example.com', 'ldapdb.backends.ldap', 'ldap://example.com', 'example.com', 'cn=admin,dc=nodomain,dc=org', 'secret', ''), # mysql://user:password@host/dbname ('mssql://enigma:secret@example.com/dbname' '?driver=ODBC Driver 13 for SQL Server', 'mssql', 'dbname', 'example.com', 'enigma', 'secret', ''), # mysql://user:password@host:port/dbname ('mssql://enigma:secret@amazonaws.com\\insnsnss:12345/dbname' '?driver=ODBC Driver 13 for SQL Server', 'mssql', 'dbname', 'amazonaws.com\\insnsnss', 'enigma', 'secret', 12345), # mysql://user:password@host:port/dbname ('mysql://enigma:><{~!@#$%^&*}[]@example.com:1234/dbname', 'django.db.backends.mysql', 'dbname', 'example.com', 'enigma', '><{~!@#$%^&*}[]', 1234), # mysql://user:password@host/dbname ('mysql://enigma:]password]@example.com/dbname', 'django.db.backends.mysql', 'dbname', 'example.com', 'enigma', ']password]', ''), ], ids=[ 'postgres', 'postgres_unix_domain', 'postgis', 'postgres_no_ports', 'cockroachdb', 'mysqlgis', 'cleardb', 'mysql_no_password', 'sqlite_empty', 'sqlite_file', 'sqlite_memory', 'ldap', 'mssql', 'mssql_port', 'mysql_password_special_chars', 'mysql_invalid_ipv6_password', ], ) def test_db_parsing(url, engine, name, host, user, passwd, port): config = Env.db_url_config(url) assert config['ENGINE'] == engine assert config['NAME'] == name if url != 'sqlite://:memory:': if config['PORT']: assert config['PORT'] == port assert config['PASSWORD'] == passwd assert config['USER'] == user assert config['HOST'] == host if engine == 'sql_server.pyodbc': assert config['OPTIONS'] == {'driver': 'ODBC Driver 13 for SQL Server'} if host == 'reconnect.com': assert config['OPTIONS'] == {'reconnect': 'true'} def test_custom_db_engine(): """Override ENGINE determined from schema.""" env_url = 'postgres://enigma:secret@example.com:5431/dbname' engine = 'mypackage.backends.whatever' url = Env.db_url_config(env_url, engine=engine) assert url['ENGINE'] == engine def test_postgres_complex_db_name_parsing(): """Make sure we can use complex postgres host.""" env_url = ( 'postgres://user:password@//cloudsql/' 'project-1234:us-central1:instance/dbname' ) url = Env.db_url_config(env_url) assert url['ENGINE'] == DJANGO_POSTGRES assert url['HOST'] == '/cloudsql/project-1234:us-central1:instance' assert url['NAME'] == 'dbname' assert url['USER'] == 'user' assert url['PASSWORD'] == 'password' assert url['PORT'] == '' @pytest.mark.parametrize( 'scheme', ['postgres', 'postgresql', 'psql', 'pgsql', 'postgis'], ) def test_postgres_like_scheme_parsing(scheme): """Verify all the postgres-like schemes parsed the same as postgres.""" env_url1 = ( 'postgres://user:password@//cloudsql/' 'project-1234:us-central1:instance/dbname' ) env_url2 = ( '{}://user:password@//cloudsql/' 'project-1234:us-central1:instance/dbname' ).format(scheme) url1 = Env.db_url_config(env_url1) url2 = Env.db_url_config(env_url2) assert url2['NAME'] == url1['NAME'] assert url2['PORT'] == url1['PORT'] assert url2['PASSWORD'] == url1['PASSWORD'] assert url2['USER'] == url1['USER'] assert url2['HOST'] == url1['HOST'] if scheme == 'postgis': assert url2['ENGINE'] == 'django.contrib.gis.db.backends.postgis' else: assert url2['ENGINE'] == url1['ENGINE'] def test_memory_sqlite_url_warns_about_netloc(recwarn): warnings.simplefilter("always") url = 'sqlite://missing-slash-path' url = Env.db_url_config(url) assert len(recwarn) == 1 assert recwarn.pop(UserWarning) assert url['ENGINE'] == 'django.db.backends.sqlite3' assert url['NAME'] == ':memory:' def test_database_options_parsing(): url = 'postgres://user:pass@host:1234/dbname?conn_max_age=600' url = Env.db_url_config(url) assert url['CONN_MAX_AGE'] == 600 url = ('postgres://user:pass@host:1234/dbname?' 'conn_max_age=None&autocommit=True&atomic_requests=False') url = Env.db_url_config(url) assert url['CONN_MAX_AGE'] is None assert url['AUTOCOMMIT'] is True assert url['ATOMIC_REQUESTS'] is False url = ('mysql://user:pass@host:1234/dbname?init_command=SET ' 'storage_engine=INNODB') url = Env.db_url_config(url) assert url['OPTIONS'] == { 'init_command': 'SET storage_engine=INNODB', }