home-automation/tests/test_app.py

import sqlite3

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

from app.config import get_settings
from app.main import create_app
from tests.conftest import _make_alembic_config, _make_poo_alembic_config


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


def test_app_starts(client: TestClient) -> None:
    response = client.get("/")
    assert response.status_code == 200


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_location_db_missing(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    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()

    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()


def test_app_start_fails_when_location_db_exists_but_is_not_adopted(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    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()

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

    get_settings.cache_clear()


def test_app_start_fails_when_location_db_revision_mismatches(
    tmp_path, monkeypatch: pytest.MonkeyPatch
) -> None:
    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()

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

    get_settings.cache_clear()