El servidor con pruebas: las tres fallas, demostradas

git clone https://github.com/rpversontec/practica-11-servidor-con-pruebas.git
cd practica-11-servidor-con-pruebas
cp .env.example .env

docker compose up -d --build
docker compose exec api uv run --no-sync pytest

tests/test_contrato.py .....                                             [ 45%]
tests/test_duplicados.py .s                                              [ 63%]
tests/test_huerfanas.py .                                                [ 72%]
tests/test_permisos.py ..F                                               [100%]

=================================== FAILURES ===================================
__________________ test_bruno_no_puede_borrar_el_aviso_de_ana __________________
…
>       assert r.status_code == 403, f"Bruno pudo borrar el aviso de Ana: el servidor respondió {r.status_code}"
E       AssertionError: Bruno pudo borrar el aviso de Ana: el servidor respondió 204
E       assert 204 == 403
…
==================== 1 failed, 9 passed, 1 skipped in 4.64s ====================

def test_bruno_no_puede_borrar_el_aviso_de_ana(cliente: TestClient, ana: Persona, bruno: Persona, publicar):
    aviso = publicar(ana)

    r = cliente.delete(f"/api/avisos/{aviso['id']}", headers=bruno.headers)

    assert r.status_code == 403, f"Bruno pudo borrar el aviso de Ana: el servidor respondió {r.status_code}"
    # Y el aviso sigue en el tablón.
    tablon = cliente.get("/api/avisos", headers=ana.headers).json()
    assert aviso["id"] in [a["id"] for a in tablon]

@router.delete("/{id}", status_code=204)
def borrar(id: int, quien: Identidad = Depends(requiere_profesor), db: Session = Depends(get_db)) -> Response:
    aviso = db.get(Aviso, id)
    if aviso is None:
        raise ApiError(404, f"No existe el aviso {id}")
    # Tener el rol no basta: el aviso tiene que ser suyo. El autor lo guardó el servidor
    # al publicar, con la identidad del token; aquí se compara con la del token de ahora.
    if aviso.autor != quien.usuario:
        raise ApiError(403, "Solo quien publicó el aviso puede borrarlo", "no_es_tuyo", f'El aviso {id} es de "{aviso.autor}".')
    db.delete(aviso)
    db.commit()
    # Primero la fila y después el objeto: si el almacén falla, queda un archivo huérfano
    # (estorba) en vez de un aviso que apunta a una imagen que ya no existe (truena).
    if aviso.imagen:
        almacen.borrar(aviso.imagen)
    return Response(status_code=204)

"""Pruebas negativas: lo que el servidor tiene que RECHAZAR.

Ya pasan desde el primer día. Son la evidencia de seguridad del reto: «demuestren
al menos algunos casos en los que el sistema rechaza una acción no permitida».
"""

import uuid

from fastapi.testclient import TestClient

from app.config import settings
from app.db import SessionLocal
from app.models import Usuario
from app.security import sign_jwt

from .conftest import PNG_CHICO, Persona


def test_un_token_vencido_no_sirve(cliente: TestClient, carla: Persona):
    # Sin esperar cinco minutos: se firma con la llave real del servidor un token que venció
    # hace uno. Firmarlo con OTRA llave probaría la firma, no el vencimiento.
    vencido = sign_jwt(carla.usuario, carla.rol, -60, settings.jwt_secret)

    r = cliente.get("/api/avisos", headers={"Authorization": f"Bearer {vencido}"})

    assert r.status_code == 401
    assert r.json()["code"] == "token_expirado"


def test_mandar_rol_profesor_sin_el_codigo_crea_un_alumno(cliente: TestClient):
    # El cliente puede mandar lo que quiera. El rol lo decide el servidor, con el código.
    cuerpo = {"usuario": f"tramposa.{uuid.uuid4().hex[:8]}", "password": "secreta123", "rol": "profesor"}

    r = cliente.post("/api/auth/register", json=cuerpo)

    assert r.status_code == 201
    assert r.json()["rol"] == "alumno"


def test_una_alumna_no_puede_subir_imagenes(cliente: TestClient, carla: Persona):
    r = cliente.post("/api/imagenes", headers=carla.headers, files={"archivo": ("foto.png", PNG_CHICO, "image/png")})

    assert r.status_code == 403


def test_la_contrasena_no_se_guarda_en_claro(carla: Persona):
    # Se lee la fila directo de PostgreSQL, como la vería quien se robe un respaldo.
    with SessionLocal() as db:
        fila = db.get(Usuario, carla.usuario)

    assert fila is not None
    assert "secreta123" not in fila.password_hash
    assert fila.password_hash.startswith("$argon2id$")

from sqlalchemy import BigInteger, DateTime, ForeignKey, String, Text, UniqueConstraint, func

