499 lines
21 KiB
Python
499 lines
21 KiB
Python
|
|
from django.urls import reverse
|
|
from rest_framework.test import APITestCase, APIClient
|
|
from rest_framework import status
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from unittest.mock import patch
|
|
from io import BytesIO
|
|
import zipfile
|
|
from api.organization.models import Organizacion
|
|
from api.licence.models import Licencia
|
|
from .models import Pedimento, TipoOperacion, ProcesamientoPedimento, EDocument
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class CustomsViewsTests(APITestCase):
|
|
def setUp(self):
|
|
self.org = Organizacion.objects.create(nombre="OrgTest", is_active=True, is_verified=True)
|
|
self.org2 = Organizacion.objects.create(nombre="OrgTest2", is_active=True, is_verified=True)
|
|
self.admin = User.objects.create_user(username="admin", password="adminpass", organizacion=self.org)
|
|
self.admin.groups.create(name="admin")
|
|
self.superuser = User.objects.create_superuser(username="superuser", password="superpass")
|
|
self.importador = User.objects.create_user(username="importador", password="importpass", organizacion=self.org2, is_importador=True, rfc="RFC123456789")
|
|
self.importador.groups.create(name="importador")
|
|
self.client = APIClient()
|
|
|
|
def test_admin_sees_only_own_pedimentos(self):
|
|
from .models import Pedimento
|
|
p1 = Pedimento.objects.create(pedimento="P1", organizacion=self.org)
|
|
p2 = Pedimento.objects.create(pedimento="P2", organizacion=self.org2)
|
|
self.client.force_authenticate(user=self.admin)
|
|
url = reverse('Pedimento-list')
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
pedimentos = [p['pedimento'] for p in response.data]
|
|
self.assertIn("P1", pedimentos)
|
|
self.assertNotIn("P2", pedimentos)
|
|
|
|
def test_superuser_sees_all_pedimentos(self):
|
|
from .models import Pedimento
|
|
p1 = Pedimento.objects.create(pedimento="P1", organizacion=self.org)
|
|
p2 = Pedimento.objects.create(pedimento="P2", organizacion=self.org2)
|
|
self.client.force_authenticate(user=self.superuser)
|
|
url = reverse('Pedimento-list')
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
pedimentos = [p['pedimento'] for p in response.data]
|
|
self.assertIn("P1", pedimentos)
|
|
self.assertIn("P2", pedimentos)
|
|
|
|
def test_importador_cannot_create_pedimento(self):
|
|
self.client.force_authenticate(user=self.importador)
|
|
url = reverse('Pedimento-list')
|
|
data = {
|
|
"pedimento": "P3",
|
|
"patente": "1234",
|
|
"aduana": "001",
|
|
"regimen": "A1",
|
|
"clave_pedimento": "A1",
|
|
"contribuyente": "ImportadorTest"
|
|
}
|
|
response = self.client.post(url, data)
|
|
self.assertNotIn(response.status_code, [status.HTTP_201_CREATED, status.HTTP_200_OK])
|
|
|
|
def test_list_tipos_operacion(self):
|
|
url = reverse('TipoOperacion-list')
|
|
self.client.force_authenticate(user=self.admin)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
def test_list_procesamientos(self):
|
|
url = reverse('ProcesamientoPedimento-list')
|
|
self.client.force_authenticate(user=self.admin)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
def test_list_edocuments(self):
|
|
url = reverse('EDocument-list')
|
|
self.client.force_authenticate(user=self.admin)
|
|
response = self.client.get(url)
|
|
self.assertEqual(response.status_code, status.HTTP_200_OK)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests de integración para bulk-create (ViewSetPedimento.bulk_create)
|
|
# Verifica que al re-cargar un pedimento existente sus documentos se actualicen
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class BulkCreateDocumentReplaceTests(APITestCase):
|
|
"""Verifica que bulk-create actualiza los documentos de pedimentos existentes
|
|
en vez de ignorarlos, y que no quedan archivos residuales en el storage."""
|
|
|
|
PEDIMENTO_APP = "24-01-3420-1234567"
|
|
|
|
def setUp(self):
|
|
self.licencia = Licencia.objects.create(nombre="Lic100GB", almacenamiento=100)
|
|
self.org = Organizacion.objects.create(
|
|
nombre="OrgBulkCreate",
|
|
licencia=self.licencia,
|
|
is_active=True,
|
|
is_verified=True,
|
|
)
|
|
self.user = User.objects.create_user(
|
|
username="bulkcreateuser", password="pass", organizacion=self.org
|
|
)
|
|
self.pedimento = Pedimento.objects.create(
|
|
organizacion=self.org,
|
|
pedimento="1234567",
|
|
pedimento_app=self.PEDIMENTO_APP,
|
|
)
|
|
from api.record.models import DocumentType, Fuente
|
|
self.doc_type = DocumentType.objects.get_or_create(nombre="Pedimento")[0]
|
|
# bulk_create usa fuente_id=4 hardcodeado; debe existir en la DB de test
|
|
Fuente.objects.get_or_create(id=4, defaults={"nombre": "Bulk Create"})
|
|
self.url = reverse("Pedimento-bulk-create")
|
|
self.client.force_authenticate(user=self.user)
|
|
|
|
def _make_zip(self, files_dict):
|
|
"""Crea un ZIP en memoria. files_dict = {nombre_archivo: contenido_bytes}"""
|
|
buf = BytesIO()
|
|
with zipfile.ZipFile(buf, "w") as zf:
|
|
for name, content in files_dict.items():
|
|
zf.writestr(name, content)
|
|
buf.seek(0)
|
|
return SimpleUploadedFile(
|
|
f"{self.PEDIMENTO_APP}.zip", buf.read(), content_type="application/zip"
|
|
)
|
|
|
|
def _post_zip(self, files_dict):
|
|
return self.client.post(
|
|
self.url,
|
|
{"contribuyente": "XAXX010101000", "archivos": [self._make_zip(files_dict)]},
|
|
format="multipart",
|
|
)
|
|
|
|
@patch("api.customs.views.storage_service")
|
|
def test_existing_pedimento_not_duplicated(self, mock_st):
|
|
"""Re-subir un pedimento existente NO debe crear un segundo Pedimento."""
|
|
mock_st.save_document_from_path.return_value = "org_1/documents/ped/informe_a1b2c3d4.pdf"
|
|
|
|
self._post_zip({"informe.pdf": b"contenido"})
|
|
|
|
self.assertEqual(
|
|
Pedimento.objects.filter(
|
|
organizacion=self.org, pedimento_app=self.PEDIMENTO_APP
|
|
).count(),
|
|
1,
|
|
)
|
|
|
|
@patch("api.customs.views.storage_service")
|
|
def test_existing_pedimento_document_replaced_not_duplicated(self, mock_st):
|
|
"""Documento existente con el mismo nombre base se reemplaza, no se duplica."""
|
|
from api.record.models import Document
|
|
|
|
old_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_a1b2c3d4.pdf"
|
|
old_doc = Document.objects.create(
|
|
organizacion=self.org,
|
|
pedimento=self.pedimento,
|
|
document_type=self.doc_type,
|
|
archivo=old_path,
|
|
size=500,
|
|
extension="pdf",
|
|
)
|
|
new_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_b5c6d7e8.pdf"
|
|
mock_st.save_document_from_path.return_value = new_path
|
|
mock_st.delete_file.return_value = True
|
|
|
|
self._post_zip({"informe.pdf": b"contenido actualizado"})
|
|
|
|
docs = Document.objects.filter(pedimento=self.pedimento)
|
|
# Sin duplicados
|
|
self.assertEqual(docs.count(), 1)
|
|
# Mismo registro
|
|
self.assertEqual(docs.first().id, old_doc.id)
|
|
# Archivo actualizado
|
|
old_doc.refresh_from_db()
|
|
self.assertEqual(old_doc.archivo.name, new_path)
|
|
|
|
@patch("api.customs.views.storage_service")
|
|
def test_existing_pedimento_stale_file_deleted_from_storage(self, mock_st):
|
|
"""Al reemplazar un documento, el archivo viejo debe eliminarse del storage."""
|
|
from api.record.models import Document
|
|
|
|
old_path = f"org_1/documents/{self.PEDIMENTO_APP}/informe_a1b2c3d4.pdf"
|
|
Document.objects.create(
|
|
organizacion=self.org,
|
|
pedimento=self.pedimento,
|
|
document_type=self.doc_type,
|
|
archivo=old_path,
|
|
size=500,
|
|
extension="pdf",
|
|
)
|
|
mock_st.save_document_from_path.return_value = f"org_1/documents/{self.PEDIMENTO_APP}/informe_b5c6d7e8.pdf"
|
|
mock_st.delete_file.return_value = True
|
|
|
|
self._post_zip({"informe.pdf": b"contenido"})
|
|
|
|
# delete_file debe haberse llamado con la ruta del archivo viejo
|
|
mock_st.delete_file.assert_called()
|
|
called_arg = str(mock_st.delete_file.call_args[0][0])
|
|
self.assertIn("informe_a1b2c3d4", called_arg)
|
|
|
|
@patch("api.customs.views.storage_service")
|
|
def test_existing_pedimento_new_file_added(self, mock_st):
|
|
"""Archivo nuevo en el ZIP se añade al pedimento existente."""
|
|
from api.record.models import Document
|
|
|
|
mock_st.save_document_from_path.return_value = "org_1/documents/ped/nuevo_b5c6d7e8.pdf"
|
|
|
|
self._post_zip({"nuevo_documento.pdf": b"contenido nuevo"})
|
|
|
|
self.assertGreaterEqual(
|
|
Document.objects.filter(pedimento=self.pedimento).count(), 1
|
|
)
|
|
|
|
@patch("api.customs.views.storage_service")
|
|
def test_already_existing_count_in_response(self, mock_st):
|
|
"""La respuesta debe indicar que el pedimento ya existía (already_existing_count >= 1)."""
|
|
mock_st.save_document_from_path.return_value = "org_1/documents/ped/f_a1b2c3d4.pdf"
|
|
|
|
response = self._post_zip({"archivo.pdf": b"contenido"})
|
|
|
|
self.assertIn(response.status_code, [status.HTTP_200_OK, status.HTTP_207_MULTI_STATUS, status.HTTP_201_CREATED])
|
|
data = response.json()
|
|
self.assertGreaterEqual(data.get("already_existing_count", 0), 1)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests del comando fix_partidas_error
|
|
# Una partida descargado=True solo es válida si alguno de sus documentos
|
|
# contiene consultarPartidaRespuesta sin tieneError=true. Partidas que solo
|
|
# tienen el REQUEST (o errores) deben volver a descargado=False.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
from io import StringIO
|
|
from types import SimpleNamespace
|
|
from django.core.management import call_command
|
|
from django.test import TestCase
|
|
|
|
|
|
XML_RESPUESTA_VALIDA = (
|
|
"<?xml version='1.0' encoding='UTF-8'?>"
|
|
'<S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body>'
|
|
'<ns9:consultarPartidaRespuesta xmlns:ns9="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpartida">'
|
|
"<tieneError>false</tieneError><ns9:partida/></ns9:consultarPartidaRespuesta>"
|
|
"</S:Body></S:Envelope>"
|
|
)
|
|
|
|
XML_ERROR_VUCEM = (
|
|
"<?xml version='1.0' encoding='UTF-8'?>"
|
|
'<S:Envelope xmlns:S="http://schemas.xmlsoap.org/soap/envelope/"><S:Body>'
|
|
'<ns9:consultarPartidaRespuesta xmlns:ns9="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpartida">'
|
|
"<tieneError>true</tieneError></ns9:consultarPartidaRespuesta>"
|
|
"</S:Body></S:Envelope>"
|
|
)
|
|
|
|
XML_ECO_REQUEST = (
|
|
"<?xml version='1.0' encoding='UTF-8'?>"
|
|
'<soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"'
|
|
' xmlns:con="http://www.ventanillaunica.gob.mx/pedimentos/ws/oxml/consultarpartida"><soapenv:Body>'
|
|
"<con:consultarPartidaPeticion><con:peticion/></con:consultarPartidaPeticion>"
|
|
"</soapenv:Body></soapenv:Envelope>"
|
|
)
|
|
|
|
|
|
class _FakeMinioObject:
|
|
"""Simula el objeto retornado por minio get_object."""
|
|
|
|
def __init__(self, content):
|
|
self._content = content
|
|
|
|
def read(self):
|
|
return self._content
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
def release_conn(self):
|
|
pass
|
|
|
|
|
|
class FixPartidasErrorCommandTests(TestCase):
|
|
PED_APP = "24-01-3420-1234567"
|
|
|
|
def setUp(self):
|
|
from api.customs.models import Partida
|
|
from api.record.models import DocumentType
|
|
|
|
self.licencia = Licencia.objects.create(nombre="LicFixPartidas", almacenamiento=100)
|
|
self.org = Organizacion.objects.create(
|
|
nombre="OrgFixPartidas", licencia=self.licencia, is_active=True, is_verified=True
|
|
)
|
|
# Pedimento VÁLIDO (no malformado): el comando ya no se limita a malformados
|
|
self.pedimento = Pedimento.objects.create(
|
|
organizacion=self.org,
|
|
pedimento="1234567",
|
|
pedimento_app=self.PED_APP,
|
|
aduana="034",
|
|
patente="3420",
|
|
numero_operacion="12345678",
|
|
)
|
|
self.partida = Partida.objects.create(
|
|
pedimento=self.pedimento,
|
|
organizacion=self.org,
|
|
numero_partida=1,
|
|
descargado=True,
|
|
)
|
|
self.type_resp = DocumentType.objects.get_or_create(id=1, defaults={"nombre": "XML"})[0]
|
|
self.type_req = DocumentType.objects.get_or_create(id=17, defaults={"nombre": "PT Request"})[0]
|
|
self.type_err = DocumentType.objects.get_or_create(id=18, defaults={"nombre": "PT Error"})[0]
|
|
|
|
# Storage simulado: dict path -> bytes
|
|
self.storage = {}
|
|
patcher = patch("api.customs.management.commands.fix_partidas_error.minio_client")
|
|
self.minio = patcher.start()
|
|
self.addCleanup(patcher.stop)
|
|
self.minio._bucket_name = "test-bucket"
|
|
self.minio.file_exists.side_effect = lambda name: name in self.storage
|
|
self.minio._client.get_object.side_effect = (
|
|
lambda bucket, name: _FakeMinioObject(self.storage[name])
|
|
)
|
|
self.minio.upload_file.side_effect = (
|
|
lambda name, file_data=None, content_type=None: self.storage.__setitem__(
|
|
name, file_data.read()
|
|
)
|
|
)
|
|
self.minio.delete_file.side_effect = lambda name: self.storage.pop(name, None)
|
|
|
|
def _doc(self, filename, doc_type, content=None):
|
|
from api.record.models import Document
|
|
|
|
path = f"org/{self.PED_APP}/{filename}"
|
|
doc = Document.objects.create(
|
|
organizacion=self.org,
|
|
pedimento=self.pedimento,
|
|
document_type=doc_type,
|
|
archivo=path,
|
|
size=100,
|
|
extension="xml",
|
|
)
|
|
if content is not None:
|
|
self.storage[path] = content.encode("utf-8")
|
|
return doc
|
|
|
|
def _run(self, **kwargs):
|
|
out = StringIO()
|
|
call_command("fix_partidas_error", stdout=out, stderr=StringIO(), **kwargs)
|
|
return out.getvalue()
|
|
|
|
def test_partida_solo_request_se_marca_no_descargada(self):
|
|
"""El caso reportado: descargado=True pero solo existe el XML del REQUEST."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST)
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertFalse(self.partida.descargado)
|
|
|
|
def test_partida_sin_documentos_se_marca_no_descargada(self):
|
|
"""descargado=True sin ningún documento tampoco es una descarga real."""
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertFalse(self.partida.descargado)
|
|
|
|
def test_partida_con_respuesta_valida_permanece_descargada(self):
|
|
"""Con consultarPartidaRespuesta sin error la partida no se toca."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST)
|
|
self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_RESPUESTA_VALIDA)
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertTrue(self.partida.descargado)
|
|
|
|
def test_doc_con_error_vucem_se_renombra_y_marca_no_descargada(self):
|
|
"""tieneError=true: doc → type 18 con sufijo _ERROR y partida → False."""
|
|
doc = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ERROR_VUCEM)
|
|
old_path = doc.archivo.name
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
doc.refresh_from_db()
|
|
self.assertFalse(self.partida.descargado)
|
|
self.assertEqual(doc.document_type_id, 18)
|
|
self.assertTrue(doc.archivo.name.endswith(f"vu_PT_{self.PED_APP}_1_ERROR.xml"))
|
|
self.assertTrue(doc.vu)
|
|
self.assertNotIn(old_path, self.storage)
|
|
self.assertIn(doc.archivo.name, self.storage)
|
|
|
|
def test_eco_de_request_guardado_como_respuesta_se_reclasifica(self):
|
|
"""Un eco de consultarPartidaPeticion guardado como respuesta se
|
|
reclasifica a type 17 sin chocar con el REQUEST real existente."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST)
|
|
doc = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ECO_REQUEST)
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
doc.refresh_from_db()
|
|
self.assertFalse(self.partida.descargado)
|
|
self.assertEqual(doc.document_type_id, 17)
|
|
# El nombre sin índice ya lo usa el REQUEST real → debe ir con _1
|
|
self.assertTrue(doc.archivo.name.endswith(f"vu_PT_{self.PED_APP}_1_REQUEST_1.xml"))
|
|
|
|
def test_doc_ausente_sin_canario_no_cambia_partida(self):
|
|
"""Archivo ausente y NINGÚN archivo del pedimento en storage: posible
|
|
storage equivocado (p. ej. dev) → sin cambios."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, content=None)
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertTrue(self.partida.descargado)
|
|
|
|
def test_registro_fantasma_con_storage_real_se_marca_no_descargada(self):
|
|
"""Document type 1 en BD sin archivo en storage, pero el REQUEST sí
|
|
existe físicamente (canario): el storage es el correcto, el registro es
|
|
fantasma → la partida no tiene XML de partida → descargado=False."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST)
|
|
fantasma = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, content=None)
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
fantasma.refresh_from_db()
|
|
self.assertFalse(self.partida.descargado)
|
|
# El registro fantasma se reporta pero no se modifica ni se borra
|
|
self.assertEqual(fantasma.document_type_id, 1)
|
|
|
|
def test_storage_inaccesible_no_cambia_partida(self):
|
|
"""Excepción al consultar storage (conexión caída): sin cambios."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ERROR_VUCEM)
|
|
self.minio.file_exists.side_effect = Exception("connection refused")
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertTrue(self.partida.descargado)
|
|
|
|
def test_naming_legacy_valida_partida(self):
|
|
"""Documentos con nomenclatura legacy (partida al final) también validan."""
|
|
self._doc("vu_PT_010Imp_034_3420_1234567_1.xml", self.type_resp, XML_RESPUESTA_VALIDA)
|
|
|
|
self._run(pedimento=str(self.pedimento.id))
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertTrue(self.partida.descargado)
|
|
|
|
def test_dry_run_no_modifica(self):
|
|
"""--dry-run reporta pero no toca BD ni storage."""
|
|
doc = self._doc(f"vu_PT_{self.PED_APP}_1.xml", self.type_resp, XML_ERROR_VUCEM)
|
|
|
|
self._run(pedimento=str(self.pedimento.id), dry_run=True)
|
|
|
|
self.partida.refresh_from_db()
|
|
doc.refresh_from_db()
|
|
self.assertTrue(self.partida.descargado)
|
|
self.assertEqual(doc.document_type_id, 1)
|
|
self.assertIn(doc.archivo.name, self.storage)
|
|
|
|
def test_universo_general_incluye_pedimentos_validos(self):
|
|
"""Sin --pedimento ni --solo-malformados también procesa pedimentos bien formados."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST)
|
|
|
|
self._run()
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertFalse(self.partida.descargado)
|
|
|
|
def test_solo_malformados_excluye_pedimentos_validos(self):
|
|
"""Con --solo-malformados un pedimento bien formado no se procesa."""
|
|
self._doc(f"vu_PT_{self.PED_APP}_1_REQUEST.xml", self.type_req, XML_ECO_REQUEST)
|
|
|
|
self._run(solo_malformados=True)
|
|
|
|
self.partida.refresh_from_db()
|
|
self.assertTrue(self.partida.descargado)
|
|
|
|
def test_no_confunde_partida_1_con_11(self):
|
|
"""La asignación por nombre no debe mezclar partida 1 con partida 11."""
|
|
from api.customs.management.commands.fix_partidas_error import Command
|
|
|
|
docs = [
|
|
SimpleNamespace(id=1, document_type_id=1, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_1.xml")),
|
|
SimpleNamespace(id=2, document_type_id=17, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_1_REQUEST.xml")),
|
|
SimpleNamespace(id=3, document_type_id=1, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_11.xml")),
|
|
SimpleNamespace(id=4, document_type_id=17, archivo=SimpleNamespace(name=f"org/x/vu_PT_{self.PED_APP}_11_REQUEST.xml")),
|
|
]
|
|
cmd = Command()
|
|
|
|
ids_p1 = {d.id for d in cmd._docs_de_partida(docs, self.PED_APP, 1)}
|
|
ids_p11 = {d.id for d in cmd._docs_de_partida(docs, self.PED_APP, 11)}
|
|
|
|
self.assertEqual(ids_p1, {1, 2})
|
|
self.assertEqual(ids_p11, {3, 4})
|