jfmcarreira
66b72d1348
App checks / changes (push) Canceled after 0s
App checks / tournament (push) Canceled after 0s
App checks / shifts (push) Canceled after 0s
42 lines
1.8 KiB
Python
42 lines
1.8 KiB
Python
from sqlalchemy import URL
|
|
|
|
from crr_common.database import create_sync_engine
|
|
from crr_common.migrations import migration_config
|
|
|
|
|
|
def ini(tmp_path):
|
|
path = tmp_path / "alembic.ini"
|
|
path.write_text("[alembic]\nscript_location = %(here)s/migrations\n")
|
|
return path
|
|
|
|
|
|
def test_configuration_resolves_paths_and_preserves_url_percent_escapes(tmp_path):
|
|
scripts = tmp_path / "migration%files"
|
|
url = "postgresql://reader:p%25ss@localhost/database"
|
|
config = migration_config(ini(tmp_path), scripts, database_url=url)
|
|
assert config.config_file_name == str(tmp_path / "alembic.ini")
|
|
assert config.get_main_option("script_location") == str(scripts)
|
|
assert config.get_main_option("sqlalchemy.url") == url
|
|
assert "connection" not in config.attributes
|
|
|
|
|
|
def test_sqlalchemy_url_objects_are_rendered_without_redacting_connection_credentials(tmp_path):
|
|
url = URL.create("postgresql", username="test-user", password="test%only", host="localhost", database="example")
|
|
config = migration_config(ini(tmp_path), tmp_path / "migrations", database_url=url)
|
|
assert config.get_main_option("sqlalchemy.url") == url.render_as_string(hide_password=False)
|
|
|
|
|
|
def test_configuration_reuses_caller_connection_without_opening_or_committing_it(tmp_path):
|
|
engine = create_sync_engine("sqlite:///:memory:")
|
|
try:
|
|
with engine.connect() as connection:
|
|
transaction = connection.begin()
|
|
config = migration_config(ini(tmp_path), tmp_path / "migrations", connection=connection)
|
|
assert config.attributes["connection"] is connection
|
|
assert connection.in_transaction()
|
|
assert not connection.closed
|
|
assert config.get_main_option("sqlalchemy.url") is None
|
|
transaction.rollback()
|
|
finally:
|
|
engine.dispose()
|