class Idempotencia(Base):
    """Una fila por publicación que llegó con `Idempotency-Key`: quién, con qué clave, qué aviso salió.

    La restricción única es la que de verdad impide el duplicado. La consulta de
    `crear` ataja el reintento normal; si dos envíos llegan al mismo tiempo, los dos
    pasan la consulta, y PostgreSQL deja guardar solo a uno.
    """

    __tablename__ = "idempotencia"
    __table_args__ = (UniqueConstraint("usuario", "clave"),)

    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    usuario: Mapped[str] = mapped_column(String(32))
    clave: Mapped[str] = mapped_column(String(64))
    # sha256 del aviso que llegó con esa clave: la misma clave con otro contenido no es un reintento.
    huella: Mapped[str] = mapped_column(String(64))
    # Si se borra el aviso, su fila se va con él (ON DELETE CASCADE, lo hace PostgreSQL).
    aviso_id: Mapped[int] = mapped_column(ForeignKey("avisos.id", ondelete="CASCADE"))
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())

from fastapi import APIRouter, Depends, Header, Request, Response
from fastapi.responses import StreamingResponse
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm import Session

from .. import almacen
from ..db import SessionLocal, get_db
from ..deps import Identidad, requiere_profesor, usuario_actual
from ..errors import ApiError
from ..models import Aviso, Idempotencia
from ..schemas import AvisoOut, NuevoAviso
from ..security import sha256

def ya_publicado(db: Session, quien: Identidad, clave: str, huella: str) -> Aviso | None:
    """El aviso que ya salió con esta clave, o None si es la primera vez que llega."""
    previo = db.scalar(select(Idempotencia).where(Idempotencia.usuario == quien.usuario, Idempotencia.clave == clave))
    if previo is None:
        return None
    if previo.huella != huella:
        raise ApiError(
            422,
            "Esa clave ya se usó para publicar otro aviso",
            "clave_reutilizada",
            "Genera una Idempotency-Key nueva por publicación; repítela solo al reintentar la misma.",
        )
    return db.get(Aviso, previo.aviso_id)


@router.post("", response_model=AvisoOut, status_code=201)
def crear(
    body: NuevoAviso,
    quien: Identidad = Depends(requiere_profesor),
    db: Session = Depends(get_db),
    # La inventa el teléfono: una por publicación, la misma en cada reintento. Es opcional:
    # sin ella, cada envío es un aviso nuevo, como antes.
    clave: str | None = Header(default=None, alias="Idempotency-Key", max_length=64),
) -> Aviso:
    huella = sha256(body.model_dump_json())
    if clave is not None and (previo := ya_publicado(db, quien, clave, huella)) is not None:
        return previo
    # Una clave con buena forma no basta: la imagen tiene que estar en el almacén.
    if body.imagen is not None and not almacen.existe(body.imagen):
        raise ApiError(422, "Esa imagen no existe", "imagen_inexistente", "Súbela primero con POST /api/imagenes y manda la clave que te devuelve.")
    aviso = Aviso(titulo=body.titulo, cuerpo=body.cuerpo, autor=quien.usuario, imagen=body.imagen)
    db.add(aviso)
    if clave is not None:
        db.flush()  # PostgreSQL asigna aviso.id sin cerrar la transacción
        db.add(Idempotencia(usuario=quien.usuario, clave=clave, huella=huella, aviso_id=aviso.id))
    try:
        db.commit()  # el aviso y su clave se guardan juntos, o ninguno
    except IntegrityError:
        # Otro envío con la misma clave se guardó primero: los dos pasaron la consulta
        # de arriba, y la restricción única dejó pasar solo a uno. Se responde el suyo.
        db.rollback()
        previo = ya_publicado(db, quien, clave, huella) if clave else None
        if previo is None:
            raise
        return previo
    db.refresh(aviso)
    return aviso

"""Tarjeta 2 · Envié una vez, aparecieron dos.

El contrato: el teléfono manda `Idempotency-Key` con un UUID que inventa al abrir
la pantalla de publicar. El mismo envío con la misma clave devuelve el aviso que ya
se guardó; la misma clave con otro contenido es un error. Sin clave, cada envío es
un aviso nuevo: publicar dos avisos con el mismo título es válido.
"""

import uuid

from fastapi.testclient import TestClient

from .conftest import Persona

CUERPO = {"titulo": "Feria de proyectos", "cuerpo": "El viernes a las 12:00 en el patio central."}


def avisos_de(cliente: TestClient, quien: Persona) -> list[dict]:
    return [a for a in cliente.get("/api/avisos", headers=quien.headers).json() if a["autor"] == quien.usuario]


def test_sin_clave_dos_envios_son_dos_avisos(ana: Persona, publicar):
    primero = publicar(ana, titulo="Feria de proyectos")
    segundo = publicar(ana, titulo="Feria de proyectos")

    assert primero["id"] != segundo["id"]


