home-automation/tests/test_app.py

import sqlite3

import anyio
import pytest
from alembic import command
from fastapi.testclient import TestClient

from app.auth_db import reset_auth_db_caches
from app.config import get_settings
from app.main import create_app
from scripts.app_db_adopt import APP_BASELINE_REVISION, adopt_or_initialize_app_db
from tests.conftest import _make_alembic_config, _make_app_alembic_config, _make_poo_alembic_config


async def _run_lifespan(app) -> None:
    async with app.router.lifespan_context(app):
        return None


def _prepare_app_db(tmp_path) -> str:
    app_database_path = tmp_path / "app_ready.db"
    app_database_url = f"sqlite:///{app_database_path}"
    command.upgrade(_make_app_alembic_config(app_database_url), "head")
    return app_database_url


def test_app_starts(client: TestClient) -> None:
    response = client.get("/", follow_redirects=False)
    assert response.status_code == 303
    assert response.headers["location"] == "/login"


def test_status_endpoint(client: TestClient) -> None:
    response = client.get("/status")
    assert response.status_code == 200
    assert response.json() == {"status": "ok"}


def test_app_start_fails_when_app_db_missing(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:
    poo_database_path = tmp_path / "poo_ready.db"
    location_database_path = tmp_path / "location_ready.db"
    command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head")
    command.upgrade(_make_alembic_config(f"sqlite:///{location_database_path}"), "head")

    monkeypatch.setenv("APP_DATABASE_URL", f"sqlite:///{tmp_path / 'missing_app.db'}")
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_database_path}")
    monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}")
    get_settings.cache_clear()
    reset_auth_db_caches()

    app = create_app()
    with pytest.raises(RuntimeError, match="Run 'python scripts/app_db_adopt.py' first"):
        anyio.run(_run_lifespan, app)

    get_settings.cache_clear()
    reset_auth_db_caches()


def test_app_db_adoption_initializes_new_database(tmp_path) -> None:
    database_url = f"sqlite:///{tmp_path / 'app_init.db'}"

    result = adopt_or_initialize_app_db(database_url)

    assert result == "initialized"
    conn = sqlite3.connect(tmp_path / "app_init.db")
    try:
        revision = conn.execute("SELECT version_num FROM alembic_version").fetchone()[0]
        assert revision == APP_BASELINE_REVISION
        tables = {
            row[0]
            for row in conn.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'"
            ).fetchall()
        }
        assert {"auth_users", "auth_sessions", "app_config", "alembic_version"} <= tables
    finally:
        conn.close()


def test_app_start_seeds_missing_config_from_env_without_overwriting_existing_values(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    app_database_url = _prepare_app_db(tmp_path)
    location_database_path = tmp_path / "location_ready.db"
    poo_database_path = tmp_path / "poo_ready.db"
    command.upgrade(_make_alembic_config(f"sqlite:///{location_database_path}"), "head")
    command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head")

    app_database_path = tmp_path / "app_ready.db"
    conn = sqlite3.connect(app_database_path)
    conn.execute(
        "INSERT INTO app_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
        ("APP_NAME", "Database Owned Name"),
    )
    conn.commit()
    conn.close()

    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    monkeypatch.setenv("APP_NAME", "Bootstrap Name")
    monkeypatch.setenv("HOME_ASSISTANT_BASE_URL", "http://bootstrap-ha.local:8123")
    monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_database_path}")
    monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}")
    get_settings.cache_clear()
    reset_auth_db_caches()

    app = create_app()
    anyio.run(_run_lifespan, app)

    conn = sqlite3.connect(app_database_path)
    try:
        rows = dict(conn.execute("SELECT key, value FROM app_config").fetchall())
    finally:
        conn.close()

    assert rows["APP_NAME"] == "Database Owned Name"
    assert rows["HOME_ASSISTANT_BASE_URL"] == "http://bootstrap-ha.local:8123"
    assert rows["AUTH_SESSION_COOKIE_NAME"] == "home_automation_session"

    get_settings.cache_clear()
    reset_auth_db_caches()


