diff --git a/alembic/versions/30a80fe36910_fiscal_idempotency_keys.py b/alembic/versions/30a80fe36910_fiscal_idempotency_keys.py new file mode 100644 index 0000000..3b25512 --- /dev/null +++ b/alembic/versions/30a80fe36910_fiscal_idempotency_keys.py @@ -0,0 +1,48 @@ +"""fiscal_idempotency_keys (Task 5): backs the `Idempotency-Key` contract of +`POST /v1/emissoes` (design spec decision #6) -- a separate table so the +Task 3 `fiscal_documents` migration stays untouched. See +`documents.models.FiscalIdempotencyKey`'s docstring for the two-layer +idempotency pattern this table's UNIQUE constraint exists for. + +Revision ID: 30a80fe36910 +Revises: 523235d06bd5 +Create Date: 2026-07-22 00:00:00.000000 + +""" +from __future__ import annotations + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "30a80fe36910" +down_revision: Union[str, Sequence[str], None] = "523235d06bd5" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "fiscal_idempotency_keys", + sa.Column("id", sa.UUID(), nullable=False), + sa.Column("product_id", sa.UUID(), nullable=False), + sa.Column("idempotency_key", sa.String(length=255), nullable=False), + sa.Column("document_id", sa.UUID(), nullable=False), + sa.Column( + "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False + ), + sa.Column( + "updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False + ), + sa.ForeignKeyConstraint(["product_id"], ["products.id"]), + sa.ForeignKeyConstraint(["document_id"], ["fiscal_documents.id"]), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint( + "product_id", "idempotency_key", name="uq_fiscal_idempotency_key_product_key" + ), + ) + + +def downgrade() -> None: + op.drop_table("fiscal_idempotency_keys") diff --git a/src/fiscal_svc/documents/models.py b/src/fiscal_svc/documents/models.py index 63d94f7..bf48d32 100644 --- a/src/fiscal_svc/documents/models.py +++ b/src/fiscal_svc/documents/models.py @@ -215,3 +215,42 @@ class FiscalDocument(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin): rejeicao_motivo: Mapped[str | None] = mapped_column(String(500), nullable=True) protocolo: Mapped[str | None] = mapped_column(String(20), nullable=True) autorizada_em: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + +class FiscalIdempotencyKey(Base, UUIDPKMixin, TimestampMixin): + """Task 5: backs the `Idempotency-Key` contract of `POST /v1/emissoes` + (design spec decision #6) -- a SEPARATE table rather than a column on + `FiscalDocument` (Task 3's table, already shipped/migrated) so this + Task never touches that migration. `UNIQUE (product_id, + idempotency_key)` is the SAME-transaction outbox partner of `emission. + service.emitir_documento`'s number-allocation + document-INSERT commit + (`documents.service.allocate_fiscal_number`'s docstring): this row is + added to the SAME session, in the SAME commit, as the `FiscalDocument` + it points to via `document_id` -- so a rollback (e.g. the idempotency + race below, or a signature failure) undoes the allocated number AND the + document row AND this key together, never just some of the three. + + The two-layer idempotency pattern this table exists for: (1) a cheap + pre-check SELECT before doing any real work (the fast path for a + genuine retry); (2) this UNIQUE constraint as the source of truth for + the RACE -- two concurrent requests carrying the SAME `Idempotency-Key` + can both pass the pre-check (`None`) before either commits; the FIRST + to commit wins, the SECOND's commit raises `IntegrityError` here, which + `emission.service` catches and translates into a RE-READ of the + winner's row (converging both callers on the SAME `FiscalDocument`, + never emitting a duplicate NF-e for one logical request). + + No soft-delete mixin: an idempotency key's history is permanent by + design -- there is no "un-claim this key" operation.""" + + __tablename__ = "fiscal_idempotency_keys" + __table_args__ = ( + UniqueConstraint( + "product_id", "idempotency_key", + name="uq_fiscal_idempotency_key_product_key", + ), + ) + + product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False) + idempotency_key: Mapped[str] = mapped_column(String(255), nullable=False) + document_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("fiscal_documents.id"), nullable=False) diff --git a/src/fiscal_svc/emission/__init__.py b/src/fiscal_svc/emission/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/fiscal_svc/emission/router.py b/src/fiscal_svc/emission/router.py new file mode 100644 index 0000000..17fb438 --- /dev/null +++ b/src/fiscal_svc/emission/router.py @@ -0,0 +1,84 @@ +import uuid + +from fastapi import APIRouter, Depends, Header, HTTPException, Query, Response, status +from sqlalchemy.ext.asyncio import AsyncSession + +from fiscal_svc.core.db import get_session +from fiscal_svc.emission import service +from fiscal_svc.emission.schemas import EmissaoRequest, FiscalDocumentRead +from fiscal_svc.shared.errors import conflict +from fiscal_svc.tenancy.deps import require_product +from fiscal_svc.tenancy.models import Product + +router = APIRouter(prefix="/v1", tags=["emissao"]) + + +@router.post("/emissoes", response_model=FiscalDocumentRead) +async def emitir_documento_endpoint( + payload: EmissaoRequest, + response: Response, + idempotency_key: str = Header(..., alias="Idempotency-Key", max_length=255, min_length=1), + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> FiscalDocumentRead: + """`Idempotency-Key` OBRIGATÓRIO (design spec decisão #6) -- ausente + vira 422 na borda do FastAPI (`Header(...)`, sem default), antes de + qualquer lógica rodar. Repetida com a MESMA `product_id` -> 200 com o + documento já existente (`created=False`); nova -> 201 (`created=True`). + Ver `emission.service.emitir_documento`'s docstring para o padrão de + duas camadas que garante isto mesmo sob duas requisições concorrentes + com a mesma chave.""" + try: + document, created = await service.emitir_documento(session, product, payload, idempotency_key) + except service.FiscalConfigMissingError as exc: + raise conflict("fiscal_config_missing", str(exc)) from exc + except service.FiscalDocumentConflictError as exc: + raise conflict("fiscal_document_conflict", str(exc)) from exc + response.status_code = status.HTTP_201_CREATED if created else status.HTTP_200_OK + return FiscalDocumentRead.model_validate(document) + + +@router.get("/documentos/{document_id}", response_model=FiscalDocumentRead) +async def get_fiscal_document_endpoint( + document_id: uuid.UUID, + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> FiscalDocumentRead: + document = await service.get_fiscal_document(session, product.id, document_id) + if document is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Documento fiscal não encontrado" + ) + return FiscalDocumentRead.model_validate(document) + + +@router.get("/documentos/{document_id}/xml") +async def get_fiscal_document_xml_endpoint( + document_id: uuid.UUID, + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> Response: + document = await service.get_fiscal_document(session, product.id, document_id) + if document is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, detail="Documento fiscal não encontrado" + ) + return Response(content=document.xml_assinado, media_type="application/xml") + + +@router.get("/documentos", response_model=list[FiscalDocumentRead]) +async def list_fiscal_documents_endpoint( + tenant_ref: str | None = Query(default=None, max_length=64), + branch_ref: str | None = Query(default=None, max_length=64), + status_filter: str | None = Query(default=None, alias="status", max_length=20), + limit: int = Query(default=50, ge=1, le=200), + offset: int = Query(default=0, ge=0), + product: Product = Depends(require_product), + session: AsyncSession = Depends(get_session), +) -> list[FiscalDocumentRead]: + documents = await service.list_fiscal_documents( + session, product.id, + tenant_ref=tenant_ref, branch_ref=branch_ref, status_=status_filter, + limit=limit, offset=offset, + ) + return [FiscalDocumentRead.model_validate(d) for d in documents] diff --git a/src/fiscal_svc/emission/schemas.py b/src/fiscal_svc/emission/schemas.py new file mode 100644 index 0000000..03129aa --- /dev/null +++ b/src/fiscal_svc/emission/schemas.py @@ -0,0 +1,170 @@ +"""Task 5: `EmissaoRequest` -- the pydantic mirror of `sowai_fiscal. +xml_builder.DadosEmissao` (+ `sowai_fiscal.resolver.FiscalResult`/ +`TributoLinha`, which the lib itself already defines as pydantic +`BaseModel`s -- these two are NOT re-mirrored here as separate payload +classes, `emission.service` constructs them straight via `FiscalResult(** +...)`/`TributoLinha(**...)`), adapted per the porte table and design spec: + + * `tenant_ref`/`branch_ref` (opaque strings, decision #4) replace what in + the auto was implicit (`sale.branch_id`). + * `document_model`/`serie` select WHICH `FiscalSeries` to allocate from + (Task 4) -- `chave_acesso`/`numero`/`cnf`/`dh_emi` are DROPPED from this + payload entirely (unlike the lib's own `DadosEmissao`, which expects + them pre-computed): this SERVICE computes all four itself (`cNF do + serviço, persistido`, plan Task 5) -- a caller can never inject its own + chave/número, closing the exact class of bug the auto's own emission + code had to defend against for `branch_id` (C1, "auditoria Fable + 2026-07-16", `auto/backend/app/modules/fiscal/emissao.py`'s module + docstring). + * `ver_proc` is REQUIRED (no default) -- design spec's "EMENDA F1": the + lib's own `DadosEmissao.ver_proc` defaults to `"sowai-auto/1b.1"`, a + default that made sense when this code lived IN the auto and is + actively WRONG for every other product calling this shared service. A + plain (no-default) pydantic field is already "required, 422 if absent" + -- no extra validator needed. + * `fiscal_result` per item is the motor de regras' OUTPUT (design spec + decision #3: "O serviço NÃO resolve imposto (FiscalResult vem no + payload)") -- this service treats it as opaque data to embed in the + XML, never recomputes it.""" +import uuid +from datetime import datetime +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class TributoLinhaPayload(BaseModel): + tax_domain: str + cst: str | None = None + csosn: str | None = None + base_calc: Decimal + base_calc_percent: Decimal + aliquota: Decimal | None = None + valor: Decimal + mva: Decimal | None = None + aliquota_st: Decimal | None = None + fcp_percent: Decimal | None = None + codigo_beneficio: str | None = None + rule_id: uuid.UUID + + +class FiscalResultPayload(BaseModel): + cfop: str + cst: str | None = None + csosn: str | None = None + origem: str | None = None + consumidor_final: bool + indicador_ie: str + tributos: list[TributoLinhaPayload] = Field(default_factory=list) + + +class EmitenteDataPayload(BaseModel): + cnpj: str = Field(min_length=14, max_length=14) + razao_social: str + nome_fantasia: str | None = None + ie: str + crt: str = Field(min_length=1, max_length=1) + address_street: str + address_number: str + address_complement: str | None = None + address_district: str + address_city: str + address_state: str = Field(min_length=2, max_length=2) + address_zip: str + address_city_ibge_code: str + fone: str | None = None + + +class DestinatarioDataPayload(BaseModel): + """`None` no `EmissaoRequest.destinatario` == consumidor final não + identificado -- espelha `sowai_fiscal.xml_builder.DestinatarioData`.""" + + nome: str + cnpj: str | None = None + cpf: str | None = None + indicador_ie: str + ie: str | None = None + address_street: str | None = None + address_number: str | None = None + address_complement: str | None = None + address_district: str | None = None + address_city: str | None = None + address_state: str | None = None + address_zip: str | None = None + address_city_ibge_code: str | None = None + email: str | None = None + + +class ItemDataPayload(BaseModel): + codigo: str + descricao: str + ncm: str + cfop: str + unidade_comercial: str + unidade_tributavel: str + quantidade: Decimal = Field(gt=0) + valor_unitario: Decimal = Field(gt=0) + fiscal_result: FiscalResultPayload + gtin: str | None = None + cest: str | None = None + peso_liquido_kg: Decimal | None = None + peso_bruto_kg: Decimal | None = None + + +class PagamentoDataPayload(BaseModel): + tpag: str + valor: Decimal = Field(gt=0) + indpag: str = "0" + + +class EmissaoRequest(BaseModel): + tenant_ref: str = Field(max_length=64) + branch_ref: str = Field(max_length=64) + # Only "55" for now -- see `series.schemas.FiscalSeriesCreate`'s + # docstring for why (`xml_builder.build_nfe` hardcodes `mod="55"`). + document_model: Literal["55"] = "55" + serie: int = Field(ge=0) + emitente: EmitenteDataPayload + itens: list[ItemDataPayload] = Field(min_length=1) + pagamento: PagamentoDataPayload + ambiente: Literal["homologacao", "producao"] + uf_destino_tipo: Literal["interna", "interestadual"] + destinatario: DestinatarioDataPayload | None = None + nat_op: str = "Venda" + tp_emis: str = Field(default="1", min_length=1, max_length=1) + ind_final: str = Field(default="1", min_length=1, max_length=1) + ind_pres: str = Field(default="1", min_length=1, max_length=1) + fin_nfe: str = Field(default="1", min_length=1, max_length=1) + # OBRIGATÓRIO -- ver o docstring do módulo (EMENDA F1). Nenhum default: + # ausente no payload -> 422 na borda do FastAPI, antes de qualquer + # lógica de negócio rodar. + ver_proc: str = Field(min_length=1, max_length=20) + + +class FiscalDocumentRead(BaseModel): + """NUNCA inclui `xml_assinado` -- o XML sai só por `GET /v1/documentos/ + {id}/xml` (`Response(media_type="application/xml")`), mesmo racional de + `FiscalCertificateRead` nunca incluir o binário do certificado. Sem + `sale_id`/`service_order_id` (porte table: este serviço não conhece o + domínio do produto chamador).""" + + model_config = ConfigDict(from_attributes=True) + + id: uuid.UUID + product_id: uuid.UUID + tenant_ref: str + branch_ref: str + series_id: uuid.UUID + document_model: str + serie: int + numero: int + chave_acesso: str + codigo_numerico: str + status: str + ambiente: str + rejeicao_codigo: str | None + rejeicao_motivo: str | None + protocolo: str | None + autorizada_em: datetime | None + created_at: datetime diff --git a/src/fiscal_svc/emission/service.py b/src/fiscal_svc/emission/service.py new file mode 100644 index 0000000..7d6456e --- /dev/null +++ b/src/fiscal_svc/emission/service.py @@ -0,0 +1,430 @@ +"""Task 5: emissão de NF-e (paridade 1b.1) -- ported from the auto's +`app/modules/fiscal/emissao.py`, porte table applied throughout. The +biggest structural change (design spec decision #3): this service receives +a COMPLETE `EmissaoRequest` -- it never collects Sale/Branch/Person/Part +rows, never calls `resolve_fiscal`, and validates only STRUCTURAL +completeness (a certificate exists for the branch_ref, a series exists for +the (document_model, serie)) rather than an entire domain's worth of +cadastro fields. The outbox invariant survives verbatim: número allocation +(`documents.service.allocate_fiscal_number`, no-commit contract) and the +`FiscalDocument` + `FiscalIdempotencyKey` INSERTs happen in the SAME +transaction, ONE commit -- any failure (including a sabotaged signature) +rolls back all three together, never burning a number with nothing to show +for it. + +ORDER (mirrors the auto's own emissao.py, same reason: `gerar_cnf(numero)` +needs the number ALREADY allocated -- cNF != nNF is NT2019.001): + 1. idempotency pre-check (`Idempotency-Key` -> existing document, if any). + 2. completeness: live certificate for branch_ref, existing series for + (document_model, serie) -- BOTH checked BEFORE touching + `allocate_fiscal_number`, so a missing config never wastes a número. + 3. `allocate_fiscal_number` (lock, no commit) -> `gerar_cnf` -> chave. + 4. `build_nfe` (pure, `sowai_fiscal`) -> serialize -> assina. + 5. `FiscalDocument(ASSINADO)` + `FiscalIdempotencyKey` added to the SAME + session -> ONE commit. A UNIQUE-violation on the idempotency key here + (the race: two concurrent requests, same key, both passed step 1 + before either committed) rolls back and RE-READS the winner's row -- + see `documents.models.FiscalIdempotencyKey`'s docstring for the full + two-layer pattern this implements.""" +import uuid +from datetime import datetime, timezone +from zoneinfo import ZoneInfo + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric.rsa import RSAPrivateKey +from cryptography.x509 import Certificate +from erpbrasil.assinatura.assinatura import Assinatura +from lxml import etree +from sqlalchemy import select +from sqlalchemy.exc import IntegrityError +from sqlalchemy.ext.asyncio import AsyncSession +from xsdata.formats.dataclass.serializers import XmlSerializer +from xsdata.formats.dataclass.serializers.config import SerializerConfig + +from fiscal_svc.certificates import crypto as certificate_lib +from fiscal_svc.documents.models import ( + FiscalCertificate, + FiscalDocument, + FiscalDocumentStatus, + FiscalIdempotencyKey, + FiscalSeries, +) +from fiscal_svc.documents.service import FiscalSeriesNotFoundError, allocate_fiscal_number +from fiscal_svc.emission.schemas import ( + DestinatarioDataPayload, + EmissaoRequest, + EmitenteDataPayload, + FiscalResultPayload, + ItemDataPayload, + PagamentoDataPayload, + TributoLinhaPayload, +) +from fiscal_svc.tenancy.models import Product +from sowai_fiscal.chave_acesso import gerar_cnf, montar_chave_acesso +from sowai_fiscal.resolver import FiscalResult, TributoLinha +from sowai_fiscal.xml_builder import ( + DadosEmissao, + DestinatarioData, + EmitenteData, + ItemData, + PagamentoData, + build_nfe, +) + +_NFE_NAMESPACE = "http://www.portalfiscal.inf.br/nfe" + +# M4 do auto (review Opus, 2026-07-16), preservado: dhEmi e o AAMM da chave +# em horário de Brasília, NÃO UTC -- na virada de mês o AAMM em UTC cairia +# no período de apuração errado e a chave divergiria do dhEmi. O Brasil não +# observa mais horário de verão (desde 2019), então o offset é sempre +# -03:00 para America/Sao_Paulo. +_TZ_EMISSAO = ZoneInfo("America/Sao_Paulo") + + +class FiscalConfigMissingError(Exception): + """Fail-closed: campos estruturais ausentes para EMITIR (certificado/ + série) -- router mapeia para 409 `fiscal_config_missing`, nomeando os + campos. Ao contrário do auto (que também validava dezenas de campos de + cadastro de Branch/Person/Part), este serviço só valida o que É DELE: + o `EmissaoRequest` inteiro já passou pela borda pydantic (campos + obrigatórios/tipos), e o `FiscalResult` por item já vem RESOLVIDO + (design spec decisão #3) -- não há cadastro para revalidar aqui.""" + + def __init__(self, missing: list[str]): + self.missing = missing + super().__init__("Configuração fiscal ausente para emissão: " + ", ".join(missing)) + + +class FiscalDocumentConflictError(Exception): + """`chave_acesso` é UNIQUE GLOBAL (`uq_fiscal_documents_chave_acesso`, + Task 3) -- uma colisão no INSERT final (cNF repetido por acaso para o + mesmo nNF, ou qualquer outra causa) estouraria `IntegrityError` direto + do driver; traduzido aqui em 409 `fiscal_document_conflict`. O + `rollback()` que acompanha desfaz a alocação do número junto (mesmo + outbox de qualquer outra falha antes do commit) -- a colisão nunca + queima um número.""" + + def __init__(self, chave_acesso: str): + self.chave_acesso = chave_acesso + super().__init__( + f"Colisão de chave de acesso ({chave_acesso}) ao gravar o documento fiscal — tente novamente" + ) + + +def _is_chave_acesso_constraint_violation(exc: IntegrityError) -> bool: + detail = str(getattr(exc.orig, "args", [""])[0]) if exc.orig else str(exc) + return "uq_fiscal_documents_chave_acesso" in detail.lower() + + +def _is_idempotency_key_constraint_violation(exc: IntegrityError) -> bool: + detail = str(getattr(exc.orig, "args", [""])[0]) if exc.orig else str(exc) + return "uq_fiscal_idempotency_key_product_key" in detail.lower() + + +class _SignerCertificado: + """Shim mínimo para `erpbrasil.assinatura.Assinatura`, que espera um + objeto `certificado` com atributos `.key`/`._cert`/`._chave`/`._senha` + (a forma de `erpbrasil.assinatura.certificado.Certificado`). Ported + verbatim from the auto's `fiscal.emissao._SignerCertificado`.""" + + def __init__(self, private_key: RSAPrivateKey, cert: Certificate): + self.key = private_key + self.cert = cert + self._cert = cert.public_bytes(encoding=serialization.Encoding.PEM) + self._chave = private_key.private_bytes( + encoding=serialization.Encoding.PEM, + format=serialization.PrivateFormat.PKCS8, + encryption_algorithm=serialization.NoEncryption(), + ) + self._senha = b"" + + +def _serialize_nfe(nfe) -> str: + config = SerializerConfig(xml_declaration=False, indent=None) + return XmlSerializer(config=config).render(nfe, ns_map={None: _NFE_NAMESPACE}) + + +def sign_nfe_xml(xml_str: str, chave_acesso: str, private_key: RSAPrivateKey, cert: Certificate) -> str: + """Assina o XML (enveloped, `erpbrasil.assinatura`/xmlsec) referenciando + `infNFe` pelo seu `Id` (`"NFe" + chave_acesso`). Função de módulo (não + inline em `emitir_documento`) DE PROPÓSITO -- é o ponto exato que o + teste-prova do outbox (`tests/emission/test_emissao.py`) monkeypatcha + para forçar uma falha DEPOIS da alocação do número e ANTES do commit.""" + assinatura = Assinatura(_SignerCertificado(private_key, cert)) + root = etree.fromstring(xml_str.encode("utf-8")) + signed = assinatura.assina_xml2(root, reference="NFe" + chave_acesso) + return signed.decode("utf-8") if isinstance(signed, bytes) else signed + + +# --- EmissaoRequest -> dataclasses da lib (sowai_fiscal.xml_builder) -------- + + +def _to_tributo(payload: TributoLinhaPayload) -> TributoLinha: + return TributoLinha(**payload.model_dump()) + + +def _to_fiscal_result(payload: FiscalResultPayload) -> FiscalResult: + return FiscalResult( + cfop=payload.cfop, + cst=payload.cst, + csosn=payload.csosn, + origem=payload.origem, + consumidor_final=payload.consumidor_final, + indicador_ie=payload.indicador_ie, + tributos=[_to_tributo(t) for t in payload.tributos], + ) + + +def _to_item(payload: ItemDataPayload) -> ItemData: + return ItemData( + codigo=payload.codigo, + descricao=payload.descricao, + ncm=payload.ncm, + cfop=payload.cfop, + unidade_comercial=payload.unidade_comercial, + unidade_tributavel=payload.unidade_tributavel, + quantidade=payload.quantidade, + valor_unitario=payload.valor_unitario, + fiscal_result=_to_fiscal_result(payload.fiscal_result), + gtin=payload.gtin, + cest=payload.cest, + peso_liquido_kg=payload.peso_liquido_kg, + peso_bruto_kg=payload.peso_bruto_kg, + ) + + +def _to_emitente(payload: EmitenteDataPayload) -> EmitenteData: + return EmitenteData(**payload.model_dump()) + + +def _to_destinatario(payload: DestinatarioDataPayload | None) -> DestinatarioData | None: + if payload is None: + return None + return DestinatarioData(**payload.model_dump()) + + +def _to_pagamento(payload: PagamentoDataPayload) -> PagamentoData: + return PagamentoData(**payload.model_dump()) + + +# --- lookups internos (duplicados, não importados de certificates.service -- +# mesmo racional "não vale o acoplamento por uma SELECT de poucas linhas" +# que o auto documenta em `fiscal.emissao._get_live_certificate`) ---------- + + +async def _get_live_certificate( + session: AsyncSession, product_id: uuid.UUID, branch_ref: str +) -> FiscalCertificate | None: + result = await session.execute( + select(FiscalCertificate).where( + FiscalCertificate.product_id == product_id, + FiscalCertificate.branch_ref == branch_ref, + FiscalCertificate.deleted_at.is_(None), + ) + ) + return result.scalar_one_or_none() + + +async def _get_series( + session: AsyncSession, + product_id: uuid.UUID, + tenant_ref: str, + branch_ref: str, + document_model: str, + serie: int, +) -> FiscalSeries | None: + result = await session.execute( + select(FiscalSeries).where( + FiscalSeries.product_id == product_id, + FiscalSeries.tenant_ref == tenant_ref, + FiscalSeries.branch_ref == branch_ref, + FiscalSeries.document_model == document_model, + FiscalSeries.serie == serie, + FiscalSeries.deleted_at.is_(None), + ) + ) + return result.scalar_one_or_none() + + +async def _get_document_by_idempotency_key( + session: AsyncSession, product_id: uuid.UUID, idempotency_key: str +) -> FiscalDocument | None: + result = await session.execute( + select(FiscalDocument) + .join(FiscalIdempotencyKey, FiscalIdempotencyKey.document_id == FiscalDocument.id) + .where( + FiscalIdempotencyKey.product_id == product_id, + FiscalIdempotencyKey.idempotency_key == idempotency_key, + ) + ) + return result.scalar_one_or_none() + + +async def emitir_documento( + session: AsyncSession, + product: Product, + payload: EmissaoRequest, + idempotency_key: str, +) -> tuple[FiscalDocument, bool]: + """Devolve `(document, created)` -- `created=False` quando `idempotency_ + key` já resolvia para um documento existente (pré-checagem OU corrida + resolvida via re-leitura), para o router escolher 200 vs 201.""" + # 1. idempotência -- pré-checagem (camada 1 do padrão de duas camadas). + existing = await _get_document_by_idempotency_key(session, product.id, idempotency_key) + if existing is not None: + return existing, False + + # 2. completude estrutural -- ANTES de alocar número. + missing: list[str] = [] + + certificate = await _get_live_certificate(session, product.id, payload.branch_ref) + if certificate is None: + missing.append("certificado A1 do branch_ref") + elif certificate.not_valid_after < datetime.now(timezone.utc): + missing.append("certificado A1 vencido") + + series = await _get_series( + session, product.id, payload.tenant_ref, payload.branch_ref, payload.document_model, payload.serie + ) + if series is None: + missing.append(f"série fiscal modelo {payload.document_model} série {payload.serie}") + + if missing: + raise FiscalConfigMissingError(missing) + + # 3. aloca o número (lock, sem commit) -> cNF -> chave -------------- + try: + numero = await allocate_fiscal_number( + session, product.id, payload.tenant_ref, payload.branch_ref, + payload.document_model, payload.serie, + ) + except FiscalSeriesNotFoundError as exc: + # Corrida rara: a série existia no pré-check acima e sumiu (soft- + # delete concorrente) antes do SELECT ... FOR UPDATE de allocate. + # Mesma família de erro que a ausência original -- 409 fiscal_ + # config_missing, não um 500. + raise FiscalConfigMissingError( + [f"série fiscal modelo {payload.document_model} série {payload.serie}"] + ) from exc + + cnf = gerar_cnf(numero) + # UM `now()` só, em horário de Brasília -- o AAMM da chave e o dhEmi + # têm que vir do MESMO instante/fuso, senão divergem na virada de mês. + agora_brasil = datetime.now(_TZ_EMISSAO) + chave_acesso = montar_chave_acesso( + uf_ibge=payload.emitente.address_city_ibge_code[:2], + aamm=agora_brasil.strftime("%y%m"), + cnpj=payload.emitente.cnpj, + modelo=payload.document_model, + serie=payload.serie, + numero=numero, + tp_emis=payload.tp_emis, + cnf=cnf, + ) + + dados = DadosEmissao( + emitente=_to_emitente(payload.emitente), + itens=[_to_item(item) for item in payload.itens], + pagamento=_to_pagamento(payload.pagamento), + ambiente=payload.ambiente, + chave_acesso=chave_acesso, + numero=numero, + serie=payload.serie, + cnf=cnf, + # `timespec="seconds"` -- o XSD não aceita fração de segundo. + dh_emi=agora_brasil.isoformat(timespec="seconds"), + uf_destino_tipo=payload.uf_destino_tipo, + destinatario=_to_destinatario(payload.destinatario), + nat_op=payload.nat_op, + tp_emis=payload.tp_emis, + ind_final=payload.ind_final, + ind_pres=payload.ind_pres, + fin_nfe=payload.fin_nfe, + ver_proc=payload.ver_proc, + ) + + # 4. monta + assina --------------------------------------------------- + nfe = build_nfe(dados) + xml_str = _serialize_nfe(nfe) + private_key, cert = certificate_lib.load_private_key_and_cert(certificate) + xml_assinado = sign_nfe_xml(xml_str, chave_acesso, private_key, cert) + + # 5. persiste -- MESMO commit da alocação acima + a chave de idempotência + document = FiscalDocument( + product_id=product.id, + tenant_ref=payload.tenant_ref, + branch_ref=payload.branch_ref, + series_id=series.id, + document_model=payload.document_model, + serie=payload.serie, + numero=numero, + chave_acesso=chave_acesso, + codigo_numerico=cnf, + status=FiscalDocumentStatus.ASSINADO.value, + ambiente=payload.ambiente, + xml_assinado=xml_assinado, + ) + session.add(document) + await session.flush() # popula document.id para o FK abaixo + + idem_row = FiscalIdempotencyKey( + product_id=product.id, idempotency_key=idempotency_key, document_id=document.id + ) + session.add(idem_row) + + try: + await session.commit() + except IntegrityError as exc: + await session.rollback() + if _is_idempotency_key_constraint_violation(exc): + # Camada 2 do padrão: um concorrente com a MESMA idempotency_key + # venceu a corrida entre a pré-checagem (passo 1) e este commit + # -- re-lê o vencedor e converge, em vez de expor a corrida como + # erro. O rollback acima já desfez a alocação do número E o + # INSERT do documento deste caller (outbox intacto). + winner = await _get_document_by_idempotency_key(session, product.id, idempotency_key) + if winner is not None: + return winner, False + raise + if _is_chave_acesso_constraint_violation(exc): + raise FiscalDocumentConflictError(chave_acesso) from exc + raise + await session.refresh(document) + return document, True + + +async def get_fiscal_document( + session: AsyncSession, product_id: uuid.UUID, document_id: uuid.UUID +) -> FiscalDocument | None: + result = await session.execute( + select(FiscalDocument).where( + FiscalDocument.id == document_id, + FiscalDocument.product_id == product_id, + FiscalDocument.deleted_at.is_(None), + ) + ) + return result.scalar_one_or_none() + + +async def list_fiscal_documents( + session: AsyncSession, + product_id: uuid.UUID, + *, + tenant_ref: str | None = None, + branch_ref: str | None = None, + status_: str | None = None, + limit: int = 50, + offset: int = 0, +) -> list[FiscalDocument]: + query = select(FiscalDocument).where( + FiscalDocument.product_id == product_id, FiscalDocument.deleted_at.is_(None) + ) + if tenant_ref is not None: + query = query.where(FiscalDocument.tenant_ref == tenant_ref) + if branch_ref is not None: + query = query.where(FiscalDocument.branch_ref == branch_ref) + if status_ is not None: + query = query.where(FiscalDocument.status == status_) + query = query.order_by(FiscalDocument.created_at.desc()).limit(limit).offset(offset) + result = await session.execute(query) + return list(result.scalars().all()) diff --git a/src/fiscal_svc/main.py b/src/fiscal_svc/main.py index 0660bee..7b9d68c 100644 --- a/src/fiscal_svc/main.py +++ b/src/fiscal_svc/main.py @@ -2,12 +2,14 @@ from fastapi import FastAPI from fiscal_svc.certificates.router import router as certificates_router from fiscal_svc.core.config import settings +from fiscal_svc.emission.router import router as emission_router from fiscal_svc.series.router import router as series_router app = FastAPI(title=settings.app_name) app.include_router(certificates_router) app.include_router(series_router) +app.include_router(emission_router) @app.get("/v1/health") diff --git a/tests/certificates/test_certificates.py b/tests/certificates/test_certificates.py index 0287414..d7d0afb 100644 --- a/tests/certificates/test_certificates.py +++ b/tests/certificates/test_certificates.py @@ -263,7 +263,7 @@ async def test_cross_tenant_ref_same_product_is_404(db_session): @pytest.mark.asyncio async def test_second_upload_replaces_the_first_soft_deleting_it(db_session): key = f"k-{uuid.uuid4().hex}" - await create_product(db_session, name="auto", api_key=key) + product = await create_product(db_session, name="auto", api_key=key) first_pfx = _build_test_pfx(cnpj="14200166000187", password="senha123", cn="PRIMEIRO:14200166000187") second_pfx = _build_test_pfx(cnpj="14200166000187", password="senha456", cn="SEGUNDO:14200166000187") @@ -280,8 +280,15 @@ async def test_second_upload_replaces_the_first_soft_deleting_it(db_session): assert get_response.json()["id"] == second_response.json()["id"] assert "SEGUNDO" in get_response.json()["subject_cn"] + # `product_id` scoped -- `tenant_ref="t1"`/`branch_ref="b1"` are literals + # reused by MANY tests in this file/suite (the `db_session` fixture only + # rolls back at teardown, it does not truncate what earlier tests already + # committed), so an unscoped query here would count every OTHER test's + # certificates for the same branch_ref too. result = await db_session.execute( - select(FiscalCertificate).where(FiscalCertificate.branch_ref == "b1") + select(FiscalCertificate).where( + FiscalCertificate.product_id == product.id, FiscalCertificate.branch_ref == "b1" + ) ) rows = result.scalars().all() assert len(rows) == 2 diff --git a/tests/emission/__init__.py b/tests/emission/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/emission/test_emissao.py b/tests/emission/test_emissao.py new file mode 100644 index 0000000..11022f0 --- /dev/null +++ b/tests/emission/test_emissao.py @@ -0,0 +1,410 @@ +"""Task 5: `POST /v1/emissoes` + `GET /v1/documentos/{id}[/xml]` -- ported +from the auto's `tests/modules/fiscal/test_emissao.py`, with the porte +table's biggest structural change applied: this service receives a +COMPLETE `EmissaoRequest` (no Sale/Branch/Person/Part collection), so the +"caminho feliz" setup here is FAR shorter -- upload a certificate + create a +series + POST a golden's `DadosEmissao` payload, no cadastro at all. + +Golden cases (`sowai_fiscal.goldens/*.input.json`) are the payload SOURCE +for the happy path -- proves the service round-trips a real, lib-shaped +`DadosEmissao` end to end. Byte-exact signature determinism against +pre-generated `.expected.xml` fixtures is Task 6's `test_signed_goldens.py`; +this file only proves the document is well-formed/valid and the invariants +(outbox, idempotency, `ver_proc` obrigatório) hold.""" +import importlib.resources +import json +import uuid +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest +from cryptography import x509 +from cryptography.fernet import Fernet +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import rsa +from cryptography.hazmat.primitives.serialization import pkcs12 +from cryptography.x509.oid import NameOID +from httpx import ASGITransport, AsyncClient +from lxml import etree +from sqlalchemy import select + +from fiscal_svc.certificates import crypto as certificate_lib +from fiscal_svc.core.db import get_session +from fiscal_svc.documents.models import FiscalDocument, FiscalSeries +from fiscal_svc.emission import service as emission_service +from fiscal_svc.emission.schemas import EmissaoRequest +from fiscal_svc.main import app +from fiscal_svc.tenancy.service import create_product + +_GOLDENS_DIR = Path(str(importlib.resources.files("sowai_fiscal") / "goldens")) +_CNPJ_EMITENTE = "12345678000190" + + +@pytest.fixture(autouse=True) +def _override_db(db_session): + async def _get_session_override(): + yield db_session + + app.dependency_overrides[get_session] = _get_session_override + yield + app.dependency_overrides.clear() + + +@pytest.fixture(autouse=True) +def _fiscal_cert_encryption_key(monkeypatch): + monkeypatch.setenv("FISCAL_CERT_ENCRYPTION_KEY", Fernet.generate_key().decode("ascii")) + certificate_lib._get_fernet.cache_clear() + yield + certificate_lib._get_fernet.cache_clear() + + +def _build_test_pfx(cnpj: str = _CNPJ_EMITENTE, password: str = "senha123") -> bytes: + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = issuer = x509.Name( + [ + x509.NameAttribute(NameOID.COMMON_NAME, f"AUTOPECAS THIAGO LTDA:{cnpj}"), + x509.NameAttribute(NameOID.SERIAL_NUMBER, cnpj), + ] + ) + now = datetime.now(timezone.utc) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(now - timedelta(days=1)) + .not_valid_after(now + timedelta(days=365)) + .sign(key, hashes.SHA256()) + ) + return pkcs12.serialize_key_and_certificates( + name=b"test", key=key, cert=cert, cas=None, + encryption_algorithm=serialization.BestAvailableEncryption(password.encode("utf-8")), + ) + + +def _headers(api_key: str) -> dict[str, str]: + return {"X-Api-Key": api_key} + + +async def _product_and_key(db_session, name="auto"): + key = f"k-{uuid.uuid4().hex}" + product = await create_product(db_session, name=name, api_key=key) + return product, key + + +def _payload_from_golden( + case: str, *, tenant_ref: str = "t1", branch_ref: str = "b1", ver_proc: str = "sowai-auto/1b.1" +) -> tuple[dict, dict]: + raw = json.loads((_GOLDENS_DIR / f"{case}.input.json").read_text(encoding="utf-8")) + payload = { + "tenant_ref": tenant_ref, + "branch_ref": branch_ref, + "document_model": "55", + "serie": raw["serie"], + "emitente": raw["emitente"], + "itens": raw["itens"], + "pagamento": raw["pagamento"], + "ambiente": raw["ambiente"], + "uf_destino_tipo": raw["uf_destino_tipo"], + "destinatario": raw.get("destinatario"), + "ver_proc": ver_proc, + } + return payload, raw + + +async def _setup_certificate_and_series( + db_session, client, key, *, tenant_ref, branch_ref, serie, next_number, cnpj=_CNPJ_EMITENTE, +): + pfx_bytes = _build_test_pfx(cnpj=cnpj) + upload_response = await client.post( + "/v1/certificados", + params={"tenant_ref": tenant_ref, "branch_ref": branch_ref}, + files={"file": ("cert.pfx", pfx_bytes, "application/x-pkcs12")}, + data={"password": "senha123", "cnpj": cnpj}, + headers=_headers(key), + ) + assert upload_response.status_code == 201, upload_response.text + + series_response = await client.post( + "/v1/series", + json={ + "tenant_ref": tenant_ref, "branch_ref": branch_ref, + "document_model": "55", "serie": serie, "next_number": next_number, + }, + headers=_headers(key), + ) + assert series_response.status_code == 201, series_response.text + return series_response.json() + + +def _xsd_schema(): + """Resolvido do pacote `nfelib` INSTALADO -- ausência é FALHA, não + skip, mesma convenção do auto (`tests/modules/fiscal/test_emissao.py:: + _xsd_schema`).""" + import nfelib + + path = Path(nfelib.__file__).parent / "nfe" / "schemas" / "v4_0" / "nfe_v4.00.xsd" + if not path.exists(): + pytest.fail(f"XSD não encontrado em {path} -- validação XSD é prova OBRIGATÓRIA") + return etree.XMLSchema(etree.parse(str(path))) + + +@pytest.mark.asyncio +async def test_emitir_documento_caminho_feliz_gera_documento_assinado_xsd_valido(db_session): + product, key = await _product_and_key(db_session) + payload, raw = _payload_from_golden("caso_padrao_intra") + + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + await _setup_certificate_and_series( + db_session, client, key, tenant_ref="t1", branch_ref="b1", + serie=raw["serie"], next_number=raw["numero"], + ) + + response = await client.post( + "/v1/emissoes", json=payload, + headers={**_headers(key), "Idempotency-Key": f"idem-{uuid.uuid4().hex}"}, + ) + + assert response.status_code == 201, response.text + body = response.json() + assert body["status"] == "ASSINADO" + assert body["numero"] == raw["numero"] + assert len(body["chave_acesso"]) == 44 and body["chave_acesso"].isdigit() + assert body["tenant_ref"] == "t1" + assert body["branch_ref"] == "b1" + assert "xml_assinado" not in body + + result = await db_session.execute( + select(FiscalDocument).where(FiscalDocument.id == uuid.UUID(body["id"])) + ) + document = result.scalar_one() + assert "