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.
128 lines
5.8 KiB
Python
128 lines
5.8 KiB
Python
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
|