X-Financial/server/tests/test_maintenance_database_target.py

from __future__ import annotations

import pytest

from app.db.maintenance_database_target import (
    MaintenanceDatabaseTargetError,
    is_disposable_maintenance_target,
    parse_maintenance_database_target,
    validate_maintenance_database_target,
)


def assert_target_error(code: str, callback) -> None:
    with pytest.raises(MaintenanceDatabaseTargetError) as exc_info:
        callback()
    assert exc_info.value.code == code


def test_parse_target_supports_hostname_and_hides_password() -> None:
    target = parse_maintenance_database_target(
        "postgresql+psycopg://finance:super-secret@db.internal:5544/x_financial"
        "?application_name=backfill",
        expected_host="DB.INTERNAL.",
        expected_database="x_financial",
    )

    assert target.host == "db.internal"
    assert target.port == 5544
    assert target.database == "x_financial"
    assert target.username == "finance"
    assert target.exact_target == "db.internal:5544/x_financial"
    assert "super-secret" not in target.sanitized_url
    assert "***" in target.sanitized_url
    assert "application_name=backfill" in target.sanitized_url
    assert target.is_disposable is False


def test_parse_target_supports_ipv4_and_defaults_postgresql_port() -> None:
    target = parse_maintenance_database_target(
        "postgresql://finance:password@127.0.0.1/x_financial",
        expected_host="127.0.0.1",
        expected_database="x_financial",
    )

    assert target.port == 5432
    assert target.exact_target == "127.0.0.1:5432/x_financial"
    assert "password" not in target.sanitized_url


def test_parse_target_redacts_sensitive_query_values() -> None:
    target = parse_maintenance_database_target(
        "postgresql://finance:authority-secret@db.internal/x_financial"
        "?sslpassword=query-secret&application_name=safe-name",
        expected_host="db.internal",
        expected_database="x_financial",
    )

    assert "authority-secret" not in target.sanitized_url
    assert "query-secret" not in target.sanitized_url
    assert "application_name=safe-name" in target.sanitized_url


@pytest.mark.parametrize(
    "routing_query",
    [
        "host=x-financial-local-postgres",
        "port=6432",
        "dbname=production",
        "user=other-user",
        "service=production-service",
        "options=-csearch_path%3Dother_schema",
    ],
)
def test_parse_target_rejects_query_parameters_that_can_override_route(
    routing_query: str,
) -> None:
    assert_target_error(
        "database_routing_query_forbidden",
        lambda: parse_maintenance_database_target(
            f"postgresql://finance:password@migration-probe/migration_probe?{routing_query}",
            expected_host="migration-probe",
            expected_database="migration_probe",
        ),
    )


@pytest.mark.parametrize(
    ("database_url", "code"),
    [
        ("", "database_url_required"),
        ("not a database url", "invalid_database_url"),
        ("sqlite:///tmp/test.db", "postgresql_required"),
        ("postgresql:///x_financial", "database_host_required"),
        ("postgresql://finance@db.internal", "database_name_required"),
    ],
)
def test_parse_target_rejects_invalid_or_incomplete_urls(database_url: str, code: str) -> None:
    assert_target_error(
        code,
        lambda: parse_maintenance_database_target(
            database_url,
            expected_host="db.internal",
            expected_database="x_financial",
        ),
    )


@pytest.mark.parametrize(
    ("expected_host", "expected_database", "code"),
    [
        ("", "x_financial", "expected_host_required"),
        ("other-db", "x_financial", "expected_host_mismatch"),
        ("db.internal", "", "expected_database_required"),
        ("db.internal", "other_database", "expected_database_mismatch"),
    ],
)
def test_parse_target_rejects_expected_target_mismatches(
    expected_host: str,
    expected_database: str,
    code: str,
) -> None:
    assert_target_error(
        code,
        lambda: parse_maintenance_database_target(
            "postgresql://finance:password@db.internal/x_financial",
            expected_host=expected_host,
            expected_database=expected_database,
        ),
    )


@pytest.mark.parametrize(
    ("host", "database", "expected"),
    [
        ("migration-probe", "migration_probe", True),
        ("x-financial-migration-probe-123", "migration_probe_clone_123", True),
        ("disposable-probe-7", "disposable_probe_7", True),
        ("x-financial-local-postgres", "migration_probe", False),
        ("migration-probe-7", "x_financial", False),
    ],
)
def test_disposable_target_requires_markers_on_host_and_database(
    host: str,
    database: str,
    expected: bool,
) -> None:
    assert is_disposable_maintenance_target(host=host, database=database) is expected


def test_dry_run_allows_non_disposable_target_without_apply_override() -> None:
    target = validate_maintenance_database_target(
        "postgresql://finance:password@db.internal/x_financial",
        expected_host="db.internal",
        expected_database="x_financial",
        apply=False,
    )

    assert target.is_disposable is False


def test_apply_allows_disposable_target_without_non_disposable_override() -> None:
    target = validate_maintenance_database_target(
        "postgresql://finance:password@x-financial-migration-probe-1/migration_probe_1",
        expected_host="x-financial-migration-probe-1",
        expected_database="migration_probe_1",
        apply=True,
    )

    assert target.is_disposable is True


def test_apply_rejects_wrong_optional_confirmation_for_disposable_target() -> None:
    assert_target_error(
        "confirm_target_mismatch",
        lambda: validate_maintenance_database_target(
            "postgresql://finance:password@migration-probe/migration_probe",
            expected_host="migration-probe",
            expected_database="migration_probe",
            apply=True,
            confirm_target="other:5432/migration_probe",
        ),
    )


def test_apply_rejects_non_disposable_target_without_explicit_override() -> None:
    assert_target_error(
        "non_disposable_apply_forbidden",
        lambda: validate_maintenance_database_target(
            "postgresql://finance:password@db.internal/x_financial",
            expected_host="db.internal",
            expected_database="x_financial",
            apply=True,
            confirm_target="db.internal:5432/x_financial",
        ),
    )


@pytest.mark.parametrize("confirm_target", [None, "", "db.internal:5432/other_database"])
def test_apply_rejects_non_disposable_target_without_exact_confirmation(
    confirm_target: str | None,
) -> None:
    assert_target_error(
        "confirm_target_mismatch",
        lambda: validate_maintenance_database_target(
            "postgresql://finance:password@db.internal/x_financial",
            expected_host="db.internal",
            expected_database="x_financial",
            apply=True,
            allow_non_disposable=True,
            confirm_target=confirm_target,
        ),
    )


def test_apply_allows_non_disposable_target_with_override_and_exact_confirmation() -> None:
    target = validate_maintenance_database_target(
        "postgresql://finance:password@db.internal/x_financial",
        expected_host="db.internal",
        expected_database="x_financial",
        apply=True,
        allow_non_disposable=True,
        confirm_target="db.internal:5432/x_financial",
    )

    assert target.exact_target == "db.internal:5432/x_financial"