51 lines
1.6 KiB
Python
51 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from cloud.database import CloudDatabase, create_database_engine, normalize_database_url
|
|
|
|
|
|
def test_sqlite_database_owns_working_repository_lifecycle(tmp_path) -> None:
|
|
database = CloudDatabase(f"sqlite:///{(tmp_path / 'cloud.sqlite3').as_posix()}")
|
|
|
|
database.repository.health_check()
|
|
assert database.engine.dialect.name == "sqlite"
|
|
|
|
database.close()
|
|
|
|
|
|
def test_in_memory_sqlite_uses_one_shared_pool() -> None:
|
|
database = CloudDatabase("sqlite:///:memory:")
|
|
try:
|
|
database.repository.health_check()
|
|
assert database.engine.pool.__class__.__name__ == "StaticPool"
|
|
finally:
|
|
database.close()
|
|
|
|
|
|
def test_postgresql_url_uses_psycopg_dialect_without_connecting() -> None:
|
|
database = CloudDatabase(
|
|
"postgresql://user:secret@localhost/cloud",
|
|
create_schema=False,
|
|
)
|
|
try:
|
|
assert database.engine.dialect.name == "postgresql"
|
|
assert database.engine.dialect.driver == "psycopg"
|
|
finally:
|
|
database.close()
|
|
|
|
|
|
def test_normalize_database_url_preserves_explicit_dialects() -> None:
|
|
assert (
|
|
normalize_database_url("postgresql://db/cloud")
|
|
== "postgresql+psycopg://db/cloud"
|
|
)
|
|
assert normalize_database_url("sqlite:///cloud.db") == "sqlite:///cloud.db"
|
|
|
|
|
|
def test_sqlite_engine_allows_cross_thread_fastapi_usage() -> None:
|
|
engine = create_database_engine("sqlite:///:memory:")
|
|
try:
|
|
assert engine.url.drivername == "sqlite"
|
|
assert engine.pool.__class__.__name__ == "StaticPool"
|
|
finally:
|
|
engine.dispose()
|