def test_reintentar_con_la_misma_clave_no_crea_otro_aviso(cliente: TestClient, ana: Persona):
    headers = {**ana.headers, "Idempotency-Key": str(uuid.uuid4())}

    primero = cliente.post("/api/avisos", headers=headers, json=CUERPO)
    reintento = cliente.post("/api/avisos", headers=headers, json=CUERPO)

    assert primero.status_code == 201
    assert reintento.status_code == 201
    assert reintento.json()["id"] == primero.json()["id"]
    assert len(avisos_de(cliente, ana)) == 1


def test_la_misma_clave_con_otro_contenido_se_rechaza(cliente: TestClient, ana: Persona):
    headers = {**ana.headers, "Idempotency-Key": str(uuid.uuid4())}
    cliente.post("/api/avisos", headers=headers, json=CUERPO)

    r = cliente.post("/api/avisos", headers=headers, json={**CUERPO, "titulo": "Otro aviso distinto"})

    assert r.status_code == 422
    assert r.json()["code"] == "clave_reutilizada"
    assert len(avisos_de(cliente, ana)) == 1


def test_la_clave_de_ana_no_le_estorba_a_bruno(cliente: TestClient, ana: Persona, bruno: Persona):
    # La clave es única por persona, no en todo el servidor: Bruno no puede «adivinar»
    # la clave de Ana y recibir su aviso, ni bloquearle una publicación.
    clave = str(uuid.uuid4())

    de_ana = cliente.post("/api/avisos", headers={**ana.headers, "Idempotency-Key": clave}, json=CUERPO)
    de_bruno = cliente.post("/api/avisos", headers={**bruno.headers, "Idempotency-Key": clave}, json=CUERPO)

    assert de_bruno.status_code == 201
    assert de_bruno.json()["id"] != de_ana.json()["id"]
    assert de_bruno.json()["autor"] == bruno.usuario

"""El almacén de objetos, en un solo archivo. Todo lo demás le habla a estas seis
funciones y no sabe que del otro lado hay S3.

Se usa boto3, el cliente oficial de Amazon S3: habla con cualquier almacén
compatible —RustFS aquí, MinIO, Cloudflare R2, Amazon S3 en producción—
cambiando solo `S3_ENDPOINT` y las llaves.

Un objeto es una clave ("3f2a….jpg"), unos bytes y su tipo. No hay carpetas, no
hay consultas, no hay transacciones: guardar, leer, preguntar si existe, listar y borrar.
Por eso las imágenes van aquí y no en PostgreSQL, y por eso la base solo guarda
la clave.
"""

from collections.abc import Iterator
from datetime import datetime
from typing import Any

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

from .config import settings

_s3 = boto3.client(
    "s3",
    endpoint_url=settings.s3_endpoint,
    aws_access_key_id=settings.s3_access_key,
    aws_secret_access_key=settings.s3_secret_key,
    # S3 exige una región aunque el almacén sea tuyo; RustFS acepta esta.
    region_name="us-east-1",
    # "path": http://almacen:9000/avisos/clave. El estilo por omisión pone el bucket en el
    # nombre del host (avisos.almacen), y ese nombre no existe en la red de Docker.
    config=Config(s3={"addressing_style": "path"}, connect_timeout=3, retries={"max_attempts": 2}),
)


def preparar() -> None:
    """Crea el bucket si no existe. Se llama una vez, al arrancar."""
    try:
        _s3.head_bucket(Bucket=settings.s3_bucket)
    except ClientError:
        _s3.create_bucket(Bucket=settings.s3_bucket)


def guardar(clave: str, datos: bytes, tipo: str) -> None:
    _s3.put_object(Bucket=settings.s3_bucket, Key=clave, Body=datos, ContentType=tipo)


def abrir(clave: str) -> dict[str, Any] | None:
    """El objeto listo para mandarse por partes, o None si no existe.

    `Body` es un flujo: los bytes salen del almacén conforme el cliente los lee,
    sin cargar la imagen completa en la memoria del servidor.
    """
    try:
        return _s3.get_object(Bucket=settings.s3_bucket, Key=clave)
    except ClientError as e:
        if e.response["Error"]["Code"] in ("NoSuchKey", "404"):
            return None
        raise


def existe(clave: str) -> bool:
    try:
        _s3.head_object(Bucket=settings.s3_bucket, Key=clave)
        return True
    except ClientError as e:
        if e.response["Error"]["Code"] in ("NoSuchKey", "404"):
            return False
        raise


def borrar(clave: str) -> None:
    # Borrar lo que no existe no es error en S3: así borrar dos veces no truena.
    _s3.delete_object(Bucket=settings.s3_bucket, Key=clave)


