Api compat

brazil_fiscal_client/fiscal_client.py

@@ -3,7 +3,10 @@
 
 from __future__ import annotations  # Python 3.8 compat
 
+import base64
+import binascii
 import logging
+import os
 import re
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
@@ -125,21 +128,37 @@ def envio_raiz(self) -> Any:
         """Backward compatibility alias for legacy API."""
         return self.request_obj
 
+    @envio_raiz.setter
+    def envio_raiz(self, value: Any) -> None:
+        self.request_obj = value
+
     @property
     def envio_xml(self) -> bytes:
         """Backward compatibility alias for legacy API."""
         return self.request_xml
 
+    @envio_xml.setter
+    def envio_xml(self, value: bytes) -> None:
+        self.request_xml = value
+
     @property
     def resposta(self) -> Any:
         """Backward compatibility alias for legacy API."""
         return self.response_obj
 
+    @resposta.setter
+    def resposta(self, value: Any) -> None:
+        self.response_obj = value
+
     @property
     def retorno(self) -> WrappedHTTPResponse:
         """Backward compatibility alias for legacy API."""
         return self.response
 
+    @retorno.setter
+    def retorno(self, value: WrappedHTTPResponse) -> None:
+        self.response = value
+
 
 class FiscalClient(Client):
     """A Brazilian fiscal client extending the xsdata SOAP wsdl client.
@@ -197,7 +216,11 @@ def __init__(
         else:
             self.session = requests.Session()
         self.versao = versao
+        # Keep the value as given (nfelib's CommonMixin.sign_xml expects the
+        # base64 form, like erpbrasil.assinatura's Certificado did); expose the
+        # normalized raw PKCS12 bytes separately for the transport.
         self.pkcs12_data = pkcs12_data
+        self.pkcs12_bytes = self._normalize_pkcs12(pkcs12_data)
         self.pkcs12_password = pkcs12_password
         self.verify_ssl = verify_ssl
         self.service = service
@@ -266,7 +289,7 @@ def send(
             self._session.mount(
                 server,
                 Pkcs12Adapter(
-                    pkcs12_data=self.pkcs12_data,
+                    pkcs12_data=self.pkcs12_bytes,
                     pkcs12_password=self.pkcs12_password,
                 ),
             )
@@ -366,6 +389,42 @@ def _post(self, location: str, data: str, headers: dict) -> bytes:
         response.raise_for_status()
         return response.content
 
+    @staticmethod
+    def _normalize_pkcs12(value: Any) -> bytes:
+        """Normalize the certificate input to raw PKCS12 bytes.
+
+        Accepts raw bytes, a base64 encoded string/bytes (as stored in Odoo
+        Binary fields and expected by erpbrasil.assinatura) or a file path,
+        mirroring the flexibility of the legacy erpbrasil
+        ``Certificado`` class.
+        """
+        if isinstance(value, bytes):
+            # Heuristic: a valid PKCS12 blob starts with 0x30 (ASN.1 SEQUENCE);
+            # a base64 encoded one would not (unless it decodes to itself,
+            # which is not possible for a well-formed PFX).
+            try:
+                raw = base64.b64decode(value, validate=True)
+                if raw[:1] == b"\x30" and value[:1] != b"\x30":
+                    return raw
+            except (binascii.Error, ValueError):
+                pass
+            return value
+        if isinstance(value, str):
+            if os.path.exists(value):
+                with open(value, "rb") as f:
+                    return f.read()
+            try:
+                return base64.b64decode(value, validate=True)
+            except (binascii.Error, ValueError) as e:
+                raise ClientValueError(
+                    "pkcs12_data should be raw bytes, base64 encoded "
+                    "certificate or a file path."
+                ) from e
+        raise ClientValueError(
+            "pkcs12_data should be raw bytes, base64 encoded certificate "
+            "or a file path."
+        )
+
     @staticmethod
     def _webservice_name(action_class: Any) -> str:
         soap_action = getattr(action_class, "soapAction", "")

tests/test_fiscal_client.py

@@ -1,3 +1,4 @@
+import base64
 from os import environ
 from unittest import TestCase, mock
 
@@ -435,3 +436,81 @@ def test_prepare_payload_without_xsdata_requires_raw_xml(self):
 
         with self.assertRaisesRegex(ClientValueError, "xsdata is not installed"):
             client.prepare_payload({"Body": {}})
+
+
+class PKCS12NormalizationTests(TestCase):
+    """pkcs12_data accepts raw bytes, base64 or a file path (erpbrasil
+    Certificado compatibility) while pkcs12_bytes is always the raw PFX."""
+
+    def _make_client(self, pkcs12_data):
+        return FiscalClient(
+            ambiente=Tamb.DEV,
+            uf=TcodUfIbge.SC,
+            versao="4.00",
+            pkcs12_data=pkcs12_data,
+            pkcs12_password="123456",
+            fake_certificate=True,
+            service="nfe",
+        )
+
+    def test_raw_bytes_kept_as_is(self):
+        pfx = b"\x30" + b"\x00" * 10  # ASN.1 SEQUENCE first byte
+        client = self._make_client(pfx)
+        self.assertEqual(client.pkcs12_data, pfx)
+        self.assertEqual(client.pkcs12_bytes, pfx)
+
+    def test_base64_bytes_decoded(self):
+        pfx = b"\x30" + b"\x00" * 10
+        b64 = base64.b64encode(pfx)
+        client = self._make_client(b64)
+        self.assertEqual(client.pkcs12_data, b64)
+        self.assertEqual(client.pkcs12_bytes, pfx)
+
+    def test_base64_str_decoded(self):
+        pfx = b"\x30" + b"\x00" * 10
+        b64 = base64.b64encode(pfx).decode("utf-8")
+        client = self._make_client(b64)
+        self.assertEqual(client.pkcs12_data, b64)
+        self.assertEqual(client.pkcs12_bytes, pfx)
+
+    def test_file_path_read(self):
+        import tempfile
+
+        pfx = b"\x30" + b"\x00" * 10
+        with tempfile.NamedTemporaryFile(suffix=".pfx", delete=False) as f:
+            f.write(pfx)
+            path = f.name
+        try:
+            client = self._make_client(path)
+            self.assertEqual(client.pkcs12_data, path)
+            self.assertEqual(client.pkcs12_bytes, pfx)
+        finally:
+            import os
+
+            os.unlink(path)
+
+    def test_invalid_value_raises(self):
+        with self.assertRaises(ClientValueError):
+            self._make_client(None)
+
+
+class WrappedResponseCompatTests(TestCase):
+    def test_legacy_aliases_are_writable(self):
+        from brazil_fiscal_client.fiscal_client import (
+            WrappedHTTPResponse,
+            WrappedResponse,
+        )
+
+        wr = WrappedResponse(
+            webservice="nfeStatusServicoNF",
+            request_obj=None,
+            request_xml=b"",
+            response_obj=None,
+            response=WrappedHTTPResponse(content=b"", status_code=200),
+        )
+        wr.resposta = "parsed"
+        wr.envio_xml = b"<xml/>"
+        wr.envio_raiz = "root"
+        self.assertEqual(wr.response_obj, "parsed")
+        self.assertEqual(wr.request_xml, b"<xml/>")
+        self.assertEqual(wr.request_obj, "root")