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
18 changes: 14 additions & 4 deletions brazil_fiscal_client/fiscal_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -362,10 +362,16 @@ def prepare_headers(self, headers: dict) -> dict:
"""Prepare request headers.

Keep xsdata behavior when available and default to plain HTTP headers
when running in lightweight mode without xsdata.
when running in lightweight mode without xsdata. Either way the
content type declares the UTF-8 charset the payload is encoded with
(see _post).
"""
if self._xsdata_available:
return super().prepare_headers(headers)
result = super().prepare_headers(headers)
content_type = result.get("content-type", "text/xml")
if "charset" not in content_type.lower():
result["content-type"] = f"{content_type}; charset=utf-8"
return result

return {
"Content-Type": "text/xml; charset=utf-8",
Expand All @@ -377,12 +383,16 @@ def _session(self):
return self.transport.session if self._xsdata_available else self.session

def _post(self, location: str, data: str, headers: dict) -> bytes:
# http.client would encode a str body as ISO-8859-1 while the XML
# declares UTF-8: SEFAZ then reads accented characters wrongly and
# rejects signed documents with cStat 297.
body = data.encode("utf-8")
if self._xsdata_available:
return self.transport.post(location, data=data, headers=headers)
return self.transport.post(location, data=body, headers=headers)

response = self.session.post(
location,
data=data,
data=body,
headers=headers,
timeout=self.timeout,
)
Expand Down
86 changes: 84 additions & 2 deletions tests/test_fiscal_client.py
Original file line number Diff line number Diff line change
Expand Up @@ -271,7 +271,7 @@ def test_send_force_soap12_request(self, mock_post):

# Check that the *request* sent used SOAP 1.2 namespace
mock_post.assert_called_once()
sent_data = mock_post.call_args.kwargs.get("data", b"")
sent_data = mock_post.call_args.kwargs.get("data", b"").decode()
self.assertIn(
SOAP12_ENV_NS, sent_data, "SOAP 1.2 Namespace not found in request"
)
Expand Down Expand Up @@ -313,7 +313,7 @@ def test_send_default_soap11_request(self, mock_post):

# Check that the *request* sent used SOAP 1.1 namespace
mock_post.assert_called_once()
sent_data = mock_post.call_args.kwargs.get("data", b"")
sent_data = mock_post.call_args.kwargs.get("data", b"").decode()
# Check based on prefix and namespace presence
self.assertIn(
'xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"', sent_data
Expand Down Expand Up @@ -437,6 +437,88 @@ def test_prepare_payload_without_xsdata_requires_raw_xml(self):
with self.assertRaisesRegex(ClientValueError, "xsdata is not installed"):
client.prepare_payload({"Body": {}})

@mock.patch.object(DefaultTransport, "post")
def test_send_encodes_request_as_utf8(self, mock_post):
"""The request goes on the wire as UTF-8 bytes declared as such.

A str body is encoded as ISO-8859-1 by http.client while the XML
declares UTF-8, so SEFAZ reads accented characters wrongly and
rejects signed documents with cStat 297 (Assinatura difere do
calculado).
"""
mock_post.return_value = response.encode()

client = FiscalClient(
ambiente=Tamb.DEV,
uf=TcodUfIbge.SC,
versao="4.00",
pkcs12_data=b"fake_cert",
pkcs12_password="123456",
fake_certificate=True,
service="nfe",
)
payload_obj = ConsStatServ(
tpAmb=NFeTamb.VALUE_2,
cUF=NFeTcodUfIbge.VALUE_42,
xServ=TconsStatServXServ.STATUS,
versao="4.00",
)
wrapped_payload = {"Body": {"nfeDadosMsg": {"content": [payload_obj]}}}
# signed content injected verbatim, as nfelib does with an NF-e
signed_content = "<NFe><xNome>Indústria e Comércio Ltda</xNome></NFe>"

client.send(
action_class=NfeStatusServico4SoapNfeStatusServicoNf,
location="http://fake.location.com/service",
wrapped_obj=wrapped_payload,
placeholder_exp=r"<consStatServ.*?</consStatServ>",
placeholder_content=signed_content,
)

sent_data = mock_post.call_args.kwargs["data"]
self.assertIsInstance(sent_data, bytes)
self.assertIn(signed_content.encode("utf-8"), sent_data)
sent_headers = {
key.lower(): value
for key, value in mock_post.call_args.kwargs["headers"].items()
}
self.assertIn("charset=utf-8", sent_headers["content-type"])

@mock.patch("brazil_fiscal_client.fiscal_client.XSDATA_AVAILABLE", False)
@mock.patch("requests.Session.post")
def test_send_without_xsdata_encodes_request_as_utf8(self, mock_post):
mock_post.return_value = mock.Mock(
content=response.encode(),
raise_for_status=mock.Mock(),
)

client = FiscalClient(
ambiente=Tamb.DEV,
uf=TcodUfIbge.SC,
versao="4.00",
pkcs12_data=b"fake_cert",
pkcs12_password="123456",
fake_certificate=True,
service="nfe",
)

raw_payload = (
"<soapenv:Envelope xmlns:soapenv='http://schemas.xmlsoap.org/soap/envelope/'>"
"<soapenv:Body><xNome>Indústria e Comércio Ltda</xNome></soapenv:Body>"
"</soapenv:Envelope>"
)
client.send(
action_class=None,
location="https://nfe-homologacao.svrs.rs.gov.br/ws/NfeStatusServico/NfeStatusServico4.asmx",
wrapped_obj=raw_payload,
)

sent_data = mock_post.call_args.kwargs["data"]
self.assertEqual(sent_data, raw_payload.encode("utf-8"))
self.assertIn(
"charset=utf-8", mock_post.call_args.kwargs["headers"]["Content-Type"]
)


class PKCS12NormalizationTests(TestCase):
"""pkcs12_data accepts raw bytes, base64 or a file path (erpbrasil
Expand Down
Loading