home-automation/tests/test_app.py

import sqlite3

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

from app.db import reset_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_app_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:
    # With SPA enabled, GET / is served by the catch-all and returns index.html (200).
    # Without SPA (e.g. SPA_DIST_DIR points to empty dir), it returns 404.
    # Either way the app started successfully — just assert it is not a server error.
    response = client.get("/", follow_redirects=False)
    assert response.status_code in (200, 404)


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:
    missing_app_path = tmp_path / "missing_app.db"

    monkeypatch.setenv("APP_DATABASE_URL", f"sqlite:///{missing_app_path}")
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    get_settings.cache_clear()
    reset_db_caches()

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

    assert not missing_app_path.exists()

    get_settings.cache_clear()
    reset_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)

    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")
    get_settings.cache_clear()
    reset_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"
    assert rows["MQTT_CLIENT_ID"] == "home-automation"

    get_settings.cache_clear()
    reset_db_caches()


def test_startup_initializes_fresh_legacy_discovery_cleanup_ledger(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    """The cleanup ledger exists before the expose UI can write its first toggle."""
    import app.main as main

    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")
    get_settings.cache_clear()
    reset_db_caches()

    main.ensure_auth_db_ready()

    conn = sqlite3.connect(tmp_path / "app_ready.db")
    try:
        value = conn.execute(
            "SELECT value FROM app_config WHERE key = ?",
            ("HA_DISCOVERY_LEGACY_THERMAL_CLEANUP_V1",),
        ).fetchone()[0]
    finally:
        conn.close()
    assert value == '{"complete": true, "inventory": [], "topics": []}'

    get_settings.cache_clear()
    reset_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)

    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")
    get_settings.cache_clear()
    reset_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_db_caches()


def test_lifespan_schedules_immediate_and_hourly_tibber_refresh(tmp_path, monkeypatch: pytest.MonkeyPatch) -> None:
    """The single Tibber interval job starts immediately while retaining its hourly trigger."""
    import app.main as main

    app_database_url = _prepare_app_db(tmp_path)
    added_jobs = []

    class _Scheduler:
        def __init__(self, **_kwargs): pass
        def add_job(self, func, **kwargs): added_jobs.append((func, kwargs))
        def start(self): pass
        def shutdown(self, **_kwargs): pass

    monkeypatch.setenv("APP_DATABASE_URL", app_database_url)
    monkeypatch.setenv("AUTH_BOOTSTRAP_USERNAME", "admin")
    monkeypatch.setenv("AUTH_BOOTSTRAP_PASSWORD", "test-password")
    monkeypatch.setattr(main, "BackgroundScheduler", _Scheduler)
    monkeypatch.setattr(main.mqtt_manager, "connect", lambda _settings: None)
    monkeypatch.setattr(main.mqtt_manager, "disconnect", lambda: None)
    monkeypatch.setattr(main, "apply_dsmr_subscription", lambda _settings: None)
    monkeypatch.setattr(main.warmtelink_worker_manager, "start", lambda: None)
    monkeypatch.setattr(main.warmtelink_worker_manager, "shutdown", lambda: None)
    get_settings.cache_clear()
    reset_db_caches()

    anyio.run(_run_lifespan, create_app())

    tibber_jobs = [kwargs for func, kwargs in added_jobs if func is main._run_scheduled_tibber_refresh]
    assert len(tibber_jobs) == 1
    assert tibber_jobs[0]["id"] == "tibber-refresh"
    assert tibber_jobs[0]["max_instances"] == 1
    assert tibber_jobs[0]["next_run_time"] is not None

    get_settings.cache_clear()
    reset_db_caches()