gendesign/backend/tests/sql/test_2464_default_profile_unique.py

"""У пользователя не может быть двух дефолтных профилей весов (#2464).

`create_profile`/`update_profile` делают «снять is_default у всех → поставить новому»
двумя отдельными операторами. Между ними инвариант нарушен, и при одновременных запросах
у пользователя может оказаться ДВА профиля с `is_default=TRUE`. Читающий `_SELECT_DEFAULT`
брал `LIMIT 1` без `ORDER BY` — выбор молча перескакивал между ними.

Правка из двух рубежей:
  • миграция 190 — частичный уникальный индекс «не более одного дефолта на пользователя»;
  • `ORDER BY id ASC` в `_SELECT_DEFAULT` — детерминированный выбор, если индекс снимут.

Проверяется ПОВЕДЕНИЕ на живом Postgres: вторая установка дефолта обязана отвергаться
базой. Тест герметичный — таблица создаётся временной в своей же сессии.
"""

from __future__ import annotations

import os

os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")

from pathlib import Path

import pytest
from sqlalchemy import create_engine, text
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import sessionmaker

_MIGRATION = (
    Path(__file__).resolve().parents[3]
    / "data"
    / "sql"
    / "190_user_weight_profiles_one_default.sql"
)


def _dsn() -> str:
    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
        "DATABASE_URL", "postgresql+psycopg://gendesign@localhost:15432/gendesign"
    )
    return (
        raw
        if raw.startswith("postgresql+")
        else raw.replace("postgresql://", "postgresql+psycopg://")
    )


def _db_reachable() -> tuple[bool, str]:
    try:
        eng = create_engine(_dsn(), connect_args={"connect_timeout": 3})
        with eng.connect() as c:
            c.execute(text("SELECT 1"))
        return True, ""
    except Exception as exc:
        return False, str(exc)


_DB_OK, _DB_ERR = _db_reachable()
pytestmark = pytest.mark.skipif(not _DB_OK, reason=f"Postgres недоступен: {_DB_ERR}")

_TABLE = """
CREATE TEMP TABLE user_weight_profiles (
    id bigserial PRIMARY KEY,
    user_id text NOT NULL,
    profile_name text,
    weights jsonb,
    is_default boolean DEFAULT false,
    description text
) ON COMMIT DROP;
"""


@pytest.fixture
def db():
    engine = create_engine(_dsn())
    session = sessionmaker(bind=engine)()
    try:
        session.execute(text(_TABLE))
        n = session.execute(text("SELECT count(*) FROM user_weight_profiles")).scalar()
        assert n == 0, f"запрос попал НЕ во временную таблицу ({n} строк)"
        yield session
    finally:
        session.rollback()
        session.close()
        engine.dispose()


def _apply_migration(db) -> None:
    sql = _MIGRATION.read_text()
    body = sql[sql.index("BEGIN;") + len("BEGIN;") : sql.rindex("COMMIT;")]
    # Комментарии снимаем ДО разбиения: «;» внутри комментария разрезала бы SQL.
    code = "\n".join(
        ln for ln in body.splitlines() if ln.strip() and not ln.lstrip().startswith("--")
    )
    for chunk in code.split(";"):
        if chunk.strip():
            db.execute(text(chunk))


def _add(db, *, user: str, name: str, default: bool) -> None:
    db.execute(
        text(
            "INSERT INTO user_weight_profiles (user_id, profile_name, is_default)"
            " VALUES (:u, :n, :d)"
        ),
        {"u": user, "n": name, "d": default},
    )


def test_second_default_is_rejected_by_the_database(db) -> None:
    """Второй дефолт у того же пользователя обязан отвергаться.

    Без миграции такая вставка проходит, и у пользователя оказывается два дефолта.
    """
    _apply_migration(db)
    _add(db, user="u1", name="первый", default=True)

    with pytest.raises(IntegrityError):
        _add(db, user="u1", name="второй", default=True)


def test_without_migration_two_defaults_slip_through(db) -> None:
    """Фальсификация: без индекса два дефолта вставляются молча.

    Без этой проверки зелёный тест выше неотличим от «оно и так не вставлялось».
    """
    _add(db, user="u1", name="первый", default=True)
    _add(db, user="u1", name="второй", default=True)
    n = db.execute(
        text("SELECT count(*) FROM user_weight_profiles WHERE user_id='u1' AND is_default")
    ).scalar()
    assert n == 2, f"без индекса дефолтов {n}, ожидалось 2 — тест выше ничего не доказывает"


def test_reassigning_default_still_works(db) -> None:
    """Контроль: штатная переустановка дефолта проходит.

    Порядок в коде — сначала снять у всех, потом поставить новому. Индекс не должен
    этому мешать, иначе пользователь не сможет сменить дефолтный профиль.
    """
    _apply_migration(db)
    _add(db, user="u1", name="первый", default=True)
    _add(db, user="u1", name="второй", default=False)

    db.execute(text("UPDATE user_weight_profiles SET is_default = FALSE WHERE user_id='u1'"))
    db.execute(
        text(
            "UPDATE user_weight_profiles SET is_default = TRUE"
            " WHERE user_id='u1' AND profile_name='второй'"
        )
    )
    name = db.execute(
        text("SELECT profile_name FROM user_weight_profiles WHERE user_id='u1' AND is_default")
    ).scalar()
    assert name == "второй"


def test_different_users_keep_their_own_defaults(db) -> None:
    """Контроль: индекс не мешает разным пользователям иметь свой дефолт."""
    _apply_migration(db)
    _add(db, user="u1", name="a", default=True)
    _add(db, user="u2", name="b", default=True)
    n = db.execute(text("SELECT count(*) FROM user_weight_profiles WHERE is_default")).scalar()
    assert n == 2