ADUANASOFT/plantillas-proyectos
13
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor test setup to utilize ephemeral tenant IDs

Browse Source 
- Introduced a new fixture, `test_tenant`, to allocate ephemeral tenant and company IDs for tests, reducing conflicts with real data.
- Updated various test functions to use the new `test_tenant` fixture, ensuring consistent tenant ID usage across tests.
- Enhanced the `ensure_tenant_company` function to accept dynamic tenant and company IDs, improving flexibility in test scenarios.
- Adjusted database interaction in tests to utilize the ephemeral IDs, streamlining the setup process and enhancing isolation.
This commit is contained in:
AlexeerCT
2026-03-24 17:24:01 -05:00
parent 1150b59dcf
commit 06b3b20fd8
8 changed files with 136 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
backend/alembic/versions/bccb7f8986c7_iva_factor.py
View File

@@ -18,11 +18,31 @@ branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def _public_table_exists(bind, name: str) -> bool:
"""Evita CREATE INDEX si la tabla no existe (BD parcial / orden atípico de migraciones)."""
return name in sa.inspect(bind).get_table_names(schema="public")
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_index(op.f('ix_public_carta_porte_code'), table_name='carta_porte_codes')
op.create_index(op.f('ix_public_carta_porte_codes_code'), 'carta_porte_codes', ['code'], unique=False, schema='public')
bind = op.get_bind()
if _public_table_exists(bind, "carta_porte_codes"):
# Coincide con 4ad64605fad2 (schema=public). if_exists: sin índice previo.
op.drop_index(
op.f("ix_public_carta_porte_code"),
table_name="carta_porte_codes",
schema="public",
if_exists=True,
)
op.create_index(
op.f("ix_public_carta_porte_codes_code"),
"carta_porte_codes",
["code"],
unique=False,
schema="public",
if_not_exists=True,
)
op.alter_column('invoice_financials', 'iva_factor',
existing_type=sa.VARCHAR(length=10),
type_=sa.Numeric(precision=23, scale=8),
@@ -40,6 +60,20 @@ def downgrade() -> None:
type_=sa.VARCHAR(length=10),
existing_nullable=True,
schema='a76')
op.drop_index(op.f('ix_public_carta_porte_codes_code'), table_name='carta_porte_codes', schema='public')
op.create_index(op.f('ix_public_carta_porte_code'), 'carta_porte_codes', ['code'], unique=False)
bind = op.get_bind()
if _public_table_exists(bind, "carta_porte_codes"):
op.drop_index(
op.f("ix_public_carta_porte_codes_code"),
table_name="carta_porte_codes",
schema="public",
if_exists=True,
)
op.create_index(
op.f("ix_public_carta_porte_code"),
"carta_porte_codes",
["code"],
unique=False,
schema="public",
if_not_exists=True,
)
# ### end Alembic commands ###

4
backend/tests/README.md
View File

@@ -44,14 +44,14 @@ flowchart TB
- **BD**: `TEST_DATABASE_URL` o `CORE_DATABASE_URL` o `settings.core_database_url`. Cada test corre dentro de una transacción; al terminar **no persiste** nada (rollback).
- **Celery**: sin Redis; `task_always_eager` para que el endpoint dispare lógica síncrona.
- **Seguridad en tests**: `validate_access_to_resource` se reemplaza por un stub que devuelve `tenant_id` del usuario fake.
- **Seguridad en tests**: `validate_access_to_resource` se reemplaza por un stub que devuelve `tenant_id` del usuario fake (alineado con el fixture `test_tenant`).
### Builders: qué se construye antes de llamar a la API
Orden típico en casi todos los tests que tocan proceso:
1. `ensure_reference_data`: tipos de factura (`TEM`, `DEF`, `MEX`, `DONAC`) y régimen `A1` si no existen.
2. `ensure_tenant_company(tenant_id=1, company_id=1)`.
2. Fixture `test_tenant`: asigna IDs efímeros altos y crea tenant + company; el `app` usa ese `tenant_id` en el usuario fake.
3. `create_business_catalogs`: proveedor/cliente/destinatario (con dirección idempotente vía `_ensure_client_provider_address`), agente aduanal, UoM, part number FA, tipo cambio, etc.
4. `create_import_invoice_with_line` / `create_export_invoice_with_line` según el caso.

