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

Add backend test job to CI workflow

Browse Source 
- 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.
This commit is contained in:
AlexeerCT
2026-03-24 12:33:21 -05:00
parent 23509cbcc9
commit b750c35522
11 changed files with 1104 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

37
.gitea/workflows/build.yml
View File

@@ -17,8 +17,45 @@ on:
- main
jobs:
# Ejecuta la suite de tests del backend antes de construir imágenes.
# Si falla, no se ejecuta build (ni push ni deploy).
test:
runs-on: self-hosted
steps:
- name: Checkout código
uses: actions/checkout@v4
- name: Configurar Python
uses: actions/setup-python@v5
with:
python-version: "3.11"
cache: "pip"
cache-dependency-path: backend/requirements.txt
- name: Instalar dependencias del backend
run: |
python -m pip install --upgrade pip
pip install -r backend/requirements.txt
- name: Comprobar TEST_DATABASE_URL
env:
TEST_DATABASE_URL: ${{ secrets.TEST_DATABASE_URL }}
run: |
if [ -z "$TEST_DATABASE_URL" ]; then
echo "::error::Define el secret TEST_DATABASE_URL en el repo (Gitea → Ajustes → Secretos)."
echo "Ejemplo: postgresql://usuario:clave@host:5432/nombre_bd"
exit 1
fi
- name: Ejecutar tests del backend
env:
TEST_DATABASE_URL: ${{ secrets.TEST_DATABASE_URL }}
working-directory: backend
run: pytest -q tests
build:
runs-on: self-hosted
needs: test
steps:
- name: Checkout código

34
backend/tests/README.md Normal file
View File

@@ -0,0 +1,34 @@
# Backend test strategy (Anexo24)
## Test suites
- `tests/e2e/`: full business flow (catalogs -> import -> export -> balances/discharges).
- `tests/integration/`: process API behavior and edge cases.
- `tests/unit/`: critical algorithms only (no trivial CRUD tests).
## Prerequisites
- PostgreSQL test database available.
- Environment variable:
- `TEST_DATABASE_URL=postgresql://user:pass@host:5432/db_name`
## Run commands
- Fast subset (CI gate):
- `pytest -q tests/unit tests/integration`
- Full suite:
- `pytest -q tests`
- `pytest tests -v -ra`
- `pytest tests -v -ra -s`
- E2E only:
- `pytest -q tests/e2e`
## CI recommendations
- Pull request gate:
- Run unit + integration on every PR.
- Nightly / main branch:
- Run full suite including E2E.
- Keep Celery eager mode for deterministic process endpoint tests:
- `task_always_eager=True`
- `task_eager_propagates=True`

1
backend/tests/__init__.py Normal file
View File

@@ -0,0 +1 @@

138
backend/tests/conftest.py Normal file
View File

@@ -0,0 +1,138 @@
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)

113
backend/tests/e2e/test_inventory_flow_anexo24.py Normal file
View File

