Implementação da consulta de CTe pela Chave

pynfe/processamento/comunicacao.py

@@ -1246,6 +1246,17 @@ def consulta_distribuicao(self, cnpj=None, cpf=None, chave=None, nsu=0, consulta
         # Monta XML para envio da requisição
         xml = self._construir_xml_soap("CTeDistribuicaoDFe", raiz)
         return self._post(url, xml)
+
+    def consulta(self, chave):
+        url = self._get_url("CONSULTA")
+        # Monta XML do corpo da requisição
+        raiz = etree.Element("consSitCTe", versao=self._versao, xmlns=NAMESPACE_CTE)
+        etree.SubElement(raiz, "tpAmb").text = str(self._ambiente)
+        etree.SubElement(raiz, "xServ").text = "CONSULTAR"
+        etree.SubElement(raiz, "chCTe").text = chave
+        # Monta XML para envio da requisição
+        xml = self._construir_xml_soap("cteConsultaCT", raiz)
+        return self._post(url, xml)
 
     def _get_url_an(self, consulta):
         ambiente = "https://www1."  # produção

pynfe/utils/webservices.py

@@ -552,41 +552,49 @@
     },
     "MT": {
         "STATUS": "sefaz.mt.gov.br/ctews/services/CteStatusServico",
+        "CONSULTA": "sefaz.mt.gov.br/ctews2/services/CTeConsultaV4?wsdl",
         "HTTPS": "https://cte.",
         "HOMOLOGACAO": "https://homologacao.",
     },
     "MS": {
         "STATUS": "cte.ms.gov.br/ws/CteStatusServico",
+        "CONSULTA": "cte.ms.gov.br/ws/CTeConsultaV4",
         "HTTPS": "https://producao.",
         "HOMOLOGACAO": "https://homologacao.",
     },
     "MG": {
         "STATUS": "fazenda.mg.gov.br/cte/services/CteStatusServico",
+        "CONSULTA": "fazenda.mg.gov.br/cte/services/CTeConsultaV4",
         "HTTPS": "https://cte.",
         "HOMOLOGACAO": "https://hcte.",
     },
     "PR": {
         "STATUS": "fazenda.pr.gov.br/cte/CteStatusServico?wsdl",
+        "CONSULTA": "fazenda.pr.gov.br/cte4/CTeConsultaV4?wsdl",
         "HTTPS": "https://cte.",
         "HOMOLOGACAO": "https://homologacao.",
     },
     "RS": {
         "STATUS": "svrs.rs.gov.br/ws/ctestatusservico/CteStatusServico.asmx",
+        "CONSULTA": "svrs.rs.gov.br/ws/CTeConsultaV4/CTeConsultaV4.asmx",
         "HTTPS": "https://cte.",
         "HOMOLOGACAO": "https://cte-homologacao.",
     },
     "SP": {
         "STATUS": "fazenda.sp.gov.br/cteWEB/services/cteStatusServico.asmx",
+        "CONSULTA": "fazenda.sp.gov.br/CTeWS/WS/CTeConsultaV4.asmx",
         "HTTPS": "https://nfe.",
         "HOMOLOGACAO": "https://homologacao.nfe.",
     },
     "SVRS": {
         "STATUS": "svrs.rs.gov.br/ws/ctestatusservico/CteStatusServico.asmx",
+        "CONSULTA": "svrs.rs.gov.br/ws/CTeConsultaV4/CTeConsultaV4.asmx",
         "HTTPS": "https://cte.",
         "HOMOLOGACAO": "https://cte-homologacao.",
     },
     "SVSP": {
         "STATUS": "fazenda.sp.gov.br/cteWEB/services/CteStatusServico.asmx",
+        "CONSULTA": "fazenda.sp.gov.br/CTeWS/WS/CTeConsultaV4.asmx",
         "HTTPS": "https://nfe.",
         "HOMOLOGACAO": "https://homologacao.nfe.",
     },

tests/processamento/test_comunicacao_cte.py

@@ -0,0 +1,45 @@
+import unittest
+from unittest.mock import MagicMock
+from pynfe.processamento.comunicacao import ComunicacaoCTe
+from pynfe.utils import etree
+
+
+class TestComunicacaoCTe(unittest.TestCase):
+    """
+    Testa a classe ComunicacaoCTe
+    """
+
+    def test_consulta(self):
+        chave = '35190912345678000123570010000000011000000018'
+        url_esperada = 'https://url.ficticia/consulta'
+        resposta_mock = '<resposta>ok</resposta>'
+
+        comunicacao = ComunicacaoCTe(
+            uf="rs",
+            certificado="./tests/certificado.pfx",
+            certificado_senha=bytes("123456", "utf-8"),
+            homologacao=True
+        )
+        comunicacao._versao = "3.00"
+        comunicacao._ambiente = 1
+        comunicacao._get_url = MagicMock(return_value=url_esperada)
+        comunicacao._construir_xml_soap = MagicMock(side_effect=lambda metodo, xml: f"<soap>{etree.tostring(xml).decode()}</soap>")
+        comunicacao._post = MagicMock(return_value=resposta_mock)
+
+        resposta = comunicacao.consulta(chave)
+        xml_gerado = comunicacao._construir_xml_soap.call_args[0][1]
+        chave_extraida = xml_gerado.find(".//{*}chCTe").text
+
+        comunicacao._get_url.assert_called_once_with("CONSULTA")
+        comunicacao._construir_xml_soap.assert_called_once()
+        comunicacao._post.assert_called_once_with(url_esperada, f"<soap>{etree.tostring(xml_gerado).decode()}</soap>")
+        self.assertEqual(resposta, resposta_mock)
+        self.assertEqual(chave_extraida, chave)
+        self.assertEqual(xml_gerado.find(".//{*}tpAmb").text, "1")
+        self.assertEqual(xml_gerado.find(".//{*}xServ").text, "CONSULTAR")
+        self.assertTrue(xml_gerado.tag.endswith("consSitCTe"))
+        self.assertIn("xmlns", xml_gerado.attrib)
+
+
+if __name__ == '__main__':
+    unittest.main()