def test_app_start_syncs_app_hostname_from_env_even_when_db_has_old_value(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    app_database_url = _prepare_app_db(tmp_path)
    location_database_path = tmp_path / "location_ready.db"
    poo_database_path = tmp_path / "poo_ready.db"
    command.upgrade(_make_alembic_config(f"sqlite:///{location_database_path}"), "head")
    command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head")

    app_database_path = tmp_path / "app_ready.db"
    conn = sqlite3.connect(app_database_path)
    conn.execute(
        "INSERT INTO app_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)",
        ("APP_HOSTNAME", "old.example.com"),
    )
    conn.commit()
    conn.close()

    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    monkeypatch.setenv("APP_HOSTNAME", "new.example.com")
    monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{location_database_path}")
    monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}")
    get_settings.cache_clear()
    reset_auth_db_caches()

    app = create_app()
    anyio.run(_run_lifespan, app)

    conn = sqlite3.connect(app_database_path)
    try:
        rows = dict(conn.execute("SELECT key, value FROM app_config").fetchall())
    finally:
        conn.close()

    assert rows["APP_HOSTNAME"] == "new.example.com"

    get_settings.cache_clear()
    reset_auth_db_caches()


def test_app_start_fails_when_location_db_missing(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    app_database_url = _prepare_app_db(tmp_path)
    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    poo_database_path = tmp_path / "poo_ready.db"
    command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head")

    monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{tmp_path / 'missing.db'}")
    monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}")
    get_settings.cache_clear()
    reset_auth_db_caches()

    app = create_app()
    with pytest.raises(RuntimeError, match="Run 'python scripts/location_db_adopt.py' first"):
        anyio.run(_run_lifespan, app)

    get_settings.cache_clear()
    reset_auth_db_caches()


def test_app_start_fails_when_location_db_exists_but_is_not_adopted(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    app_database_url = _prepare_app_db(tmp_path)
    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    poo_database_path = tmp_path / "poo_ready.db"
    command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head")

    database_path = tmp_path / "legacy_only.db"
    conn = sqlite3.connect(database_path)
    conn.execute(
        """
        CREATE TABLE location (
          person TEXT NOT NULL,
          datetime TEXT NOT NULL,
          latitude REAL NOT NULL,
          longitude REAL NOT NULL,
          altitude REAL,
          PRIMARY KEY (person, datetime)
        )
        """
    )
    conn.execute("PRAGMA user_version = 2")
    conn.commit()
    conn.close()

    monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{database_path}")
    monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}")
    get_settings.cache_clear()
    reset_auth_db_caches()

    app = create_app()
    with pytest.raises(RuntimeError, match="is not yet Alembic-managed"):
        anyio.run(_run_lifespan, app)

    get_settings.cache_clear()
    reset_auth_db_caches()


def test_app_start_fails_when_location_db_revision_mismatches(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    app_database_url = _prepare_app_db(tmp_path)
    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    poo_database_path = tmp_path / "poo_ready.db"
    command.upgrade(_make_poo_alembic_config(f"sqlite:///{poo_database_path}"), "head")

    database_path = tmp_path / "wrong_revision.db"
    command.upgrade(_make_alembic_config(f"sqlite:///{database_path}"), "head")

    conn = sqlite3.connect(database_path)
    conn.execute("UPDATE alembic_version SET version_num = 'wrong_revision'")
    conn.commit()
    conn.close()

    monkeypatch.setenv("LOCATION_DATABASE_URL", f"sqlite:///{database_path}")
    monkeypatch.setenv("POO_DATABASE_URL", f"sqlite:///{poo_database_path}")
    get_settings.cache_clear()
    reset_auth_db_caches()

    app = create_app()
    with pytest.raises(RuntimeError, match="Location DB revision mismatch"):
        anyio.run(_run_lifespan, app)

    get_settings.cache_clear()
    reset_auth_db_caches()