SQL con parámetros y el mock boundary

"""Settings: the ONLY module that reads environment variables (as in m-b-core)."""

import os


class Settings:
    ENV: str = os.getenv("ENV", "dev")
    LEGACY_DB_URL: str = os.getenv("LEGACY_DB_URL", "sqlite+aiosqlite:///./legacy.db")


settings = Settings()

"""Read access to the legacy DB.

In m-b-core this module wraps aiomysql (legacy MySQL) / asyncpg (Postgres replica).
Here it wraps SQLite via aiosqlite so the practices run on any laptop, but the
contract is the same: `legacy_db_read(sql, params) -> list[dict]`, SQL always
with named parameters (`:name`), never interpolated values.
"""

from __future__ import annotations

from typing import Any

from sqlalchemy import text
from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine

from shared.settings import settings

_engine: AsyncEngine | None = None


def get_engine() -> AsyncEngine:
    global _engine
    if _engine is None:
        _engine = create_async_engine(settings.LEGACY_DB_URL)
    return _engine


async def legacy_db_read(sql: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
    """Run one read-only query and return rows as dicts. The default for reads."""
    async with get_engine().connect() as conn:
        result = await conn.execute(text(sql), params or {})
        return [dict(row) for row in result.mappings().all()]

from .legacy_db import legacy_db_read

__all__ = ["legacy_db_read"]

"""Create the legacy tables in SQLite and load the practice dataset."""

import asyncio
from typing import Any

from sqlalchemy import text

from shared.legacy_db.legacy_db import get_engine

DDL = [
    "DROP TABLE IF EXISTS Local",
    "DROP TABLE IF EXISTS Ubigeo",
    "DROP TABLE IF EXISTS Reserva",
    "DROP TABLE IF EXISTS Mesa",
    """CREATE TABLE Local (Id INTEGER PRIMARY KEY, Nombre TEXT, Pais TEXT, Depa TEXT,
       EstadoId INTEGER, ZonaHoraria TEXT)""",
    "CREATE TABLE Ubigeo (Pais TEXT, Depa TEXT, ZonaHoraria TEXT, EstadoId INTEGER)",
    """CREATE TABLE Reserva (Id INTEGER PRIMARY KEY, LocalId INTEGER, FechaReserva TEXT,
       Horario TEXT, EstadoId INTEGER, Pax INTEGER)""",
    "CREATE TABLE Mesa (Id INTEGER PRIMARY KEY, LocalId INTEGER, Capacidad INTEGER, DisponibleReserva INTEGER)",
]

ROWS: list[tuple[str, list[dict[str, Any]]]] = [
    (
        "INSERT INTO Local VALUES (:id, :n, :p, :d, :e, :tz)",
        [
            {"id": 11, "n": "La Mar", "p": "PE", "d": "LIM", "e": 1, "tz": "America/Lima"},
            {"id": 873, "n": "Boragó", "p": "CL", "d": "RM", "e": 1, "tz": None},
            {"id": 2246, "n": "Zazu", "p": "EC", "d": "PIC", "e": 1, "tz": None},
            {"id": 999, "n": "Cerrado", "p": "PE", "d": "LIM", "e": 0, "tz": "America/Lima"},
        ],
    ),
    (
        "INSERT INTO Ubigeo VALUES (:p, :d, :tz, :e)",
        [
            {"p": "PE", "d": "LIM", "tz": "America/Lima", "e": 1},
            {"p": "CL", "d": "RM", "tz": "America/Santiago", "e": 1},
            {"p": "EC", "d": "PIC", "tz": None, "e": 1},
        ],
    ),
    (
        "INSERT INTO Reserva VALUES (:id, :l, :f, :h, :e, :pax)",
        [
            {"id": 1, "l": 11, "f": "2026-08-20", "h": "13:00", "e": 1, "pax": 2},
            {"id": 2, "l": 11, "f": "2026-08-20", "h": "20:30", "e": 1, "pax": 4},
            {"id": 3, "l": 11, "f": "2026-08-20", "h": "21:00", "e": 3, "pax": 2},
            {"id": 4, "l": 11, "f": "2026-08-21", "h": "13:00", "e": 1, "pax": 6},
            {"id": 5, "l": 873, "f": "2026-08-20", "h": "20:00", "e": 1, "pax": 2},
        ],
    ),
    (
        "INSERT INTO Mesa VALUES (:id, :l, :c, :d)",
        [
            {"id": 1, "l": 11, "c": 2, "d": 1},
            {"id": 2, "l": 11, "c": 4, "d": 1},
            {"id": 3, "l": 11, "c": 6, "d": 0},
            {"id": 4, "l": 873, "c": 2, "d": 1},
        ],
    ),
]


async def main() -> None:
    engine = get_engine()
    async with engine.begin() as conn:
        for stmt in DDL:
            await conn.execute(text(stmt))
        for sql, rows in ROWS:
            await conn.execute(text(sql), rows)
    async with engine.connect() as conn:
        n = (await conn.execute(text("SELECT COUNT(*) FROM Local"))).scalar_one()
    print(f"seed listo: {n} locales en {engine.url.database}")


if __name__ == "__main__":
    asyncio.run(main())

seed:
	$(VENV)/bin/python scripts/seed_legacy.py

"""I/O: reads from the legacy DB. No business logic here."""

from __future__ import annotations

from typing import Any

from shared.legacy_db import legacy_db_read

# Constant SQL fragment: identifier/filter text written by us, never user input.
_ACTIVE_FILTER_SQL = "AND l.EstadoId = :estado_activo"  # identifier, not input
ESTADO_ACTIVO = 1


async def fetch_local(local_id: int) -> dict[str, Any] | None:
    """One local with its Ubigeo timezone (LEFT JOIN so a missing Ubigeo still returns the row)."""
    rows = await legacy_db_read(
        """
            SELECT l.Id AS id, l.Nombre AS nombre, l.Pais AS pais,
                   u.ZonaHoraria AS zona_horaria
            FROM Local l
            LEFT JOIN Ubigeo u ON u.Pais = l.Pais AND u.Depa = l.Depa AND u.EstadoId = 1
            WHERE l.Id = :local_id
            LIMIT 1
        """,
        {"local_id": local_id},
    )
    return rows[0] if rows else None


async def fetch_locales(q: str | None = None, pais: str | None = None) -> list[dict[str, Any]]:
    """Active locals, optionally filtered by name (LIKE) and country."""
    return await legacy_db_read(
        f"""
            SELECT l.Id AS id, l.Nombre AS nombre, l.Pais AS pais,
                   u.ZonaHoraria AS zona_horaria
            FROM Local l
            LEFT JOIN Ubigeo u ON u.Pais = l.Pais AND u.Depa = l.Depa AND u.EstadoId = 1
            WHERE (:q IS NULL OR l.Nombre LIKE :q_like)
              AND (:pais IS NULL OR l.Pais = :pais)
              {_ACTIVE_FILTER_SQL}
            ORDER BY l.Nombre
        """,  # noqa: S608 — solo interpola _ACTIVE_FILTER_SQL (constante); los valores van por :param
        {"q": q, "q_like": f"%{q}%" if q else None, "pais": pais, "estado_activo": ESTADO_ACTIVO},
    )


async def fetch_reservas_del_dia(local_id: int, fecha: str) -> list[dict[str, Any]]:
    """Active reservations of a local for one date (YYYY-MM-DD), ordered by time."""
    return await legacy_db_read(
        """
            SELECT r.Id AS id, r.Horario AS horario, r.Pax AS pax
            FROM Reserva r
            WHERE r.LocalId = :local_id
              AND r.FechaReserva = :fecha
              AND r.EstadoId = :estado_activo
            ORDER BY r.Horario
        """,
        {"local_id": local_id, "fecha": fecha, "estado_activo": ESTADO_ACTIVO},
    )

"""GET /v1/locales, /v1/locales/{id}, /v1/locales/{id}/reservas?fecha=YYYY-MM-DD."""

from __future__ import annotations

from typing import Annotated

from fastapi import APIRouter, Query
from fastapi.responses import JSONResponse

from shared.errors import error_response
from shared.errors.codes import LOCAL_NOT_FOUND

from .queries import fetch_local, fetch_locales, fetch_reservas_del_dia
from .schemas import LocalDTO, LocalesResponse, ReservasDelDiaResponse
from .utils import to_local_dto, to_reserva_dto

router = APIRouter(tags=["locales"])


@router.get("/locales", response_model=LocalesResponse)
async def list_locales(
    q: Annotated[str | None, Query(min_length=2, max_length=60)] = None,
    pais: Annotated[str | None, Query(min_length=2, max_length=2)] = None,
) -> LocalesResponse:
    rows = await fetch_locales(q=q, pais=pais)
    return LocalesResponse(locales=[to_local_dto(r) for r in rows])


@router.get("/locales/{local_id}", response_model=LocalDTO)
async def get_local(local_id: int) -> LocalDTO | JSONResponse:
    row = await fetch_local(local_id)
    if row is None:
        return JSONResponse(
            status_code=404,
            content=error_response(LOCAL_NOT_FOUND, f"Local {local_id} no existe"),
        )
    return to_local_dto(row)


@router.get("/locales/{local_id}/reservas", response_model=ReservasDelDiaResponse)
async def reservas_del_dia(
    local_id: int,
    fecha: Annotated[str, Query(pattern=r"^\d{4}-\d{2}-\d{2}$")],
) -> ReservasDelDiaResponse | JSONResponse:
    if await fetch_local(local_id) is None:
        return JSONResponse(
            status_code=404,
            content=error_response(LOCAL_NOT_FOUND, f"Local {local_id} no existe"),
        )
    rows = await fetch_reservas_del_dia(local_id, fecha)
    return ReservasDelDiaResponse(
        local_id=local_id, fecha=fecha, reservas=[to_reserva_dto(r) for r in rows]
    )

"""Por qué NUNCA se interpola un valor en el SQL: la misma consulta, con y sin parámetro."""

import asyncio

from shared.legacy_db import legacy_db_read


async def main() -> None:
    entrada = "' OR 1=1 --"  # lo que un cliente malicioso manda como ?q=

    print("1) f-string con el valor dentro (MAL):")
    sql_mal = f"SELECT Id, Nombre FROM Local WHERE Nombre = '{entrada}'"  # noqa: S608
    print("   SQL enviado:", sql_mal)
    print("   filas:", await legacy_db_read(sql_mal))

    print("2) parámetro nombrado (BIEN):")
    sql_bien = "SELECT Id, Nombre FROM Local WHERE Nombre = :nombre"
    print("   SQL enviado:", sql_bien, "| params:", {"nombre": entrada})
    print("   filas:", await legacy_db_read(sql_bien, {"nombre": entrada}))


if __name__ == "__main__":
    asyncio.run(main())

"""Shared fixtures (same contract as m-b-core services/api/app/conftest.py)."""

from __future__ import annotations

import pytest


class LegacyReadMock:
    """Async callable that records calls to `legacy_db_read`.

    Usage:
        async def test_X(legacy_read_mock, monkeypatch):
            legacy_read_mock.return_value = [{"Id": 1}]
            monkeypatch.setattr("services.api.app.locales.queries.legacy_db_read", legacy_read_mock)
            await fetch_X("PE")
            assert legacy_read_mock.last_params == {"country_code": "PE"}
    """

    def __init__(self) -> None:
        self.return_value: list[dict] = []
        self.side_effect: BaseException | None = None
        self.call_count: int = 0
        self.calls: list[tuple[str, dict | None]] = []

    @property
    def last_sql(self) -> str:
        if not self.calls:
            raise AssertionError("legacy_db_read was never called")
        return self.calls[-1][0]

    @property
    def last_params(self) -> dict | None:
        if not self.calls:
            raise AssertionError("legacy_db_read was never called")
        return self.calls[-1][1]

    async def __call__(self, sql, params=None):
        self.call_count += 1
        self.calls.append((str(sql), params))
        if self.side_effect is not None:
            raise self.side_effect
        return self.return_value


@pytest.fixture()
def legacy_read_mock():
    """Yield a fresh LegacyReadMock per test. Wiring it into the SUT module is the test's job."""
    return LegacyReadMock()

"""queries.py → patch `legacy_db_read` at the module's local binding (mock boundary rule)."""

from services.api.app.locales import queries as q

LIMA = {"id": 11, "nombre": "La Mar", "pais": "PE", "zona_horaria": "America/Lima"}


async def test_fetch_local_pasa_el_id_como_parametro(legacy_read_mock, monkeypatch):
    legacy_read_mock.return_value = [LIMA]
    monkeypatch.setattr(q, "legacy_db_read", legacy_read_mock)

    row = await q.fetch_local(11)

    assert row == LIMA
    assert legacy_read_mock.call_count == 1
    assert legacy_read_mock.last_params == {"local_id": 11}
    assert "WHERE l.Id = :local_id" in legacy_read_mock.last_sql
    assert "LEFT JOIN Ubigeo" in legacy_read_mock.last_sql


async def test_fetch_local_devuelve_none_sin_filas(legacy_read_mock, monkeypatch):
    legacy_read_mock.return_value = []
    monkeypatch.setattr(q, "legacy_db_read", legacy_read_mock)
    assert await q.fetch_local(999) is None


async def test_fetch_locales_filtra_por_pais_y_texto(legacy_read_mock, monkeypatch):
    legacy_read_mock.return_value = [LIMA]
    monkeypatch.setattr(q, "legacy_db_read", legacy_read_mock)

    await q.fetch_locales(q="mar", pais="PE")

    assert legacy_read_mock.last_params == {
        "q": "mar",
        "q_like": "%mar%",
        "pais": "PE",
        "estado_activo": 1,
    }
    assert "l.Pais = :pais" in legacy_read_mock.last_sql
    assert "l.EstadoId = :estado_activo" in legacy_read_mock.last_sql
    assert "'" not in legacy_read_mock.last_sql.replace("''", "")  # no literales interpolados


async def test_fetch_reservas_del_dia_solo_activas(legacy_read_mock, monkeypatch):
    legacy_read_mock.return_value = [{"id": 1, "horario": "13:00", "pax": 2}]
    monkeypatch.setattr(q, "legacy_db_read", legacy_read_mock)

    rows = await q.fetch_reservas_del_dia(11, "2026-08-20")

    assert rows[0]["horario"] == "13:00"
    assert legacy_read_mock.last_params == {
        "local_id": 11,
        "fecha": "2026-08-20",
        "estado_activo": 1,
    }
    assert "r.EstadoId = :estado_activo" in legacy_read_mock.last_sql
    assert "ORDER BY r.Horario" in legacy_read_mock.last_sql


async def test_fetch_local_propaga_error_de_db(legacy_read_mock, monkeypatch):
    legacy_read_mock.side_effect = RuntimeError("legacy caído")
    monkeypatch.setattr(q, "legacy_db_read", legacy_read_mock)
    try:
        await q.fetch_local(11)
    except RuntimeError as e:
        assert "legacy caído" in str(e)
    else:
        raise AssertionError("debió propagar")

"""End-to-end against the seeded SQLite (marked e2e: excluded from default runs)."""

import pytest

from services.api.app.locales import queries as q

pytestmark = pytest.mark.e2e


async def test_fetch_local_11_trae_zona_de_ubigeo():
    row = await q.fetch_local(11)
    assert row is not None and row["zona_horaria"] == "America/Lima"


async def test_fetch_local_873_sin_zona_en_ubigeo_devuelve_none_en_zona():
    row = await q.fetch_local(873)
    assert row is not None and row["zona_horaria"] == "America/Santiago"


async def test_reservas_del_dia_excluye_canceladas():
    rows = await q.fetch_reservas_del_dia(11, "2026-08-20")
    assert [r["horario"] for r in rows] == ["13:00", "20:30"]

from unittest.mock import AsyncMock, patch

from fastapi.testclient import TestClient

from services.api.app.main import app

client = TestClient(app)


def test_get_local_200_mockeando_la_ruta():
    with patch(
        "services.api.app.locales.routes.fetch_local",
        new=AsyncMock(return_value={"id": 11, "nombre": "La Mar", "pais": "PE", "zona_horaria": None}),
    ):
        assert client.get("/v1/locales/11").status_code == 200

[tool.ruff.lint]
select = ["E", "W", "F", "I", "B", "C4", "UP", "S608"]

        """,  # noqa: S608 — solo interpola _ACTIVE_FILTER_SQL (constante); los valores van por :param

# Testing — mesa-core-practicas (resumen del TESTING.md de m-b-core)

**Principio 0:** la cobertura es consecuencia, no meta. Antes de escribir un test:
"¿qué bug de producción atraparía si se rompiera?". Si ninguno, no se escribe.

| Forma del código | Estrategia | Qué se mockea |
|---|---|---|
| Función pura (sin I/O) | `parametrize` de bordes | nada |
| `*queries.py` | patch de `legacy_db_read` **en el binding local del módulo** | `legacy_db_read` |
| Orquestador (llama varias funciones) | `AsyncMock` de cada subfunción | las subfunciones que importa |
| Ruta FastAPI | `TestClient` + patch de las funciones que la ruta importa | `services.api.app.<mod>.routes.<fn>` |

**Regla del mock boundary:** mockea el nivel más bajo de la dependencia; nunca la función
bajo prueba. Mockear un nivel arriba deja la query sin cubrir y "pasa" con el SQL roto.

**Asserts sobre SQL:** pinear cláusulas que llevan contrato (`"WHERE l.Id = :local_id" in sql`)
y `last_params`; nunca igualdad del string entero.