@@ -0,0 +1,113 @@
from decimal import Decimal
from api.v1.modules.a24.balance_movements.models import BalanceMovement, MovementType
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,
)
def _patch_import_pipeline(monkeypatch, import_line):
monkeypatch.setattr(import_main, "pre_validators", lambda *args, **kwargs: [import_line])
monkeypatch.setattr(import_main, "review_classes", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_exchange_rate", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_kgs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_lbs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_sisimp_limits", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_lines", lambda *args, **kwargs: ([], {}))
def _patch_export_pipeline(monkeypatch, export_line):
monkeypatch.setattr(export_main, "pre_validators", lambda *args, **kwargs: [export_line])
monkeypatch.setattr(export_main, "review_class", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_exchange_rate", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "assign_values", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_qty_vs_weight", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_unit_cost", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_qty_series", lambda *args, **kwargs: None)
def test_e2e_inventory_flow_import_then_export(client, db_session, monkeypatch):
"""
E2E principal:
- Alta catálogos/base
- Procesa factura IMP -> genera entry
- Procesa factura EXP -> genera consumo/descarga
- Valida decremento de saldo sin negativos
"""
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)
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")
)
_patch_import_pipeline(monkeypatch, import_line)
response = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id=1")
assert response.status_code == 200
entry_movements = (
db_session.query(BalanceMovement)
.filter(
BalanceMovement.import_invoice_id == import_invoice.id,
BalanceMovement.import_item_line_id == import_line.id,
BalanceMovement.movement_type == MovementType.ENTRY,
)
.all()
)
assert entry_movements
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")
)
_patch_export_pipeline(monkeypatch, export_line)
response = client.post(f"/api/v1/a76/invoices/{export_invoice.id}/process?company_id=1")
assert response.status_code == 200
consumptions = (
db_session.query(BalanceMovement)
.filter(
BalanceMovement.source_invoice_id == export_invoice.id,
BalanceMovement.movement_type == MovementType.CONSUMPTION,
)
.all()
)
assert consumptions, "El proceso de exportación debe generar consumos"
discharge_headers = (
db_session.query(DischargeHeader)
.filter(DischargeHeader.source_invoice_id == export_invoice.id)
.all()
)
assert discharge_headers
discharge_details = (
db_session.query(DischargeDetail)
.filter(DischargeDetail.export_item_line_id == export_line.id)
.all()
)
assert discharge_details
assert all(detail.movement_id is not None for detail in discharge_details)
net_balance = db_session.query(BalanceMovement).filter(
BalanceMovement.import_item_line_id == import_line.id
).all()
signed = Decimal("0")
for mov in net_balance:
qty = Decimal(str(mov.quantity or 0))
if mov.movement_type in {MovementType.CONSUMPTION, MovementType.WASTE, MovementType.SCRAP, MovementType.DESTRUCTION}:
signed -= qty
else:
signed += qty
assert signed >= 0
assert signed < Decimal("10")

1
backend/tests/fixtures/__init__.py vendored Normal file
View File

@@ -0,0 +1 @@

396
backend/tests/fixtures/builders.py vendored Normal file
View File

