consultoria-as
a386ca5b4a
- F0 blindaje: Redis/Celery, pg_hba sin trust LAN, secretos a env 600, backups, webhook auth, jobs muertos - F1 caja: bugs POS, reembolsos al corte, apartados, CFDI, topes de descuento en cotizacion/apartado - F2/F7 IA de catalogo: autocategorizacion, dedup de SKUs con embeddings, OCR de facturas, descripciones, claves SAT, limpieza de fitment - F3 IA en el flujo: vision Yonke, preguntas MeLi, venta cruzada, anomalias, sinonimos, busqueda semantica (rerank/embeddings) - F4/F8 escala y cartera: /metrics+Prometheus+Grafana, pool, score de credito, churn, metricas multi-tenant - Nexus Yonke (M1-M6): modelo de datos 1-de-1, alta por donador, vitrina publica, MeLi usado, consignacion, garantia por dias, reportes, etiquetas - Marketplace B2B entre yonkes (catalogo + pedidos cross-tenant) - Seguridad: auth en /api/admin/* (35 endpoints), alertas por tenant - Migraciones v4.27/v4.28; timers expire-demos y rollup-metrics
269 lines
10 KiB
Python
269 lines
10 KiB
Python
"""Unit tests for Facturapi service with mocked HTTP calls.
|
|
|
|
These tests do not require PostgreSQL or network access.
|
|
"""
|
|
|
|
import base64
|
|
from unittest import mock
|
|
|
|
import pytest
|
|
|
|
# Import must not trigger DB connections.
|
|
from pos.services import facturapi_service
|
|
|
|
|
|
@pytest.fixture
|
|
def user_key():
|
|
"""Patch USER_KEY for the duration of the test."""
|
|
with mock.patch.object(facturapi_service, "USER_KEY", "sk_user_abc"):
|
|
yield
|
|
|
|
|
|
class TestGetApiKey:
|
|
def test_prefers_secret_key(self):
|
|
config = {
|
|
"facturapi_secret_key": "sk_secret_123",
|
|
"facturapi_key": "sk_test_456",
|
|
}
|
|
assert facturapi_service.get_api_key(config) == "sk_secret_123"
|
|
|
|
def test_falls_back_to_facturapi_key(self):
|
|
config = {"facturapi_key": "sk_test_456"}
|
|
assert facturapi_service.get_api_key(config) == "sk_test_456"
|
|
|
|
def test_falls_back_to_cfdi_prefixed_key(self):
|
|
config = {"cfdi_facturapi_key": "sk_test_789"}
|
|
assert facturapi_service.get_api_key(config) == "sk_test_789"
|
|
|
|
def test_falls_back_to_user_key_env(self, user_key):
|
|
assert facturapi_service.get_api_key({}) == "sk_user_abc"
|
|
|
|
def test_raises_when_nothing_configured(self):
|
|
with (
|
|
mock.patch.object(facturapi_service, "USER_KEY", ""),
|
|
pytest.raises(facturapi_service.FacturapiError),
|
|
):
|
|
facturapi_service.get_api_key({})
|
|
|
|
|
|
class TestGetOrgId:
|
|
def test_prefers_short_key(self):
|
|
config = {"facturapi_org_id": "org_123", "cfdi_facturapi_org_id": "org_456"}
|
|
assert facturapi_service._get_org_id(config) == "org_123"
|
|
|
|
def test_falls_back_to_cfdi_prefixed_key(self):
|
|
config = {"cfdi_facturapi_org_id": "org_789"}
|
|
assert facturapi_service._get_org_id(config) == "org_789"
|
|
|
|
def test_returns_none_when_missing(self):
|
|
assert facturapi_service._get_org_id({}) is None
|
|
|
|
|
|
class TestRequest:
|
|
@mock.patch("pos.services.facturapi_service.requests.request")
|
|
def test_successful_request_returns_json(self, mock_request):
|
|
mock_response = mock.Mock()
|
|
mock_response.ok = True
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"id": "inv_1"}'
|
|
mock_response.json.return_value = {"id": "inv_1"}
|
|
mock_request.return_value = mock_response
|
|
|
|
result = facturapi_service._request("GET", "/invoices", "sk_test")
|
|
|
|
assert result == {"id": "inv_1"}
|
|
mock_request.assert_called_once()
|
|
_, kwargs = mock_request.call_args
|
|
assert kwargs["auth"] == ("sk_test", "")
|
|
|
|
@mock.patch("pos.services.facturapi_service.requests.request")
|
|
def test_failed_request_raises_facturapi_error(self, mock_request):
|
|
mock_response = mock.Mock()
|
|
mock_response.ok = False
|
|
mock_response.status_code = 400
|
|
mock_response.text = "Bad request"
|
|
mock_request.return_value = mock_response
|
|
|
|
with pytest.raises(facturapi_service.FacturapiError) as exc_info:
|
|
facturapi_service._request("POST", "/invoices", "sk_test", json_payload={})
|
|
|
|
assert exc_info.value.status_code == 400
|
|
|
|
|
|
class TestCreateOrganization:
|
|
@mock.patch("pos.services.facturapi_service.update_organization_legal")
|
|
@mock.patch("pos.services.facturapi_service.find_organization_by_rfc")
|
|
@mock.patch("pos.services.facturapi_service._request")
|
|
def test_creates_organization_and_generates_live_key(self, mock_request, mock_find, mock_legal, user_key):
|
|
mock_find.return_value = None
|
|
mock_request.side_effect = [
|
|
{"id": "org_123"}, # POST /organizations
|
|
{"key": "sk_live_abc"}, # PUT /organizations/org_123/apikeys/live
|
|
]
|
|
|
|
result = facturapi_service.create_organization({"rfc": "ABC010101AAA", "razon_social": "Test SA"})
|
|
|
|
assert result == {"org_id": "org_123", "api_key": "sk_live_abc", "legal_updated": True}
|
|
assert mock_request.call_count == 2
|
|
mock_legal.assert_called_once()
|
|
|
|
@mock.patch("pos.services.facturapi_service.update_organization_legal")
|
|
@mock.patch("pos.services.facturapi_service.find_organization_by_rfc")
|
|
@mock.patch("pos.services.facturapi_service._request")
|
|
def test_reuses_existing_organization_by_rfc(self, mock_request, mock_find, mock_legal, user_key):
|
|
mock_find.return_value = {"id": "org_existing"}
|
|
mock_request.return_value = {"key": "sk_live_existing"}
|
|
|
|
result = facturapi_service.create_organization({"rfc": "ABC010101AAA", "razon_social": "Test SA"})
|
|
|
|
assert result["org_id"] == "org_existing"
|
|
assert result["legal_updated"] is True
|
|
mock_request.assert_called_once()
|
|
|
|
|
|
class TestUploadCsd:
|
|
@mock.patch("pos.services.facturapi_service.requests.put")
|
|
def test_uploads_csd(self, mock_put):
|
|
mock_response = mock.Mock()
|
|
mock_response.ok = True
|
|
mock_response.json.return_value = {"certificate": {"has_certificate": True}}
|
|
mock_put.return_value = mock_response
|
|
|
|
config = {"facturapi_key": "sk_test", "facturapi_org_id": "org_1"}
|
|
cer_b64 = base64.b64encode(b"fake-cer").decode("ascii")
|
|
key_b64 = base64.b64encode(b"fake-key").decode("ascii")
|
|
|
|
result = facturapi_service.upload_csd(config, cer_b64, key_b64, "password")
|
|
|
|
assert result["certificate"]["has_certificate"] is True
|
|
mock_put.assert_called_once()
|
|
args, kwargs = mock_put.call_args
|
|
assert args[0].endswith("/organizations/org_1/certificate")
|
|
assert kwargs["auth"] == ("sk_test", "")
|
|
assert set(kwargs["files"]) == {"cer", "key", "password"}
|
|
assert kwargs["files"]["cer"][1] == b"fake-cer"
|
|
assert kwargs["files"]["key"][1] == b"fake-key"
|
|
assert kwargs["files"]["password"][1] == "password"
|
|
|
|
|
|
class TestCreateInvoice:
|
|
@mock.patch("pos.services.facturapi_service._request")
|
|
def test_creates_invoice(self, mock_request):
|
|
mock_request.return_value = {"id": "inv_1", "uuid": "uuid-1"}
|
|
|
|
config = {"facturapi_key": "sk_test"}
|
|
payload = {"customer": {"tax_id": "XAXX010101000"}}
|
|
|
|
result = facturapi_service.create_invoice(config, payload)
|
|
|
|
assert result["uuid"] == "uuid-1"
|
|
mock_request.assert_called_once_with("POST", "/invoices", "sk_test", json_payload=payload, timeout=90)
|
|
|
|
|
|
class TestCancelInvoice:
|
|
@mock.patch("pos.services.facturapi_service._request")
|
|
def test_cancel_invoice_with_replacement(self, mock_request):
|
|
mock_request.return_value = {"status": "canceled"}
|
|
|
|
config = {"facturapi_key": "sk_test"}
|
|
result = facturapi_service.cancel_invoice(config, "inv_1", "01", replacement_uuid="uuid-2")
|
|
|
|
assert result["status"] == "canceled"
|
|
mock_request.assert_called_once_with(
|
|
"DELETE",
|
|
"/invoices/inv_1",
|
|
"sk_test",
|
|
params={"motive": "01", "replacement": "uuid-2"},
|
|
timeout=60,
|
|
)
|
|
|
|
|
|
class TestDownloadXml:
|
|
@mock.patch("pos.services.facturapi_service.requests.request")
|
|
def test_downloads_xml(self, mock_request):
|
|
mock_response = mock.Mock()
|
|
mock_response.ok = True
|
|
mock_response.content = b"<xml/>"
|
|
mock_request.return_value = mock_response
|
|
|
|
config = {"facturapi_key": "sk_test"}
|
|
result = facturapi_service.download_xml(config, "inv_1")
|
|
|
|
assert result == b"<xml/>"
|
|
|
|
|
|
class TestGetOrgStatus:
|
|
@mock.patch("pos.services.facturapi_service.get_organization")
|
|
def test_returns_configured_with_csd(self, mock_get_org):
|
|
mock_get_org.return_value = {
|
|
"legal": {"name": "Test SA", "tax_id": "ABC010101AAA"},
|
|
"certificate": {"has_certificate": True},
|
|
"pending_steps": [],
|
|
}
|
|
|
|
config = {"facturapi_key": "sk_test", "cfdi_facturapi_org_id": "org_1"}
|
|
result = facturapi_service.get_org_status(config)
|
|
|
|
assert result["configured"] is True
|
|
assert result["has_csd"] is True
|
|
assert result["has_org_id"] is True
|
|
|
|
@mock.patch("pos.services.facturapi_service.get_organization")
|
|
def test_handles_organization_without_certificate(self, mock_get_org):
|
|
mock_get_org.return_value = {
|
|
"legal": {"name": "Test SA", "tax_id": "ABC010101AAA"},
|
|
"certificate": None,
|
|
}
|
|
|
|
config = {"facturapi_key": "sk_test", "cfdi_facturapi_org_id": "org_1"}
|
|
result = facturapi_service.get_org_status(config)
|
|
|
|
assert result["has_csd"] is False
|
|
assert result["configured"] is True
|
|
|
|
def test_returns_error_without_key(self):
|
|
with mock.patch.object(facturapi_service, "USER_KEY", ""):
|
|
result = facturapi_service.get_org_status({})
|
|
assert result["has_key"] is False
|
|
assert "not configured" in result["error"].lower()
|
|
|
|
def test_returns_error_without_org_id(self):
|
|
with mock.patch.object(facturapi_service, "USER_KEY", ""):
|
|
result = facturapi_service.get_org_status({"facturapi_key": "sk_test"})
|
|
assert result["has_org_id"] is False
|
|
assert "organization" in result["error"].lower()
|
|
|
|
@mock.patch("pos.services.facturapi_service.get_organization")
|
|
def test_reports_error_without_raising_when_fetch_fails(self, mock_get_org):
|
|
mock_get_org.side_effect = facturapi_service.FacturapiError("boom", status_code=500)
|
|
|
|
config = {"facturapi_key": "sk_live_x", "cfdi_facturapi_org_id": "org_1"}
|
|
with mock.patch.object(facturapi_service, "USER_KEY", ""):
|
|
result = facturapi_service.get_org_status(config)
|
|
|
|
assert result["has_csd"] is False
|
|
assert "boom" in result["error"]
|
|
|
|
|
|
class TestCreateOrUpdateCustomer:
|
|
@mock.patch("pos.services.facturapi_service._request")
|
|
def test_creates_new_customer_when_not_found(self, mock_request):
|
|
mock_request.side_effect = [
|
|
{"data": []}, # search
|
|
{"id": "cus_1"}, # create
|
|
]
|
|
|
|
config = {"facturapi_key": "sk_test"}
|
|
customer_id = facturapi_service.create_or_update_customer(
|
|
config,
|
|
{
|
|
"legal_name": "Test",
|
|
"tax_id": "ABC010101AAA",
|
|
"tax_system": "601",
|
|
"email": "test@example.com",
|
|
"zip": "01000",
|
|
},
|
|
)
|
|
|
|
assert customer_id == "cus_1"
|