Kairo-Jinkendo/backend/tests/test_migrations.py

"""Migration runner tests (require PostgreSQL)."""

import os
from pathlib import Path

import psycopg2
import pytest

import run_migrations


def _db_available() -> bool:
    try:
        conn = psycopg2.connect(
            host=os.getenv("DB_HOST", "localhost"),
            port=os.getenv("DB_PORT", "5432"),
            dbname=os.getenv("DB_NAME", "kairo_dev"),
            user=os.getenv("DB_USER", "kairo_dev"),
            password=os.getenv("DB_PASSWORD", "dev_password"),
        )
        conn.close()
        return True
    except psycopg2.OperationalError:
        return False


pytestmark = pytest.mark.skipif(not _db_available(), reason="PostgreSQL nicht erreichbar")


def test_migration_runner_finds_migrations():
    migrations_dir = run_migrations.migrations_directory()
    files = run_migrations.migration_files(migrations_dir)
    names = [name for name, _ in files]
    assert "001_init_core" in names
    assert "002_auth_identity_tenant_actor" in names


def test_migration_runner_is_idempotent():
    assert run_migrations.main() == 0
    assert run_migrations.main() == 0

    conn = run_migrations.connect_with_retry(max_retries=5)
    executed = run_migrations.executed_migrations(conn)
    conn.close()
    assert "001_init_core" in executed
    assert "002_auth_identity_tenant_actor" in executed


def test_core_table_exists():
    conn = run_migrations.connect_with_retry(max_retries=5)
    with conn.cursor() as cur:
        cur.execute(
            """
            SELECT COUNT(*) FROM information_schema.tables
            WHERE table_schema = 'public' AND table_name = 'kairo_app_meta'
            """
        )
        assert cur.fetchone()[0] == 1
    conn.close()