21
backend/tests/conftest.py
View File

@@ -17,6 +17,11 @@ 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 = (
@@ -49,6 +54,14 @@ def db_session() -> Generator[Session, None, None]:
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
@@ -77,7 +90,11 @@ def celery_eager() -> Generator[None, None, None]:
@pytest.fixture
def app(db_session: Session, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
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")
@@ -85,7 +102,7 @@ def app(db_session: Session, monkeypatch: pytest.MonkeyPatch) -> FastAPI:
yield db_session
async def _override_current_user():
return {"sub": "test-user", "tenant_id": 1}
return {"sub": "test-user", "tenant_id": test_tenant.tenant_id}
# validate_access_to_resource is imported directly in the routes module.
monkeypatch.setattr(

16
backend/tests/e2e/test_inventory_flow_anexo24.py
View File

@@ -4,13 +4,11 @@ from api.v1.modules.a24.balance_movements.models import BalanceMovement, Movemen
from api.v1.modules.a24.discharges.models import DischargeDetail, DischargeHeader
from api.v1.modules.a76.invoices.imports.process import main_process as import_main
from api.v1.modules.a76.invoices.exports.process import main_process as export_main
from api.v1.modules.a76.invoices.exports.process import task as export_task
from tests.fixtures.builders import (
create_business_catalogs,
create_export_invoice_with_line,
create_import_invoice_with_line,
ensure_reference_data,
ensure_tenant_company,
)
@@ -34,7 +32,7 @@ def _patch_export_pipeline(monkeypatch, export_line):
monkeypatch.setattr(export_main, "review_qty_series", lambda *args, **kwargs: None)
def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch, test_tenant):
"""
E2E principal:
- Alta catálogos/base
@@ -42,16 +40,16 @@ def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
- Procesa factura EXP -> genera consumo/descarga
- Valida decremento de saldo sin negativos
"""
tid, cid = test_tenant.tenant_id, test_tenant.company_id
ensure_reference_data(db_session)
ensure_tenant_company(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
import_invoice, import_line = create_import_invoice_with_line(
db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-E2E-01", qty=Decimal("10")
db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-E2E-01", qty=Decimal("10")
)
_patch_import_pipeline(monkeypatch, import_line)
response = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id=1")
response = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id={cid}")
assert response.status_code == 200
entry_movements = (
@@ -67,11 +65,11 @@ def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
assert sum(Decimal(str(m.quantity or 0)) for m in entry_movements) > 0
export_invoice, export_line = create_export_invoice_with_line(
db_session, 1, 1, catalogs, import_invoice, import_line, qty=Decimal("4")
db_session, tid, cid, catalogs, import_invoice, import_line, qty=Decimal("4")
)
_patch_export_pipeline(monkeypatch, export_line)
response = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
response = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id={cid}")
assert response.status_code == 200
consumptions = (

31
backend/tests/fixtures/builders.py vendored
View File

@@ -1,7 +1,9 @@
from __future__ import annotations
import uuid
from datetime import date, datetime
from decimal import Decimal
from typing import NamedTuple
from sqlalchemy.orm import Session
@@ -34,6 +36,31 @@ from api.v1.modules.public.reference_data.invoice_types.models import InvoiceTyp
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
class EphemeralOrgIds(NamedTuple):
"""Tenant + company creados por test (IDs altos, baja colisión con datos reales / CI)."""
tenant_id: int
company_id: int
def allocate_ephemeral_tenant_company_ids(db: Session) -> tuple[int, int]:
"""
Genera par (tenant_id, company_id) no usados en esta sesión.
Mismo entero para ambas PKs es válido (tablas distintas).
"""
for _ in range(64):
n = uuid.uuid4().int
tid = 1_500_000_000 + (n % 99_000_000)
if db.get(Tenant, tid) is not None:
continue
if db.get(Company, tid) is not None:
continue
return tid, tid
raise RuntimeError(
"No se pudo asignar tenant/company efímero para tests (reintentar o revisar datos en BD)."
)
def ensure_reference_data(db: Session) -> None:
for key, desc in [
("TEM", "IMPORTACION TEMPORAL"),
@@ -63,13 +90,13 @@ def ensure_reference_data(db: Session) -> None:
db.flush()
def ensure_tenant_company(db: Session, tenant_id: int = 1, company_id: int = 1) -> Company:
def ensure_tenant_company(db: Session, tenant_id: int, company_id: int) -> Company:
tenant = db.get(Tenant, tenant_id)
if tenant is None:
tenant = Tenant(
id=tenant_id,
name=f"Tenant {tenant_id}",
slug=f"tenant-{tenant_id}",
slug=f"pytest-{tenant_id}-{uuid.uuid4().hex[:10]}",
type=TenantType.SHARED,
keycloak_realm="test",
is_active=True,

46
backend/tests/integration/test_export_process_api.py
View File

@@ -7,11 +7,11 @@ from api.v1.modules.a76.invoices.exports.process import main_process as export_m
from api.v1.modules.a76.invoices.imports.process import main_process as import_main
from core.exceptions import ValidationException
from tests.fixtures.builders import (
EphemeralOrgIds,
create_business_catalogs,
create_export_invoice_with_line,
create_import_invoice_with_line,
ensure_reference_data,
ensure_tenant_company,
)
@@ -44,27 +44,30 @@ def _patch_export_pipeline(monkeypatch, export_line):
monkeypatch.setattr(export_main, "review_qty_series", lambda *args, **kwargs: None)
def _prepare_inventory(client, db_session, monkeypatch):
def _prepare_inventory(client, db_session, monkeypatch, test_tenant: EphemeralOrgIds):
tid, cid = test_tenant.tenant_id, test_tenant.company_id
ensure_reference_data(db_session)
ensure_tenant_company(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
import_invoice, import_line = create_import_invoice_with_line(
db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-EXP-01", qty=Decimal("10")
db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-EXP-01", qty=Decimal("10")
)
_patch_import_pipeline(monkeypatch, import_line)
resp = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id=1")
resp = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id={cid}")
assert resp.status_code == 200
return catalogs, import_invoice, import_line
def test_process_export_endpoint_consumes_existing_balances(client, db_session, monkeypatch):
catalogs, import_invoice, import_line = _prepare_inventory(client, db_session, monkeypatch)
def test_process_export_endpoint_consumes_existing_balances(client, db_session, monkeypatch, test_tenant):
catalogs, import_invoice, import_line = _prepare_inventory(
client, db_session, monkeypatch, test_tenant
)
tid, cid = test_tenant.tenant_id, test_tenant.company_id
export_invoice, export_line = create_export_invoice_with_line(
db_session, 1, 1, catalogs, import_invoice, import_line, qty=Decimal("3")
db_session, tid, cid, catalogs, import_invoice, import_line, qty=Decimal("3")
)
_patch_export_pipeline(monkeypatch, export_line)
resp = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
resp = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id={cid}")
assert resp.status_code == 200
consumptions = (
@@ -78,15 +81,18 @@ def test_process_export_endpoint_consumes_existing_balances(client, db_session,
assert consumptions
def test_process_export_prevents_negative_balance(client, db_session, monkeypatch):
catalogs, import_invoice, import_line = _prepare_inventory(client, db_session, monkeypatch)
def test_process_export_prevents_negative_balance(client, db_session, monkeypatch, test_tenant):
catalogs, import_invoice, import_line = _prepare_inventory(
client, db_session, monkeypatch, test_tenant
)
tid, cid = test_tenant.tenant_id, test_tenant.company_id
export_invoice, export_line = create_export_invoice_with_line(
db_session, 1, 1, catalogs, import_invoice, import_line, qty=Decimal("99")
db_session, tid, cid, catalogs, import_invoice, import_line, qty=Decimal("99")
)
_patch_export_pipeline(monkeypatch, export_line)
with pytest.raises(ValidationException):
client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id={cid}")
# Debe no crear consumos al existir insuficiencia de saldo.
consumptions = (
@@ -100,19 +106,19 @@ def test_process_export_prevents_negative_balance(client, db_session, monkeypatc
assert consumptions == []
def test_process_endpoint_prevents_double_processing_import(client, db_session, monkeypatch):
def test_process_endpoint_prevents_double_processing_import(client, db_session, monkeypatch, test_tenant):
tid, cid = test_tenant.tenant_id, test_tenant.company_id
ensure_reference_data(db_session)
ensure_tenant_company(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
invoice, line = create_import_invoice_with_line(
db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-DOUBLE-01", qty=Decimal("5")
db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-DOUBLE-01", qty=Decimal("5")
)
_patch_import_pipeline_without_prevalidators(monkeypatch)
first = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
first = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
assert first.status_code == 200
with pytest.raises(ValidationException):
client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
# No debe duplicar entradas para la misma factura/línea procesada.
entries = (

21
backend/tests/integration/test_import_process_api.py
View File

@@ -6,7 +6,6 @@ from tests.fixtures.builders import (
create_business_catalogs,
create_import_invoice_with_line,
ensure_reference_data,
ensure_tenant_company,
)
@@ -20,16 +19,16 @@ def _patch_import_pipeline(monkeypatch, import_line):
monkeypatch.setattr(import_main, "_validate_lines", lambda *args, **kwargs: ([], {}))
def test_process_import_endpoint_creates_balance_entries(client, db_session, monkeypatch):
def test_process_import_endpoint_creates_balance_entries(client, db_session, monkeypatch, test_tenant):
tid, cid = test_tenant.tenant_id, test_tenant.company_id
ensure_reference_data(db_session)
ensure_tenant_company(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
invoice, line = create_import_invoice_with_line(
db_session, 1, 1, catalogs, invoice_type="TEM", invoice_number="IMP-INT-01", qty=Decimal("7")
db_session, tid, cid, catalogs, invoice_type="TEM", invoice_number="IMP-INT-01", qty=Decimal("7")
)
_patch_import_pipeline(monkeypatch, line)
resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
assert resp.status_code == 200
entries = (
@@ -44,16 +43,16 @@ def test_process_import_endpoint_creates_balance_entries(client, db_session, mon
assert len(entries) >= 1
def test_process_import_def_does_not_create_balance_entries(client, db_session, monkeypatch):
def test_process_import_def_does_not_create_balance_entries(client, db_session, monkeypatch, test_tenant):
tid, cid = test_tenant.tenant_id, test_tenant.company_id
ensure_reference_data(db_session)
ensure_tenant_company(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
invoice, line = create_import_invoice_with_line(
db_session, 1, 1, catalogs, invoice_type="DEF", invoice_number="IMP-DEF-01", qty=Decimal("7")
db_session, tid, cid, catalogs, invoice_type="DEF", invoice_number="IMP-DEF-01", qty=Decimal("7")
)
_patch_import_pipeline(monkeypatch, line)
resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
resp = client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id={cid}")
assert resp.status_code == 200
entries = (

11
backend/tests/unit/invoices/test_balance_algorithm.py
View File

@@ -15,22 +15,21 @@ from tests.fixtures.builders import (
create_business_catalogs,
create_import_invoice_with_line,
ensure_reference_data,
ensure_tenant_company,
)
def test_net_balance_accounts_for_returns_and_entry_void(db_session):
def test_net_balance_accounts_for_returns_and_entry_void(db_session, test_tenant):
"""
Balance neto esperado:
entry 10 - consumption 4 + return 1 - entry_void 2 = 5
"""
tid, cid = test_tenant.tenant_id, test_tenant.company_id
ensure_reference_data(db_session)
ensure_tenant_company(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=1, company_id=1)
catalogs = create_business_catalogs(db_session, tenant_id=tid, company_id=cid)
import_invoice, import_line = create_import_invoice_with_line(
db_session,
1,
1,
tid,
cid,
catalogs,
invoice_type="TEM",
invoice_number="IMP-UNIT-01",