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