@@ -0,0 +1,396 @@
from __future__ import annotations
from datetime import date, datetime
from decimal import Decimal
from sqlalchemy.orm import Session
from api.v1.modules.a24.fa.fa_item_lines.models import FaLineItem
from api.v1.modules.a76.clients_and_providers.models import (
ClientOrProviderEnum,
ClientProvider,
ClientProviderAddress,
)
from api.v1.modules.a76.customs_brokers.models import CustomsBroker
from api.v1.modules.a76.general_catalogs.company.models import Company
from api.v1.modules.a76.general_catalogs.exchange_rate.models import ExchangeRate
from api.v1.modules.a76.general_catalogs.units_of_measure.models import UnitOfMeasure
from api.v1.modules.a76.invoices.models import (
Currency,
InvoiceComplianceMx,
InvoiceFinancials,
InvoiceHeader,
InvoiceLogistics,
InvoiceStatus,
OperationType,
)
from api.v1.modules.a76.items.line_customs.models import LineCustom
from api.v1.modules.a76.items.line_descriptions.models import LineDescription
from api.v1.modules.a76.items.line_financials.models import LineFinancial
from api.v1.modules.a76.items.line_quantities.models import LineQuantity
from api.v1.modules.a76.items.models import LineItem
from api.v1.modules.core.tenants.models import Tenant, TenantType
from api.v1.modules.public.reference_data.invoice_types.models import InvoiceType
from api.v1.modules.public.reference_data.pedimento_regimens.models import RegimenPedimento
def ensure_reference_data(db: Session) -> None:
for key, desc in [
("TEM", "IMPORTACION TEMPORAL"),
("DEF", "IMPORTACION DEFINITIVA"),
("MEX", "COMPRAS MEXICANAS"),
("DONAC", "DONACION"),
]:
if db.get(InvoiceType, key) is None:
operation = "exp" if key == "DONAC" else "imp"
db.add(
InvoiceType(
key=key,
description=desc,
note="seed for tests",
type="both",
operation=operation,
)
)
if db.get(RegimenPedimento, "A1") is None:
db.add(
RegimenPedimento(
code="A1",
description="Regimen de prueba",
)
)
db.flush()
def ensure_tenant_company(db: Session, tenant_id: int = 1, company_id: int = 1) -> 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}",
type=TenantType.SHARED,
keycloak_realm="test",
is_active=True,
)
db.add(tenant)
company = db.get(Company, company_id)
if company is None:
company = Company(
id=company_id,
tenant_id=tenant_id,
name=f"Company {company_id}",
rfc="TST010101AAA",
prosec=False,
)
db.add(company)
db.flush()
return company
def create_business_catalogs(db: Session, tenant_id: int, company_id: int) -> dict:
provider = (
db.query(ClientProvider)
.filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
ClientProvider.name == "Proveedor Test",
)
.first()
)
if provider is None:
provider = ClientProvider(
tenant_id=tenant_id,
company_id=company_id,
name="Proveedor Test",
client_or_provider=ClientOrProviderEnum.BOTH,
is_active=True,
)
db.add(provider)
db.flush()
if provider.address is None:
provider.address = ClientProviderAddress(
id=provider.id,
client_id=provider.id,
tenant_id=tenant_id,
company_id=company_id,
country="MEX",
)
sold_to = (
db.query(ClientProvider)
.filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
ClientProvider.name == "Cliente Test",
)
.first()
)
if sold_to is None:
sold_to = ClientProvider(
tenant_id=tenant_id,
company_id=company_id,
name="Cliente Test",
client_or_provider=ClientOrProviderEnum.BOTH,
is_active=True,
)
db.add(sold_to)
db.flush()
if sold_to.address is None:
sold_to.address = ClientProviderAddress(
id=sold_to.id,
client_id=sold_to.id,
tenant_id=tenant_id,
company_id=company_id,
country="USA",
)
shipped_to = (
db.query(ClientProvider)
.filter(
ClientProvider.tenant_id == tenant_id,
ClientProvider.company_id == company_id,
ClientProvider.name == "Destinatario Test",
)
.first()
)
if shipped_to is None:
shipped_to = ClientProvider(
tenant_id=tenant_id,
company_id=company_id,
name="Destinatario Test",
client_or_provider=ClientOrProviderEnum.BOTH,
is_active=True,
)
db.add(shipped_to)
db.flush()
if shipped_to.address is None:
shipped_to.address = ClientProviderAddress(
id=shipped_to.id,
client_id=shipped_to.id,
tenant_id=tenant_id,
company_id=company_id,
country="USA",
)
broker = (
db.query(CustomsBroker)
.filter(
CustomsBroker.tenant_id == tenant_id,
CustomsBroker.company_id == company_id,
CustomsBroker.broker_key == "A1234",
)
.first()
)
if broker is None:
broker = CustomsBroker(
tenant_id=tenant_id,
company_id=company_id,
broker_key="A1234",
name="Agente Test",
)
db.add(broker)
db.flush()
uom = (
db.query(UnitOfMeasure)
.filter(
UnitOfMeasure.tenant_id == tenant_id,
UnitOfMeasure.company_id == company_id,
UnitOfMeasure.code == "PZA",
)
.first()
)
if uom is None:
uom = UnitOfMeasure(
tenant_id=tenant_id,
company_id=company_id,
code="PZA",
description="Pieza",
)
db.add(uom)
db.flush()
exr = (
db.query(ExchangeRate)
.filter(
ExchangeRate.tenant_id == tenant_id,
ExchangeRate.company_id == company_id,
ExchangeRate.date == datetime(2026, 3, 17, 0, 0, 0),
)
.first()
)
if exr is None:
exr = ExchangeRate(
tenant_id=tenant_id,
company_id=company_id,
date=datetime(2026, 3, 17, 0, 0, 0),
value=Decimal("17.250000"),
local_currency="MN",
foreign_currency="ME",
)
db.add(exr)
db.flush()
return {
"provider_id": provider.id,
"sold_to_id": sold_to.id,
"shipped_to_id": shipped_to.id,
"broker_id": broker.id,
"uom_id": uom.id,
}
def create_import_invoice_with_line(
db: Session,
tenant_id: int,
company_id: int,
catalogs: dict,
invoice_type: str = "TEM",
invoice_number: str = "IMP-0001",
qty: Decimal = Decimal("10"),
) -> tuple[InvoiceHeader, LineItem]:
invoice = InvoiceHeader(
tenant_id=tenant_id,
company_id=company_id,
system="fixed_asset",
operation_type=OperationType.IMP,
invoice_type=invoice_type,
document_type="A1",
invoice_number=invoice_number,
invoice_date=date(2026, 3, 17),
status=InvoiceStatus.PENDING,
)
invoice.compliance_mx = InvoiceComplianceMx(
tenant_id=tenant_id,
company_id=company_id,
provider_id=catalogs["provider_id"],
sold_to_id=catalogs["sold_to_id"],
shipped_to_id=catalogs["shipped_to_id"],
customs_broker_id=catalogs["broker_id"],
is_pedimento_pending=True,
was_reviewed_by_company=True,
)
invoice.financials = InvoiceFinancials(
tenant_id=tenant_id,
company_id=company_id,
currency=Currency.FOREIGN,
exchange_rate=Decimal("17.250000"),
iva_factor="16",
)
invoice.logistics = InvoiceLogistics(
tenant_id=tenant_id,
company_id=company_id,
weight_type="kgs",
)
db.add(invoice)
db.flush()
line = LineItem(
tenant_id=tenant_id,
company_id=company_id,
invoice_id=invoice.id,
line_number=1,
unit_of_measure=catalogs["uom_id"],
)
line.quantity = LineQuantity(
quantity=qty,
net_weight=Decimal("50"),
gross_weight=Decimal("55"),
package_quantity=1,
)
line.financial = LineFinancial(
unit_cost_capture=Decimal("10"),
value_usd=Decimal("100"),
value_mxn=Decimal("1725"),
)
line.customs = LineCustom(origin_country="MEX", fraction_type="GENERAL")
line.description = LineDescription(has_serial=False)
db.add(line)
db.flush()
return invoice, line
def create_export_invoice_with_line(
db: Session,
tenant_id: int,
company_id: int,
catalogs: dict,
source_import_invoice: InvoiceHeader,
source_import_line: LineItem,
qty: Decimal = Decimal("4"),
) -> tuple[InvoiceHeader, LineItem]:
invoice = InvoiceHeader(
tenant_id=tenant_id,
company_id=company_id,
system="fixed_asset",
operation_type=OperationType.EXP,
invoice_type="DONAC",
document_type="A1",
invoice_number="EXP-0001",
invoice_date=date(2026, 3, 18),
status=InvoiceStatus.PENDING,
)
invoice.compliance_mx = InvoiceComplianceMx(
tenant_id=tenant_id,
company_id=company_id,
provider_id=catalogs["provider_id"],
sold_to_id=catalogs["sold_to_id"],
shipped_to_id=catalogs["shipped_to_id"],
customs_broker_id=catalogs["broker_id"],
is_pedimento_pending=True,
was_reviewed_by_company=True,
)
invoice.financials = InvoiceFinancials(
tenant_id=tenant_id,
company_id=company_id,
currency=Currency.FOREIGN,
exchange_rate=Decimal("17.250000"),
iva_factor="16",
)
invoice.logistics = InvoiceLogistics(
tenant_id=tenant_id,
company_id=company_id,
weight_type="kgs",
)
db.add(invoice)
db.flush()
line = LineItem(
tenant_id=tenant_id,
company_id=company_id,
invoice_id=invoice.id,
line_number=1,
unit_of_measure=catalogs["uom_id"],
)
line.quantity = LineQuantity(
quantity=qty,
net_weight=Decimal("20"),
gross_weight=Decimal("22"),
package_quantity=1,
)
line.financial = LineFinancial(
unit_cost_capture=Decimal("12"),
value_usd=Decimal("48"),
value_mxn=Decimal("828"),
)
line.customs = LineCustom(origin_country="MEX", origin_procedure="TEM", fraction_type="GENERAL")
line.description = LineDescription(has_serial=False)
db.add(line)
db.flush()
db.add(
FaLineItem(
id=line.id,
tenant_id=tenant_id,
company_id=company_id,
discharge=True,
search_invoice=source_import_invoice.invoice_number,
search_line=source_import_line.line_number,
is_subitem=False,
)
)
db.flush()
return invoice, line

