diff --git a/l10n_br_nfse/constants/nfse.py b/l10n_br_nfse/constants/nfse.py index 01bc8fec59ce..0081bf94af39 100644 --- a/l10n_br_nfse/constants/nfse.py +++ b/l10n_br_nfse/constants/nfse.py @@ -32,3 +32,14 @@ ("5", "Microempresario Individual(MEI)"), ("6", "Microempresario e Empresa de Pequeno Porte(ME EPP)"), ] + + +ISSQN_TO_TRIBUTACAO_ISS = { + "1": "1", # Exigível → Operação tributável + "2": "4", # Não incidência → Não Incidência + "3": "4", # Isenção → Não Incidência + "4": "3", # Exportação → Exportação de serviço + "5": "2", # Imunidade → Imunidade + "6": "1", # Suspensa (Judicial) → Operação tributável + "7": "1", # Suspensa (Administrativo) → Operação tributável +} diff --git a/l10n_br_nfse/models/document.py b/l10n_br_nfse/models/document.py index d9521de78c51..37e21c6403b6 100644 --- a/l10n_br_nfse/models/document.py +++ b/l10n_br_nfse/models/document.py @@ -20,6 +20,7 @@ ) from ..constants.nfse import ( + ISSQN_TO_TRIBUTACAO_ISS, NFSE_ENVIRONMENTS, OPERATION_NATURE, RPS_TYPE, @@ -167,6 +168,8 @@ def _prepare_dados_servico(self): cbs_aliquota = 0 ibs_uf_valor = 0 cbs_valor = 0 + base_calculo_pis = 0 + base_calculo_cofins = 0 for line in lines: result_line.update(line._prepare_line_service()) @@ -195,6 +198,15 @@ def _prepare_dados_servico(self): cbs_aliquota += result_line.get("cbs_aliquota") or 0 ibs_uf_valor += result_line.get("ibs_uf_valor") or 0 cbs_valor += result_line.get("cbs_valor") or 0 + situacao_tributaria_pis = result_line.get("situacao_tributaria_pis") + situacao_tributaria_cofins = result_line.get("situacao_tributaria_cofins") + base_calculo_pis += result_line.get("base_calculo_pis", 0) + base_calculo_cofins += result_line.get("base_calculo_cofins", 0) + aliquota_pis = result_line.get("aliquota_pis") or 0 + aliquota_cofins = result_line.get("aliquota_cofins") or 0 + tipo_retencao_pis_cofins = ( + result_line.get("tipo_retencao_pis_cofins") or "2" + ) result = { "valor_servicos": valor_servicos, @@ -219,10 +231,14 @@ def _prepare_dados_servico(self): "valor_liquido_nfse": valor_liquido_nfse, "item_lista_servico": self.fiscal_line_ids[0].service_type_id.code and self.fiscal_line_ids[0].service_type_id.code.replace(".", ""), + "codigo_tributacao_nacional": self.fiscal_line_ids[ + 0 + ].national_taxation_code_id.code + or None, "codigo_tributacao_municipio": self.fiscal_line_ids[ 0 ].city_taxation_code_id.code - or "", + or None, "municipio_prestacao_servico": self.fiscal_line_ids[ 0 ].issqn_fg_city_id.ibge_code @@ -246,6 +262,16 @@ def _prepare_dados_servico(self): "ibs_uf_valor": ibs_uf_valor if ibs_uf_valor else None, "ibs_mun_valor": 0.0, "cbs_valor": cbs_valor if cbs_valor else None, + "situacao_tributaria_pis": situacao_tributaria_pis, + "situacao_tributaria_cofins": situacao_tributaria_cofins, + "base_calculo_pis": round(base_calculo_pis, 2), + "base_calculo_cofins": round(base_calculo_cofins, 2), + "aliquota_pis": round(aliquota_pis, 2), + "aliquota_cofins": round(aliquota_cofins, 2), + "tipo_retencao_pis_cofins": tipo_retencao_pis_cofins, + "codigo_tributacao_iss": ISSQN_TO_TRIBUTACAO_ISS[ + self.fiscal_line_ids[0].issqn_eligibility + ], } result.update(self.company_id._prepare_company_service()) diff --git a/l10n_br_nfse/models/document_line.py b/l10n_br_nfse/models/document_line.py index 46b3970d132e..a591d2d6b371 100644 --- a/l10n_br_nfse/models/document_line.py +++ b/l10n_br_nfse/models/document_line.py @@ -92,7 +92,8 @@ def _prepare_line_service(self): "valor_liquido_nfse": round(self.amount_taxed, 2), "item_lista_servico": self.service_type_id.code and self.service_type_id.code.replace(".", ""), - "codigo_tributacao_municipio": self.city_taxation_code_id.code or "", + "codigo_tributacao_nacional": self.national_taxation_code_id.code or None, + "codigo_tributacao_municipio": self.city_taxation_code_id.code or None, "municipio_prestacao_servico": self.issqn_fg_city_id.ibge_code or "", "discriminacao": str(self.name[:2000] or ""), "codigo_cnae": misc.punctuation_rm(self.cnae_id.code) or None, @@ -109,4 +110,15 @@ def _prepare_line_service(self): "ibs_uf_valor": round(self.ibs_value, 2) if self.ibs_value else None, "ibs_mun_valor": 0.0, "cbs_valor": round(self.cbs_value, 2) if self.cbs_value else None, + "situacao_tributaria_pis": self.pis_cst_code or "", + "situacao_tributaria_cofins": self.cofins_cst_code or "", + "base_calculo_pis": round(self.pis_base, 2), + "base_calculo_cofins": round(self.cofins_base, 2), + "aliquota_pis": round(self.pis_percent, 2) if self.pis_percent else 0.0, + "aliquota_cofins": ( + round(self.cofins_percent, 2) if self.cofins_percent else 0.0 + ), + "tipo_retencao_pis_cofins": ( + "1" if (self.pis_wh_value or self.cofins_wh_value) else "2" + ), } diff --git a/l10n_br_nfse_focus/README.rst b/l10n_br_nfse_focus/README.rst index 0a09efd974b1..e0b6da18b9c4 100644 --- a/l10n_br_nfse_focus/README.rst +++ b/l10n_br_nfse_focus/README.rst @@ -1,7 +1,3 @@ -.. image:: https://odoo-community.org/readme-banner-image - :target: https://odoo-community.org/get-involved?utm_source=readme - :alt: Odoo Community Association - ================ NFS-e (FocusNFE) ================ @@ -17,7 +13,7 @@ NFS-e (FocusNFE) .. |badge1| image:: https://img.shields.io/badge/maturity-Beta-yellow.png :target: https://odoo-community.org/page/development-status :alt: Beta -.. |badge2| image:: https://img.shields.io/badge/license-AGPL--3-blue.png +.. |badge2| image:: https://img.shields.io/badge/licence-AGPL--3-blue.png :target: http://www.gnu.org/licenses/agpl-3.0-standalone.html :alt: License: AGPL-3 .. |badge3| image:: https://img.shields.io/badge/github-OCA%2Fl10n--brazil-lightgray.png?logo=github @@ -32,10 +28,16 @@ NFS-e (FocusNFE) |badge1| |badge2| |badge3| |badge4| |badge5| -Esse módulo integra a emissão de Notas Fiscais de Serviços(NFSe) com a +Esse módulo integra a emissão de Notas Fiscais de Serviços (NFSe) com a API da FocusNFE permitindo assim, a criação, transmissão, consulta e cancelamento de documentos fiscais do tipo NFSe. +O módulo suporta dois tipos de emissão: + +- **NFSe Municipal:** Emissão de NFSe através da API municipal da + FocusNFE +- **NFSe Nacional:** Emissão de NFSe através da API nacional da FocusNFE + Para mais informações, acesse: https://focusnfe.com.br/ **Table of contents** @@ -68,22 +70,59 @@ para a empresa desejada: - **Provedor NFS-e:** Selecione a opção FocusNFE - **FocusNFe Token:** Informe o token de acesso da empresa. Obs. Este token é obtido através da plataforma da FocusNFE + + - **Token de Produção:** Token para ambiente de produção + (visível quando Ambiente NFS-e = Produção) + - **Token de Homologação:** Token para ambiente de homologação + (visível quando Ambiente NFS-e = Homologação) + + - **Tipo FocusNFe NFSe:** Selecione o tipo de API a ser utilizada: + + - **NFSe:** Para emissão de NFSe Municipal (padrão) + - **NFSe Nacional:** Para emissão de NFSe Nacional + - **Valor Tipo de Serviço:** Se necessário configure o campo que - deve preencher o valor de tipo de serviço + deve preencher o valor de tipo de serviço (Service Type ou City + Taxation Code) - **Valor Código CNAE:** Se necessário configure o campo que deve - preencher o valor do Código CNAE + preencher o valor do Código CNAE (CNAE Code ou City Taxation + Code) + - **Formato Taxa:** Selecione o formato da taxa (Decimal ou + Percentage) + - **Incluir Documentos Autorizados na Verificação de Status:** Se + marcado, documentos autorizados serão incluídos na verificação + de status + - **Forçar DANFSE Odoo:** Se marcado, o sistema sempre usará o + DANFSE do Odoo ao invés do DANFSE da FocusNFE Usage ===== Para usar este módulo: -1. Crie uma fatura com o tipo de documento fiscal 'SE'. -2. Preencha os detalhes necessários, como o código tributário da cidade, - impostos e informações correlatas. -3. Valide o documento. -4. Envie o Documento Fiscal. -5. Acompanhe o status de processamento do documento. +1. Configure a empresa conforme descrito na seção de Configuração. + +2. Crie uma fatura com o tipo de documento fiscal 'SE'. + +3. Preencha os detalhes necessários: + + - Para **NFSe Municipal:** Preencha o código tributário municipal, + impostos e informações correlatas + - Para **NFSe Nacional:** Preencha o código tributário nacional + (NBS), código tributário municipal (se aplicável), impostos e + informações correlatas + +4. Valide o documento fiscal. + +5. Envie o Documento Fiscal através do botão "Enviar Documento Fiscal". + +6. Acompanhe o status de processamento do documento. O sistema + verificará automaticamente o status através de um cron job, ou você + pode verificar manualmente através do botão "Verificar Status". + +7. Após a autorização, o DANFSE (Documento Auxiliar da Nota Fiscal de + Serviço Eletrônica) será gerado automaticamente, a menos que a opção + "Forçar DANFSE Odoo" esteja marcada na configuração da empresa. Bug Tracker =========== diff --git a/l10n_br_nfse_focus/models/__init__.py b/l10n_br_nfse_focus/models/__init__.py index 166dd3bd2493..3736cf1ed389 100644 --- a/l10n_br_nfse_focus/models/__init__.py +++ b/l10n_br_nfse_focus/models/__init__.py @@ -1,2 +1,7 @@ +from . import base +from . import constants from . import document +from . import helpers +from . import nfse_municipal +from . import nfse_nacional from . import res_company diff --git a/l10n_br_nfse_focus/models/base.py b/l10n_br_nfse_focus/models/base.py new file mode 100644 index 000000000000..29735519be44 --- /dev/null +++ b/l10n_br_nfse_focus/models/base.py @@ -0,0 +1,60 @@ +# Copyright 2023 - TODAY, Marcel Savegnago +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +"""Base class for FocusNFE NFSe operations.""" + +import requests + +from odoo import _, models +from odoo.exceptions import UserError + + +class FocusnfeNfseBase(models.AbstractModel): + """Base class for FocusNFE NFSe operations with shared HTTP request logic.""" + + _name = "focusnfe.nfse.base" + _description = "FocusNFE NFSE Base" + + def _make_focus_nfse_http_request( + self, method, url, token, data=None, params=None, service_name="NFSe" + ): + """Perform a generic HTTP request. + + Args: + method (str): The HTTP method to use (e.g., 'GET', 'POST'). + url (str): The URL to which the request is sent. + token (str): The authentication token for the service. + data (dict, optional): The payload to send in the request body. + Defaults to None. + params (dict, optional): The URL parameters to append to the URL. + Defaults to None. + service_name (str): Name of the service for error messages. + + Returns: + requests.Response: The response object from the requests library. + + Raises: + UserError: If the HTTP request fails with a 4xx/5xx response. + """ + auth = (token, "") + try: + response = requests.request( # pylint: disable=external-request-timeout + method, + url, + data=data, + params=params, + auth=auth, + ) + if response.status_code == 422: + payload = response.json() + msg = payload.get("mensagem") or "" + raise UserError( + f"Error communicating with {service_name} service: {msg}" + ) + response.raise_for_status() # Raises an error for 4xx/5xx responses + return response + except requests.HTTPError as e: + raise UserError( + _("Error communicating with %(service)s service: %(error)s") + % {"service": service_name, "error": e} + ) from e diff --git a/l10n_br_nfse_focus/models/constants.py b/l10n_br_nfse_focus/models/constants.py new file mode 100644 index 000000000000..0e2945d8375f --- /dev/null +++ b/l10n_br_nfse_focus/models/constants.py @@ -0,0 +1,41 @@ +# Copyright 2023 - TODAY, Marcel Savegnago +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +"""Constants for FocusNFE NFSe integration.""" + +NFSE_URL = { + "1": "https://api.focusnfe.com.br", + "2": "https://homologacao.focusnfe.com.br", +} + +API_ENDPOINT = { + "envio": "/v2/nfse?", + "status": "/v2/nfse/", + "resposta": "/v2/nfse/", + "cancelamento": "/v2/nfse/", +} + +API_ENDPOINT_NACIONAL = { + "envio": "/v2/nfsen", + "status": "/v2/nfsen/", + "resposta": "/v2/nfsen/", + "cancelamento": "/v2/nfsen/", +} + +TIMEOUT = 60 # 60 seconds + +# Constants for document status +STATUS_AUTORIZADO = "autorizado" +STATUS_CANCELADO = "cancelado" +STATUS_ERRO_AUTORIZACAO = "erro_autorizacao" +STATUS_PROCESSANDO_AUTORIZACAO = "processando_autorizacao" +CODE_NFE_CANCELADA = "nfe_cancelada" +CODE_NFE_AUTORIZADA = "nfe_autorizada" + +# CPF/CNPJ length constants +CPF_LENGTH = 11 +CNPJ_LENGTH = 14 + +# PDF validation constants +PDF_HEADER = b"%PDF-" +PDF_FOOTER = b"%%EOF" diff --git a/l10n_br_nfse_focus/models/document.py b/l10n_br_nfse_focus/models/document.py index c5009fd3a4e9..4fdd0db21084 100644 --- a/l10n_br_nfse_focus/models/document.py +++ b/l10n_br_nfse_focus/models/document.py @@ -2,8 +2,9 @@ # Copyright 2023 - TODAY, Marcel Savegnago # License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). +"""Document model for FocusNFE NFSe integration.""" + import base64 -import json import logging from datetime import datetime @@ -24,294 +25,29 @@ from odoo.addons.l10n_br_fiscal_edi.models.document import Document as FiscalDocument from odoo.addons.l10n_br_nfse.models.document import filter_processador_edoc_nfse -NFSE_URL = { - "1": "https://api.focusnfe.com.br", - "2": "https://homologacao.focusnfe.com.br", -} - -API_ENDPOINT = { - "envio": "/v2/nfse?", - "status": "/v2/nfse/", - "resposta": "/v2/nfse/", - "cancelamento": "/v2/nfse/", -} - -TIMEOUT = 60 # 60 seconds +from .constants import ( + CODE_NFE_AUTORIZADA, + CODE_NFE_CANCELADA, + NFSE_URL, + STATUS_AUTORIZADO, + STATUS_CANCELADO, + STATUS_ERRO_AUTORIZACAO, + STATUS_PROCESSANDO_AUTORIZACAO, + TIMEOUT, +) +from .helpers import ( + _is_valid_pdf, + filter_focusnfe, + filter_focusnfe_municipal, + filter_focusnfe_nacional, +) _logger = logging.getLogger(__name__) -def filter_focusnfe(record): - return record.company_id.provedor_nfse == "focusnfe" - - -class FocusnfeNfse(models.AbstractModel): - _name = "focusnfe.nfse" - _description = "FocusNFE NFSE" - - def _make_focus_nfse_http_request(self, method, url, token, data=None, params=None): - """Perform a generic HTTP request. - - Args: - method (str): The HTTP method to use (e.g., 'GET', 'POST'). - url (str): The URL to which the request is sent. - token (str): The authentication token for the service. - data (dict, optional): The payload to send in the request body. - Defaults to None. - params (dict, optional): The URL parameters to append to the URL. - Defaults to None. - - Returns: - requests.Response: The response object from the requests library. - - Raises: - UserError: If the HTTP request fails with a 4xx/5xx response. - """ - auth = (token, "") - try: - response = requests.request( # pylint: disable=external-request-timeout - method, - url, - data=data, - params=params, - auth=auth, - ) - if response.status_code == 422: - payload = response.json() - msg = payload.get("mensagem") or "" - raise UserError(f"Error communicating with NFSe service: {msg}") - response.raise_for_status() # Raises an error for 4xx/5xx responses - return response - except requests.HTTPError as e: - raise UserError(_("Error communicating with NFSe service: %s") % e) from e - - def _identify_service_recipient(self, recipient): - """Identify whether the service recipient is a CPF or CNPJ. - - Args: - recipient (dict): A dictionary containing either 'cpf' or 'cnpj' keys. - - Returns: - dict: A dictionary with either a 'cpf' or 'cnpj' key and its value. - """ - return ( - {"cpf": recipient.get("cpf")} - if recipient.get("cpf") - else {"cnpj": recipient.get("cnpj")} - ) - - @api.model - def process_focus_nfse_document(self, edoc, ref, company, environment): - """Process the electronic fiscal document. - - Args: - edoc (tuple): The electronic document data. - ref (str): The document reference. - company (recordset): The company record. - - Returns: - requests.Response: The response from the NFSe service. - """ - token = company.get_focusnfe_token() - data = self._prepare_payload(*edoc, company) - payload = json.dumps(data) - url = f"{NFSE_URL[environment]}{API_ENDPOINT['envio']}" - ref = {"ref": ref} - return self._make_focus_nfse_http_request( - "POST", url, token, data=payload, params=ref - ) - - def _prepare_payload(self, rps, service, recipient, company): - """Construct the NFSe payload. - - Args: - rps (dict): Information about the RPS. - service (dict): Details of the service provided. - recipient (dict): Information about the service recipient. - company (recordset): The company record. - - Returns: - dict: The complete payload for the NFSe request. - """ - rps_info = rps.get("rps") - service_info = service.get("service") - recipient_info = recipient.get("recipient") - recipient_identification = self._identify_service_recipient(recipient_info) - - vals = { - "prestador": self._prepare_provider_data(rps_info, company), - "servico": self._prepare_service_data(service_info, company), - "tomador": self._prepare_recipient_data( - recipient_info, recipient_identification, company - ), - "razao_social": company.name, - "data_emissao": rps_info.get("data_emissao"), - "incentivador_cultural": rps_info.get("incentivador_cultural", False), - "natureza_operacao": rps_info.get("natureza_operacao"), - "optante_simples_nacional": rps_info.get("optante_simples_nacional", False), - "status": rps_info.get("status"), - "informacoes_adicionais_contribuinte": ( - rps_info.get("customer_additional_data", False)[:256] - if rps_info.get("customer_additional_data") - else False - ), - } - codigo_obra = rps_info.get("codigo_obra", False) - art = rps_info.get("art", False) - - if codigo_obra: - vals["codigo_obra"] = codigo_obra - - if art: - vals["art"] = art - - return vals - - def _prepare_provider_data(self, rps, company): - """Construct the provider section of the payload. - - Args: - rps (dict): Information about the RPS. - company (recordset): The company record. - - Returns: - dict: The provider section of the payload. - """ - return { - "cnpj": rps.get("cnpj"), - "inscricao_municipal": rps.get("inscricao_municipal"), - "codigo_municipio": company.city_id.ibge_code, - } - - def _prepare_service_data(self, service, company): - """Construct the service section of the payload. - - Args: - service (dict): Details of the service provided. - company (recordset): The company record. - - Returns: - dict: The service section of the payload. - """ - return { - "aliquota": service.get("aliquota") - if company.focusnfe_tax_rate_format == "decimal" - else round(service.get("aliquota", 0.0) * 100, 1), - "base_calculo": round(service.get("base_calculo", 0), 2), - "discriminacao": service.get("discriminacao"), - "iss_retido": service.get("iss_retido"), - "codigo_municipio": service.get("municipio_prestacao_servico"), - "item_lista_servico": service.get(company.focusnfe_nfse_service_type_value), - "codigo_cnae": service.get(company.focusnfe_nfse_cnae_code_value), - "valor_iss": round(service.get("valor_iss", 0), 2), - "valor_iss_retido": round(service.get("valor_iss_retido", 0), 2), - "valor_pis": round(service.get("valor_pis_retido", 0), 2), - "valor_cofins": round(service.get("valor_cofins_retido", 0), 2), - "valor_inss": round(service.get("valor_inss_retido", 0), 2), - "valor_ir": round(service.get("valor_ir_retido", 0), 2), - "valor_csll": round(service.get("valor_csll_retido", 0), 2), - "valor_deducoes": round(service.get("valor_deducoes", 0), 2), - "fonte_total_tributos": service.get("fonte_total_tributos", "IBPT"), - "desconto_incondicionado": round( - service.get("valor_desconto_incondicionado", 0), 2 - ), - "desconto_condicionado": round(service.get("desconto_condicionado", 0), 2), - "outras_retencoes": round(service.get("outras_retencoes", 0), 2), - "valor_servicos": round(service.get("valor_servicos", 0), 2), - "valor_liquido": round(service.get("valor_liquido_nfse", 0), 2), - "codigo_tributario_municipio": service.get("codigo_tributacao_municipio"), - "codigo_nbs": service.get("codigo_nbs"), - "codigo_indicador_operacao": service.get("codigo_indicador_operacao"), - "codigo_classificacao_tributaria": service.get( - "codigo_classificacao_tributaria" - ), - "codigo_situacao_tributaria": service.get("codigo_situacao_tributaria"), - "ibs_cbs_base_calculo": service.get("ibs_cbs_base_calculo"), - "ibs_uf_aliquota": round(service.get("ibs_uf_aliquota", 0), 2) - if service.get("ibs_uf_aliquota") - else None, - "ibs_mun_aliquota": 0.0, - "cbs_aliquota": round(service.get("cbs_aliquota", 0), 2) - if service.get("cbs_aliquota") - else None, - "ibs_uf_valor": round(service.get("ibs_uf_valor", 0), 2) - if service.get("ibs_uf_valor") - else None, - "ibs_mun_valor": 0.0, - "cbs_valor": round(service.get("cbs_valor", 0), 2) - if service.get("cbs_valor") - else None, - } - - def _prepare_recipient_data(self, recipient, identification, company): - """Construct the recipient section of the payload. - - Args: - recipient (dict): Information about the service recipient. - identification (dict): The recipient's identification (CPF or CNPJ). - company (recordset): The company record. - Returns: - dict: The recipient section of the payload. - """ - - if recipient.get("nif"): - recipient["codigo_municipio"] = company.city_id.ibge_code - - return { - **identification, - "nif": recipient.get("nif"), - "nif_motivo_ausencia": recipient.get("nif_motivo_ausencia"), - "razao_social": recipient.get("razao_social"), - "email": recipient.get("email"), - "endereco": { - "bairro": recipient.get("bairro"), - "cep": recipient.get("cep"), - "codigo_municipio": recipient.get("codigo_municipio"), - "logradouro": recipient.get("endereco"), - "numero": recipient.get("numero"), - "uf": recipient.get("uf"), - }, - } - - @api.model - def query_focus_nfse_by_rps(self, ref, complete, company, environment): - """Query NFSe by RPS. - - Args: - ref (str): The RPS reference. - complete (bool): Whether to return complete information. - company (recordset): The company record. - - Returns: - requests.Response: The response from the NFSe service. - """ - token = company.get_focusnfe_token() - url = f"{NFSE_URL[environment]}{API_ENDPOINT['status']}{ref}" - return self._make_focus_nfse_http_request( - "GET", url, token, params={"completa": complete} - ) - - @api.model - def cancel_focus_nfse_document(self, ref, cancel_reason, company, environment): - """Cancel an electronic fiscal document. - - Args: - ref (str): The document reference. - cancel_reason (str): The reason for cancellation. - company (recordset): The company record. - - Returns: - requests.Response: The response from the NFSe service. - """ - token = company.get_focusnfe_token() - data = {"justificativa": cancel_reason} - url = f"{NFSE_URL[environment]}{API_ENDPOINT['cancelamento']}{ref}" - return self._make_focus_nfse_http_request( - "DELETE", url, token, data=json.dumps(data) - ) - - class Document(models.Model): + """Document model with FocusNFE NFSe integration.""" + _inherit = "l10n_br_fiscal.document" def make_focus_nfse_pdf(self, content): @@ -355,8 +91,19 @@ def _serialize(self, edocs): NFSe-specific information. """ edocs = super()._serialize(edocs) + # Handle NFSe Nacional for record in self.filtered(filter_processador_edoc_nfse).filtered( - filter_focusnfe + filter_focusnfe_nacional + ): + edoc = { + "rps": record._prepare_lote_rps(), + "service": record._prepare_dados_servico(), + "recipient": record._prepare_dados_tomador(), + } + edocs.append(edoc) + # Handle NFSe Municipal (original) + for record in self.filtered(filter_processador_edoc_nfse).filtered( + filter_focusnfe_municipal ): edoc = [] edoc.append({"rps": record._prepare_lote_rps()}) @@ -394,6 +141,251 @@ def _document_export(self, pretty_print=True): record.authorization_event_id = event_id return result + def _parse_authorization_datetime(self, json_data): + """Parse authorization datetime from JSON data. + + Args: + json_data (dict): JSON response data. + + Returns: + datetime: Naive datetime in UTC. + """ + aware_datetime = datetime.strptime( + json_data["data_emissao"], "%Y-%m-%dT%H:%M:%S%z" + ) + utc_datetime = aware_datetime.astimezone(pytz.utc) + return utc_datetime.replace(tzinfo=None) + + def _fetch_xml_from_path(self, record, xml_path): + """Fetch XML content from the given path. + + Args: + record: The document record. + xml_path (str): Path to XML file. + + Returns: + str: XML content as string, empty if path is invalid. + """ + if not xml_path: + return "" + try: + return requests.get( + NFSE_URL[record.nfse_environment] + xml_path, + timeout=TIMEOUT, + verify=record.company_id.nfse_ssl_verify, + ).content.decode("utf-8") + except Exception as e: + _logger.warning("Failed to fetch XML from %s: %s", xml_path, e) + return "" + + def _fetch_pdf_from_urls(self, record, json_data, use_url_first=False): + """Fetch PDF content from URLs in JSON data. + + Args: + record: The document record. + json_data (dict): JSON response data. + use_url_first (bool): If True, try 'url' first, then 'url_danfse'. + If False, only try 'url_danfse'. + + Returns: + bytes: PDF content, or None if not found or invalid. + """ + if record.company_id.focusnfe_nfse_force_odoo_danfse: + return None + + pdf_url = None + if use_url_first: + pdf_url = json_data.get("url") + if pdf_url: + try: + pdf_content = requests.get( + pdf_url, + timeout=TIMEOUT, + verify=record.company_id.nfse_ssl_verify, + ).content + if _is_valid_pdf(pdf_content): + return pdf_content + except Exception as e: + _logger.warning("Failed to fetch PDF from %s: %s", pdf_url, e) + + pdf_url = json_data.get("url_danfse", "") + if pdf_url: + try: + pdf_content = requests.get( + pdf_url, + timeout=TIMEOUT, + verify=record.company_id.nfse_ssl_verify, + ).content + if _is_valid_pdf(pdf_content): + return pdf_content + except Exception as e: + _logger.warning("Failed to fetch PDF from %s: %s", pdf_url, e) + + return None + + def _process_authorized_status_base( + self, + record, + json_data, + verify_code_key="codigo_verificacao", + use_url_first=False, + xml_required=True, + ): + """Base method to process authorized status. + + Args: + record: The document record. + json_data (dict): JSON response data. + verify_code_key (str): Key to get verification code from json_data. + use_url_first (bool): Whether to try 'url' first for PDF. + xml_required (bool): Whether XML path is required (municipal) + or optional (nacional). + """ + naive_datetime = self._parse_authorization_datetime(json_data) + verify_code = ( + json_data.get(verify_code_key, "") + if verify_code_key + else json_data.get("codigo_verificacao", "") + ) + document_number = json_data.get("numero", "") + + record.write( + { + "verify_code": verify_code, + "document_number": document_number, + "authorization_date": naive_datetime, + } + ) + + xml_path = json_data.get("caminho_xml_nota_fiscal", "") + if xml_required and not xml_path: + # Will raise KeyError if not present + xml_path = json_data.get("caminho_xml_nota_fiscal") + + xml = self._fetch_xml_from_path(record, xml_path) if xml_path else "" + + if not record.authorization_event_id: + record._document_export() + + if record.authorization_event_id: + # For municipal, xml is required; for nacional, only if available + if xml_required or xml: + record.authorization_event_id.set_done( + status_code=4, + response=_("Successfully Processed"), + protocol_date=record.authorization_date, + protocol_number=record.authorization_protocol, + file_response_xml=xml, + ) + record._change_state(SITUACAO_EDOC_AUTORIZADA) + + if record.company_id.focusnfe_nfse_force_odoo_danfse: + record.make_pdf() + else: + pdf_content = self._fetch_pdf_from_urls( + record, json_data, use_url_first + ) + if pdf_content: + record.make_focus_nfse_pdf(pdf_content) + + def _process_authorized_status_nacional(self, record, json_data): + """Process authorized status for NFSe Nacional.""" + self._process_authorized_status_base( + record, + json_data, + verify_code_key="codigo_verificacao", + use_url_first=False, + xml_required=False, + ) + + def _process_authorized_status_municipal(self, record, json_data): + """Process authorized status for NFSe Municipal.""" + self._process_authorized_status_base( + record, + json_data, + verify_code_key="codigo_verificacao", + use_url_first=True, + xml_required=True, + ) + + def _process_error_status(self, record, json_data): + """Process error authorization status.""" + erros = json_data.get("erros", []) + error_msg = erros[0]["mensagem"] if erros else _("Authorization error") + record.write( + { + "edoc_error_message": error_msg, + } + ) + record._change_state(SITUACAO_EDOC_REJEITADA) + + def _process_status_nacional(self, record): + """Process status check for NFSe Nacional.""" + ref = str(record.rps_number) + response = record.env[ + "focusnfe.nfse.nacional" + ].query_focus_nfse_nacional_by_ref( + ref, record.company_id, record.nfse_environment + ) + + json = response.json() + + edoc_states = ["a_enviar", "enviada", "rejeitada"] + if record.company_id.focusnfe_nfse_update_authorized_document_status: + edoc_states.append("autorizada") + + if response.status_code == 200: + if record.state in edoc_states: + if ( + json["status"] == STATUS_AUTORIZADO + and record.state_edoc != SITUACAO_EDOC_AUTORIZADA + ): + self._process_authorized_status_nacional(record, json) + elif json["status"] == STATUS_ERRO_AUTORIZACAO: + self._process_error_status(record, json) + elif json["status"] == STATUS_CANCELADO: + if record.state_edoc != SITUACAO_EDOC_CANCELADA: + record._document_cancel(record.cancel_reason) + + return _(json["status"]) + + return "Unable to retrieve the document status." + + def _process_status_municipal(self, record): + """Process status check for NFSe Municipal.""" + ref = "rps" + record.rps_number + response = record.env["focusnfe.nfse"].query_focus_nfse_by_rps( + ref, 0, record.company_id, record.nfse_environment + ) + + json = response.json() + + edoc_states = ["a_enviar", "enviada", "rejeitada"] + if record.company_id.focusnfe_nfse_update_authorized_document_status: + edoc_states.append("autorizada") + + if response.status_code == 200: + if record.state in edoc_states: + if ( + json["status"] == STATUS_AUTORIZADO + and record.state_edoc != SITUACAO_EDOC_AUTORIZADA + ): + self._process_authorized_status_municipal(record, json) + elif json["status"] == STATUS_ERRO_AUTORIZACAO: + record.write( + { + "edoc_error_message": json["erros"][0]["mensagem"], + } + ) + record._change_state(SITUACAO_EDOC_REJEITADA) + elif json["status"] == STATUS_CANCELADO: + if record.state_edoc != SITUACAO_EDOC_CANCELADA: + record._document_cancel(record.cancel_reason) + + return _(json["status"]) + + return "Unable to retrieve the document status." + def _document_status(self): """Check and update the status of the NFSe document. @@ -404,92 +396,16 @@ def _document_status(self): A string indicating the current status of the document. """ result = super()._document_status() + # Handle NFSe Nacional for record in self.filtered(filter_processador_edoc_nfse).filtered( - filter_focusnfe + filter_focusnfe_nacional ): - ref = "rps" + record.rps_number - response = record.env["focusnfe.nfse"].query_focus_nfse_by_rps( - ref, 0, record.company_id, record.nfse_environment - ) - - json = response.json() - - edoc_states = ["a_enviar", "enviada", "rejeitada"] - if record.company_id.focusnfe_nfse_update_authorized_document_status: - edoc_states.append("autorizada") - - if response.status_code == 200: - if record.state in edoc_states: - if ( - json["status"] == "autorizado" - and record.state_edoc != SITUACAO_EDOC_AUTORIZADA - ): - aware_datetime = datetime.strptime( - json["data_emissao"], "%Y-%m-%dT%H:%M:%S%z" - ) - utc_datetime = aware_datetime.astimezone(pytz.utc) - naive_datetime = utc_datetime.replace(tzinfo=None) - record.write( - { - "verify_code": json["codigo_verificacao"], - "document_number": json["numero"], - "authorization_date": naive_datetime, - } - ) - - xml = requests.get( - NFSE_URL[record.nfse_environment] - + json["caminho_xml_nota_fiscal"], - timeout=TIMEOUT, - ).content.decode("utf-8") - - if not record.authorization_event_id: - record._document_export() - - if record.authorization_event_id: - record.authorization_event_id.set_done( - status_code=4, - response=_("Successfully Processed"), - protocol_date=record.authorization_date, - protocol_number=record.authorization_protocol, - file_response_xml=xml, - ) - record._change_state(SITUACAO_EDOC_AUTORIZADA) - if record.company_id.focusnfe_nfse_force_odoo_danfse: - record.make_pdf() - else: - pdf_content = requests.get( - json["url"], - timeout=TIMEOUT, - verify=record.company_id.nfse_ssl_verify, - ).content - if not pdf_content.startswith( - b"%PDF-" - ) and not pdf_content.strip().endswith(b"%%EOF"): - pdf_content = requests.get( - json["url_danfse"], - timeout=TIMEOUT, - verify=record.company_id.nfse_ssl_verify, - ).content - if pdf_content.startswith( - b"%PDF-" - ) and pdf_content.strip().endswith(b"%%EOF"): - record.make_focus_nfse_pdf(pdf_content) - - elif json["status"] == "erro_autorizacao": - record.write( - { - "edoc_error_message": json["erros"][0]["mensagem"], - } - ) - record._change_state(SITUACAO_EDOC_REJEITADA) - elif json["status"] == "cancelado": - if record.state_edoc != SITUACAO_EDOC_CANCELADA: - record._document_cancel(record.cancel_reason) - - result = _(json["status"]) - else: - result = "Unable to retrieve the document status." + result = self._process_status_nacional(record) + # Handle NFSe Municipal (original) + for record in self.filtered(filter_processador_edoc_nfse).filtered( + filter_focusnfe_municipal + ): + result = self._process_status_municipal(record) return result @@ -502,11 +418,14 @@ def create_cancel_event(self, status_json, record): Returns: The created event. """ - - xml = requests.get( - NFSE_URL[record.nfse_environment] + status_json["caminho_xml_cancelamento"], - timeout=TIMEOUT, - ).content.decode("utf-8") + xml_path = status_json.get("caminho_xml_cancelamento", "") + xml = "" + if xml_path: + xml = requests.get( + NFSE_URL[record.nfse_environment] + xml_path, + timeout=TIMEOUT, + verify=record.company_id.nfse_ssl_verify, + ).content.decode("utf-8") event = record.event_ids.create_event_save_xml( company_id=record.company_id, @@ -541,17 +460,160 @@ def fetch_and_verify_pdf_content(self, status_json, record): timeout=TIMEOUT, verify=record.company_id.nfse_ssl_verify, ).content - if not pdf_content.startswith(b"%PDF-") and not pdf_content.strip().endswith( - b"%%EOF" - ): + if not _is_valid_pdf(pdf_content): pdf_content = requests.get( status_json["url_danfse"], timeout=TIMEOUT, verify=record.company_id.nfse_ssl_verify, ).content - if pdf_content.startswith(b"%PDF-") and pdf_content.strip().endswith(b"%%EOF"): + if _is_valid_pdf(pdf_content): record.make_focus_nfse_pdf(pdf_content) + def _handle_cancelled_status(self, record, status_json, use_url_first=False): + """Handle already cancelled status. + + Args: + record: The document record. + status_json (dict): Status JSON response. + use_url_first (bool): Whether to try 'url' first for PDF. + """ + record.cancel_event_id = record.create_cancel_event(status_json, record) + if record.company_id.focusnfe_nfse_force_odoo_danfse: + record.make_pdf() + else: + if use_url_first: + record.fetch_and_verify_pdf_content(status_json, record) + else: + url_danfse = status_json.get("url_danfse", "") + if url_danfse: + pdf_content = requests.get( + url_danfse, + timeout=TIMEOUT, + verify=record.company_id.nfse_ssl_verify, + ).content + if _is_valid_pdf(pdf_content): + record.make_focus_nfse_pdf(pdf_content) + + def _process_cancel_base( + self, + record, + ref, + query_method, + cancel_method, + use_url_first=False, + apply_barueri_hack=False, + ): + """Base method to process cancellation. + + Args: + record: The document record. + ref (str): Document reference. + query_method: Method to query document status. + cancel_method: Method to cancel document. + use_url_first (bool): Whether to try 'url' first for PDF. + apply_barueri_hack (bool): Whether to apply Barueri-specific hack. + + Returns: + requests.Response: The cancellation response. + """ + # Check current status + status_response = query_method(ref, record.company_id, record.nfse_environment) + status_json = status_response.json() + + if status_response.status_code == 200: + status = ( + status_json.get("status", "") + if isinstance(status_json, dict) + else status_json.get("status", "") + ) + if ( + status == STATUS_CANCELADO + and record.state_edoc != SITUACAO_EDOC_CANCELADA + ): + self._handle_cancelled_status(record, status_json, use_url_first) + return status_response + + # Perform cancellation + response = cancel_method( + ref, record.cancel_reason, record.company_id, record.nfse_environment + ) + json_data = response.json() + + if response.status_code in [200, 400]: + code = json_data.get("codigo", "") + status = json_data.get("status", "") + + if not code: + code = json_data.get("erros", [{}])[0].get("codigo", "") + if code == "OK200" or (not code and status == STATUS_CANCELADO): + code = CODE_NFE_CANCELADA + + if code == CODE_NFE_CANCELADA or status == STATUS_CANCELADO: + # Query status again after cancellation + status_rps = query_method( + ref, record.company_id, record.nfse_environment + ) + status_json = status_rps.json() + self._handle_cancelled_status(record, status_json, use_url_first) + return response + + raise UserError( + _( + "%(code)s - %(status)s", + code=code or response.status_code, + status=status, + ) + ) + + raise UserError( + _( + "%(code)s - %(msg)s", + code=response.status_code, + msg=json_data.get("mensagem", ""), + ) + ) + + def _process_cancel_nacional(self, record): + """Process cancellation for NFSe Nacional.""" + ref = str(record.rps_number) + nfse_nacional = record.env["focusnfe.nfse.nacional"] + + def query_method(ref, company, environment): + return nfse_nacional.query_focus_nfse_nacional_by_ref( + ref, company, environment + ) + + def cancel_method(ref, cancel_reason, company, environment): + return nfse_nacional.cancel_focus_nfse_nacional_document( + ref, cancel_reason, company, environment + ) + + return self._process_cancel_base( + record, ref, query_method, cancel_method, use_url_first=False + ) + + def _process_cancel_municipal(self, record): + """Process cancellation for NFSe Municipal.""" + ref = "rps" + record.rps_number + nfse = record.env["focusnfe.nfse"] + + def query_method(ref, company, environment): + return nfse.query_focus_nfse_by_rps(ref, 0, company, environment) + + def cancel_method(ref, cancel_reason, company, environment): + return nfse.cancel_focus_nfse_document( + ref, cancel_reason, company, environment + ) + + return self._process_cancel_base( + record, + ref, + query_method, + cancel_method, + use_url_first=True, + apply_barueri_hack=True, + ) + def cancel_document_focus(self): """Cancel a NFSe document with the Focus NFSe provider. @@ -561,97 +623,74 @@ def cancel_document_focus(self): Returns: The response regarding the cancellation request. """ + # Handle NFSe Nacional for record in self.filtered(filter_processador_edoc_nfse).filtered( - filter_focusnfe + filter_focusnfe_nacional ): - ref = "rps" + record.rps_number - - status_response = record.env["focusnfe.nfse"].query_focus_nfse_by_rps( - ref, 0, record.company_id, record.nfse_environment + return self._process_cancel_nacional(record) + # Handle NFSe Municipal (original) + for record in self.filtered(filter_processador_edoc_nfse).filtered( + filter_focusnfe_municipal + ): + return self._process_cancel_municipal(record) + + def _process_send_nacional(self, record): + """Process document send for NFSe Nacional.""" + for edoc in record.serialize(): + ref = str(record.rps_number) + response = self.env[ + "focusnfe.nfse.nacional" + ].process_focus_nfse_nacional_document( + edoc, ref, record.company_id, record.nfse_environment ) - status_json = status_response.json() + json = response.json() - if status_response.status_code == 200: - if ( - status_json["status"] == "cancelado" - and record.state_edoc != SITUACAO_EDOC_CANCELADA - ): - record.cancel_event_id = record.create_cancel_event( - status_json, record - ) - if record.company_id.focusnfe_nfse_force_odoo_danfse: - record.make_pdf() + if response.status_code == 202: + if json["status"] == STATUS_PROCESSANDO_AUTORIZACAO: + if record.state == "rejeitada": + record.state_edoc = SITUACAO_EDOC_ENVIADA else: - record.fetch_and_verify_pdf_content(status_json, record) - return status_response + record._change_state(SITUACAO_EDOC_ENVIADA) + elif response.status_code == 422: + code = json.get("codigo", "") + if code == CODE_NFE_AUTORIZADA and record.state in [ + "a_enviar", + "enviada", + "rejeitada", + ]: + record._document_status() + else: + record._change_state(SITUACAO_EDOC_REJEITADA) + else: + record._change_state(SITUACAO_EDOC_REJEITADA) - response = record.env["focusnfe.nfse"].cancel_focus_nfse_document( - ref, record.cancel_reason, record.company_id, record.nfse_environment + def _process_send_municipal(self, record): + """Process document send for NFSe Municipal.""" + for edoc in record.serialize(): + ref = "rps" + record.rps_number + response = self.env["focusnfe.nfse"].process_focus_nfse_document( + edoc, ref, record.company_id, record.nfse_environment ) - - code = False - status = False - json = response.json() - if response.status_code in [200, 400]: - try: - code = json["codigo"] - response = True - except Exception: - _logger.error( - _("HTTP status is 200 or 400 but unable to read json['codigo']") - ) - try: - status = json["status"] - except Exception: - _logger.error( - _("HTTP status is 200 or 400 but unable to read json['status']") - ) - - # hack barueri - provisório - if not code and record.company_id.city_id.ibge_code == "3505708": - try: - code = json["erros"][0].get("codigo") - except Exception: - _logger.error( - _("HTTP status is 200 or 400 but unable to read error code") - ) - if code == "OK200": - code = "nfe_cancelada" - - if code == "nfe_cancelada" or status == "cancelado": - status_rps = record.env["focusnfe.nfse"].query_focus_nfse_by_rps( - ref, 0, record.company_id, record.nfse_environment - ) - status_json = status_rps.json() - - record.cancel_event_id = record.create_cancel_event( - status_json, record - ) - if record.company_id.focusnfe_nfse_force_odoo_danfse: - record.make_pdf() + if response.status_code == 202: + if json["status"] == STATUS_PROCESSANDO_AUTORIZACAO: + if record.state == "rejeitada": + record.state_edoc = SITUACAO_EDOC_ENVIADA else: - record.fetch_and_verify_pdf_content(status_json, record) - - return response - + record._change_state(SITUACAO_EDOC_ENVIADA) + elif response.status_code == 422: + code = json.get("codigo", "") + if code == CODE_NFE_AUTORIZADA and record.state in [ + "a_enviar", + "enviada", + "rejeitada", + ]: + record._document_status() else: - raise UserError( - _( - "%(code)s - %(status)s", - code=response.status_code, - status=status, - ) - ) + record._change_state(SITUACAO_EDOC_REJEITADA) else: - raise UserError( - _( - "%(code)s - %(msg)s", - code=response.status_code, - msg=json["mensagem"], - ) - ) + record._change_state(SITUACAO_EDOC_REJEITADA) def _eletronic_document_send(self): """Send the electronic document to the NFSe provider. @@ -663,38 +702,16 @@ def _eletronic_document_send(self): None. Updates the document's status based on the response. """ res = super()._eletronic_document_send() + # Handle NFSe Nacional for record in self.filtered(filter_processador_edoc_nfse).filtered( - filter_focusnfe + filter_focusnfe_nacional ): - for edoc in record.serialize(): - ref = "rps" + record.rps_number - response = self.env["focusnfe.nfse"].process_focus_nfse_document( - edoc, ref, record.company_id, record.nfse_environment - ) - json = response.json() - - if response.status_code == 202: - if json["status"] == "processando_autorizacao": - if record.state == "rejeitada": - record.state_edoc = SITUACAO_EDOC_ENVIADA - else: - record._change_state(SITUACAO_EDOC_ENVIADA) - elif response.status_code == 422: - try: - code = json["codigo"] - except Exception: - code = "" - - if code == "nfe_autorizada" and record.state in [ - "a_enviar", - "enviada", - "rejeitada", - ]: - record._document_status() - else: - record._change_state(SITUACAO_EDOC_REJEITADA) - else: - record._change_state(SITUACAO_EDOC_REJEITADA) + self._process_send_nacional(record) + # Handle NFSe Municipal (original) + for record in self.filtered(filter_processador_edoc_nfse).filtered( + filter_focusnfe_municipal + ): + self._process_send_municipal(record) return res def _exec_before_SITUACAO_EDOC_CANCELADA(self, old_state, new_state): @@ -726,5 +743,7 @@ def _cron_document_status_focus(self): .filtered(filter_processador_edoc_nfse) .filtered(filter_focusnfe) ) - if records: - records._document_status() + # Iterate over each record individually, as _document_status() + # may expect a singleton in some cases + for record in records: + record._document_status() diff --git a/l10n_br_nfse_focus/models/helpers.py b/l10n_br_nfse_focus/models/helpers.py new file mode 100644 index 000000000000..41123c802cb3 --- /dev/null +++ b/l10n_br_nfse_focus/models/helpers.py @@ -0,0 +1,70 @@ +# Copyright 2023 - TODAY, Marcel Savegnago +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +"""Helper functions for FocusNFE NFSe integration.""" + +from .constants import CNPJ_LENGTH, CPF_LENGTH, PDF_FOOTER, PDF_HEADER + + +def filter_focusnfe(record): + """Filter records with FocusNFE provider.""" + return record.company_id.provedor_nfse == "focusnfe" + + +def filter_focusnfe_nacional(record): + """Filter records with FocusNFE Nacional type.""" + return ( + record.company_id.provedor_nfse == "focusnfe" + and record.company_id.focusnfe_nfse_type == "nfse_nacional" + ) + + +def filter_focusnfe_municipal(record): + """Filter records with FocusNFE Municipal type.""" + return ( + record.company_id.provedor_nfse == "focusnfe" + and record.company_id.focusnfe_nfse_type == "nfse" + ) + + +def _clean_cpf_cnpj(value): + """Remove formatting from CPF/CNPJ string. + + Args: + value (str): CPF or CNPJ string with formatting. + + Returns: + str: Cleaned CPF/CNPJ string with only digits. + """ + if not value: + return "" + return value.replace(".", "").replace("/", "").replace("-", "") + + +def _identify_cpf_cnpj(cpf, cnpj): + """Identify if the provided values are CPF or CNPJ. + + Args: + cpf (str): CPF value. + cnpj (str): CNPJ value. + + Returns: + tuple: (is_cpf, is_cnpj, cleaned_cpf, cleaned_cnpj) + """ + cleaned_cpf = _clean_cpf_cnpj(cpf) if cpf else "" + cleaned_cnpj = _clean_cpf_cnpj(cnpj) if cnpj else "" + is_cpf = bool(cleaned_cpf and len(cleaned_cpf) == CPF_LENGTH) + is_cnpj = bool(cleaned_cnpj and len(cleaned_cnpj) == CNPJ_LENGTH) + return is_cpf, is_cnpj, cleaned_cpf, cleaned_cnpj + + +def _is_valid_pdf(content): + """Check if content is a valid PDF. + + Args: + content (bytes): PDF content to validate. + + Returns: + bool: True if content is a valid PDF, False otherwise. + """ + return content.startswith(PDF_HEADER) and content.strip().endswith(PDF_FOOTER) diff --git a/l10n_br_nfse_focus/models/nfse_municipal.py b/l10n_br_nfse_focus/models/nfse_municipal.py new file mode 100644 index 000000000000..a7db6cb7dc9f --- /dev/null +++ b/l10n_br_nfse_focus/models/nfse_municipal.py @@ -0,0 +1,269 @@ +# Copyright 2023 - TODAY, Marcel Savegnago +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +"""NFSe Municipal integration with FocusNFE.""" + +import json + +from odoo import api + +from .base import FocusnfeNfseBase +from .constants import API_ENDPOINT, NFSE_URL + + +class FocusnfeNfse(FocusnfeNfseBase): + """FocusNFE NFSe Municipal implementation.""" + + _name = "focusnfe.nfse" + _description = "FocusNFE NFSE" + + def _make_focus_nfse_http_request(self, method, url, token, data=None, params=None): + """Perform a generic HTTP request. + + Args: + method (str): The HTTP method to use (e.g., 'GET', 'POST'). + url (str): The URL to which the request is sent. + token (str): The authentication token for the service. + data (dict, optional): The payload to send in the request body. + Defaults to None. + params (dict, optional): The URL parameters to append to the URL. + Defaults to None. + + Returns: + requests.Response: The response object from the requests library. + + Raises: + UserError: If the HTTP request fails with a 4xx/5xx response. + """ + return super()._make_focus_nfse_http_request( + method, url, token, data, params, service_name="NFSe" + ) + + def _identify_service_recipient(self, recipient): + """Identify whether the service recipient is a CPF or CNPJ. + + Args: + recipient (dict): A dictionary containing either 'cpf' or 'cnpj' keys. + + Returns: + dict: A dictionary with either a 'cpf' or 'cnpj' key and its value. + """ + return ( + {"cpf": recipient.get("cpf")} + if recipient.get("cpf") + else {"cnpj": recipient.get("cnpj")} + ) + + @api.model + def process_focus_nfse_document(self, edoc, ref, company, environment): + """Process the electronic fiscal document. + + Args: + edoc (tuple): The electronic document data. + ref (str): The document reference. + company (recordset): The company record. + environment (str): The environment (1=production, 2=homologation). + + Returns: + requests.Response: The response from the NFSe service. + """ + token = company.get_focusnfe_token() + data = self._prepare_payload(*edoc, company) + payload = json.dumps(data) + url = f"{NFSE_URL[environment]}{API_ENDPOINT['envio']}" + ref = {"ref": ref} + return self._make_focus_nfse_http_request( + "POST", url, token, data=payload, params=ref + ) + + def _prepare_payload(self, rps, service, recipient, company): + """Construct the NFSe payload. + + Args: + rps (dict): Information about the RPS. + service (dict): Details of the service provided. + recipient (dict): Information about the service recipient. + company (recordset): The company record. + + Returns: + dict: The complete payload for the NFSe request. + """ + rps_info = rps.get("rps") + service_info = service.get("service") + recipient_info = recipient.get("recipient") + recipient_identification = self._identify_service_recipient(recipient_info) + + vals = { + "prestador": self._prepare_provider_data(rps_info, company), + "servico": self._prepare_service_data(service_info, company), + "tomador": self._prepare_recipient_data( + recipient_info, recipient_identification, company + ), + "razao_social": company.name, + "data_emissao": rps_info.get("data_emissao"), + "incentivador_cultural": rps_info.get("incentivador_cultural", False), + "natureza_operacao": rps_info.get("natureza_operacao"), + "optante_simples_nacional": rps_info.get("optante_simples_nacional", False), + "status": rps_info.get("status"), + "informacoes_adicionais_contribuinte": ( + rps_info.get("customer_additional_data", False)[:256] + if rps_info.get("customer_additional_data") + else False + ), + } + codigo_obra = rps_info.get("codigo_obra", False) + art = rps_info.get("art", False) + + if codigo_obra: + vals["codigo_obra"] = codigo_obra + + if art: + vals["art"] = art + + return vals + + def _prepare_provider_data(self, rps, company): + """Construct the provider section of the payload. + + Args: + rps (dict): Information about the RPS. + company (recordset): The company record. + + Returns: + dict: The provider section of the payload. + """ + return { + "cnpj": rps.get("cnpj"), + "inscricao_municipal": rps.get("inscricao_municipal"), + "codigo_municipio": company.city_id.ibge_code, + } + + def _prepare_service_data(self, service, company): + """Construct the service section of the payload. + + Args: + service (dict): Details of the service provided. + company (recordset): The company record. + + Returns: + dict: The service section of the payload. + """ + return { + "aliquota": service.get("aliquota") + if company.focusnfe_tax_rate_format == "decimal" + else round(service.get("aliquota", 0.0) * 100, 1), + "base_calculo": round(service.get("base_calculo", 0), 2), + "discriminacao": service.get("discriminacao"), + "iss_retido": service.get("iss_retido"), + "codigo_municipio": service.get("municipio_prestacao_servico"), + "codigo_municipio_incidencia": service.get("municipio_prestacao_servico"), + "item_lista_servico": service.get(company.focusnfe_nfse_service_type_value), + "codigo_cnae": service.get(company.focusnfe_nfse_cnae_code_value), + "valor_iss": round(service.get("valor_iss", 0), 2), + "valor_iss_retido": round(service.get("valor_iss_retido", 0), 2), + "valor_pis": round(service.get("valor_pis_retido", 0), 2), + "valor_cofins": round(service.get("valor_cofins_retido", 0), 2), + "valor_inss": round(service.get("valor_inss_retido", 0), 2), + "valor_ir": round(service.get("valor_ir_retido", 0), 2), + "valor_csll": round(service.get("valor_csll_retido", 0), 2), + "valor_deducoes": round(service.get("valor_deducoes", 0), 2), + "fonte_total_tributos": service.get("fonte_total_tributos", "IBPT"), + "desconto_incondicionado": round( + service.get("valor_desconto_incondicionado", 0), 2 + ), + "desconto_condicionado": round(service.get("desconto_condicionado", 0), 2), + "outras_retencoes": round(service.get("outras_retencoes", 0), 2), + "valor_servicos": round(service.get("valor_servicos", 0), 2), + "valor_liquido": round(service.get("valor_liquido_nfse", 0), 2), + "codigo_tributario_municipio": service.get("codigo_tributacao_municipio"), + "codigo_nbs": service.get("codigo_nbs"), + "codigo_indicador_operacao": service.get("codigo_indicador_operacao"), + "codigo_classificacao_tributaria": service.get( + "codigo_classificacao_tributaria" + ), + "codigo_situacao_tributaria": service.get("codigo_situacao_tributaria"), + "ibs_cbs_base_calculo": service.get("ibs_cbs_base_calculo"), + "ibs_uf_aliquota": round(service.get("ibs_uf_aliquota", 0), 2) + if service.get("ibs_uf_aliquota") + else None, + "ibs_mun_aliquota": 0.0, + "cbs_aliquota": round(service.get("cbs_aliquota", 0), 2) + if service.get("cbs_aliquota") + else None, + "ibs_uf_valor": round(service.get("ibs_uf_valor", 0), 2) + if service.get("ibs_uf_valor") + else None, + "ibs_mun_valor": 0.0, + "cbs_valor": round(service.get("cbs_valor", 0), 2) + if service.get("cbs_valor") + else None, + } + + def _prepare_recipient_data(self, recipient, identification, company): + """Construct the recipient section of the payload. + + Args: + recipient (dict): Information about the service recipient. + identification (dict): The recipient's identification (CPF or CNPJ). + company (recordset): The company record. + + Returns: + dict: The recipient section of the payload. + """ + if recipient.get("nif"): + recipient["codigo_municipio"] = company.city_id.ibge_code + + return { + **identification, + "nif": recipient.get("nif"), + "nif_motivo_ausencia": recipient.get("nif_motivo_ausencia"), + "razao_social": recipient.get("razao_social"), + "email": recipient.get("email"), + "endereco": { + "bairro": recipient.get("bairro"), + "cep": recipient.get("cep"), + "codigo_municipio": recipient.get("codigo_municipio"), + "logradouro": recipient.get("endereco"), + "numero": recipient.get("numero"), + "uf": recipient.get("uf"), + }, + } + + @api.model + def query_focus_nfse_by_rps(self, ref, complete, company, environment): + """Query NFSe by RPS. + + Args: + ref (str): The RPS reference. + complete (bool): Whether to return complete information. + company (recordset): The company record. + environment (str): The environment (1=production, 2=homologation). + + Returns: + requests.Response: The response from the NFSe service. + """ + token = company.get_focusnfe_token() + url = f"{NFSE_URL[environment]}{API_ENDPOINT['status']}{ref}" + return self._make_focus_nfse_http_request( + "GET", url, token, params={"completa": complete} + ) + + @api.model + def cancel_focus_nfse_document(self, ref, cancel_reason, company, environment): + """Cancel an electronic fiscal document. + + Args: + ref (str): The document reference. + cancel_reason (str): The reason for cancellation. + company (recordset): The company record. + environment (str): The environment (1=production, 2=homologation). + + Returns: + requests.Response: The response from the NFSe service. + """ + token = company.get_focusnfe_token() + data = {"justificativa": cancel_reason} + url = f"{NFSE_URL[environment]}{API_ENDPOINT['cancelamento']}{ref}" + return self._make_focus_nfse_http_request( + "DELETE", url, token, data=json.dumps(data) + ) diff --git a/l10n_br_nfse_focus/models/nfse_nacional.py b/l10n_br_nfse_focus/models/nfse_nacional.py new file mode 100644 index 000000000000..758f5d4cc9f3 --- /dev/null +++ b/l10n_br_nfse_focus/models/nfse_nacional.py @@ -0,0 +1,362 @@ +# Copyright 2023 - TODAY, Marcel Savegnago +# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl). + +"""NFSe Nacional integration with FocusNFE.""" + +import json + +from odoo import api + +from .base import FocusnfeNfseBase +from .constants import API_ENDPOINT_NACIONAL, NFSE_URL +from .helpers import _identify_cpf_cnpj + + +class FocusnfeNfseNacional(FocusnfeNfseBase): + """FocusNFE NFSe Nacional implementation.""" + + _name = "focusnfe.nfse.nacional" + _description = "FocusNFE NFSe Nacional" + + def _make_focus_nfse_http_request(self, method, url, token, data=None, params=None): + """Perform a generic HTTP request. + + Args: + method (str): The HTTP method to use (e.g., 'GET', 'POST'). + url (str): The URL to which the request is sent. + token (str): The authentication token for the service. + data (dict, optional): The payload to send in the request body. + Defaults to None. + params (dict, optional): The URL parameters to append to the URL. + Defaults to None. + + Returns: + requests.Response: The response object from the requests library. + + Raises: + UserError: If the HTTP request fails with a 4xx/5xx response. + """ + return super()._make_focus_nfse_http_request( + method, url, token, data, params, service_name="NFSe Nacional" + ) + + @api.model + def process_focus_nfse_nacional_document(self, edoc, ref, company, environment): + """Process the electronic fiscal document for NFSe Nacional. + + Args: + edoc (dict): The electronic document data. + ref (str): The document reference. + company (recordset): The company record. + environment (str): The environment (1=production, 2=homologation). + + Returns: + requests.Response: The response from the NFSe Nacional service. + """ + token = company.get_focusnfe_token() + data = self._prepare_payload_nacional(edoc, company) + payload = json.dumps(data) + url = f"{NFSE_URL[environment]}{API_ENDPOINT_NACIONAL['envio']}" + ref_params = {"ref": ref} + return self._make_focus_nfse_http_request( + "POST", url, token, data=payload, params=ref_params + ) + + def _prepare_dates_nacional(self, rps_info): + """Prepare emission and competence dates for NFSe Nacional. + + Args: + rps_info (dict): RPS information. + + Returns: + tuple: (emission_date, competence_date) + """ + emission_date = rps_info.get("data_emissao", "") + if emission_date and not emission_date.endswith(("-0300", "-0200", "+0000")): + # Add timezone if not present (assuming -0300 for Brazil) + emission_date = emission_date + "-0300" + + competence_date = ( + rps_info.get("data_emissao", "")[:10] + if rps_info.get("data_emissao") + else "" + ) + + return emission_date, competence_date + + def _prepare_provider_nacional(self, rps_info, company): + """Prepare provider data for NFSe Nacional. + + Args: + rps_info (dict): RPS information. + company (recordset): The company record. + + Returns: + dict: Provider data with CPF/CNPJ identification. + """ + cnpj_prestador = rps_info.get("cnpj", "") + cpf_prestador = rps_info.get("cpf", "") + ( + is_cpf_prestador, + is_cnpj_prestador, + cpf_prestador_limpo, + cnpj_prestador_limpo, + ) = _identify_cpf_cnpj(cpf_prestador, cnpj_prestador) + + optante_simples = rps_info.get("optante_simples_nacional", "1") + codigo_opcao_simples_nacional = "2" if optante_simples == "1" else "1" + + regime_especial_tributacao = ( + rps_info.get("regime_especial_tributacao", "0") or "0" + ) + + return { + "is_cpf": is_cpf_prestador, + "is_cnpj": is_cnpj_prestador, + "cpf_limpo": cpf_prestador_limpo, + "cnpj_limpo": cnpj_prestador_limpo, + "codigo_opcao_simples_nacional": codigo_opcao_simples_nacional, + "regime_especial_tributacao": regime_especial_tributacao, + "codigo_municipio_emissora": str(company.city_id.ibge_code or ""), + } + + def _prepare_recipient_nacional(self, recipient_info): + """Prepare recipient data for NFSe Nacional. + + Args: + recipient_info (dict): Recipient information. + + Returns: + dict: Recipient data with CPF/CNPJ identification. + """ + cnpj_tomador = recipient_info.get("cnpj", "") + cpf_tomador = recipient_info.get("cpf", "") + is_cpf, is_cnpj, cpf_limpo, cnpj_limpo = _identify_cpf_cnpj( + cpf_tomador, cnpj_tomador + ) + + cep_tomador = recipient_info.get("cep", "") + if isinstance(cep_tomador, int): + cep_tomador = str(cep_tomador) + + return { + "is_cpf": is_cpf, + "is_cnpj": is_cnpj, + "cpf_limpo": cpf_limpo, + "cnpj_limpo": cnpj_limpo, + "razao_social": recipient_info.get("razao_social", ""), + "codigo_municipio": str(recipient_info.get("codigo_municipio", "")), + "cep": cep_tomador or "", + "logradouro": recipient_info.get("endereco", ""), + "numero": recipient_info.get("numero", ""), + "complemento": recipient_info.get("complemento", ""), + "bairro": recipient_info.get("bairro", ""), + "telefone": recipient_info.get("telefone", ""), + "email": recipient_info.get("email", ""), + } + + def _prepare_service_basic_nacional(self, service_info): + """Prepare basic service data for NFSe Nacional. + + Args: + service_info (dict): Service information. + + Returns: + dict: Basic service data. + """ + codigo_municipio_prestacao = service_info.get("municipio_prestacao_servico", "") + + codigo_tributacao_nacional = service_info.get("codigo_tributacao_nacional", "") + + codigo_tributacao_municipio = service_info.get( + "codigo_tributacao_municipio", "" + ) + + tributacao_iss = service_info.get("codigo_tributacao_iss", "") + + # TODO: improve logic to get ISS retention code + tipo_retencao_iss = "2" if service_info.get("iss_retido") == "1" else "1" + + return { + "codigo_municipio_prestacao": str(codigo_municipio_prestacao), + "codigo_tributacao_nacional": codigo_tributacao_nacional, + "codigo_tributacao_municipio": codigo_tributacao_municipio, + "descricao": service_info.get("discriminacao", ""), + "valor": round(service_info.get("valor_servicos", 0), 2), + "tributacao_iss": str(tributacao_iss), + "tipo_retencao_iss": str(tipo_retencao_iss), + } + + def _prepare_tax_data_nacional(self, service_info, valor_servico): + """Prepare tax data (PIS/COFINS, etc.) for NFSe Nacional. + + Args: + service_info (dict): Service information. + valor_servico (float): Service value. + + Returns: + dict: Tax data. + """ + # PIS/COFINS tax situation + situacao_tributaria_pis_cofins = ( + service_info.get("situacao_tributaria_pis", "") + or service_info.get("situacao_tributaria_cofins", "") + or "" + ) + if situacao_tributaria_pis_cofins == "99": + situacao_tributaria_pis_cofins = "00" + + # PIS/COFINS calculation base + base_calculo_pis = service_info.get("base_calculo_pis", 0) + base_calculo_cofins = service_info.get("base_calculo_cofins", 0) + base_calculo_pis_cofins = round( + base_calculo_pis if base_calculo_pis else base_calculo_cofins, 2 + ) + + if situacao_tributaria_pis_cofins: + if situacao_tributaria_pis_cofins in ["00", "08", "09"]: + base_calculo_pis_cofins = 0.0 + else: + if not base_calculo_pis_cofins or base_calculo_pis_cofins == 0: + base_calculo_pis_cofins = round(valor_servico, 2) + + # Format rates as strings with 2 decimal places + aliquota_pis_raw = round(service_info.get("aliquota_pis", 0), 2) + aliquota_pis = f"{aliquota_pis_raw:.2f}" + aliquota_cofins_raw = round(service_info.get("aliquota_cofins", 0), 2) + aliquota_cofins = f"{aliquota_cofins_raw:.2f}" + + return { + "situacao_tributaria_pis_cofins": situacao_tributaria_pis_cofins or "", + "base_calculo_pis_cofins": round(base_calculo_pis_cofins, 2), + "aliquota_pis": aliquota_pis, + "aliquota_cofins": aliquota_cofins, + "valor_pis": round(service_info.get("valor_pis", 0), 2), + "valor_cofins": round(service_info.get("valor_cofins", 0), 2), + "tipo_retencao_pis_cofins": service_info.get( + "tipo_retencao_pis_cofins", "2" + ), + "valor_cp": round(service_info.get("valor_inss_retido", 0), 2), + "valor_irrf": round(service_info.get("valor_ir_retido", 0), 2), + "valor_csll": round(service_info.get("valor_csll_retido", 0), 2), + } + + def _prepare_payload_nacional(self, edoc, company): + """Construct the NFSe Nacional payload. + + Args: + edoc (dict): The electronic document data containing rps, + service, recipient. + company (recordset): The company record. + + Returns: + dict: The complete payload for the NFSe Nacional request. + """ + rps_info = edoc.get("rps", {}) + service_info = edoc.get("service", {}) + recipient_info = edoc.get("recipient", {}) + + # Prepare dates + emission_date, competence_date = self._prepare_dates_nacional(rps_info) + + # Prepare provider data + provider_data = self._prepare_provider_nacional(rps_info, company) + + # Prepare recipient data + recipient_data = self._prepare_recipient_nacional(recipient_info) + + # Prepare service data + service_basic = self._prepare_service_basic_nacional(service_info) + tax_data = self._prepare_tax_data_nacional(service_info, service_basic["valor"]) + + # Build payload + payload = { + "data_emissao": emission_date, + "data_competencia": competence_date, + "codigo_municipio_emissora": provider_data["codigo_municipio_emissora"], + **( + {"cnpj_prestador": provider_data["cnpj_limpo"]} + if provider_data["is_cnpj"] + else {} + ), + **( + {"cpf_prestador": provider_data["cpf_limpo"]} + if provider_data["is_cpf"] + else {} + ), + "codigo_opcao_simples_nacional": provider_data[ + "codigo_opcao_simples_nacional" + ], + "regime_especial_tributacao": provider_data["regime_especial_tributacao"], + **( + {"cnpj_tomador": recipient_data["cnpj_limpo"]} + if recipient_data["is_cnpj"] + else {} + ), + **( + {"cpf_tomador": recipient_data["cpf_limpo"]} + if recipient_data["is_cpf"] + else {} + ), + "razao_social_tomador": recipient_data["razao_social"], + "codigo_municipio_tomador": recipient_data["codigo_municipio"], + "cep_tomador": recipient_data["cep"], + "logradouro_tomador": recipient_data["logradouro"], + "numero_tomador": recipient_data["numero"], + "complemento_tomador": recipient_data["complemento"], + "bairro_tomador": recipient_data["bairro"], + "telefone_tomador": recipient_data["telefone"], + "email_tomador": recipient_data["email"], + "codigo_municipio_prestacao": service_basic["codigo_municipio_prestacao"], + "codigo_tributacao_nacional_iss": service_basic[ + "codigo_tributacao_nacional" + ], + "codigo_tributacao_municipal_iss": service_basic[ + "codigo_tributacao_municipio" + ], + "descricao_servico": service_basic["descricao"], + "valor_servico": service_basic["valor"], + "tributacao_iss": service_basic["tributacao_iss"], + "tipo_retencao_iss": service_basic["tipo_retencao_iss"], + **tax_data, + } + + return payload + + @api.model + def query_focus_nfse_nacional_by_ref(self, ref, company, environment): + """Query NFSe Nacional by reference. + + Args: + ref (str): The document reference. + company (recordset): The company record. + environment (str): The environment (1=production, 2=homologation). + + Returns: + requests.Response: The response from the NFSe Nacional service. + """ + token = company.get_focusnfe_token() + url = f"{NFSE_URL[environment]}{API_ENDPOINT_NACIONAL['status']}{ref}" + return self._make_focus_nfse_http_request("GET", url, token) + + @api.model + def cancel_focus_nfse_nacional_document( + self, ref, cancel_reason, company, environment + ): + """Cancel an electronic fiscal document for NFSe Nacional. + + Args: + ref (str): The document reference. + cancel_reason (str): The reason for cancellation. + company (recordset): The company record. + environment (str): The environment (1=production, 2=homologation). + + Returns: + requests.Response: The response from the NFSe Nacional service. + """ + token = company.get_focusnfe_token() + data = {"justificativa": cancel_reason} + url = f"{NFSE_URL[environment]}{API_ENDPOINT_NACIONAL['cancelamento']}{ref}" + return self._make_focus_nfse_http_request( + "DELETE", url, token, data=json.dumps(data) + ) diff --git a/l10n_br_nfse_focus/models/res_company.py b/l10n_br_nfse_focus/models/res_company.py index fb03f08b1db3..25449af2908f 100644 --- a/l10n_br_nfse_focus/models/res_company.py +++ b/l10n_br_nfse_focus/models/res_company.py @@ -26,6 +26,7 @@ class ResCompany(models.Model): [ ("item_lista_servico", "Service Type"), ("codigo_tributacao_municipio", "City Taxation Code"), + ("codigo_tributacao_nacional", "National Taxation Code"), ], string="NFSE Service Type Value", default="item_lista_servico", @@ -60,6 +61,16 @@ class ResCompany(models.Model): [("decimal", "Decimal"), ("percentage", "Percentage")], default="decimal" ) + focusnfe_nfse_type = fields.Selection( + [ + ("nfse", "NFSe"), + ("nfse_nacional", "NFSe Nacional"), + ], + string="FocusNFe NFSe Type", + default="nfse", + help="Select whether to use NFSe (municipal) or NFSe Nacional (national) API", + ) + def get_focusnfe_token(self): """ Retrieve the appropriate FocusNFe API token based on the current NFSe diff --git a/l10n_br_nfse_focus/readme/CONFIGURE.md b/l10n_br_nfse_focus/readme/CONFIGURE.md index 84b04fc99ba8..9a2a45d103d9 100644 --- a/l10n_br_nfse_focus/readme/CONFIGURE.md +++ b/l10n_br_nfse_focus/readme/CONFIGURE.md @@ -14,7 +14,15 @@ para a empresa desejada: > - **Provedor NFS-e:** Selecione a opção FocusNFE > - **FocusNFe Token:** Informe o token de acesso da empresa. Obs. > Este token é obtido através da plataforma da FocusNFE + > - **Token de Produção:** Token para ambiente de produção (visível quando Ambiente NFS-e = Produção) + > - **Token de Homologação:** Token para ambiente de homologação (visível quando Ambiente NFS-e = Homologação) + > - **Tipo FocusNFe NFSe:** Selecione o tipo de API a ser utilizada: + > - **NFSe:** Para emissão de NFSe Municipal (padrão) + > - **NFSe Nacional:** Para emissão de NFSe Nacional > - **Valor Tipo de Serviço:** Se necessário configure o campo que - > deve preencher o valor de tipo de serviço + > deve preencher o valor de tipo de serviço (Service Type ou City Taxation Code) > - **Valor Código CNAE:** Se necessário configure o campo que deve - > preencher o valor do Código CNAE + > preencher o valor do Código CNAE (CNAE Code ou City Taxation Code) + > - **Formato Taxa:** Selecione o formato da taxa (Decimal ou Percentage) + > - **Incluir Documentos Autorizados na Verificação de Status:** Se marcado, documentos autorizados serão incluídos na verificação de status + > - **Forçar DANFSE Odoo:** Se marcado, o sistema sempre usará o DANFSE do Odoo ao invés do DANFSE da FocusNFE diff --git a/l10n_br_nfse_focus/readme/DESCRIPTION.md b/l10n_br_nfse_focus/readme/DESCRIPTION.md index 3846fbf0de2b..f8568ec2d639 100644 --- a/l10n_br_nfse_focus/readme/DESCRIPTION.md +++ b/l10n_br_nfse_focus/readme/DESCRIPTION.md @@ -1,5 +1,9 @@ -Esse módulo integra a emissão de Notas Fiscais de Serviços(NFSe) com a +Esse módulo integra a emissão de Notas Fiscais de Serviços (NFSe) com a API da FocusNFE permitindo assim, a criação, transmissão, consulta e cancelamento de documentos fiscais do tipo NFSe. +O módulo suporta dois tipos de emissão: +- **NFSe Municipal:** Emissão de NFSe através da API municipal da FocusNFE +- **NFSe Nacional:** Emissão de NFSe através da API nacional da FocusNFE + Para mais informações, acesse: diff --git a/l10n_br_nfse_focus/readme/USAGE.md b/l10n_br_nfse_focus/readme/USAGE.md index 5c3fad8166f8..ca356c25bb57 100644 --- a/l10n_br_nfse_focus/readme/USAGE.md +++ b/l10n_br_nfse_focus/readme/USAGE.md @@ -1,8 +1,17 @@ Para usar este módulo: -1. Crie uma fatura com o tipo de documento fiscal 'SE'. -2. Preencha os detalhes necessários, como o código tributário da - cidade, impostos e informações correlatas. -3. Valide o documento. -4. Envie o Documento Fiscal. -5. Acompanhe o status de processamento do documento. +1. Configure a empresa conforme descrito na seção de Configuração. + +2. Crie uma fatura com o tipo de documento fiscal 'SE'. + +3. Preencha os detalhes necessários: + - Para **NFSe Municipal:** Preencha o código tributário municipal, impostos e informações correlatas + - Para **NFSe Nacional:** Preencha o código tributário nacional (NBS), código tributário municipal (se aplicável), impostos e informações correlatas + +4. Valide o documento fiscal. + +5. Envie o Documento Fiscal através do botão "Enviar Documento Fiscal". + +6. Acompanhe o status de processamento do documento. O sistema verificará automaticamente o status através de um cron job, ou você pode verificar manualmente através do botão "Verificar Status". + +7. Após a autorização, o DANFSE (Documento Auxiliar da Nota Fiscal de Serviço Eletrônica) será gerado automaticamente, a menos que a opção "Forçar DANFSE Odoo" esteja marcada na configuração da empresa. diff --git a/l10n_br_nfse_focus/static/description/index.html b/l10n_br_nfse_focus/static/description/index.html index 37c25a7cc73e..107de66098cf 100644 --- a/l10n_br_nfse_focus/static/description/index.html +++ b/l10n_br_nfse_focus/static/description/index.html @@ -3,7 +3,7 @@ -README.rst +NFS-e (FocusNFE) -
+
+

NFS-e (FocusNFE)

- - -Odoo Community Association - -
-

NFS-e (FocusNFE)

-

Beta License: AGPL-3 OCA/l10n-brazil Translate me on Weblate Try me on Runboat

-

Esse módulo integra a emissão de Notas Fiscais de Serviços(NFSe) com a +

Beta License: AGPL-3 OCA/l10n-brazil Translate me on Weblate Try me on Runboat

+

Esse módulo integra a emissão de Notas Fiscais de Serviços (NFSe) com a API da FocusNFE permitindo assim, a criação, transmissão, consulta e cancelamento de documentos fiscais do tipo NFSe.

+

O módulo suporta dois tipos de emissão:

+
    +
  • NFSe Municipal: Emissão de NFSe através da API municipal da +FocusNFE
    • +
  • NFSe Nacional: Emissão de NFSe através da API nacional da FocusNFE
    • +

Para mais informações, acesse: https://focusnfe.com.br/

Table of contents

@@ -395,12 +396,12 @@

NFS-e (FocusNFE)

-

Installation

+

Installation

Para instalar esta funcionalidde, simplesmente instale o módulo e faça as devidas configurações.

-

Configuration

+

Configuration

Após a instalação do módulo, siga os seguintes passos para configurá-lo para a empresa desejada:

    @@ -417,30 +418,62 @@

    Configuration

    (Produção, Homologação)
  1. Provedor NFS-e: Selecione a opção FocusNFE
  2. FocusNFe Token: Informe o token de acesso da empresa. Obs. -Este token é obtido através da plataforma da FocusNFE
    3. +Este token é obtido através da plataforma da FocusNFE
      +
    • Token de Produção: Token para ambiente de produção +(visível quando Ambiente NFS-e = Produção)
      • +
    • Token de Homologação: Token para ambiente de homologação +(visível quando Ambiente NFS-e = Homologação)
      • +
    + +
  3. Tipo FocusNFe NFSe: Selecione o tipo de API a ser utilizada:
      +
    • NFSe: Para emissão de NFSe Municipal (padrão)
      • +
    • NFSe Nacional: Para emissão de NFSe Nacional
      • +
    +
  4. Valor Tipo de Serviço: Se necessário configure o campo que -deve preencher o valor de tipo de serviço
    5. +deve preencher o valor de tipo de serviço (Service Type ou City +Taxation Code)
  5. Valor Código CNAE: Se necessário configure o campo que deve -preencher o valor do Código CNAE
    6. +preencher o valor do Código CNAE (CNAE Code ou City Taxation +Code) +
  6. Formato Taxa: Selecione o formato da taxa (Decimal ou +Percentage)
    7. +
  7. Incluir Documentos Autorizados na Verificação de Status: Se +marcado, documentos autorizados serão incluídos na verificação +de status
    8. +
  8. Forçar DANFSE Odoo: Se marcado, o sistema sempre usará o +DANFSE do Odoo ao invés do DANFSE da FocusNFE
-

Usage

+

Usage

Para usar este módulo:

    +
  1. Configure a empresa conforme descrito na seção de Configuração.
  2. Crie uma fatura com o tipo de documento fiscal ‘SE’.
    3. -
  3. Preencha os detalhes necessários, como o código tributário da cidade, -impostos e informações correlatas.
    4. -
  4. Valide o documento.
    5. -
  5. Envie o Documento Fiscal.
    6. -
  6. Acompanhe o status de processamento do documento.
    7. +
  7. Preencha os detalhes necessários:
      +
    • Para NFSe Municipal: Preencha o código tributário municipal, +impostos e informações correlatas
      • +
    • Para NFSe Nacional: Preencha o código tributário nacional +(NBS), código tributário municipal (se aplicável), impostos e +informações correlatas
      • +
    +
    8. +
  8. Valide o documento fiscal.
    9. +
  9. Envie o Documento Fiscal através do botão “Enviar Documento Fiscal”.
    10. +
  10. Acompanhe o status de processamento do documento. O sistema +verificará automaticamente o status através de um cron job, ou você +pode verificar manualmente através do botão “Verificar Status”.
    11. +
  11. Após a autorização, o DANFSE (Documento Auxiliar da Nota Fiscal de +Serviço Eletrônica) será gerado automaticamente, a menos que a opção +“Forçar DANFSE Odoo” esteja marcada na configuração da empresa.
-

Bug Tracker

+

Bug Tracker

Bugs are tracked on GitHub Issues. In case of trouble, please check there if your issue has already been reported. If you spotted it first, help us to smash it by providing a detailed and welcomed @@ -448,16 +481,16 @@

Bug Tracker

Do not contact contributors directly about support or help with technical issues.

-

Credits

+

Credits

-

Authors

+

Authors

  • KMEE
  • Escodoo
-

Contributors

+

Contributors

-

Maintainers

+

Maintainers

This module is maintained by the OCA.

Odoo Community Association @@ -486,6 +519,5 @@

Maintainers

-
diff --git a/l10n_br_nfse_focus/tests/test_l10n_br_nfse_focus.py b/l10n_br_nfse_focus/tests/test_l10n_br_nfse_focus.py index 33bc3dcca8cd..b25410eec825 100644 --- a/l10n_br_nfse_focus/tests/test_l10n_br_nfse_focus.py +++ b/l10n_br_nfse_focus/tests/test_l10n_br_nfse_focus.py @@ -25,7 +25,13 @@ # Importing necessary models and functions for NFSe processing from ... import l10n_br_nfse_focus -from ..models.document import API_ENDPOINT, NFSE_URL, Document, filter_focusnfe +from ..models.constants import API_ENDPOINT, NFSE_URL +from ..models.document import Document +from ..models.helpers import ( + filter_focusnfe, + filter_focusnfe_municipal, + filter_focusnfe_nacional, +) # Mock path for testing purposes MOCK_PATH = "odoo.addons.l10n_br_nfse_focus" @@ -199,7 +205,7 @@ def test_filter_focusnfe(self): record.company_id.provedor_nfse, "focusnfe" ) # Asserting provider is set to focusnfe - @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.request") + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") def test_processar_documento(self, mock_post): """Tests document processing with mocked POST request.""" mock_post.return_value.status_code = 200 # Simulating successful POST request @@ -216,7 +222,7 @@ def test_processar_documento(self, mock_post): result.json(), {"status": "simulado"} ) # Asserting expected JSON response - @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.request") + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") def test_make_focus_nfse_http_request_generic(self, mock_request): """ Tests generic HTTP request for Focus NFSe operations with mocked responses. @@ -283,7 +289,7 @@ def mock_response_based_on_method(method, data): result_delete.json(), {"status": "success"} ) # Asserting JSON response - @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.request") + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") def test_consulta_nfse_rps(self, mock_get): """Tests NFSe query by RPS with mocked GET request.""" mock_get.return_value.status_code = 200 # Simulating successful GET request @@ -301,7 +307,7 @@ def test_consulta_nfse_rps(self, mock_get): result.json(), {"status": "success", "data": {"nfse_info": "…"}} ) # Asserting expected JSON response - @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.request") + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") def test_cancela_documento(self, mock_delete): """Tests document cancellation with mocked DELETE request.""" mock_delete.return_value.status_code = ( @@ -313,7 +319,7 @@ def test_cancela_documento(self, mock_delete): self.assertEqual(result.status_code, 204) # Asserting status code 204 - @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.request") + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") def test_make_focus_nfse_http_request_422(self, mock_request): """Tests handling of HTTP 422 with Focus NFSe error message.""" mock_request.return_value.status_code = 422 @@ -422,7 +428,7 @@ def test_document_export(self): self.assertTrue(record.authorization_event_id) @patch( - "odoo.addons.l10n_br_nfse_focus.models.document.FocusnfeNfse.query_focus_nfse_by_rps" + "odoo.addons.l10n_br_nfse_focus.models.nfse_municipal.FocusnfeNfse.query_focus_nfse_by_rps" ) def test_document_status(self, mock_query): """Tests querying document status.""" @@ -446,7 +452,7 @@ def test_document_status(self, mock_query): ) # Asserting result message @patch( - "odoo.addons.l10n_br_nfse_focus.models.document.FocusnfeNfse._make_focus_nfse_http_request" # noqa: B950 + "odoo.addons.l10n_br_nfse_focus.models.base.FocusnfeNfseBase._make_focus_nfse_http_request" # noqa: B950 ) def test_cancel_document_focus_with_error(self, mock_request): """Tests document cancellation with simulated error.""" @@ -461,12 +467,15 @@ def test_cancel_document_focus_with_error(self, mock_request): document.document_type_id.code = ( MODELO_FISCAL_NFSE # Setting document type to NFSe ) + document.company_id.provedor_nfse = "focusnfe" # Setting provider + document.company_id.focusnfe_nfse_type = "nfse" # Setting to municipal document.document_date = datetime.strptime( "2024-01-01T05:10:12", "%Y-%m-%dT%H:%M:%S" ) # Setting document date document.date_in_out = datetime.strptime( "2024-01-01T05:10:12", "%Y-%m-%dT%H:%M:%S" ) # Setting date in/out + document.cancel_reason = "Teste de cancelamento" with self.assertRaises(UserError) as context: document.cancel_document_focus() # Attempting to cancel document @@ -478,7 +487,7 @@ def test_cancel_document_focus_with_error(self, mock_request): ) @patch( - "odoo.addons.l10n_br_nfse_focus.models.document.FocusnfeNfse.process_focus_nfse_document" # noqa: B950 + "odoo.addons.l10n_br_nfse_focus.models.nfse_municipal.FocusnfeNfse.process_focus_nfse_document" # noqa: B950 ) def test_eletronic_document_send(self, mock_process_focus_nfse_document): """Tests sending of electronic document with simulated responses.""" @@ -597,3 +606,959 @@ def test_exec_before_SITUACAO_EDOC_CANCELADA(self, mock_cancel_document_focus): self.assertEqual( result, mock_cancel_document_focus.return_value ) # Asserting expected result + + def test_filter_focusnfe_nacional(self): + """Tests filtering of NFSe Nacional documents.""" + record = self.nfse_demo + record.company_id.provedor_nfse = "focusnfe" + record.company_id.focusnfe_nfse_type = "nfse_nacional" + + result = filter_focusnfe_nacional(record) + + self.assertEqual(record.company_id.provedor_nfse, "focusnfe") + self.assertEqual(record.company_id.focusnfe_nfse_type, "nfse_nacional") + self.assertEqual(result, True) + + record.company_id.focusnfe_nfse_type = "nfse" + result = filter_focusnfe_nacional(record) + self.assertEqual(result, False) + + def test_filter_focusnfe_municipal(self): + """Tests filtering of NFSe Municipal documents.""" + record = self.nfse_demo + record.company_id.provedor_nfse = "focusnfe" + record.company_id.focusnfe_nfse_type = "nfse" + + result = filter_focusnfe_municipal(record) + + self.assertEqual(record.company_id.provedor_nfse, "focusnfe") + self.assertEqual(record.company_id.focusnfe_nfse_type, "nfse") + self.assertEqual(result, True) + + record.company_id.focusnfe_nfse_type = "nfse_nacional" + result = filter_focusnfe_municipal(record) + self.assertEqual(result, False) + + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") + def test_process_focus_nfse_nacional_document(self, mock_post): + """Tests NFSe Nacional document processing with mocked POST request.""" + mock_post.return_value.status_code = 202 + mock_post.return_value.json.return_value = {"status": "processando_autorizacao"} + + nfse_nacional = self.env["focusnfe.nfse.nacional"] + edoc = { + "rps": PAYLOAD[0]["rps"], + "service": PAYLOAD[1]["service"], + "recipient": PAYLOAD[2]["recipient"], + } + + result = nfse_nacional.process_focus_nfse_nacional_document( + edoc, "12345", self.company, self.tpAmb + ) + + self.assertEqual(result.status_code, 202) + self.assertEqual(result.json(), {"status": "processando_autorizacao"}) + + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") + def test_query_focus_nfse_nacional_by_ref(self, mock_get): + """Tests NFSe Nacional query by reference with mocked GET request.""" + mock_get.return_value.status_code = 200 + mock_get.return_value.json.return_value = { + "status": "autorizado", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + } + + nfse_nacional = self.env["focusnfe.nfse.nacional"] + + result = nfse_nacional.query_focus_nfse_nacional_by_ref( + "12345", self.company, self.tpAmb + ) + + self.assertEqual(result.status_code, 200) + self.assertEqual(result.json()["status"], "autorizado") + + @patch("odoo.addons.l10n_br_nfse_focus.models.base.requests.request") + def test_cancel_focus_nfse_nacional_document(self, mock_delete): + """Tests NFSe Nacional document cancellation with mocked DELETE request.""" + mock_delete.return_value.status_code = 200 + mock_delete.return_value.json.return_value = {"status": "cancelado"} + + nfse_nacional = self.env["focusnfe.nfse.nacional"] + + result = nfse_nacional.cancel_focus_nfse_nacional_document( + "12345", "Teste de cancelamento", self.company, self.tpAmb + ) + + self.assertEqual(result.status_code, 200) + self.assertEqual(result.json()["status"], "cancelado") + + def test_prepare_payload_nacional(self): + """Tests NFSe Nacional payload preparation.""" + nfse_nacional = self.env["focusnfe.nfse.nacional"] + edoc = { + "rps": PAYLOAD[0]["rps"], + "service": PAYLOAD[1]["service"], + "recipient": PAYLOAD[2]["recipient"], + } + + # Set company city for IBGE code + self.company.city_id = self.env.ref("l10n_br_base.city_3550308") + + payload = nfse_nacional._prepare_payload_nacional(edoc, self.company) + + self.assertIn("data_emissao", payload) + self.assertIn("data_competencia", payload) + self.assertIn("cnpj_prestador", payload) + self.assertIn("cnpj_tomador", payload) + self.assertIn("codigo_tributacao_nacional_iss", payload) + self.assertIn("valor_servico", payload) + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.process_focus_nfse_nacional_document" # noqa: B950 + ) + def test_eletronic_document_send_nacional(self, mock_process): + """Tests sending electronic document for NFSe Nacional.""" + mock_response = MagicMock() + mock_response.status_code = 202 + mock_response.json.return_value = {"status": "processando_autorizacao"} + mock_process.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.document_date = datetime.strptime( + "2024-01-01T05:10:12", "%Y-%m-%dT%H:%M:%S" + ) + document.date_in_out = datetime.strptime( + "2024-01-01T05:10:12", "%Y-%m-%dT%H:%M:%S" + ) + + document._eletronic_document_send() + + self.assertEqual(document.state, SITUACAO_EDOC_ENVIADA) + mock_process.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + def test_document_status_nacional(self, mock_query): + """Tests document status check for NFSe Nacional.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + } + mock_query.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.state = "enviada" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.requests.get" + ) as mock_get: + mock_get.return_value.content.decode.return_value = "test" + result = document._document_status() + + self.assertIn("autorizado", result) + mock_query.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.cancel_focus_nfse_nacional_document" # noqa: B950 + ) + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + def test_cancel_document_focus_nacional(self, mock_query, mock_cancel): + """Tests document cancellation for NFSe Nacional.""" + # Mock query response - not cancelled yet + mock_query_response = MagicMock() + mock_query_response.status_code = 200 + mock_query_response.json.return_value = {"status": "autorizado"} + mock_query.return_value = mock_query_response + + # Mock cancel response + mock_cancel_response = MagicMock() + mock_cancel_response.status_code = 200 + mock_cancel_response.json.return_value = {"status": "cancelado"} + mock_cancel.return_value = mock_cancel_response + + # Mock query response after cancellation + mock_query_response_after = MagicMock() + mock_query_response_after.status_code = 200 + mock_query_response_after.json.return_value = { + "status": "cancelado", + "url_danfse": "https://example.com/danfse.pdf", + } + mock_query.side_effect = [ + mock_query_response, + mock_query_response_after, + ] + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.cancel_reason = "Teste de cancelamento" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.requests.get" + ) as mock_get: + mock_get.return_value.content = b"%PDF-test%%EOF" + result = document.cancel_document_focus() + + self.assertEqual(result.status_code, 200) + mock_cancel.assert_called_once() + + def test_parse_authorization_datetime(self): + """Tests parsing authorization datetime from JSON data.""" + document = self.nfse_demo + json_data = {"data_emissao": "2024-01-01T05:10:12-03:00"} + + result = document._parse_authorization_datetime(json_data) + + self.assertIsInstance(result, datetime) + self.assertEqual(result.year, 2024) + self.assertEqual(result.month, 1) + self.assertEqual(result.day, 1) + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_xml_from_path_success(self, mock_get): + """Tests successful XML fetch from path.""" + document = self.nfse_demo + document.nfse_environment = "1" + xml_path = "/v2/nfse/12345.xml" + mock_get.return_value.content.decode.return_value = "test" + + result = document._fetch_xml_from_path(document, xml_path) + + self.assertEqual(result, "test") + mock_get.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_xml_from_path_error(self, mock_get): + """Tests XML fetch from path with error.""" + document = self.nfse_demo + document.nfse_environment = "1" + xml_path = "/v2/nfse/12345.xml" + mock_get.side_effect = Exception("Connection error") + + result = document._fetch_xml_from_path(document, xml_path) + + self.assertEqual(result, "") + mock_get.assert_called_once() + + def test_fetch_xml_from_path_empty(self): + """Tests XML fetch with empty path.""" + document = self.nfse_demo + result = document._fetch_xml_from_path(document, "") + self.assertEqual(result, "") + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_pdf_from_urls_url_first(self, mock_get): + """Tests PDF fetch using url first.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.company_id.nfse_ssl_verify = True + json_data = {"url": "https://example.com/pdf.pdf"} + mock_get.return_value.content = b"%PDF-test%%EOF" + + result = document._fetch_pdf_from_urls(document, json_data, use_url_first=True) + + self.assertEqual(result, b"%PDF-test%%EOF") + mock_get.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_pdf_from_urls_url_danfse(self, mock_get): + """Tests PDF fetch using url_danfse.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.company_id.nfse_ssl_verify = True + json_data = {"url_danfse": "https://example.com/danfse.pdf"} + mock_get.return_value.content = b"%PDF-test%%EOF" + + result = document._fetch_pdf_from_urls(document, json_data, use_url_first=False) + + self.assertEqual(result, b"%PDF-test%%EOF") + mock_get.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_pdf_from_urls_invalid_pdf(self, mock_get): + """Tests PDF fetch with invalid PDF content.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.company_id.nfse_ssl_verify = True + json_data = {"url": "https://example.com/invalid.pdf"} + mock_get.return_value.content = b"invalid content" + + result = document._fetch_pdf_from_urls(document, json_data, use_url_first=True) + + self.assertIsNone(result) + + def test_fetch_pdf_from_urls_force_odoo_danfse(self): + """Tests PDF fetch when force_odoo_danfse is True.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = True + json_data = {"url": "https://example.com/pdf.pdf"} + + result = document._fetch_pdf_from_urls(document, json_data, use_url_first=True) + + self.assertIsNone(result) + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_pdf_from_urls_error(self, mock_get): + """Tests PDF fetch with error.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.company_id.nfse_ssl_verify = True + json_data = {"url": "https://example.com/pdf.pdf"} + mock_get.side_effect = Exception("Connection error") + + result = document._fetch_pdf_from_urls(document, json_data, use_url_first=True) + + self.assertIsNone(result) + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_process_authorized_status_nacional(self, mock_get, mock_query): + """Tests processing authorized status for NFSe Nacional.""" + mock_query_response = MagicMock() + mock_query_response.status_code = 200 + mock_query_response.json.return_value = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + } + mock_query.return_value = mock_query_response + mock_get.return_value.content.decode.return_value = "test" + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.state = "enviada" + document.nfse_environment = "1" + json_data = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + "url_danfse": "https://example.com/danfse.pdf", + "caminho_xml_nota_fiscal": "/v2/nfsen/12345.xml", + } + + # Create authorization event to trigger the state change + # Create a real event in the database + auth_event = self.env["l10n_br_fiscal.event"].create( + { + "type": "0", + "company_id": document.company_id.id, + "document_type_id": document.document_type_id.id, + "document_serie_id": document.document_serie_id.id, + "document_number": document.document_number or "TEST001", + } + ) + document.authorization_event_id = auth_event + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ) as mock_change_state: + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.make_focus_nfse_pdf" + ) as mock_make_pdf: + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_event.Event.set_done" + ) as mock_set_done: + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._fetch_xml_from_path" + ) as mock_fetch_xml: + mock_fetch_xml.return_value = b"test" + mock_get.return_value.content = b"%PDF-test%%EOF" + document._process_authorized_status_nacional( + document, + json_data, + ) + + mock_change_state.assert_called_once() + mock_make_pdf.assert_called_once() + mock_set_done.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_municipal.FocusnfeNfse.query_focus_nfse_by_rps" # noqa: B950 + ) + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_process_authorized_status_municipal(self, mock_get, mock_query): + """Tests processing authorized status for NFSe Municipal.""" + mock_query_response = MagicMock() + mock_query_response.status_code = 200 + mock_query_response.json.return_value = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + } + mock_query.return_value = mock_query_response + mock_get.return_value.content.decode.return_value = "test" + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse" + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.state = "enviada" + document.nfse_environment = "1" + json_data = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + "caminho_xml_nota_fiscal": "/v2/nfse/12345.xml", + "url": "https://example.com/pdf.pdf", + } + + # Create authorization event to trigger the state change + # Create a real event in the database + auth_event = self.env["l10n_br_fiscal.event"].create( + { + "type": "0", + "company_id": document.company_id.id, + "document_type_id": document.document_type_id.id, + "document_serie_id": document.document_serie_id.id, + "document_number": document.document_number or "TEST001", + } + ) + document.authorization_event_id = auth_event + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ) as mock_change_state: + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.make_focus_nfse_pdf" + ) as mock_make_pdf: + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_event.Event.set_done" + ) as mock_set_done: + mock_get.return_value.content = b"%PDF-test%%EOF" + document._process_authorized_status_municipal(document, json_data) + + mock_change_state.assert_called_once() + mock_make_pdf.assert_called_once() + mock_set_done.assert_called_once() + + def test_process_error_status(self): + """Tests processing error status.""" + document = self.nfse_demo + json_data = {"erros": [{"mensagem": "Error message"}]} + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ) as mock_change_state: + document._process_error_status(document, json_data) + + self.assertEqual(document.edoc_error_message, "Error message") + mock_change_state.assert_called_once() + + def test_process_error_status_no_errors(self): + """Tests processing error status without errors list.""" + document = self.nfse_demo + json_data = {} + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ) as mock_change_state: + document._process_error_status(document, json_data) + + mock_change_state.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_create_cancel_event(self, mock_get): + """Tests creating cancel event.""" + document = self.nfse_demo + document.nfse_environment = "1" + status_json = { + "caminho_xml_cancelamento": "/v2/nfse/12345_cancel.xml", + } + mock_get.return_value.content.decode.return_value = "cancel" + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_event.Event.create_event_save_xml" + ) as mock_create_event: + mock_event = MagicMock() + mock_create_event.return_value = mock_event + mock_event.set_done = MagicMock() + + result = document.create_cancel_event(status_json, document) + + self.assertEqual(result, mock_event) + mock_event.set_done.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_and_verify_pdf_content(self, mock_get): + """Tests fetching and verifying PDF content.""" + document = self.nfse_demo + document.company_id.nfse_ssl_verify = True + status_json = { + "url": "https://example.com/pdf.pdf", + "url_danfse": "https://example.com/danfse.pdf", + } + mock_get.return_value.content = b"%PDF-test%%EOF" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.make_focus_nfse_pdf" + ) as mock_make_pdf: + document.fetch_and_verify_pdf_content(status_json, document) + + mock_make_pdf.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_fetch_and_verify_pdf_content_fallback(self, mock_get): + """Tests fetching PDF with fallback to url_danfse.""" + document = self.nfse_demo + document.company_id.nfse_ssl_verify = True + status_json = { + "url": "https://example.com/invalid.pdf", + "url_danfse": "https://example.com/danfse.pdf", + } + # First call returns invalid PDF, second returns valid + mock_get.side_effect = [ + MagicMock(content=b"invalid"), + MagicMock(content=b"%PDF-test%%EOF"), + ] + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.make_focus_nfse_pdf" + ) as mock_make_pdf: + document.fetch_and_verify_pdf_content(status_json, document) + + mock_make_pdf.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_handle_cancelled_status_force_odoo(self, mock_get): + """Tests handling cancelled status with force_odoo_danfse.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = True + status_json = {"url": "https://example.com/pdf.pdf"} + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.create_cancel_event" + ) as mock_create_event: + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document.Document.make_pdf" + ) as mock_make_pdf: + # Create a real event in the database for the mock to return + mock_event = self.env["l10n_br_fiscal.event"].create( + { + "type": "2", + "company_id": document.company_id.id, + "document_type_id": document.document_type_id.id, + "document_serie_id": document.document_serie_id.id, + "document_number": document.document_number or "TEST001", + } + ) + mock_create_event.return_value = mock_event + + document._handle_cancelled_status( + document, status_json, use_url_first=False + ) + + mock_make_pdf.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_handle_cancelled_status_use_url_first(self, mock_get): + """Tests handling cancelled status with use_url_first.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.company_id.nfse_ssl_verify = True + status_json = { + "url": "https://example.com/pdf.pdf", + "url_danfse": "https://example.com/danfse.pdf", + } + mock_get.return_value.content = b"%PDF-test%%EOF" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.create_cancel_event" + ) as mock_create_event: + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.fetch_and_verify_pdf_content" # noqa: B950 + ) as mock_fetch_pdf: + # Create a real event in the database for the mock to return + mock_event = self.env["l10n_br_fiscal.event"].create( + { + "type": "2", + "company_id": document.company_id.id, + "document_type_id": document.document_type_id.id, + "document_serie_id": document.document_serie_id.id, + "document_number": document.document_number or "TEST001", + } + ) + mock_create_event.return_value = mock_event + + document._handle_cancelled_status( + document, status_json, use_url_first=True + ) + + mock_fetch_pdf.assert_called_once() + + @patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get") + def test_handle_cancelled_status_url_danfse(self, mock_get): + """Tests handling cancelled status with url_danfse.""" + document = self.nfse_demo + document.company_id.focusnfe_nfse_force_odoo_danfse = False + document.company_id.nfse_ssl_verify = True + status_json = {"url_danfse": "https://example.com/danfse.pdf"} + mock_get.return_value.content = b"%PDF-test%%EOF" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.create_cancel_event" + ) as mock_create_event: + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document.make_focus_nfse_pdf" + ) as mock_make_pdf: + # Create a real event in the database for the mock to return + mock_event = self.env["l10n_br_fiscal.event"].create( + { + "type": "2", + "company_id": document.company_id.id, + "document_type_id": document.document_type_id.id, + "document_serie_id": document.document_serie_id.id, + "document_number": document.document_number or "TEST001", + } + ) + mock_create_event.return_value = mock_event + + document._handle_cancelled_status( + document, status_json, use_url_first=False + ) + + mock_make_pdf.assert_called_once() + + def test_serialize_nacional(self): + """Tests serialization for NFSe Nacional.""" + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + + with patch( + "odoo.addons.l10n_br_nfse.models.document.Document._prepare_lote_rps", + return_value={"rps": "data"}, + ) as mock_rps: + with patch( + "odoo.addons.l10n_br_nfse.models.document.Document._prepare_dados_servico", + return_value={"service": "data"}, + ) as mock_service: + with patch( + "odoo.addons.l10n_br_nfse.models.document.Document._prepare_dados_tomador", + return_value={"recipient": "data"}, + ) as mock_recipient: + edocs = [] + result = document._serialize(edocs) + + self.assertIsInstance(result, list) + mock_rps.assert_called_once() + mock_service.assert_called_once() + mock_recipient.assert_called_once() + + def test_serialize_municipal(self): + """Tests serialization for NFSe Municipal.""" + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse" + + with patch( + "odoo.addons.l10n_br_nfse.models.document.Document._prepare_lote_rps", + return_value={"rps": "data"}, + ) as mock_rps: + with patch( + "odoo.addons.l10n_br_nfse.models.document.Document._prepare_dados_servico", + return_value={"service": "data"}, + ) as mock_service: + with patch( + "odoo.addons.l10n_br_nfse.models.document.Document._prepare_dados_tomador", + return_value={"recipient": "data"}, + ) as mock_recipient: + edocs = [] + result = document._serialize(edocs) + + self.assertIsInstance(result, list) + mock_rps.assert_called_once() + mock_service.assert_called_once() + mock_recipient.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + def test_process_status_nacional_autorizado(self, mock_query): + """Tests process status nacional with autorizado status.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + } + mock_query.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.state = "enviada" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._process_authorized_status_nacional" # noqa: B950 + ) as mock_process: + with patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get"): + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ): + result = document._process_status_nacional(document) + + self.assertIn("autorizado", result) + mock_process.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + def test_process_status_nacional_erro(self, mock_query): + """Tests process status nacional with erro status.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "status": "erro_autorizacao", + "erros": [{"mensagem": "Error message"}], + } + mock_query.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.state = "enviada" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._process_error_status" + ) as mock_process: + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ): + result = document._process_status_nacional(document) + + self.assertIn("erro_autorizacao", result) + mock_process.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + def test_process_status_nacional_cancelado(self, mock_query): + """Tests process status nacional with cancelado status.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"status": "cancelado"} + mock_query.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.state = "enviada" + document.cancel_reason = "Teste" + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._document_cancel" + ) as mock_cancel: + result = document._process_status_nacional(document) + + self.assertIn("cancelado", result) + mock_cancel.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_municipal.FocusnfeNfse.query_focus_nfse_by_rps" # noqa: B950 + ) + def test_process_status_municipal_autorizado(self, mock_query): + """Tests process status municipal with autorizado status.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "status": "autorizado", + "data_emissao": "2024-01-01T05:10:12-03:00", + "numero": "12345", + "codigo_verificacao": "12345678901234567890", + } + mock_query.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse" + document.state = "enviada" + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._process_authorized_status_municipal" # noqa: B950 + ) as mock_process: + with patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get"): + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ): + result = document._process_status_municipal(document) + + self.assertIn("autorizado", result) + mock_process.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_municipal.FocusnfeNfse.query_focus_nfse_by_rps" # noqa: B950 + ) + def test_process_status_municipal_erro(self, mock_query): + """Tests process status municipal with erro status.""" + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = { + "status": "erro_autorizacao", + "erros": [{"mensagem": "Error message"}], + } + mock_query.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse" + document.state = "enviada" + + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ) as mock_change_state: + result = document._process_status_municipal(document) + + self.assertIn("erro_autorizacao", result) + mock_change_state.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.process_focus_nfse_nacional_document" # noqa: B950 + ) + def test_process_send_nacional_422_autorizada(self, mock_process): + """Tests process send nacional with 422 and autorizada code.""" + mock_response = MagicMock() + mock_response.status_code = 422 + mock_response.json.return_value = {"codigo": "nfe_autorizada"} + mock_process.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.state = "enviada" + document.document_date = datetime.now() + document.date_in_out = datetime.now() + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._document_status" + ) as mock_status: + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ): + document._process_send_nacional(document) + + mock_status.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_municipal.FocusnfeNfse.process_focus_nfse_document" # noqa: B950 + ) + def test_process_send_municipal_422_autorizada(self, mock_process): + """Tests process send municipal with 422 and autorizada code.""" + mock_response = MagicMock() + mock_response.status_code = 422 + mock_response.json.return_value = {"codigo": "nfe_autorizada"} + mock_process.return_value = mock_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse" + document.state = "enviada" + document.document_date = datetime.now() + document.date_in_out = datetime.now() + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._document_status" + ) as mock_status: + with patch( + "odoo.addons.l10n_br_fiscal_edi.models.document_workflow.DocumentWorkflow._change_state" + ): + document._process_send_municipal(document) + + mock_status.assert_called_once() + + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.query_focus_nfse_nacional_by_ref" # noqa: B950 + ) + @patch( + "odoo.addons.l10n_br_nfse_focus.models.nfse_nacional.FocusnfeNfseNacional.cancel_focus_nfse_nacional_document" # noqa: B950 + ) + def test_process_cancel_base_already_cancelled(self, mock_cancel, mock_query): + """Tests process cancel base when already cancelled.""" + # First query returns cancelled + mock_query_response = MagicMock() + mock_query_response.status_code = 200 + mock_query_response.json.return_value = { + "status": "cancelado", + "url_danfse": "https://example.com/danfse.pdf", + } + mock_query.return_value = mock_query_response + + document = self.nfse_demo + document.processador_edoc = PROCESSADOR_OCA + document.document_type_id.code = MODELO_FISCAL_NFSE + document.company_id.provedor_nfse = "focusnfe" + document.company_id.focusnfe_nfse_type = "nfse_nacional" + document.cancel_reason = "Teste" + + def query_method(ref, company, environment): + return mock_query_response + + def cancel_method(ref, cancel_reason, company, environment): + mock_cancel_response = MagicMock() + mock_cancel_response.status_code = 200 + mock_cancel_response.json.return_value = {"status": "cancelado"} + return mock_cancel_response + + with patch( + "odoo.addons.l10n_br_nfse_focus.models.document.Document._handle_cancelled_status" + ) as mock_handle: + with patch("odoo.addons.l10n_br_nfse_focus.models.document.requests.get"): + result = document._process_cancel_base( + document, + "12345", + query_method, + cancel_method, + use_url_first=False, + ) + + mock_handle.assert_called_once() + self.assertEqual(result.status_code, 200) diff --git a/l10n_br_nfse_focus/views/res_company.xml b/l10n_br_nfse_focus/views/res_company.xml index 9ea956c7103c..e240213a6640 100644 --- a/l10n_br_nfse_focus/views/res_company.xml +++ b/l10n_br_nfse_focus/views/res_company.xml @@ -39,6 +39,10 @@ name="focusnfe_tax_rate_format" attrs="{'invisible': [('provedor_nfse','!=', 'focusnfe')]}" /> + 0