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 1cd4dd77d808..0e0c788e898c 100644 --- a/l10n_br_nfse/models/document.py +++ b/l10n_br_nfse/models/document.py @@ -21,6 +21,7 @@ ) from ..constants.nfse import ( + ISSQN_TO_TRIBUTACAO_ISS, NFSE_ENVIRONMENTS, OPERATION_NATURE, RPS_TYPE, @@ -171,6 +172,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_servico()) @@ -199,6 +202,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, @@ -223,10 +235,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 @@ -251,6 +267,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_servico()) diff --git a/l10n_br_nfse/models/document_line.py b/l10n_br_nfse/models/document_line.py index eed12dc1c629..744e060b620d 100644 --- a/l10n_br_nfse/models/document_line.py +++ b/l10n_br_nfse/models/document_line.py @@ -90,7 +90,8 @@ def prepare_line_servico(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, @@ -108,4 +109,15 @@ def prepare_line_servico(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/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..6d43be44fa09 100644 --- a/l10n_br_nfse_focus/models/document.py +++ b/l10n_br_nfse_focus/models/document.py @@ -2,10 +2,11 @@ # 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 +from datetime import date, datetime import pytz import requests @@ -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()}) @@ -382,6 +129,15 @@ def _document_export(self, pretty_print=True): for record in self.filtered(filter_processador_edoc_nfse).filtered( filter_focusnfe ): + # Ensure document_date is set before creating event + if not record.document_date or not isinstance( + record.document_date, (date, datetime) + ): + # Use authorization_date if available, otherwise use current date + if record.authorization_date: + record.document_date = record.authorization_date.date() + else: + record.document_date = fields.Date.today() event_id = record.event_ids.create_event_save_xml( company_id=record.company_id, environment=( @@ -394,6 +150,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 +405,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 +427,24 @@ def create_cancel_event(self, status_json, record): Returns: The created event. """ + 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") - xml = requests.get( - NFSE_URL[record.nfse_environment] + status_json["caminho_xml_cancelamento"], - timeout=TIMEOUT, - ).content.decode("utf-8") + # Ensure document_date is set before creating event + if not record.document_date or not isinstance( + record.document_date, (date, datetime) + ): + # Use authorization_date if available, otherwise use current date + if record.authorization_date: + record.document_date = record.authorization_date.date() + else: + record.document_date = fields.Date.today() event = record.event_ids.create_event_save_xml( company_id=record.company_id, @@ -541,17 +479,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 +642,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 +721,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 +762,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/tests/test_l10n_br_nfse_focus.py b/l10n_br_nfse_focus/tests/test_l10n_br_nfse_focus.py index 6bb157ef3b5d..eec7cbf56e59 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" @@ -197,7 +203,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 @@ -214,7 +220,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. @@ -281,7 +287,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 @@ -299,7 +305,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 = ( @@ -311,7 +317,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 @@ -420,7 +426,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.""" @@ -444,7 +450,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: E501 + "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.""" @@ -459,12 +465,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 @@ -476,7 +485,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: E501 + "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.""" @@ -595,3 +604,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