127
backend/tests/integration/test_export_process_api.py Normal file
View File

@@ -0,0 +1,127 @@
from decimal import Decimal
import pytest
from api.v1.modules.a24.balance_movements.models import BalanceMovement, MovementType
from api.v1.modules.a76.invoices.exports.process import main_process as export_main
from api.v1.modules.a76.invoices.imports.process import main_process as import_main
from core.exceptions import ValidationException
from tests.fixtures.builders import (
create_business_catalogs,
create_export_invoice_with_line,
create_import_invoice_with_line,
ensure_reference_data,
ensure_tenant_company,
)
def _patch_import_pipeline(monkeypatch, import_line):
monkeypatch.setattr(import_main, "pre_validators", lambda *args, **kwargs: [import_line])
monkeypatch.setattr(import_main, "review_classes", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_exchange_rate", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_kgs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_lbs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_sisimp_limits", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_lines", lambda *args, **kwargs: ([], {}))
def _patch_import_pipeline_without_prevalidators(monkeypatch):
monkeypatch.setattr(import_main, "review_classes", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_exchange_rate", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_kgs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_lbs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_sisimp_limits", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_lines", lambda *args, **kwargs: ([], {}))
def _patch_export_pipeline(monkeypatch, export_line):
monkeypatch.setattr(export_main, "pre_validators", lambda *args, **kwargs: [export_line])
monkeypatch.setattr(export_main, "review_class", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_exchange_rate", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "assign_values", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_qty_vs_weight", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_unit_cost", lambda *args, **kwargs: None)
monkeypatch.setattr(export_main, "review_qty_series", lambda *args, **kwargs: None)
def _prepare_inventory(client, db_session, monkeypatch):
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)
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")
)
_patch_import_pipeline(monkeypatch, import_line)
resp = client.post(f"/api/v1/a76/invoices/{import_invoice.id}/process?company_id=1")
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)
export_invoice, export_line = create_export_invoice_with_line(
db_session, 1, 1, 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")
assert resp.status_code == 200
consumptions = (
db_session.query(BalanceMovement)
.filter(
BalanceMovement.source_invoice_id == export_invoice.id,
BalanceMovement.movement_type == MovementType.CONSUMPTION,
)
.all()
)
assert consumptions
def test_process_export_prevents_negative_balance(client, db_session, monkeypatch):
catalogs, import_invoice, import_line = _prepare_inventory(client, db_session, monkeypatch)
export_invoice, export_line = create_export_invoice_with_line(
db_session, 1, 1, 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")
# Debe no crear consumos al existir insuficiencia de saldo.
consumptions = (
db_session.query(BalanceMovement)
.filter(
BalanceMovement.source_invoice_id == export_invoice.id,
BalanceMovement.movement_type == MovementType.CONSUMPTION,
)
.all()
)
assert consumptions == []
def test_process_endpoint_prevents_double_processing_import(client, db_session, monkeypatch):
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)
invoice, line = create_import_invoice_with_line(
db_session, 1, 1, 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")
assert first.status_code == 200
with pytest.raises(ValidationException):
client.post(f"/api/v1/a76/invoices/{invoice.id}/process?company_id=1")
# No debe duplicar entradas para la misma factura/línea procesada.
entries = (
db_session.query(BalanceMovement)
.filter(
BalanceMovement.import_invoice_id == invoice.id,
BalanceMovement.import_item_line_id == line.id,
BalanceMovement.movement_type == MovementType.ENTRY,
)
.all()
)
assert len(entries) == 1

64
backend/tests/integration/test_import_process_api.py Normal file
View File

@@ -0,0 +1,64 @@
from decimal import Decimal
from api.v1.modules.a24.balance_movements.models import BalanceMovement, MovementType
from api.v1.modules.a76.invoices.imports.process import main_process as import_main
from tests.fixtures.builders import (
create_business_catalogs,
create_import_invoice_with_line,
ensure_reference_data,
ensure_tenant_company,
)
def _patch_import_pipeline(monkeypatch, import_line):
monkeypatch.setattr(import_main, "pre_validators", lambda *args, **kwargs: [import_line])
monkeypatch.setattr(import_main, "review_classes", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_exchange_rate", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_kgs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "review_weights_lbs", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_sisimp_limits", lambda *args, **kwargs: None)
monkeypatch.setattr(import_main, "_validate_lines", lambda *args, **kwargs: ([], {}))
def test_process_import_endpoint_creates_balance_entries(client, db_session, monkeypatch):
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)
invoice, line = create_import_invoice_with_line(
db_session, 1, 1, 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")
assert resp.status_code == 200
entries = (
db_session.query(BalanceMovement)
.filter(
BalanceMovement.import_invoice_id == invoice.id,
BalanceMovement.import_item_line_id == line.id,
BalanceMovement.movement_type == MovementType.ENTRY,
)
.all()
)
assert len(entries) >= 1
def test_process_import_def_does_not_create_balance_entries(client, db_session, monkeypatch):
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)
invoice, line = create_import_invoice_with_line(
db_session, 1, 1, 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")
assert resp.status_code == 200
entries = (
db_session.query(BalanceMovement)
.filter(BalanceMovement.import_invoice_id == invoice.id)
.all()
)
assert entries == []

121
backend/tests/unit/invoices/test_balance_algorithm.py Normal file
View File

@@ -0,0 +1,121 @@
from datetime import date
from decimal import Decimal
from api.v1.modules.a24.balance_movements.models import BalanceMovement, MovementType
from api.v1.modules.a76.invoices.exports.process.sub_process.compare_balances import compare_balances
from api.v1.modules.a76.invoices.exports.process.sub_process.discharge_types import (
AvailableLot,
DownloadEntry,
)
from api.v1.modules.a76.invoices.exports.process.sub_process.fill_available_balances import (
_net_balance_for_lot,
)
from core.exceptions import ErrorCollector
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):
"""
Balance neto esperado:
entry 10 - consumption 4 + return 1 - entry_void 2 = 5
"""
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)
import_invoice, import_line = create_import_invoice_with_line(
db_session,
1,
1,
catalogs,
invoice_type="TEM",
invoice_number="IMP-UNIT-01",
qty=Decimal("10"),
)
line_id = import_line.id
tenant_id = import_invoice.tenant_id
company_id = import_invoice.company_id
import_invoice_id = import_invoice.id
for idx, (mtype, qty) in enumerate(
[
(MovementType.ENTRY, Decimal("10")),
(MovementType.CONSUMPTION, Decimal("4")),
(MovementType.RETURN, Decimal("1")),
(MovementType.ENTRY_VOID, Decimal("2")),
],
start=1,
):
db_session.add(
BalanceMovement(
id=10000 + idx,
tenant_id=tenant_id,
company_id=company_id,
import_invoice_id=import_invoice_id,
import_item_line_id=line_id,
movement_type=mtype,
quantity=qty,
order_peps=10000 + idx,
operation_date=date(2026, 3, 20),
)
)
db_session.flush()
balance = _net_balance_for_lot(db_session, line_id, date(2026, 3, 21))
assert balance == Decimal("5")
def test_fifo_consumption_algorithm_uses_oldest_lots_first(db_session, monkeypatch):
"""
Verifica distribución FIFO (PEPS):
- lote1 (order=1, qty=3), lote2 (order=2, qty=5), demanda=6
- resultado: lote1 consume 3, lote2 consume 3, faltante 0
"""
monkeypatch.setattr(
"api.v1.modules.a76.invoices.exports.process.sub_process.compare_balances._resolve_import_uom",
lambda db, import_item_line_id: "PZA",
)
entry = DownloadEntry(
origin_procedure="TEM",
export_line=1,
part_number="PN-1",
class_code="CLS",
quantity=Decimal("6"),
quantity_used=Decimal("0"),
unit_of_measure="PZA",
import_invoice="IMP-1",
import_line=1,
line_item_id=1,
available_lots=[
AvailableLot(
import_item_line_id=11,
import_invoice_id=101,
part_number_id=None,
available_qty=Decimal("3"),
value_me=Decimal("0"),
value_mn=Decimal("0"),
order_peps=1,
),
AvailableLot(
import_item_line_id=12,
import_invoice_id=101,
part_number_id=None,
available_qty=Decimal("5"),
value_me=Decimal("0"),
value_mn=Decimal("0"),
order_peps=2,
),
],
)
errors = ErrorCollector()
compare_balances(db_session, export_invoice=None, to_discharge=[entry], errors=errors) # type: ignore[arg-type]
assert not errors.has_errors()
assert entry.quantity_used == Decimal("6")
assert entry.available_lots[0].consumed_qty == Decimal("3")
assert entry.available_lots[1].consumed_qty == Decimal("3")

