All checks were successful
Build and Publish Container Image / build-and-push (push) Successful in 41s
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a02b22-9638-76d2-bbff-b7ea1770b118
57 lines
1.7 KiB
Python
57 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
from sbom_nexus import database
|
|
|
|
|
|
def test_migration_role_commits_autobegin_before_alembic_transaction() -> None:
|
|
events: list[str] = []
|
|
|
|
class Connection:
|
|
def exec_driver_sql(self, statement: str) -> None:
|
|
events.append(statement)
|
|
|
|
def commit(self) -> None:
|
|
events.append("COMMIT")
|
|
|
|
database.assume_migration_role(Connection(), "sbom_nexus_owner")
|
|
|
|
assert events == ['SET ROLE "sbom_nexus_owner"', "COMMIT"]
|
|
|
|
|
|
def test_dynamic_connection_factory_rereads_mounted_url(monkeypatch, tmp_path: Path) -> None:
|
|
secret = tmp_path / "url"
|
|
observed: list[str] = []
|
|
|
|
def fake_connect(dsn: str):
|
|
observed.append(dsn)
|
|
return object()
|
|
|
|
monkeypatch.setattr(database.psycopg, "connect", fake_connect)
|
|
factory = database._dynamic_connection_factory(secret)
|
|
secret.write_text("postgresql://first@db/sbom\n", encoding="utf-8")
|
|
factory()
|
|
secret.write_text("postgresql://second@db/sbom\n", encoding="utf-8")
|
|
factory()
|
|
|
|
assert observed == [
|
|
"postgresql://first@db/sbom",
|
|
"postgresql://second@db/sbom",
|
|
]
|
|
|
|
|
|
def test_dynamic_engine_keeps_credential_out_of_engine_url(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
secret = tmp_path / "url"
|
|
secret.write_text("postgresql://username:password@db/sbom\n", encoding="utf-8")
|
|
monkeypatch.setenv("SBOM_NEXUS_DATABASE_POOL_RECYCLE_SECONDS", "60")
|
|
|
|
engine = database.create_database_engine(
|
|
secret.read_text().strip(), database_url_file=secret
|
|
)
|
|
|
|
assert "username" not in str(engine.url)
|
|
assert "password" not in str(engine.url)
|
|
assert engine.pool._recycle == 60
|