Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 9 additions & 2 deletions nfelib/nfe/client/v4_0/dfe.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ def __init__(self, **kwargs: Any):
self.mod = kwargs.pop("mod", "55")
super().__init__(
service="nfe",
versao="4.00",
versao="1.01",
**kwargs,
)

Expand Down Expand Up @@ -121,8 +121,15 @@ def send(
**kwargs,
)

# wrap_response=True keeps backward compatibility with erpbrasil.edoc
# so that consumers migrating to brazil_fiscal_client don't need a
# large refactor. In that mode the parent returns a WrappedResponse
# whose parsed SOAP envelope is in .resposta; otherwise it returns
# the parsed SOAP envelope directly.
soap_envelope = response.resposta if self.wrap_response else response

result_container = (
response.body.nfeDistDFeInteresseResponse.nfeDistDFeInteresseResult
soap_envelope.body.nfeDistDFeInteresseResponse.nfeDistDFeInteresseResult
)

if not self.wrap_response:
Expand Down
55 changes: 55 additions & 0 deletions tests/nfe/test_client_dfe.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
from unittest import TestCase, mock, skipIf

from erpbrasil.assinatura import misc
from brazil_fiscal_client.fiscal_client import WrappedResponse
from xsdata.formats.dataclass.transports import DefaultTransport

# --- Conditional Imports for Python 3.9+ ---
Expand Down Expand Up @@ -97,3 +98,57 @@ def test_consultar_distribuicao_mocked(self, mock_post):

# Verify that the mock was called
mock_post.assert_called_once()


@skipIf(sys.version_info < (3, 9), "DfeClient requires Python 3.9+")
class DfeClientWrapResponseTest(TestCase):
"""Tests DfeClient with wrap_response=True (erpbrasil.edoc compatibility)."""

@classmethod
def setUpClass(cls):
super().setUpClass()
cls.cert_password = "testpassword"
cls.cert_data = misc.create_fake_certificate_file(
valid=True,
passwd=cls.cert_password,
issuer="TEST ISSUER",
country="BR",
subject="TEST SUBJECT",
)

cls.client = DfeClient(
ambiente="1",
uf="35",
pkcs12_data=cls.cert_data,
pkcs12_password=cls.cert_password,
fake_certificate=True,
verify_ssl=False,
wrap_response=True,
)

@mock.patch.object(DefaultTransport, "post")
def test_consultar_distribuicao_wrap_response(self, mock_post):
"""
Tests that wrap_response=True returns a WrappedResponse with the
parsed content in .resposta and HTTP data in .retorno.
"""
mock_post.return_value = response_sucesso_multiplos

result = self.client.consultar_distribuicao(
cnpj_cpf="00000000000191", ultimo_nsu="000000000000000"
)

# Must return a WrappedResponse, not a raw RetDistDfeInt
self.assertIsInstance(result, WrappedResponse)

# .resposta must contain the parsed DF-e content
self.assertIsInstance(result.resposta, RetDistDfeInt)
self.assertEqual(result.resposta.cStat, "138")
self.assertEqual(result.resposta.xMotivo, "Documento(s) localizado(s)")
self.assertEqual(result.resposta.ultNSU, "000000000000201")
self.assertIsNotNone(result.resposta.loteDistDFeInt)
self.assertEqual(len(result.resposta.loteDistDFeInt.docZip), 2)

# .retorno must contain the HTTP response data
self.assertEqual(result.retorno.status_code, 200)
self.assertIsNotNone(result.retorno.content)
Loading