Entorno, funciones puras y tests

[tool.ruff]
target-version = "py312"
line-length = 100
extend-exclude = [".venv", "alembic/versions", "__pycache__"]

[tool.ruff.lint]
select = [
    "E",  # pycodestyle errors
    "W",  # pycodestyle warnings
    "F",  # pyflakes
    "I",  # isort
    "B",  # flake8-bugbear
    "C4", # flake8-comprehensions
    "UP", # pyupgrade
]
ignore = [
    "E501",  # line too long — lo maneja el formatter
    "B008",  # function calls in argument defaults — Depends() de FastAPI
]

[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]

[tool.ruff.format]
quote-style = "double"
indent-style = "space"

[tool.mypy]
python_version = "3.12"
ignore_missing_imports = true
check_untyped_defs = true
warn_unused_ignores = true
warn_return_any = true

[pytest]
testpaths = services/api/app shared
python_files = test_*.py
python_classes = Test*
python_functions = test_*
addopts = -q --tb=short -m "not e2e"
asyncio_mode = auto
markers =
    unit: funciones puras sin I/O (rápidas, sin fixtures)
    integration: HTTP / motor con dependencias externas simuladas
    e2e: pega a servicios reales; excluido por defecto (pytest -m e2e)

# Runtime mínimo (versiones exactas, como en m-b-core)
numpy==2.2.3

-r base.txt
# Tests
pytest==8.2.2
pytest-asyncio==0.23.7
pytest-cov==6.1.1
# Lint / format / tipos
ruff==0.6.9
mypy==1.14.1