72
backend/tests/unit/invoices/test_currency_conversion_imports.py Normal file
View File

@@ -0,0 +1,72 @@
from decimal import Decimal
from types import SimpleNamespace
from api.v1.modules.a76.invoices.models import Currency
from api.v1.modules.a76.invoices.imports.process.sub_process.assing_values_def_mex import (
assign_values_iva_lines,
)
def _mk_line(qty: str, capture: str):
return SimpleNamespace(
quantity=SimpleNamespace(quantity=Decimal(qty)),
financial=SimpleNamespace(
unit_cost_capture=Decimal(capture),
unit_cost_usd=None,
unit_cost_mxn=None,
unit_cost_mc=None,
sub_import_value_usd=None,
sub_import_value_mxn=None,
sub_import_value_mc=None,
vat_usd=None,
vat_mxn=None,
vat_mc=None,
value_usd=None,
value_mxn=None,
value_mc=None,
),
)
def _mk_invoice(currency: Currency):
return SimpleNamespace(
financials=SimpleNamespace(
currency=currency,
exchange_rate=Decimal("17.25"),
exchange_rate_mm=Decimal("1.10"),
iva_factor="16",
)
)
def test_assign_values_iva_lines_currency_me():
invoice = _mk_invoice(Currency.FOREIGN)
line = _mk_line("10", "2")
assign_values_iva_lines(invoice, [line])
assert line.financial.sub_import_value_usd == Decimal("20")
assert line.financial.vat_usd == Decimal("3.2")
assert line.financial.value_usd == Decimal("23.2")
assert line.financial.unit_cost_mxn == Decimal("34.50")
def test_assign_values_iva_lines_currency_mn():
invoice = _mk_invoice(Currency.LOCAL)
line = _mk_line("5", "100")
assign_values_iva_lines(invoice, [line])
assert line.financial.sub_import_value_mxn == Decimal("500")
assert line.financial.vat_mxn == Decimal("80")
assert line.financial.value_mxn == Decimal("580")
assert line.financial.unit_cost_usd == Decimal("5.797101449275362318840579710")
def test_assign_values_iva_lines_currency_mc():
invoice = _mk_invoice(Currency.MANUAL)
line = _mk_line("4", "3")
assign_values_iva_lines(invoice, [line])
assert line.financial.sub_import_value_mc == Decimal("12")
assert line.financial.vat_mc == Decimal("1.92")
assert line.financial.value_mc == Decimal("13.92")
assert line.financial.unit_cost_usd == Decimal("3.30")