hw/tests/test_app_bootstrap.py

from __future__ import annotations

from pathlib import Path

from sqlalchemy import Table, UniqueConstraint
from sqlalchemy.dialects import mysql, sqlite

from app import create_app
from app.extensions import db
from app.models.base import ID_TYPE


EXPECTED_TABLES = {
    "accounts",
    "audit_logs",
    "gift_records",
    "household_members",
    "households",
    "option_items",
    "share_links",
}


def get_table(table_name: str) -> Table:
    return db.metadata.tables[table_name]


def test_create_app_uses_testing_config() -> None:
    app = create_app("testing")

    assert app.config["TESTING"] is True
    assert app.config["SQLALCHEMY_DATABASE_URI"] == "sqlite:///:memory:"


def test_healthcheck_route() -> None:
    app = create_app("testing")
    client = app.test_client()

    response = client.get("/health")

    assert response.status_code == 200
    assert response.get_json() == {
        "project": "HappyWedding",
        "short_name": "hw",
        "status": "ok",
    }


def test_auth_related_blueprints_are_registered() -> None:
    app = create_app("testing")

    assert {"auth", "admin", "main", "quick_entry", "health"}.issubset(app.blueprints.keys())


def test_metadata_contains_expected_tables() -> None:
    app = create_app("testing")

    with app.app_context():
        db.create_all()
        table_names = set(db.metadata.tables.keys())

    assert EXPECTED_TABLES.issubset(table_names)


def test_household_table_contains_core_columns() -> None:
    household_columns = get_table("households").columns.keys()

    assert "household_code" in household_columns
    assert "tag_option_ids_json" in household_columns
    assert "total_gift_amount" in household_columns
    assert "version" in household_columns
    assert "deleted_at" in household_columns


def test_account_table_contains_tracking_columns() -> None:
    account_columns = get_table("accounts").columns.keys()

    assert "username" in account_columns
    assert "password_hash" in account_columns
    assert "role" in account_columns
    assert "status" in account_columns
    assert "created_by" in account_columns
    assert "updated_by" in account_columns


def test_relationship_foreign_keys_exist() -> None:
    household_fks = {fk.parent.name for fk in get_table("households").foreign_keys}
    gift_record_fks = {fk.parent.name for fk in get_table("gift_records").foreign_keys}
    member_fks = {fk.parent.name for fk in get_table("household_members").foreign_keys}
    option_item_fks = {fk.parent.name for fk in get_table("option_items").foreign_keys}
    audit_log_fks = {fk.parent.name for fk in get_table("audit_logs").foreign_keys}

    assert {
        "relation_category_option_id",
        "relation_detail_option_id",
        "gift_method_option_id",
        "gift_scene_option_id",
        "created_by",
        "updated_by",
    }.issubset(household_fks)
    assert {"household_id", "method_option_id", "scene_option_id", "created_by", "updated_by"}.issubset(
        gift_record_fks,
    )
    assert {"household_id", "created_by", "updated_by"}.issubset(member_fks)
    assert {"parent_id", "created_by", "updated_by"}.issubset(option_item_fks)
    assert {"actor_user_id"}.issubset(audit_log_fks)


def test_unique_constraints_exist_on_key_tables() -> None:
    account_unique_columns = [
        list(constraint.columns.keys())
        for constraint in get_table("accounts").constraints
        if isinstance(constraint, UniqueConstraint)
    ]
    household_unique_columns = [
        list(constraint.columns.keys())
        for constraint in get_table("households").constraints
        if isinstance(constraint, UniqueConstraint)
    ]
    option_item_unique_columns = [
        list(constraint.columns.keys())
        for constraint in get_table("option_items").constraints
        if isinstance(constraint, UniqueConstraint)
    ]

    assert ["username"] in account_unique_columns
    assert ["household_code"] in household_unique_columns
    assert ["option_group", "option_code"] in option_item_unique_columns


def test_id_type_compiles_for_sqlite_and_mysql() -> None:
    assert ID_TYPE.compile(dialect=sqlite.dialect()) == "INTEGER"
    assert ID_TYPE.compile(dialect=mysql.dialect()) == "BIGINT"


def test_sqlite_connections_apply_runtime_pragmas(monkeypatch, tmp_path: Path) -> None:
    db_path = tmp_path / "pragma-check.db"
    monkeypatch.setenv("DATABASE_URL", f"sqlite:///{db_path}")

    app = create_app("development")

    with app.app_context():
        db.create_all()
        connection = db.session.connection()
        journal_mode = connection.exec_driver_sql("PRAGMA journal_mode;").scalar()
        busy_timeout = connection.exec_driver_sql("PRAGMA busy_timeout;").scalar()
        temp_store = connection.exec_driver_sql("PRAGMA temp_store;").scalar()
        synchronous = connection.exec_driver_sql("PRAGMA synchronous;").scalar()
        foreign_keys = connection.exec_driver_sql("PRAGMA foreign_keys;").scalar()

    assert str(journal_mode).lower() == "wal"
    assert busy_timeout == 5000
    assert temp_store == 2
    assert synchronous == 1
    assert foreign_keys == 1