plantillas-proyectos/backend/tests/conftest.py

from collections.abc import Generator

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import Session, sessionmaker

# Importar modelos con relaciones string para registrar mappers antes de tests.
# Evita errores tipo: expression 'Pedimentos' failed to locate a name.
from api.v1.modules.a76.pedmientos.models.pedimentos import Pedimentos  # noqa: F401
from api.v1.modules.a76.invoices.models import InvoiceHeader
from api.v1.modules.a76.invoices.imports.process import routes as process_routes
from api.v1.modules.a76.invoices.imports.process import main_process as import_main_process
from api.v1.modules.a76.invoices.exports.process import main_process as export_main_process
from core.celery_app import celery_app
from core.config import settings
from core.database import get_core_db
from core.security import get_current_user
from tests.fixtures.builders import (
    EphemeralOrgIds,
    allocate_ephemeral_tenant_company_ids,
    ensure_tenant_company,
)


TEST_DB_URL = (
    __import__("os").environ.get("TEST_DATABASE_URL")
    or __import__("os").environ.get("CORE_DATABASE_URL")
    or settings.core_database_url
)

engine = create_engine(TEST_DB_URL, future=True)
TestingSessionLocal = sessionmaker(
    bind=engine,
    autoflush=False,
    autocommit=False,
    expire_on_commit=False,
    class_=Session,
    join_transaction_mode="create_savepoint",
)


@pytest.fixture
def db_session() -> Generator[Session, None, None]:
    connection = engine.connect()
    transaction = connection.begin()
    session = TestingSessionLocal(bind=connection)
    try:
        yield session
    finally:
        session.close()
        transaction.rollback()
        connection.close()


@pytest.fixture
def test_tenant(db_session: Session) -> EphemeralOrgIds:
    """Tenant y company dedicados por test (IDs efímeros en rango alto)."""
    tid, cid = allocate_ephemeral_tenant_company_ids(db_session)
    ensure_tenant_company(db_session, tenant_id=tid, company_id=cid)
    return EphemeralOrgIds(tenant_id=tid, company_id=cid)


@pytest.fixture(autouse=True)
def celery_eager() -> Generator[None, None, None]:
    prev_broker = celery_app.conf.broker_url
    prev_backend = celery_app.conf.result_backend
    prev_always_eager = celery_app.conf.task_always_eager
    prev_propagates = celery_app.conf.task_eager_propagates
    prev_store_result = celery_app.conf.task_store_eager_result
    prev_ignore_result = celery_app.conf.task_ignore_result

    # Aislar Celery de infraestructura externa en tests (sin Redis/Valkey).
    celery_app.conf.broker_url = "memory://"
    celery_app.conf.result_backend = "cache+memory://"
    celery_app.conf.task_always_eager = True
    celery_app.conf.task_eager_propagates = True
    celery_app.conf.task_store_eager_result = False
    celery_app.conf.task_ignore_result = True
    try:
        yield
    finally:
        celery_app.conf.broker_url = prev_broker
        celery_app.conf.result_backend = prev_backend
        celery_app.conf.task_always_eager = prev_always_eager
        celery_app.conf.task_eager_propagates = prev_propagates
        celery_app.conf.task_store_eager_result = prev_store_result
        celery_app.conf.task_ignore_result = prev_ignore_result


@pytest.fixture
def app(
    db_session: Session,
    test_tenant: EphemeralOrgIds,
    monkeypatch: pytest.MonkeyPatch,
) -> FastAPI:
    test_app = FastAPI()
    test_app.include_router(process_routes.router, prefix="/api/v1/a76")

    def _override_get_db() -> Generator[Session, None, None]:
        yield db_session

    async def _override_current_user():
        return {"sub": "test-user", "tenant_id": test_tenant.tenant_id}

    # validate_access_to_resource is imported directly in the routes module.
    monkeypatch.setattr(
        process_routes,
        "validate_access_to_resource",
        lambda db, company_id, current_user: int(current_user["tenant_id"]),
    )

    class _InlineResult:
        def __init__(self, task_id: str):
            self.id = task_id

    def _run_import_inline(*, args=None, **kwargs):
        if args is None:
            args = []
        invoice_id, tenant_id, company_id = args
        invoice = db_session.get(InvoiceHeader, int(invoice_id))
        if invoice is None:
            return _InlineResult("missing-import-invoice")
        import_main_process.main_process(db_session, invoice, str(tenant_id), str(company_id))
        db_session.flush()
        return _InlineResult(f"inline-import-{invoice_id}")

    def _run_export_inline(*, args=None, **kwargs):
        if args is None:
            args = []
        invoice_id, tenant_id, company_id = args
        invoice = db_session.get(InvoiceHeader, int(invoice_id))
        if invoice is None:
            return _InlineResult("missing-export-invoice")
        export_main_process.main_process(db_session, invoice, str(tenant_id), str(company_id))
        db_session.flush()
        return _InlineResult(f"inline-export-{invoice_id}")

    class _InlineTask:
        def __init__(self, runner):
            self.apply_async = runner

    # Evita worker/redis y obliga ejecución inline con la misma sesión.
    monkeypatch.setattr(process_routes, "process_invoice_task", _InlineTask(_run_import_inline))
    monkeypatch.setattr(process_routes, "process_export_invoice_task", _InlineTask(_run_export_inline))

    test_app.dependency_overrides[get_core_db] = _override_get_db
    test_app.dependency_overrides[get_current_user] = _override_current_user
    return test_app


@pytest.fixture
def client(app: FastAPI) -> TestClient:
    return TestClient(app)