def listar() -> Iterator[tuple[str, datetime]]:
    """Cada objeto del bucket: su clave y cuándo se guardó. S3 los entrega de mil en mil."""
    for pagina in _s3.get_paginator("list_objects_v2").paginate(Bucket=settings.s3_bucket):
        for objeto in pagina.get("Contents", []):
            yield objeto["Key"], objeto["LastModified"]

"""La limpieza de imágenes huérfanas: las que están en el almacén y ningún aviso usa.

Publicar con imagen son dos peticiones, y no hay una transacción que abarque a
PostgreSQL y a RustFS: si el aviso falla, la imagen ya subió. Eso no se puede
impedir; se limpia después. Se corre a mano, o cada noche desde un cron:

    docker compose exec api uv run --no-sync python -m app.limpieza
"""

from datetime import datetime, timedelta, timezone

from sqlalchemy import select

from . import almacen
from .db import SessionLocal
from .models import Aviso

# El margen protege a quien está publicando en este momento: su imagen ya subió y
# su aviso todavía no llega. Solo se borra lo que lleva más de una hora sin aviso.
MARGEN = timedelta(hours=1)


def limpiar_huerfanas(margen: timedelta = MARGEN, ahora: datetime | None = None) -> list[str]:
    """Borra las imágenes que ningún aviso usa y que son más viejas que `margen`. Devuelve sus claves.

    `ahora` es un parámetro para que una prueba decida qué hora es, como
    `tiempoRelativo` en la app de la Práctica 11.
    """
    ahora = ahora or datetime.now(timezone.utc)
    with SessionLocal() as db:
        usadas = set(db.scalars(select(Aviso.imagen).where(Aviso.imagen.is_not(None))))
    borradas = []
    for clave, guardada in almacen.listar():
        if clave not in usadas and ahora - guardada > margen:
            almacen.borrar(clave)
            borradas.append(clave)
    return borradas


if __name__ == "__main__":
    borradas = limpiar_huerfanas()
    print(f"{len(borradas)} imágenes huérfanas borradas")

"""Tarjeta 3 · La imagen que nadie usa.

Publicar con imagen son dos peticiones: primero la imagen, después el aviso. Si la
segunda falla, la primera ya pasó, y eso no cambia. Lo que cambia es que ahora
alguien limpia: `app/limpieza.py`.
"""

from datetime import datetime, timedelta, timezone

from fastapi.testclient import TestClient

from app import almacen
from app.limpieza import limpiar_huerfanas

from .conftest import Persona


def test_si_el_aviso_falla_la_imagen_se_queda_en_el_almacen(cliente: TestClient, ana: Persona, subir):
    clave = subir(ana)

    # El aviso no pasa la validación del servidor (título de una letra).
    r = cliente.post("/api/avisos", headers=ana.headers, json={"titulo": "X", "cuerpo": "Un cuerpo válido para el aviso.", "imagen": clave})
    assert r.status_code == 422

    try:
        assert almacen.existe(clave), "La imagen ya no está en el almacén"
    finally:
        almacen.borrar(clave)  # la prueba limpia lo que dejó


def test_la_limpieza_borra_la_huerfana_y_respeta_la_que_usa_un_aviso(ana: Persona, subir, publicar):
    usada = subir(ana)
    publicar(ana, imagen=usada)
    huerfana = subir(ana)

    # Dentro de dos horas: la huérfana ya pasó el margen de una hora.
    limpiar_huerfanas(ahora=datetime.now(timezone.utc) + timedelta(hours=2))

    assert almacen.existe(usada), "Se borró una imagen que un aviso sí usa"
    assert not almacen.existe(huerfana), "La huérfana sigue en el almacén"


def test_la_limpieza_no_toca_una_imagen_recien_subida(ana: Persona, subir):
    # Quien está publicando ahora mismo: su imagen ya subió y su aviso todavía no llega.
    clave = subir(ana)

    try:
        limpiar_huerfanas()
        assert almacen.existe(clave), "Se borró la imagen de alguien que estaba publicando"
    finally:
        almacen.borrar(clave)

docker compose exec api uv run --no-sync python -m app.limpieza

0 imágenes huérfanas borradas

mkdir docs ; mkdir docs/evidencia
docker compose exec -T api uv run --no-sync pytest -v | Out-File -Encoding utf8 docs/evidencia/pytest-servidor.txt

docker compose exec -T api uv run --no-sync pytest -v > docs/evidencia/pytest-servidor.txt

tests/test_permisos.py::test_ana_puede_borrar_su_propio_aviso PASSED     [ 68%]
tests/test_permisos.py::test_una_alumna_no_puede_borrar_avisos PASSED    [ 73%]
tests/test_permisos.py::test_bruno_no_puede_borrar_el_aviso_de_ana PASSED [ 78%]
…
============================== 19 passed in 2.94s ==============================

git commit -m "checkpoint b4: el reintento no duplica" -m "Co-Authored-By: Claude <noreply@anthropic.com>"