AlexeerCT
b750c35522
- Introduced a new job in the CI workflow to run backend tests before the build process. - Configured Python environment and installed dependencies from the backend requirements. - Added a check for the TEST_DATABASE_URL secret to ensure it is defined before running tests. - The test job must pass for the build job to execute, enhancing the reliability of the CI pipeline.
139 lines
5.0 KiB
Python
139 lines
5.0 KiB
Python
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
|
|
|
|
|
|
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(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, 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": 1}
|
|
|
|
# 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)
|