From dfa8d37961d5a983b1f6ef0081cd230a19f1fb19 Mon Sep 17 00:00:00 2001 From: Marcel Savegnago Date: Fri, 7 Nov 2025 18:08:41 -0300 Subject: [PATCH 1/2] [IMP] l10n_br_nfse: add values on _prepare_line_service --- l10n_br_nfse/models/document.py | 18 ++++++++++++++++++ l10n_br_nfse/models/document_line.py | 11 +++++++++++ 2 files changed, 29 insertions(+) diff --git a/l10n_br_nfse/models/document.py b/l10n_br_nfse/models/document.py index d9521de78c51..766adcd979a8 100644 --- a/l10n_br_nfse/models/document.py +++ b/l10n_br_nfse/models/document.py @@ -167,6 +167,8 @@ def _prepare_dados_servico(self): cbs_aliquota = 0 ibs_uf_valor = 0 cbs_valor = 0 + base_calculo_pis = 0 + base_calculo_cofins = 0 for line in lines: result_line.update(line._prepare_line_service()) @@ -195,6 +197,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, @@ -246,6 +257,13 @@ 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, } result.update(self.company_id._prepare_company_service()) diff --git a/l10n_br_nfse/models/document_line.py b/l10n_br_nfse/models/document_line.py index 46b3970d132e..59082fed2886 100644 --- a/l10n_br_nfse/models/document_line.py +++ b/l10n_br_nfse/models/document_line.py @@ -109,4 +109,15 @@ def _prepare_line_service(self): "ibs_uf_valor": round(self.ibs_value, 2) if self.ibs_value else None, "ibs_mun_valor": 0.0, "cbs_valor": round(self.cbs_value, 2) if self.cbs_value else None, + "situacao_tributaria_pis": self.pis_cst_code or "", + "situacao_tributaria_cofins": self.cofins_cst_code or "", + "base_calculo_pis": round(self.pis_base, 2), + "base_calculo_cofins": round(self.cofins_base, 2), + "aliquota_pis": round(self.pis_percent, 2) if self.pis_percent else 0.0, + "aliquota_cofins": ( + round(self.cofins_percent, 2) if self.cofins_percent else 0.0 + ), + "tipo_retencao_pis_cofins": ( + "1" if (self.pis_wh_value or self.cofins_wh_value) else "2" + ), } From 03d4f1175b54874ab8d0d7c8d8344c70265069e2 Mon Sep 17 00:00:00 2001 From: Marcel Savegnago Date: Fri, 7 Nov 2025 18:09:21 -0300 Subject: [PATCH 2/2] [IMP] l10n_br_nfse_focus: add nfse nacional --- l10n_br_nfse_focus/models/document.py | 1073 ++++++++++++++--- l10n_br_nfse_focus/models/res_company.py | 10 + .../tests/test_l10n_br_nfse_focus.py | 222 +++- l10n_br_nfse_focus/views/res_company.xml | 4 + 4 files changed, 1107 insertions(+), 202 deletions(-) diff --git a/l10n_br_nfse_focus/models/document.py b/l10n_br_nfse_focus/models/document.py index c5009fd3a4e9..73092240043a 100644 --- a/l10n_br_nfse_focus/models/document.py +++ b/l10n_br_nfse_focus/models/document.py @@ -45,6 +45,20 @@ def filter_focusnfe(record): return record.company_id.provedor_nfse == "focusnfe" +def filter_focusnfe_nacional(record): + return ( + record.company_id.provedor_nfse == "focusnfe" + and record.company_id.focusnfe_nfse_type == "nfse_nacional" + ) + + +def filter_focusnfe_municipal(record): + return ( + record.company_id.provedor_nfse == "focusnfe" + and record.company_id.focusnfe_nfse_type == "nfse" + ) + + class FocusnfeNfse(models.AbstractModel): _name = "focusnfe.nfse" _description = "FocusNFE NFSE" @@ -311,6 +325,435 @@ def cancel_focus_nfse_document(self, ref, cancel_reason, company, environment): ) +API_ENDPOINT_NACIONAL = { + "envio": "/v2/nfsen", + "status": "/v2/nfsen/", + "resposta": "/v2/nfsen/", + "cancelamento": "/v2/nfsen/", +} + + +class FocusnfeNfseNacional(models.AbstractModel): + _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. + """ + 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 Nacional 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 Nacional service: %s") % e + ) from e + + @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_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 emission date with timezone + 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" + + # Prepare competence date (YYYY-MM-DD) + competence_date = ( + rps_info.get("data_emissao", "")[:10] + if rps_info.get("data_emissao") + else "" + ) + + # Get municipality code + codigo_municipio_emissora = company.city_id.ibge_code or "" + + # Prepare provider data + cnpj_prestador = rps_info.get("cnpj", "") + cpf_prestador = rps_info.get("cpf", "") + # Determinar se é CPF ou CNPJ baseado no tamanho + # CPF tem 11 dígitos, CNPJ tem 14 dígitos + # Limpar formatação para verificar o tamanho + cpf_prestador_limpo = ( + cpf_prestador.replace(".", "").replace("-", "") if cpf_prestador else "" + ) + cnpj_prestador_limpo = ( + cnpj_prestador.replace(".", "").replace("/", "").replace("-", "") + if cnpj_prestador + else "" + ) + is_cpf_prestador = bool(cpf_prestador_limpo and len(cpf_prestador_limpo) == 11) + is_cnpj_prestador = bool( + cnpj_prestador_limpo and len(cnpj_prestador_limpo) == 14 + ) + + # TODO: aparentemente nao é enviado + # inscricao_municipal_prestador = rps_info.get("inscricao_municipal", "") + + # Get simple national option code + # TODO: melhorar a lógica para obter o código da opção simples nacional + # codigo_opcao_simples_nacional + # Tag XML opSimpNac + # obrigatório + # Situação perante Simples Nacional: + # 1: Não Optante; + # 2: Optante - Microempreendedor Individual (MEI); + # 3: Optante - Microempresa ou Empresa de Pequeno Porte (ME/EPP). + optante_simples = rps_info.get("optante_simples_nacional", "1") + codigo_opcao_simples_nacional = "2" if optante_simples == "1" else "1" + + # Get special taxation regime + # TODO: melhorar a lógica para obter o código do regime especial de tributação + # regime_especial_tributacao + # Tag XML regEspTrib + # obrigatório + # Tipos de Regimes Especiais de Tributação Municipal: + # 0: Nenhum; + # 1: Ato Cooperado (Cooperativa); + # 2: Estimativa; + # 3: Microempresa Municipal; + # 4: Notário ou Registrador; + # 5: Profissional Autônomo; + # 6: Sociedade de Profissionais. + regime_especial_tributacao = ( + rps_info.get("regime_especial_tributacao", "0") or "0" + ) + + # Prepare recipient data + cnpj_tomador = recipient_info.get("cnpj", "") + cpf_tomador = recipient_info.get("cpf", "") + # Determinar se é CPF ou CNPJ baseado no tamanho + # CPF tem 11 dígitos, CNPJ tem 14 dígitos + # Limpar formatação para verificar o tamanho + cpf_limpo = cpf_tomador.replace(".", "").replace("-", "") if cpf_tomador else "" + cnpj_limpo = ( + cnpj_tomador.replace(".", "").replace("/", "").replace("-", "") + if cnpj_tomador + else "" + ) + is_cpf = bool(cpf_limpo and len(cpf_limpo) == 11) + is_cnpj = bool(cnpj_limpo and len(cnpj_limpo) == 14) + razao_social_tomador = recipient_info.get("razao_social", "") + codigo_municipio_tomador = recipient_info.get("codigo_municipio", "") + # CEP can be int or string, convert to string + cep_tomador = recipient_info.get("cep", "") + if isinstance(cep_tomador, int): + cep_tomador = str(cep_tomador) + logradouro_tomador = recipient_info.get("endereco", "") + numero_tomador = recipient_info.get("numero", "") + complemento_tomador = recipient_info.get("complemento", "") + bairro_tomador = recipient_info.get("bairro", "") + # Telefone is not in recipient_info, leave empty + telefone_tomador = recipient_info.get("telefone", "") + email_tomador = recipient_info.get("email", "") + + # Prepare service data + codigo_municipio_prestacao = service_info.get("municipio_prestacao_servico", "") + # For NFSe Nacional, we need codigo_tributacao_nacional_iss + # This should come from the service type or city taxation code + codigo_tributacao_nacional_iss = service_info.get( + "codigo_tributacao_nacional_iss", "" + ) + if not codigo_tributacao_nacional_iss: + # Try to get from city taxation code + codigo_tributacao_nacional_iss = service_info.get( + "codigo_tributacao_municipio", "" + ) + if not codigo_tributacao_nacional_iss: + # Fallback to item_lista_servico (service type code) + codigo_tributacao_nacional_iss = service_info.get("item_lista_servico", "") + + descricao_servico = service_info.get("discriminacao", "") + valor_servico = round(service_info.get("valor_servicos", 0), 2) + + # TODO: melhorar a lógica para obter o código da tributação ISS + # tributacao_iss + # Tag XML tribISSQN + # obrigatório + # Tributação do ISSQN sobre o serviço prestado: + # 1: Operação tributável; + # 2: Imunidade; + # 3: Exportação de serviço; + # 4: Não Incidência. + tributacao_iss = 1 + + # TODO: melhorar a lógica para obter o código da retencao ISS + # tipo_retencao_iss + # Tag XML tpRetISSQN + # Tipo de retencao do ISSQN: + # 1: Não Retido; + # 2: Retido pelo Tomador; + # 3: Retido pelo Intermediario. + tipo_retencao_iss = "2" if service_info.get("iss_retido") == "1" else "1" + + # TODO: tratar percentual_aliquota_relativa_municipio - aparentemente + # nao é enviado + # # Campos adicionais para NFSe Nacional + # # percentual_aliquota_relativa_municipio - usar a alíquota do serviço + # aliquota_servico = service_info.get("aliquota", 0) + # # A alíquota vem em decimal (0-1), converter para percentual (0-100) + # percentual_aliquota_relativa_municipio = ( + # round(aliquota_servico * 100, 2) if aliquota_servico else 0.0 + # ) + + # Situação tributária PIS/COFINS - usar PIS se disponível, senão COFINS + situacao_tributaria_pis_cofins = ( + service_info.get("situacao_tributaria_pis", "") + or service_info.get("situacao_tributaria_cofins", "") + or "" + ) + # Ajuste: Se a situação tributária for 99, alterar para 00 + if situacao_tributaria_pis_cofins == "99": + situacao_tributaria_pis_cofins = "00" + + # Base de cálculo PIS/COFINS - usar de qualquer um que tenha valor + 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 + ) + # Validação: Se o CST for 0, 8 ou 9, a base de cálculo deve ser 0 + # Se o CST for diferente de 0, 8 ou 9, a base de cálculo deve ser informada + if situacao_tributaria_pis_cofins: + if situacao_tributaria_pis_cofins in ["00", "08", "09"]: + # CST 0, 8 ou 9: base de cálculo deve ser 0 + base_calculo_pis_cofins = 0.0 + else: + # CST diferente de 0, 8 ou 9: base de cálculo deve ser informada + # Se não houver base de cálculo, usar o valor do serviço + if not base_calculo_pis_cofins or base_calculo_pis_cofins == 0: + base_calculo_pis_cofins = round(valor_servico, 2) + # Alíquotas PIS e COFINS devem ter sempre 2 casas decimais + # (padrão da API: 0|0\.[0-9]{2}|[1-9]{1}[0-9]{0,1}(\.[0-9]{2})?) + # Formatamos como string para garantir 2 casas decimais na serialização JSON + 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}" + 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 (Contribuição Previdenciária) - geralmente é o INSS + valor_cp = round(service_info.get("valor_inss_retido", 0), 2) + # Valor IRRF + valor_irrf = round(service_info.get("valor_ir_retido", 0), 2) + # Valor CSLL + valor_csll = round(service_info.get("valor_csll_retido", 0), 2) + + # TODO: Aparentemente nao sao enviados + # # Campos de total de tributos (opcionais - podem vir do IBPT + # # ou ser calculados) + # # Por enquanto, deixamos vazios ou calculamos se disponível + # valor_total_tributos_federais = round( + # valor_pis + valor_cofins + valor_irrf + valor_csll + valor_cp, 2 + # ) + # valor_total_tributos_estaduais = ( + # 0.0 + # ) # Para NFSe geralmente não há tributos estaduais + # valor_total_tributos_municipais = + # round(service_info.get("valor_iss", 0), 2) + # # Percentuais (calculados se base disponível) + # percentual_total_tributos_federais = ( + # round((valor_total_tributos_federais / valor_servico * 100), 2) + # if valor_servico > 0 + # else 0.0 + # ) + # percentual_total_tributos_estaduais = 0.0 + # percentual_total_tributos_municipais = ( + # round((valor_total_tributos_municipais / valor_servico * 100), 2) + # if valor_servico > 0 + # else 0.0 + # ) + # # Indicador de total de tributação (0 = Não informar valores estimados) + # indicador_total_tributacao = "0" + # # Percentual total tributos Simples Nacional (opcional) + # percentual_total_tributos_simples_nacional = 0.0 + + payload = { + "data_emissao": emission_date, + "data_competencia": competence_date, + "codigo_municipio_emissora": str(codigo_municipio_emissora) + if codigo_municipio_emissora + else "", + # Enviar CNPJ ou CPF conforme o tipo do prestador + # Enviar sem formatação (apenas números) + # Só incluir o campo no payload se for do tipo correto + **({"cnpj_prestador": cnpj_prestador_limpo} if is_cnpj_prestador else {}), + **({"cpf_prestador": cpf_prestador_limpo} if is_cpf_prestador else {}), + # TODO: aparentemente nao é enviado + # "inscricao_municipal_prestador": inscricao_municipal_prestador or "", + "codigo_opcao_simples_nacional": codigo_opcao_simples_nacional, + "regime_especial_tributacao": regime_especial_tributacao, + # Enviar CNPJ ou CPF conforme o tipo do tomador + # Enviar sem formatação (apenas números) + # Só incluir o campo no payload se for do tipo correto + **({"cnpj_tomador": cnpj_limpo} if is_cnpj else {}), + **({"cpf_tomador": cpf_limpo} if is_cpf else {}), + "razao_social_tomador": razao_social_tomador, + "codigo_municipio_tomador": str(codigo_municipio_tomador) + if codigo_municipio_tomador + else "", + "cep_tomador": cep_tomador or "", + "logradouro_tomador": logradouro_tomador, + "numero_tomador": numero_tomador or "", + "complemento_tomador": complemento_tomador or "", + "bairro_tomador": bairro_tomador, + "telefone_tomador": telefone_tomador or "", + "email_tomador": email_tomador or "", + "codigo_municipio_prestacao": str(codigo_municipio_prestacao) + if codigo_municipio_prestacao + else "", + "codigo_tributacao_nacional_iss": codigo_tributacao_nacional_iss, + "descricao_servico": descricao_servico, + "valor_servico": valor_servico, + "tributacao_iss": str(tributacao_iss), + "tipo_retencao_iss": str(tipo_retencao_iss), + # TODO: tratar percentual_aliquota_relativa_municipio + # percentual_aliquota_relativa_municipio deve ser em percentual (0-100) + # percentual_aliquota_relativa_municipio + # Decimal[1.2]Tag XML pAliq + # Valor da alíquota (%) do serviço + # prestado relativo ao município sujeito + # ativo (município de incidência) do ISSQN. Se + # o município de incidência pertence ao Sistema + # Nacional NFS-e a alíquota estará parametrizada e, portanto, + # será fornecida pelo sistema. Se o município de incidência + # não pertence ao Sistema Nacional NFS-e a alíquota não estará + # parametrizada e, por isso, deverá ser fornecida pelo emitente. + # "percentual_aliquota_relativa_municipio": ( + # percentual_aliquota_relativa_municipio + # ), + "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": valor_pis, + "valor_cofins": valor_cofins, + "tipo_retencao_pis_cofins": tipo_retencao_pis_cofins, + "valor_cp": valor_cp, + "valor_irrf": valor_irrf, + "valor_csll": valor_csll, + # TODO: aparentemente nao é enviado + # "valor_total_tributos_federais": valor_total_tributos_federais, + # "valor_total_tributos_estaduais": valor_total_tributos_estaduais, + # "valor_total_tributos_municipais": + # valor_total_tributos_municipais, + # "percentual_total_tributos_federais": + # percentual_total_tributos_federais, + # "percentual_total_tributos_estaduais": + # percentual_total_tributos_estaduais, + # "percentual_total_tributos_municipais": ( + # percentual_total_tributos_municipais + # ), + # "indicador_total_tributacao": indicador_total_tributacao, + # "percentual_total_tributos_simples_nacional": ( + # percentual_total_tributos_simples_nacional + # ), + } + + 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) + ) + + class Document(models.Model): _inherit = "l10n_br_fiscal.document" @@ -355,8 +798,19 @@ def _serialize(self, edocs): NFSe-specific information. """ edocs = super()._serialize(edocs) + # Handle NFSe Nacional for record in self.filtered(filter_processador_edoc_nfse).filtered( - filter_focusnfe + filter_focusnfe_nacional + ): + edoc = { + "rps": record._prepare_lote_rps(), + "service": record._prepare_dados_servico(), + "recipient": record._prepare_dados_tomador(), + } + edocs.append(edoc) + # Handle NFSe Municipal (original) + for record in self.filtered(filter_processador_edoc_nfse).filtered( + filter_focusnfe_municipal ): edoc = [] edoc.append({"rps": record._prepare_lote_rps()}) @@ -394,6 +848,186 @@ def _document_export(self, pretty_print=True): record.authorization_event_id = event_id return result + def _process_authorized_status_nacional(self, record, json_data): + """Process authorized status for NFSe Nacional.""" + aware_datetime = datetime.strptime( + json_data["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_data.get("codigo_verificacao", ""), + "document_number": json_data.get("numero", ""), + "authorization_date": naive_datetime, + } + ) + + xml_path = json_data.get("caminho_xml_nota_fiscal", "") + if xml_path: + xml = requests.get( + NFSE_URL[record.nfse_environment] + xml_path, + 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: + url_danfse = json_data.get("url_danfse", "") + if url_danfse: + pdf_content = requests.get( + 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) + + def _process_authorized_status_municipal(self, record, json_data): + """Process authorized status for NFSe Municipal.""" + aware_datetime = datetime.strptime( + json_data["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_data["codigo_verificacao"], + "document_number": json_data["numero"], + "authorization_date": naive_datetime, + } + ) + + xml = requests.get( + NFSE_URL[record.nfse_environment] + json_data["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_data["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_data["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) + + 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 "Erro na autorização" + 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"] == "autorizado" + and record.state_edoc != SITUACAO_EDOC_AUTORIZADA + ): + self._process_authorized_status_nacional(record, json) + elif json["status"] == "erro_autorizacao": + self._process_error_status(record, json) + elif json["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"] == "autorizado" + and record.state_edoc != SITUACAO_EDOC_AUTORIZADA + ): + self._process_authorized_status_municipal(record, json) + 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) + + 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 +1038,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 +1060,13 @@ def create_cancel_event(self, status_json, record): Returns: The created event. """ - - xml = requests.get( - NFSE_URL[record.nfse_environment] + status_json["caminho_xml_cancelamento"], - timeout=TIMEOUT, - ).content.decode("utf-8") + xml_path = status_json.get("caminho_xml_cancelamento", "") + xml = "" + if xml_path: + xml = requests.get( + NFSE_URL[record.nfse_environment] + xml_path, + timeout=TIMEOUT, + ).content.decode("utf-8") event = record.event_ids.create_event_save_xml( company_id=record.company_id, @@ -552,6 +1112,160 @@ def fetch_and_verify_pdf_content(self, status_json, record): if pdf_content.startswith(b"%PDF-") and pdf_content.strip().endswith(b"%%EOF"): record.make_focus_nfse_pdf(pdf_content) + def _process_cancel_nacional(self, record): + """Process cancellation for NFSe Nacional.""" + ref = str(record.rps_number) + + status_response = record.env[ + "focusnfe.nfse.nacional" + ].query_focus_nfse_nacional_by_ref( + ref, record.company_id, record.nfse_environment + ) + status_json = status_response.json() + + if status_response.status_code == 200: + if ( + status_json.get("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() + 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 pdf_content.startswith( + b"%PDF-" + ) and pdf_content.strip().endswith(b"%%EOF"): + record.make_focus_nfse_pdf(pdf_content) + return status_response + + response = record.env[ + "focusnfe.nfse.nacional" + ].cancel_focus_nfse_nacional_document( + ref, record.cancel_reason, record.company_id, record.nfse_environment + ) + + json = response.json() + + if response.status_code in [200, 400]: + code = json.get("codigo", "") + status = json.get("status", "") + + if code == "nfe_cancelada" or status == "cancelado": + status_rps = record.env[ + "focusnfe.nfse.nacional" + ].query_focus_nfse_nacional_by_ref( + ref, 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() + 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 pdf_content.startswith( + b"%PDF-" + ) and pdf_content.strip().endswith(b"%%EOF"): + record.make_focus_nfse_pdf(pdf_content) + + return response + + raise UserError( + _( + "%(code)s - %(status)s", + code=response.status_code, + status=status, + ) + ) + + raise UserError( + _( + "%(code)s - %(msg)s", + code=response.status_code, + msg=json.get("mensagem", ""), + ) + ) + + def _process_cancel_municipal(self, record): + """Process cancellation for NFSe Municipal.""" + ref = "rps" + record.rps_number + + status_response = record.env["focusnfe.nfse"].query_focus_nfse_by_rps( + ref, 0, record.company_id, record.nfse_environment + ) + status_json = status_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() + else: + record.fetch_and_verify_pdf_content(status_json, record) + return status_response + + response = record.env["focusnfe.nfse"].cancel_focus_nfse_document( + ref, record.cancel_reason, record.company_id, record.nfse_environment + ) + + json = response.json() + + if response.status_code in [200, 400]: + code = json.get("codigo", "") + status = json.get("status", "") + + # hack barueri - provisório + if not code and record.company_id.city_id.ibge_code == "3505708": + code = json.get("erros", [{}])[0].get("codigo", "") + 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() + else: + record.fetch_and_verify_pdf_content(status_json, record) + + return response + + raise UserError( + _( + "%(code)s - %(status)s", + code=response.status_code, + status=status, + ) + ) + + raise UserError( + _( + "%(code)s - %(msg)s", + code=response.status_code, + msg=json.get("mensagem", ""), + ) + ) + def cancel_document_focus(self): """Cancel a NFSe document with the Focus NFSe provider. @@ -561,97 +1275,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"] == "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 == "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"] == "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 == "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 +1354,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 +1395,7 @@ def _cron_document_status_focus(self): .filtered(filter_processador_edoc_nfse) .filtered(filter_focusnfe) ) - if records: - records._document_status() + # Iterar sobre cada registro individualmente, pois _document_status() + # pode esperar um singleton em alguns casos + for record in records: + record._document_status() diff --git a/l10n_br_nfse_focus/models/res_company.py b/l10n_br_nfse_focus/models/res_company.py index fb03f08b1db3..102d73976edc 100644 --- a/l10n_br_nfse_focus/models/res_company.py +++ b/l10n_br_nfse_focus/models/res_company.py @@ -60,6 +60,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 33bc3dcca8cd..36434d53bc9a 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,14 @@ # 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.document import ( + API_ENDPOINT, + NFSE_URL, + Document, + filter_focusnfe, + filter_focusnfe_municipal, + filter_focusnfe_nacional, +) # Mock path for testing purposes MOCK_PATH = "odoo.addons.l10n_br_nfse_focus" @@ -597,3 +604,216 @@ 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.document.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.document.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.document.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.document.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.document.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.document.FocusnfeNfseNacional.cancel_focus_nfse_nacional_document" # noqa: B950 + ) + @patch( + "odoo.addons.l10n_br_nfse_focus.models.document.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() 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