.venv/
__pycache__/
.pytest_cache/
.ruff_cache/
.mypy_cache/
*.pyc
.env
env/*.env

PY ?= python3
VENV := .venv
PIP := $(VENV)/bin/pip
PYTEST := $(VENV)/bin/pytest
RUFF := $(VENV)/bin/ruff
MYPY := $(VENV)/bin/mypy

# Todo import es absoluto desde la raíz del repo (services.api.app…, shared…)
export PYTHONPATH := .

.PHONY: venv install tests lint typecheck check

venv:
	$(PY) -m venv $(VENV)
	$(PIP) install --upgrade pip

install: venv
	$(PIP) install -r requirements/local.txt

tests:
	$(PYTEST)

lint:
	$(RUFF) check .
	$(RUFF) format --check .

typecheck:
	$(MYPY) services/api/app/availability_engine/

check: lint typecheck tests

"""Objetos de dominio y constantes del motor de disponibilidad (mini).

Todo valor que se repite vive aquí con nombre (guideline 1 de m-b-core:
"No magic strings/numbers"). El motor real está en
services/api/app/availability_engine/models.py de m-b-core.
"""

from __future__ import annotations

from dataclasses import dataclass
from enum import IntEnum, unique

# ---------------------------------------------------------------------------
# Constantes
# ---------------------------------------------------------------------------

ESTADO_ACTIVO = 1
MINUTES_PER_SLOT = 15
MINUTES_PER_HOUR = 60
MINUTES_PER_DAY = 24 * MINUTES_PER_HOUR  # 1440
SLOTS_PER_DAY = MINUTES_PER_DAY // MINUTES_PER_SLOT  # 96
DEFAULT_TIMEZONE = "America/Lima"


# ---------------------------------------------------------------------------
# Enums
# ---------------------------------------------------------------------------


@unique
class SlotStatus(IntEnum):
    """Estado de una celda de la matriz. Negativo = bloqueado, 0 = libre, positivo = ocupado."""

    BLOCKED = -2  # bloqueada por una regla (cierre, capacidad)
    OUTSIDE = -1  # fuera del horario de atención
    AVAILABLE = 0  # libre para reservar
    OCCUPIED = 1  # ya reservada


# ---------------------------------------------------------------------------
# Objetos de dominio
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class Table:
    """Mesa física con sus límites de capacidad."""

    id: int
    min_pax: int
    max_pax: int
    disponible_reserva: bool = True


@dataclass(frozen=True)
class Slot:
    """Un intervalo de MINUTES_PER_SLOT minutos, identificado por su índice en el día."""

    index: int

    @property
    def minute_of_day(self) -> int:
        return self.index * MINUTES_PER_SLOT

    @property
    def label(self) -> str:
        """'HH:MM' del inicio del slot."""
        hours, minutes = divmod(self.minute_of_day, MINUTES_PER_HOUR)
        return f"{hours:02d}:{minutes:02d}"

"""Matriz del día: una fila por mesa, una columna por slot de 15 minutos.

Es la versión mínima de services/api/app/availability_engine/day_matrix.py
de m-b-core: el motor real construye una grilla por día donde TODAS las
reservas y cierres se proyectan, y las reglas la mutan en sitio.
"""

from __future__ import annotations

import numpy as np

from .models import MINUTES_PER_DAY, MINUTES_PER_SLOT, SLOTS_PER_DAY, SlotStatus, Table


def slot_index(minute_of_day: int) -> int:
    """Índice de columna para un minuto del día (0..1439)."""
    if not 0 <= minute_of_day < MINUTES_PER_DAY:
        raise ValueError(f"minute_of_day fuera de rango: {minute_of_day}")
    return minute_of_day // MINUTES_PER_SLOT


def build_day_matrix(tables: list[Table]) -> np.ndarray:
    """Devuelve una matriz (len(tables), 96) con todo AVAILABLE.

    Las mesas conservan el orden de la lista: la fila i es tables[i].
    """
    if not tables:
        raise ValueError("no hay mesas: la matriz sería vacía")
    return np.full((len(tables), SLOTS_PER_DAY), SlotStatus.AVAILABLE, dtype=np.int8)


def available_slots(matrix: np.ndarray, row: int) -> list[int]:
    """Índices de slot libres para la fila (mesa) indicada."""
    return [int(i) for i in np.flatnonzero(matrix[row] == SlotStatus.AVAILABLE)]

"""Reglas de negocio como funciones puras que mutan la matriz en sitio.

Misma firma en todas (guideline 2 de m-b-core, ver
services/api/app/availability_engine/rules.py):
    apply_*(matrix, tables, ...) -> None
Reciben entradas explícitas, no tocan estado global, y son fáciles de testear
con parametrize porque su única salida es la matriz que reciben.
"""

from __future__ import annotations

import numpy as np

from .day_matrix import slot_index
from .models import MINUTES_PER_DAY, SLOTS_PER_DAY, SlotStatus, Table


def _mark(matrix: np.ndarray, rows: np.ndarray, cols: slice, status: SlotStatus) -> None:
    """Pone `status` en las celdas indicadas que aún estén AVAILABLE."""
    sub = matrix[rows, cols]
    matrix[rows, cols] = np.where(sub == SlotStatus.AVAILABLE, status, sub)


def apply_opening_hours(matrix: np.ndarray, open_minute: int, close_minute: int) -> None:
    """Marca OUTSIDE todo lo que cae antes de la apertura o desde el cierre.

    Cierre exclusivo: si cierra a las 23:00, el slot 23:00 ya está fuera.
    """
    if not open_minute < close_minute:
        raise ValueError("open_minute debe ser menor que close_minute")
    all_rows = np.arange(matrix.shape[0])
    _mark(matrix, all_rows, slice(0, slot_index(open_minute)), SlotStatus.OUTSIDE)
    _mark(matrix, all_rows, slice(slot_index(close_minute), None), SlotStatus.OUTSIDE)


def apply_min_capacity(matrix: np.ndarray, tables: list[Table], pax: int) -> None:
    """Bloquea las mesas que no admiten `pax` personas (fuera de [min_pax, max_pax])."""
    if pax < 1:
        raise ValueError("pax debe ser >= 1")
    rows = np.array(
        [i for i, t in enumerate(tables) if not (t.min_pax <= pax <= t.max_pax)],
        dtype=int,
    )
    if rows.size:
        _mark(matrix, rows, slice(None), SlotStatus.BLOCKED)


def apply_reservation(matrix: np.ndarray, row: int, start_minute: int, duration_minutes: int) -> None:
    """Ocupa los slots de una reserva existente en la mesa `row`."""
    if duration_minutes <= 0:
        raise ValueError("duration_minutes debe ser > 0")
    first = slot_index(start_minute)
    last = slot_index(min(start_minute + duration_minutes - 1, MINUTES_PER_DAY - 1)) + 1
    matrix[row, first:last] = SlotStatus.OCCUPIED

import numpy as np
import pytest

from services.api.app.availability_engine.day_matrix import (
    available_slots,
    build_day_matrix,
    slot_index,
)
from services.api.app.availability_engine.models import SLOTS_PER_DAY, SlotStatus, Table

MESA_2 = Table(id=1, min_pax=1, max_pax=2)
MESA_4 = Table(id=2, min_pax=2, max_pax=4)


@pytest.mark.parametrize(
    "minute, esperado",
    [(0, 0), (14, 0), (15, 1), (12 * 60, 48), (23 * 60 + 59, 95)],
)
def test_slot_index(minute, esperado):
    assert slot_index(minute) == esperado


@pytest.mark.parametrize("minute", [-1, 24 * 60])
def test_slot_index_fuera_de_rango(minute):
    with pytest.raises(ValueError):
        slot_index(minute)


def test_build_day_matrix_forma_y_estado():
    m = build_day_matrix([MESA_2, MESA_4])
    assert m.shape == (2, SLOTS_PER_DAY)
    assert m.dtype == np.int8
    assert (m == SlotStatus.AVAILABLE).all()


def test_build_day_matrix_sin_mesas_falla():
    with pytest.raises(ValueError):
        build_day_matrix([])


def test_available_slots_devuelve_todos_al_inicio():
    m = build_day_matrix([MESA_2])
    assert available_slots(m, 0) == list(range(SLOTS_PER_DAY))

import pytest

from services.api.app.availability_engine.day_matrix import available_slots, build_day_matrix
from services.api.app.availability_engine.models import SlotStatus, Table
from services.api.app.availability_engine.rules import (
    apply_min_capacity,
    apply_opening_hours,
    apply_reservation,
)

MESA_2 = Table(id=1, min_pax=1, max_pax=2)
MESA_4 = Table(id=2, min_pax=2, max_pax=4)
MESA_8 = Table(id=3, min_pax=5, max_pax=8)


@pytest.mark.parametrize(
    "open_h, close_h, libres_esperados",
    [
        (12, 23, 44),  # 11 horas x 4 slots
        (0, 24, 96),  # abierto todo el día
        (19, 20, 4),  # una hora
    ],
)
def test_apply_opening_hours_deja_solo_el_horario(open_h, close_h, libres_esperados):
    m = build_day_matrix([MESA_2])
    apply_opening_hours(m, open_h * 60, close_h * 60)
    assert len(available_slots(m, 0)) == libres_esperados


def test_apply_opening_hours_marca_outside_no_blocked():
    m = build_day_matrix([MESA_2])
    apply_opening_hours(m, 12 * 60, 23 * 60)
    assert m[0, 0] == SlotStatus.OUTSIDE
    assert m[0, 95] == SlotStatus.OUTSIDE
    assert m[0, 48] == SlotStatus.AVAILABLE  # 12:00


def test_apply_opening_hours_invalido():
    m = build_day_matrix([MESA_2])
    with pytest.raises(ValueError):
        apply_opening_hours(m, 20 * 60, 12 * 60)


@pytest.mark.parametrize(
    "pax, filas_bloqueadas",
    [
        (1, {1, 2}),  # solo MESA_2 admite 1
        (2, {2}),  # MESA_2 y MESA_4 admiten 2
        (4, {0, 2}),  # solo MESA_4
        (6, {0, 1}),  # solo MESA_8
        (9, {0, 1, 2}),  # nadie
    ],
)
def test_apply_min_capacity(pax, filas_bloqueadas):
    tables = [MESA_2, MESA_4, MESA_8]
    m = build_day_matrix(tables)
    apply_min_capacity(m, tables, pax)
    bloqueadas = {i for i in range(len(tables)) if (m[i] == SlotStatus.BLOCKED).all()}
    assert bloqueadas == filas_bloqueadas


def test_apply_min_capacity_pax_invalido():
    m = build_day_matrix([MESA_2])
    with pytest.raises(ValueError):
        apply_min_capacity(m, [MESA_2], 0)


def test_apply_reservation_ocupa_los_slots_de_la_duracion():
    m = build_day_matrix([MESA_2])
    apply_reservation(m, row=0, start_minute=20 * 60, duration_minutes=90)  # 20:00–21:30
    ocupados = [i for i in range(96) if m[0, i] == SlotStatus.OCCUPIED]
    assert ocupados == list(range(80, 86))  # 20:00,20:15,…,21:15


def test_las_reglas_no_pisan_lo_que_ya_estaba_ocupado():
    m = build_day_matrix([MESA_2])
    apply_reservation(m, 0, 20 * 60, 60)
    apply_opening_hours(m, 12 * 60, 21 * 60)  # el cierre 21:00 cae encima de la reserva
    assert m[0, 80] == SlotStatus.OCCUPIED  # 20:00 sigue ocupado, no OUTSIDE

def apply_opening_hours(matrix: np.ndarray, open_minute: int, close_minute: int) -> None:
    """Marca OUTSIDE todo lo que cae antes de la apertura o desde el cierre.

    Cierre exclusivo: si cierra a las 23:00, el slot 23:00 ya está fuera.
    """
    if not 0 <= open_minute < close_minute <= MINUTES_PER_DAY:
        raise ValueError("se requiere 0 <= open_minute < close_minute <= 1440")
    all_rows = np.arange(matrix.shape[0])
    # El cierre a medianoche (1440) no es un slot válido: es "hasta el final".
    close_idx = SLOTS_PER_DAY if close_minute == MINUTES_PER_DAY else slot_index(close_minute)
    _mark(matrix, all_rows, slice(0, slot_index(open_minute)), SlotStatus.OUTSIDE)
    _mark(matrix, all_rows, slice(close_idx, None), SlotStatus.OUTSIDE)

typecheck:
	# --explicit-package-bases: services/ y services/api/ no tienen __init__.py a propósito
	# (el paquete raíz es app/); sin el flag mypy ve cada módulo bajo dos nombres.
	$(MYPY) services/api/app/availability_engine/ --explicit-package-bases