diff --git a/alembic/env.py b/alembic/env.py
index 24d93e2..4c9e521 100644
--- a/alembic/env.py
+++ b/alembic/env.py
@@ -13,7 +13,7 @@ from fiscal_svc.core.db import Base
# Uncomment as modules gain SQLAlchemy models, so autogenerate can see them
# (mirrors auto/backend/alembic/env.py's own convention):
from fiscal_svc.tenancy import models as tenancy_models # noqa: F401
-# from fiscal_svc.documents import models as documents_models # noqa: F401
+from fiscal_svc.documents import models as documents_models # noqa: F401
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
diff --git a/alembic/versions/523235d06bd5_fiscal_series_certificates_documents.py b/alembic/versions/523235d06bd5_fiscal_series_certificates_documents.py
new file mode 100644
index 0000000..7518c8d
--- /dev/null
+++ b/alembic/versions/523235d06bd5_fiscal_series_certificates_documents.py
@@ -0,0 +1,126 @@
+"""fiscal_series + fiscal_certificates + fiscal_documents (Task 3): the
+three tables that "change owner" per the design spec (decision #2) --
+ported from the auto's `fiscal_document_series` (tenants), `fiscal_
+certificates`/`fiscal_documents` (fiscal), with product_id/tenant_ref/
+branch_ref replacing organization_id/branch_id (porte table).
+
+Revision ID: 523235d06bd5
+Revises: fd0ffe65c993
+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 = "523235d06bd5"
+down_revision: Union[str, Sequence[str], None] = "fd0ffe65c993"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+ op.create_table(
+ "fiscal_series",
+ sa.Column("id", sa.UUID(), nullable=False),
+ sa.Column("product_id", sa.UUID(), nullable=False),
+ sa.Column("tenant_ref", sa.String(length=64), nullable=False),
+ sa.Column("branch_ref", sa.String(length=64), nullable=False),
+ sa.Column("document_model", sa.String(length=2), nullable=False),
+ sa.Column("serie", sa.Integer(), nullable=False),
+ sa.Column("next_number", sa.Integer(), 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.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+ sa.ForeignKeyConstraint(["product_id"], ["products.id"]),
+ sa.PrimaryKeyConstraint("id"),
+ sa.UniqueConstraint(
+ "product_id", "tenant_ref", "branch_ref", "document_model", "serie",
+ name="uq_fiscal_series_product_tenant_branch_model_serie",
+ ),
+ )
+
+ op.create_table(
+ "fiscal_certificates",
+ sa.Column("id", sa.UUID(), nullable=False),
+ sa.Column("product_id", sa.UUID(), nullable=False),
+ sa.Column("tenant_ref", sa.String(length=64), nullable=False),
+ sa.Column("branch_ref", sa.String(length=64), nullable=False),
+ sa.Column("cnpj", sa.String(length=14), nullable=False),
+ sa.Column("pfx_encrypted", sa.LargeBinary(), nullable=False),
+ sa.Column("password_encrypted", sa.LargeBinary(), nullable=False),
+ sa.Column("subject_cn", sa.String(length=255), nullable=False),
+ sa.Column("cnpj_certificado", sa.String(length=14), nullable=False),
+ sa.Column("not_valid_before", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("not_valid_after", sa.DateTime(timezone=True), 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.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+ sa.ForeignKeyConstraint(["product_id"], ["products.id"]),
+ sa.PrimaryKeyConstraint("id"),
+ )
+ op.create_index(
+ "ix_fiscal_certificates_product_branch_live",
+ "fiscal_certificates",
+ ["product_id", "branch_ref"],
+ unique=True,
+ postgresql_where=sa.text("deleted_at IS NULL"),
+ )
+
+ op.create_table(
+ "fiscal_documents",
+ sa.Column("id", sa.UUID(), nullable=False),
+ sa.Column("product_id", sa.UUID(), nullable=False),
+ sa.Column("tenant_ref", sa.String(length=64), nullable=False),
+ sa.Column("branch_ref", sa.String(length=64), nullable=False),
+ sa.Column("series_id", sa.UUID(), nullable=False),
+ sa.Column("document_model", sa.String(length=2), nullable=False),
+ sa.Column("serie", sa.Integer(), nullable=False),
+ sa.Column("numero", sa.Integer(), nullable=False),
+ sa.Column("chave_acesso", sa.String(length=44), nullable=False),
+ sa.Column("codigo_numerico", sa.String(length=8), nullable=False),
+ sa.Column("status", sa.String(length=20), nullable=False),
+ sa.Column("ambiente", sa.String(length=12), nullable=False),
+ sa.Column("xml_assinado", sa.Text(), nullable=False),
+ sa.Column("rejeicao_codigo", sa.String(length=10), nullable=True),
+ sa.Column("rejeicao_motivo", sa.String(length=500), nullable=True),
+ sa.Column("protocolo", sa.String(length=20), nullable=True),
+ sa.Column("autorizada_em", sa.DateTime(timezone=True), nullable=True),
+ 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.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
+ sa.ForeignKeyConstraint(["product_id"], ["products.id"]),
+ sa.ForeignKeyConstraint(["series_id"], ["fiscal_series.id"]),
+ sa.PrimaryKeyConstraint("id"),
+ sa.UniqueConstraint("chave_acesso", name="uq_fiscal_documents_chave_acesso"),
+ )
+ op.create_index(
+ "ix_fiscal_documents_product_tenant_branch_status",
+ "fiscal_documents",
+ ["product_id", "tenant_ref", "branch_ref", "status"],
+ )
+
+
+def downgrade() -> None:
+ op.drop_index("ix_fiscal_documents_product_tenant_branch_status", table_name="fiscal_documents")
+ op.drop_table("fiscal_documents")
+
+ op.drop_index("ix_fiscal_certificates_product_branch_live", table_name="fiscal_certificates")
+ op.drop_table("fiscal_certificates")
+
+ op.drop_table("fiscal_series")
diff --git a/src/fiscal_svc/documents/__init__.py b/src/fiscal_svc/documents/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/src/fiscal_svc/documents/models.py b/src/fiscal_svc/documents/models.py
new file mode 100644
index 0000000..63d94f7
--- /dev/null
+++ b/src/fiscal_svc/documents/models.py
@@ -0,0 +1,217 @@
+"""Task 3: the three fiscal tables that "change owner" per the design spec
+(decision #2) -- they used to live in the auto (`app/modules/tenants/
+models.py`'s `FiscalDocumentSeries`, `app/modules/fiscal/models.py`'s
+`FiscalCertificate`/`FiscalDocument`) and now live HERE, as the source of
+truth this service was built to be. Ported with the porte table
+(2026-07-17-fiscal-svc-f2-servico.md) applied throughout:
+
+ | No auto | No serviço |
+ |-----------------------------------|-----------------------------------|
+ | organization_id (FK organizations)| product_id (FK products) + |
+ | | tenant_ref: String(64) |
+ | branch_id (FK branches) | branch_ref: String(64) + cnpj |
+ | | (the strong link) |
+ | Sale/rotas por venda | não existem (sale_id/ |
+ | | service_order_id DROPPED -- |
+ | | emissão recebe DadosEmissao |
+ | | completo, sem correlação a uma |
+ | | linha de venda deste lado) |
+
+Global Constraints (zero enum PG): unlike the auto's `FiscalDocumentSeries.
+document_model` (a REAL Postgres `Enum` -- a documented, deliberately
+un-repeated debt there), `document_model` here is plain `String(2)` +
+Python enum at the Pydantic border ONLY, same convention the auto's own
+`FiscalDocument.document_model`/`.status` already used (see their
+docstrings) -- this port does not re-introduce the PG enum anywhere."""
+import enum
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+ DateTime,
+ ForeignKey,
+ Index,
+ Integer,
+ LargeBinary,
+ String,
+ Text,
+ UniqueConstraint,
+ text,
+)
+from sqlalchemy.orm import Mapped, mapped_column
+
+from fiscal_svc.core.db import Base
+from fiscal_svc.shared.base_model import SoftDeleteMixin, TimestampMixin, UUIDPKMixin
+
+
+class FiscalDocumentModel(str, enum.Enum):
+ """Ported from `auto/backend/app/modules/tenants/models.py`. Python
+ enum only -- see this module's docstring for why the auto's Postgres
+ `Enum` column for this same enum is NOT repeated here."""
+
+ NFE_55 = "55"
+ NFCE_65 = "65"
+
+
+class FiscalDocumentStatus(str, enum.Enum):
+ """Ported verbatim from `auto/backend/app/modules/fiscal/models.py` --
+ already `String` + Python enum there (ZERO Postgres enum), so no
+ adaptation needed. State machine: `ASSINADO -> TRANSMITINDO ->
+ AUTORIZADA | REJEITADA | PENDENTE_CONSULTA` (+ `CANCELADA`/`DENEGADA`).
+ Task 3 (this port) only ever WRITES `ASSINADO` -- transmission (F4)
+ writes the rest; the full vocabulary exists now so a later phase never
+ needs a column-shape migration for a new status."""
+
+ ASSINADO = "ASSINADO"
+ TRANSMITINDO = "TRANSMITINDO"
+ AUTORIZADA = "AUTORIZADA"
+ REJEITADA = "REJEITADA"
+ PENDENTE_CONSULTA = "PENDENTE_CONSULTA"
+ CANCELADA = "CANCELADA"
+ DENEGADA = "DENEGADA"
+
+
+class FiscalSeries(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+ """Ported from the auto's `FiscalDocumentSeries` (`tenants/models.py`) --
+ per the design spec this table is the NEW owner of fiscal numbering
+ (decision #2: "mudam de dono"). `product_id`/`tenant_ref`/`branch_ref`
+ replace `organization_id`/`branch_id` per the porte table (this
+ module's docstring).
+
+ Atomic allocation (`SELECT ... FOR UPDATE` + `populate_existing`) is
+ `documents.service.allocate_fiscal_number` (Task 3, ported with its
+ exact no-commit contract -- see that function's docstring). The
+ "guard retroativo" (`next_number` can never regress below the highest
+ `numero` this series has already emitted, per `FiscalDocument.numero`)
+ is deliberately NOT implemented in this task -- it belongs to the
+ `PATCH /v1/series` endpoint (Task 4), which needs this exact model
+ plus `FiscalDocument` (both now exist, Task 3) to check against. Same
+ "documented now, wired later" shape as the auto's own
+ `FiscalDocumentSeries` docstring ("a emissão em si nasce no módulo
+ fiscal da 1b")."""
+
+ __tablename__ = "fiscal_series"
+ __table_args__ = (
+ UniqueConstraint(
+ "product_id", "tenant_ref", "branch_ref", "document_model", "serie",
+ name="uq_fiscal_series_product_tenant_branch_model_serie",
+ ),
+ )
+
+ product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False)
+ tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+ branch_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+ document_model: Mapped[str] = mapped_column(String(2), nullable=False)
+ serie: Mapped[int] = mapped_column(Integer, nullable=False)
+ next_number: Mapped[int] = mapped_column(Integer, nullable=False)
+
+
+class FiscalCertificate(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+ """Ported from the auto's `FiscalCertificate` (`app/modules/fiscal/
+ models.py`) -- A1 certificate (.pfx) for a `branch_ref`. Per the porte
+ table, the emitter is `(product_id, branch_ref)`, with `cnpj` carried
+ on the row itself (the "vínculo forte", design spec decision #4: the
+ CNPJ is what's validated against the certificate at upload and against
+ the emitente at emission -- `branch_ref` alone is an opaque string this
+ service never interprets). No fallback to a tenant-level or
+ product-level certificate -- fail-closed by construction, same as the
+ auto: a branch_ref without its own live certificate cannot emit.
+
+ One VIVO (`deleted_at IS NULL`) row per `(product_id, branch_ref)`: a
+ second upload soft-deletes the previous live row (Task 4's
+ `certificates.service`). `ix_fiscal_certificates_product_branch_live`
+ is a PARTIAL UNIQUE index enforcing that at the database level (same
+ "two concurrent uploads must not both land a live row" reasoning as the
+ auto's own migration `a1b2c3d4e5f6`'s fix) -- scoped by `product_id` in
+ ADDITION to `branch_ref` (the auto's version only needed `branch_id`,
+ already product-scoped by being a real FK; here `branch_ref` is an
+ OPAQUE string owned by the calling product, so two DIFFERENT products
+ could coincidentally pick the identical string for two DIFFERENT real
+ branches -- scoping the uniqueness by `product_id` too is what keeps
+ that from cross-contaminating one product's certificate slot with
+ another's).
+
+ `pfx_encrypted`/`password_encrypted` are Fernet ciphertext (Task 4,
+ `FISCAL_CERT_ENCRYPTION_KEY` env var, own key -- never
+ `app.core.config.Settings`, same "secrets never live in Settings"
+ convention as the auto)."""
+
+ __tablename__ = "fiscal_certificates"
+ __table_args__ = (
+ Index(
+ "ix_fiscal_certificates_product_branch_live",
+ "product_id",
+ "branch_ref",
+ unique=True,
+ postgresql_where=text("deleted_at IS NULL"),
+ ),
+ )
+
+ product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False)
+ tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+ branch_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+ cnpj: Mapped[str] = mapped_column(String(14), nullable=False)
+ pfx_encrypted: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
+ password_encrypted: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
+ subject_cn: Mapped[str] = mapped_column(String(255), nullable=False)
+ cnpj_certificado: Mapped[str] = mapped_column(String(14), nullable=False)
+ not_valid_before: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+ not_valid_after: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+
+class FiscalDocument(Base, UUIDPKMixin, TimestampMixin, SoftDeleteMixin):
+ """Ported from the auto's `FiscalDocument` (`app/modules/fiscal/
+ models.py`). Per the porte table, `sale_id`/`service_order_id` are
+ DROPPED -- this service has no notion of a Sale/ServiceOrder (decision
+ #3 of the design spec: it receives a complete `DadosEmissao`, never
+ resolves or references the caller's own domain objects). A product's
+ own correlation to ITS sale/OS is the product's problem, carried via
+ the `Idempotency-Key` (Task 5) it sends, not a column here.
+
+ Invariants preserved from the auto (contract, not mere convention):
+ (a) a row is only inserted in the SAME commit that allocates `numero`
+ via `documents.service.allocate_fiscal_number` (which does not
+ commit by itself -- see its docstring) -- outbox transactional,
+ exactly as ported;
+ (b) documento é IMUTÁVEL após `ASSINADO` -- correção é cancelar (evento
+ futuro) e emitir outro, nunca um UPDATE no XML/chave já assinados;
+ (c) `chave_acesso` is UNIQUE GLOBALLY (not just per product) -- it
+ already embeds the emitente's CNPJ in its own digits, so two
+ DIFFERENT products (or tenants) can never legitimately produce the
+ same one; a collision here can only be a generation bug, never a
+ false positive between tenants OR between products.
+
+ `ix_fiscal_documents_product_tenant_branch_status` replaces the auto's
+ `ix_fiscal_documents_org_branch_status` (porte table: anti-oracle 404
+ is now by `(product_id, tenant_ref)`, and `GET /v1/documentos` filters
+ by `tenant_ref`/`branch_ref`/`status` per the design spec's API
+ table) -- the index covers exactly that filter shape."""
+
+ __tablename__ = "fiscal_documents"
+ __table_args__ = (
+ Index(
+ "ix_fiscal_documents_product_tenant_branch_status",
+ "product_id",
+ "tenant_ref",
+ "branch_ref",
+ "status",
+ ),
+ UniqueConstraint("chave_acesso", name="uq_fiscal_documents_chave_acesso"),
+ )
+
+ product_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("products.id"), nullable=False)
+ tenant_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+ branch_ref: Mapped[str] = mapped_column(String(64), nullable=False)
+ series_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("fiscal_series.id"), nullable=False)
+ document_model: Mapped[str] = mapped_column(String(2), nullable=False)
+ serie: Mapped[int] = mapped_column(Integer, nullable=False)
+ numero: Mapped[int] = mapped_column(Integer, nullable=False)
+ chave_acesso: Mapped[str] = mapped_column(String(44), nullable=False)
+ codigo_numerico: Mapped[str] = mapped_column(String(8), nullable=False)
+ status: Mapped[str] = mapped_column(String(20), nullable=False)
+ ambiente: Mapped[str] = mapped_column(String(12), nullable=False)
+ xml_assinado: Mapped[str] = mapped_column(Text, nullable=False)
+ rejeicao_codigo: Mapped[str | None] = mapped_column(String(10), nullable=True)
+ 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)
diff --git a/src/fiscal_svc/documents/service.py b/src/fiscal_svc/documents/service.py
new file mode 100644
index 0000000..c2cf351
--- /dev/null
+++ b/src/fiscal_svc/documents/service.py
@@ -0,0 +1,124 @@
+import uuid
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from fiscal_svc.documents.models import FiscalDocumentModel, FiscalSeries
+
+
+class FiscalSeriesNotFoundError(Exception):
+ """Ported from the auto's `tenants.service.FiscalSeriesNotFoundError`.
+ Raised when a `(product_id, tenant_ref, branch_ref, document_model,
+ serie)` tuple does not resolve to a live row -- deliberately raised
+ both when the series truly does not exist and when it belongs to
+ another product/tenant/branch, so a cross-tenant lookup can't be used
+ to distinguish "not found" from "not yours" (anti-oracle, porte table:
+ "por org" -> "por (product_id, tenant_ref)")."""
+
+ def __init__(
+ self,
+ product_id: uuid.UUID,
+ tenant_ref: str,
+ branch_ref: str,
+ document_model: FiscalDocumentModel | str,
+ serie: int,
+ ):
+ self.product_id = product_id
+ self.tenant_ref = tenant_ref
+ self.branch_ref = branch_ref
+ self.document_model = document_model
+ self.serie = serie
+ super().__init__(
+ f"Série fiscal não encontrada para product_id={product_id}, "
+ f"tenant_ref={tenant_ref}, branch_ref={branch_ref}, "
+ f"modelo={document_model}, série={serie}"
+ )
+
+
+async def allocate_fiscal_number(
+ session: AsyncSession,
+ product_id: uuid.UUID,
+ tenant_ref: str,
+ branch_ref: str,
+ document_model: FiscalDocumentModel | str,
+ serie: int,
+) -> int:
+ """Ported verbatim (mechanism + contract) from the auto's
+ `tenants.service.allocate_fiscal_number`, with the porte table's
+ tenancy adaptation applied to the lookup (`organization_id`/`branch_id`
+ -> `product_id`/`tenant_ref`/`branch_ref`). Every invariant from the
+ original is preserved:
+
+ `SELECT ... FOR UPDATE` locks the SERIES ROW: two concurrent emissions
+ for the same `(product_id, tenant_ref, branch_ref, document_model,
+ serie)` never receive the same number -- the second waits for the
+ first's lock and reads the already-incremented `next_number`.
+
+ `.execution_options(populate_existing=True)` is OBRIGATÓRIO: the lock
+ happens in Postgres, but if this `FiscalSeries` object is already in
+ THIS session's identity map (e.g. an earlier unlocked read of the same
+ row, same session), SQLAlchemy's default behavior hands back the
+ CACHED object without repopulating it from the freshly-(re)locked row
+ -- even though the `FOR UPDATE` really did lock the real row. Without
+ this, `next_number` can stay pinned to a stale value and two callers
+ can allocate the SAME number. `fiscal_svc.core.db`'s
+ `expire_on_commit=False` makes this worse (the object never expires on
+ its own after a commit either). See
+ `auto/backend/app/modules/tenants/service.py::allocate_fiscal_number`'s
+ docstring and `tests/modules/tenants/test_fiscal_series.py::
+ test_allocate_fiscal_number_repopulates_identity_mapped_object` for the
+ original reproduction this ports the fix (and the test) for.
+
+ CONTRATO -- ESTA FUNÇÃO NÃO COMMITA. Only `flush()`s: the number is
+ allocated in memory (and the row stays LOCKED by this transaction's
+ `FOR UPDATE`) but nothing is durable until the CALLER commits. This is
+ deliberate: the real emission flow (Task 5) is "alocar número -> montar
+ o XML -> assinar -> persistir o `FiscalDocument` (com esse número)",
+ and allocation + the document INSERT must commit TOGETHER, atomically,
+ in the SAME transaction -- otherwise a failure between allocating and
+ inserting the document would burn a number with no record of what it
+ was for (and, worse, no `FiscalDocument` row to even know it needs
+ inutilização at SEFAZ later). A caller that never commits (a test that
+ only calls this and rolls back, for instance) leaves the allocation
+ with zero durable effect -- expected, not a bug.
+
+ Not an endpoint -- an INTERNAL function the emission service (Task 5)
+ calls. Each caller must pass its OWN `AsyncSession` (one transaction per
+ call): it's one session's `FOR UPDATE` blocking another session's
+ `SELECT` that actually serializes two concurrent allocations; two calls
+ sharing one session/transaction would have no real DB concurrency to
+ serialize."""
+ 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.value
+ if isinstance(document_model, FiscalDocumentModel)
+ else document_model
+ ),
+ FiscalSeries.serie == serie,
+ FiscalSeries.deleted_at.is_(None),
+ )
+ .with_for_update()
+ .execution_options(populate_existing=True)
+ )
+ series = result.scalar_one_or_none()
+ if series is None:
+ raise FiscalSeriesNotFoundError(
+ product_id=product_id,
+ tenant_ref=tenant_ref,
+ branch_ref=branch_ref,
+ document_model=document_model,
+ serie=serie,
+ )
+ allocated = series.next_number
+ series.next_number = allocated + 1
+ # NÃO commita -- ver a seção "CONTRATO" do docstring. O caller commita,
+ # idealmente junto com o INSERT do FiscalDocument que consome este
+ # número. `flush()` garante que o UPDATE já foi enviado ao Postgres (o
+ # lock do FOR UPDATE segue travado até o commit do caller).
+ await session.flush()
+ return allocated
diff --git a/tests/documents/__init__.py b/tests/documents/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/documents/test_allocate_fiscal_number.py b/tests/documents/test_allocate_fiscal_number.py
new file mode 100644
index 0000000..62a3061
--- /dev/null
+++ b/tests/documents/test_allocate_fiscal_number.py
@@ -0,0 +1,274 @@
+"""Task 3: `documents.service.allocate_fiscal_number` -- ported from the
+auto's `tenants.service.allocate_fiscal_number`
+(`tests/modules/tenants/test_fiscal_series.py`), same mechanism/contract,
+tenancy adapted to `(product_id, tenant_ref, branch_ref)`. Covers: sequential
+allocation, the no-commit contract (Fix 1's two directions), the genuine
+N=10 concurrency guarantee, the identity-map staleness repro
+`populate_existing` exists to fix, and not-found (missing/soft-deleted/
+wrong product/wrong tenant/wrong branch -- every dimension of the new
+tenancy, not just product)."""
+import asyncio
+import uuid
+
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from fiscal_svc.documents.models import FiscalDocumentModel, FiscalSeries
+from fiscal_svc.documents.service import FiscalSeriesNotFoundError, allocate_fiscal_number
+from fiscal_svc.tenancy.service import create_product
+
+
+async def _product(db_session, name="auto"):
+ return await create_product(db_session, name=name, api_key=f"k-{uuid.uuid4().hex}")
+
+
+async def _series(db_session, product, *, tenant_ref="tenant-1", branch_ref="branch-1", next_number=1014):
+ series = FiscalSeries(
+ product_id=product.id,
+ tenant_ref=tenant_ref,
+ branch_ref=branch_ref,
+ document_model=FiscalDocumentModel.NFE_55.value,
+ serie=1,
+ next_number=next_number,
+ )
+ db_session.add(series)
+ await db_session.commit()
+ await db_session.refresh(series)
+ return series
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_sequential_calls_increment(db_session):
+ product = await _product(db_session)
+ series = await _series(db_session, product)
+
+ first = await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+ )
+ await db_session.commit()
+ second = await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+ )
+ await db_session.commit()
+
+ assert first == 1014
+ assert second == 1015
+ await db_session.refresh(series)
+ assert series.next_number == 1016
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_accepts_the_raw_string_document_model(db_session):
+ """`document_model` is a plain `String(2)` column (zero enum PG) -- the
+ function must accept either the `FiscalDocumentModel` enum member or
+ its raw `.value` string, since callers (Task 4/5) may hold either."""
+ product = await _product(db_session)
+ series = await _series(db_session, product)
+
+ allocated = await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, series.branch_ref, "55", 1
+ )
+ await db_session.commit()
+
+ assert allocated == 1014
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_without_commit_does_not_persist(test_engine, db_session):
+ """Fix 1's contract, safe direction: a caller that allocates and then
+ rolls back (or never commits) leaves NO durable trace -- the next
+ allocation must hand out the SAME number again."""
+ product = await _product(db_session)
+ series = await _series(db_session, product, next_number=100)
+
+ session_maker = async_sessionmaker(test_engine, expire_on_commit=False)
+
+ rollback_session = session_maker()
+ try:
+ allocated = await allocate_fiscal_number(
+ rollback_session, product.id, series.tenant_ref, series.branch_ref,
+ FiscalDocumentModel.NFE_55, 1,
+ )
+ assert allocated == 100
+ await rollback_session.rollback()
+ finally:
+ await rollback_session.close()
+
+ await db_session.refresh(series)
+ assert series.next_number == 100, (
+ f"expected next_number to revert to 100 after rollback, got "
+ f"{series.next_number} -- the un-committed allocation leaked a "
+ "durable side effect"
+ )
+
+ reallocated = await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+ )
+ await db_session.commit()
+ assert reallocated == 100
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_missing_series_raises(db_session):
+ product = await _product(db_session)
+
+ with pytest.raises(FiscalSeriesNotFoundError):
+ await allocate_fiscal_number(
+ db_session, product.id, "tenant-1", "branch-1", FiscalDocumentModel.NFE_55, 1
+ )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_soft_deleted_series_raises(db_session):
+ from datetime import datetime, timezone
+
+ product = await _product(db_session)
+ series = await _series(db_session, product)
+ series.deleted_at = datetime.now(timezone.utc)
+ await db_session.commit()
+
+ with pytest.raises(FiscalSeriesNotFoundError):
+ await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+ )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_other_product_raises(db_session):
+ """Tenancy-scoping guard (porte adaptation): the SAME
+ tenant_ref/branch_ref/document_model/serie under a DIFFERENT product_id
+ must never allocate from a series that belongs to a different product --
+ the anti-oracle boundary is (product_id, tenant_ref), so this exercises
+ the product_id half of it."""
+ product_a = await _product(db_session, name="auto")
+ series = await _series(db_session, product_a)
+ product_b = await _product(db_session, name="crm")
+
+ with pytest.raises(FiscalSeriesNotFoundError):
+ await allocate_fiscal_number(
+ db_session, product_b.id, series.tenant_ref, series.branch_ref,
+ FiscalDocumentModel.NFE_55, 1,
+ )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_other_tenant_ref_raises(db_session):
+ """Same tenancy boundary, tenant_ref half: two tenants of the SAME
+ product must not share a series just because branch_ref/model/serie
+ coincide."""
+ product = await _product(db_session)
+ series = await _series(db_session, product, tenant_ref="tenant-a")
+
+ with pytest.raises(FiscalSeriesNotFoundError):
+ await allocate_fiscal_number(
+ db_session, product.id, "tenant-b", series.branch_ref, FiscalDocumentModel.NFE_55, 1
+ )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_other_branch_ref_raises(db_session):
+ """Same tenancy boundary, branch_ref half -- the auto's own guard test
+ (`test_allocate_fiscal_number_other_org_raises`) had no branch-level
+ equivalent since `branch_id` there is a real FK scoped by
+ `organization_id`; here `branch_ref` is its own opaque dimension."""
+ product = await _product(db_session)
+ series = await _series(db_session, product, branch_ref="branch-a")
+
+ with pytest.raises(FiscalSeriesNotFoundError):
+ await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, "branch-b", FiscalDocumentModel.NFE_55, 1
+ )
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_concurrent_calls_produce_distinct_sequential_numbers(
+ test_engine, db_session
+):
+ """The regression guard this whole feature exists for, ported verbatim
+ from the auto's own concurrency test: N=10 GENUINELY concurrent
+ `allocate_fiscal_number` calls against the SAME series must produce 10
+ DISTINCT, SEQUENTIAL numbers. Each call uses its OWN
+ `AsyncSession(test_engine)` (a real, separate DB connection) so the
+ event loop can genuinely interleave their network round-trips."""
+ product = await _product(db_session)
+ series = await _series(db_session, product, next_number=100)
+
+ async def _allocate_and_commit(session):
+ allocated = await allocate_fiscal_number(
+ session, product.id, series.tenant_ref, series.branch_ref,
+ FiscalDocumentModel.NFE_55, 1,
+ )
+ await session.commit()
+ return allocated
+
+ session_maker = async_sessionmaker(test_engine, expire_on_commit=False)
+ sessions = [session_maker() for _ in range(10)]
+ try:
+ results = await asyncio.gather(
+ *(_allocate_and_commit(session) for session in sessions),
+ return_exceptions=True,
+ )
+ finally:
+ for session in sessions:
+ await session.close()
+
+ errors = [r for r in results if isinstance(r, BaseException)]
+ assert not errors, f"expected all 10 concurrent allocations to succeed, got errors={errors!r}"
+
+ allocated = sorted(results)
+ assert allocated == list(range(100, 110)), (
+ f"expected 10 distinct sequential numbers 100..109, got {allocated!r} "
+ "(a repeat means two concurrent callers received the same NF-e "
+ "number; a gap means one was skipped)"
+ )
+
+ await db_session.refresh(series)
+ assert series.next_number == 110
+
+
+@pytest.mark.asyncio
+async def test_allocate_fiscal_number_repopulates_identity_mapped_object(test_engine, db_session):
+ """`SELECT ... FOR UPDATE` locks the ROW in Postgres, but if the calling
+ session already has that row's Python object in its IDENTITY MAP, plain
+ SQLAlchemy does NOT repopulate that object's attributes from the newly
+ fetched row by default. Reproduced with ordinary sequential awaits
+ across two sessions (no `asyncio.gather` needed) -- ported from the
+ auto's own reproduction."""
+ from fiscal_svc.documents.models import FiscalSeries as _FiscalSeries
+ from sqlalchemy import select
+
+ product = await _product(db_session)
+ series = await _series(db_session, product, next_number=100)
+
+ # Step 1: put the series into S1 (db_session)'s identity map with
+ # next_number=100.
+ loaded = (
+ await db_session.execute(select(_FiscalSeries).where(_FiscalSeries.id == series.id))
+ ).scalar_one()
+ assert loaded.next_number == 100
+
+ # Step 2: a genuinely separate session allocates 100 and commits -- the
+ # DB row now holds 101.
+ session_maker = async_sessionmaker(test_engine, expire_on_commit=False)
+ other_session = session_maker()
+ try:
+ other_allocated = await allocate_fiscal_number(
+ other_session, product.id, series.tenant_ref, series.branch_ref,
+ FiscalDocumentModel.NFE_55, 1,
+ )
+ await other_session.commit()
+ finally:
+ await other_session.close()
+ assert other_allocated == 100
+
+ # Step 3: S1 allocates next -- MUST be 101 (the real, post-S2 value),
+ # never 100 again (that would be a duplicate NF-e number).
+ allocated = await allocate_fiscal_number(
+ db_session, product.id, series.tenant_ref, series.branch_ref, FiscalDocumentModel.NFE_55, 1
+ )
+ await db_session.commit()
+ assert allocated == 101, (
+ f"expected 101 (the value S2 already committed), got {allocated} -- "
+ "S1's allocate_fiscal_number returned a STALE identity-map-cached "
+ "next_number instead of repopulating from the FOR UPDATE-locked row"
+ )
diff --git a/tests/migrations/test_fiscal_documents_schema.py b/tests/migrations/test_fiscal_documents_schema.py
new file mode 100644
index 0000000..ca99a23
--- /dev/null
+++ b/tests/migrations/test_fiscal_documents_schema.py
@@ -0,0 +1,306 @@
+"""Migration test (Task 3): `fiscal_series` + `fiscal_certificates` +
+`fiscal_documents` -- table/columns exist via `information_schema` over the
+`create_all` schema (Teste A), and each round-trips (INSERT cru via the
+real ORM) on a database built PURELY by `alembic upgrade head` (Teste B) --
+same two-test precedent as `test_products_schema.py` (Task 2) and the
+auto's own `tests/migrations/test_fiscal_*_schema.py`. Also proves the
+constraints Global Constraints calls out by name: UNIQUE `chave_acesso`,
+the partial-unique cert-vivo-per-`(product_id, branch_ref)` index, and the
+UNIQUE `(product_id, tenant_ref, branch_ref, document_model, serie)` on
+`fiscal_series`."""
+from datetime import datetime, timedelta, timezone
+
+import pytest
+import pytest_asyncio
+from sqlalchemy import select, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+# Import de registro: garante que os modelos relevantes estão registrados em
+# Base.metadata antes do create_all da fixture test_engine, ao rodar este
+# arquivo isolado.
+from fiscal_svc.documents import models as _documents_models # noqa: F401
+from fiscal_svc.tenancy import models as _tenancy_models # noqa: F401
+from tests.migrations._helpers import run_alembic, run_psql
+
+_MIGRATION_DB_NAME = "fiscal_svc_test_fiscal_documents_schema"
+_MIGRATION_DB_URL = (
+ f"postgresql+asyncpg://postgres:postgres@localhost:5432/{_MIGRATION_DB_NAME}"
+)
+
+
+@pytest.mark.asyncio
+async def test_fiscal_series_table_and_columns_exist(db_session):
+ rows = {
+ r[0]
+ for r in (
+ await db_session.execute(
+ text(
+ "select column_name from information_schema.columns "
+ "where table_name='fiscal_series'"
+ )
+ )
+ )
+ }
+ assert {
+ "id", "product_id", "tenant_ref", "branch_ref", "document_model",
+ "serie", "next_number", "created_at", "updated_at", "deleted_at",
+ } <= rows
+
+
+@pytest.mark.asyncio
+async def test_fiscal_certificates_table_and_columns_exist(db_session):
+ rows = {
+ r[0]: r[1]
+ for r in (
+ await db_session.execute(
+ text(
+ "select column_name, is_nullable from information_schema.columns "
+ "where table_name='fiscal_certificates'"
+ )
+ )
+ )
+ }
+ for required in (
+ "product_id", "tenant_ref", "branch_ref", "cnpj", "pfx_encrypted",
+ "password_encrypted", "subject_cn", "cnpj_certificado",
+ "not_valid_before", "not_valid_after", "deleted_at",
+ ):
+ assert required in rows, f"coluna {required} ausente em fiscal_certificates"
+ assert rows["pfx_encrypted"] == "NO"
+ assert rows["deleted_at"] == "YES"
+
+
+@pytest.mark.asyncio
+async def test_fiscal_documents_table_and_columns_exist(db_session):
+ rows = {
+ r[0]: r[1]
+ for r in (
+ await db_session.execute(
+ text(
+ "select column_name, is_nullable from information_schema.columns "
+ "where table_name='fiscal_documents'"
+ )
+ )
+ )
+ }
+ for required in (
+ "product_id", "tenant_ref", "branch_ref", "series_id", "document_model",
+ "serie", "numero", "chave_acesso", "codigo_numerico", "status",
+ "ambiente", "xml_assinado", "rejeicao_codigo", "rejeicao_motivo",
+ "protocolo", "autorizada_em", "deleted_at",
+ ):
+ assert required in rows, f"coluna {required} ausente em fiscal_documents"
+ assert rows["chave_acesso"] == "NO"
+ # sale_id/service_order_id are DELIBERATELY absent -- porte table.
+ assert "sale_id" not in rows
+ assert "service_order_id" not in rows
+
+
+@pytest_asyncio.fixture
+async def migration_database():
+ run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
+ run_psql("-c", f"CREATE DATABASE {_MIGRATION_DB_NAME};")
+ yield
+ run_psql("-c", f"DROP DATABASE IF EXISTS {_MIGRATION_DB_NAME} WITH (FORCE);")
+
+
+async def _make_product(session, name="auto"):
+ from fiscal_svc.tenancy.service import create_product
+
+ return await create_product(session, name=name, api_key=f"key-{name}-{id(session)}")
+
+
+@pytest.mark.asyncio
+async def test_fiscal_series_and_certificate_and_document_round_trip_on_a_real_migrated_database(
+ migration_database,
+):
+ result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+ assert result.returncode == 0, result.stderr
+
+ from fiscal_svc.documents.models import FiscalCertificate, FiscalDocument, FiscalSeries
+
+ engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+ session_maker = async_sessionmaker(engine, expire_on_commit=False)
+ try:
+ async with session_maker() as session:
+ product = await _make_product(session)
+
+ series = FiscalSeries(
+ product_id=product.id,
+ tenant_ref="tenant-1",
+ branch_ref="branch-1",
+ document_model="55",
+ serie=1,
+ next_number=1014,
+ )
+ session.add(series)
+ await session.flush()
+
+ certificate = FiscalCertificate(
+ product_id=product.id,
+ tenant_ref="tenant-1",
+ branch_ref="branch-1",
+ cnpj="14200166000187",
+ pfx_encrypted=b"\x00\x01ciphertext-pfx",
+ password_encrypted=b"\x00\x02ciphertext-pw",
+ subject_cn="EMPRESA TESTE LTDA:14200166000187",
+ cnpj_certificado="14200166000187",
+ not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
+ not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
+ )
+ session.add(certificate)
+
+ document = FiscalDocument(
+ product_id=product.id,
+ tenant_ref="tenant-1",
+ branch_ref="branch-1",
+ series_id=series.id,
+ document_model="55",
+ serie=1,
+ numero=1014,
+ chave_acesso="4" * 44,
+ codigo_numerico="12345678",
+ status="ASSINADO",
+ ambiente="homologacao",
+ xml_assinado="",
+ )
+ session.add(document)
+
+ # Would crash here (UndefinedColumn/DataError) before the fix.
+ await session.commit()
+
+ series_id, certificate_id, document_id = series.id, certificate.id, document.id
+
+ async with session_maker() as session:
+ reloaded_series = await session.get(FiscalSeries, series_id)
+ assert reloaded_series.next_number == 1014
+
+ reloaded_cert = await session.get(FiscalCertificate, certificate_id)
+ assert reloaded_cert.cnpj_certificado == "14200166000187"
+ assert reloaded_cert.pfx_encrypted == b"\x00\x01ciphertext-pfx"
+
+ reloaded_doc = await session.get(FiscalDocument, document_id)
+ assert reloaded_doc.status == "ASSINADO"
+ assert reloaded_doc.chave_acesso == "4" * 44
+
+ result_q = await session.execute(
+ select(FiscalDocument).where(FiscalDocument.chave_acesso == "4" * 44)
+ )
+ assert result_q.scalar_one().id == document_id
+ finally:
+ await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_chave_acesso_unique_constraint_holds_on_real_migration(migration_database):
+ result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+ assert result.returncode == 0, result.stderr
+
+ from fiscal_svc.documents.models import FiscalDocument, FiscalSeries
+
+ engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+ session_maker = async_sessionmaker(engine, expire_on_commit=False)
+ try:
+ async with session_maker() as session:
+ product = await _make_product(session, name="auto-chave")
+ series = FiscalSeries(
+ product_id=product.id, tenant_ref="t1", branch_ref="b1",
+ document_model="55", serie=1, next_number=1,
+ )
+ session.add(series)
+ await session.flush()
+
+ def _doc(numero):
+ return FiscalDocument(
+ product_id=product.id, tenant_ref="t1", branch_ref="b1",
+ series_id=series.id, document_model="55", serie=1, numero=numero,
+ chave_acesso="9" * 44, codigo_numerico="12345678",
+ status="ASSINADO", ambiente="homologacao", xml_assinado="",
+ )
+
+ session.add(_doc(1))
+ await session.commit()
+
+ session.add(_doc(2))
+ with pytest.raises(IntegrityError):
+ await session.commit()
+ await session.rollback()
+ finally:
+ await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_two_live_certificates_for_same_product_branch_violate_unique_index_on_real_migration(
+ migration_database,
+):
+ """Same fix/reasoning as the auto's `a1b2c3d4e5f6` migration: a PARTIAL
+ UNIQUE index (here on `(product_id, branch_ref) WHERE deleted_at IS
+ NULL`) makes two concurrently-uploaded LIVE certificates for the same
+ slot structurally impossible, not just avoided by the service layer."""
+ result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+ assert result.returncode == 0, result.stderr
+
+ from fiscal_svc.documents.models import FiscalCertificate
+
+ engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+ session_maker = async_sessionmaker(engine, expire_on_commit=False)
+ try:
+ async with session_maker() as session:
+ product = await _make_product(session, name="auto-cert-corrida")
+
+ def _cert():
+ return FiscalCertificate(
+ product_id=product.id,
+ tenant_ref="t1",
+ branch_ref="b1",
+ cnpj="14200166000280",
+ pfx_encrypted=b"\x00pfx",
+ password_encrypted=b"\x00pw",
+ subject_cn="EMPRESA TESTE LTDA:14200166000280",
+ cnpj_certificado="14200166000280",
+ not_valid_before=datetime.now(timezone.utc) - timedelta(days=1),
+ not_valid_after=datetime.now(timezone.utc) + timedelta(days=365),
+ )
+
+ session.add(_cert())
+ await session.commit()
+
+ session.add(_cert())
+ with pytest.raises(IntegrityError):
+ await session.commit()
+ await session.rollback()
+ finally:
+ await engine.dispose()
+
+
+@pytest.mark.asyncio
+async def test_duplicate_fiscal_series_tuple_violates_unique_constraint_on_real_migration(
+ migration_database,
+):
+ result = run_alembic(_MIGRATION_DB_URL, "upgrade", "head")
+ assert result.returncode == 0, result.stderr
+
+ from fiscal_svc.documents.models import FiscalSeries
+
+ engine = create_async_engine(_MIGRATION_DB_URL, echo=False)
+ session_maker = async_sessionmaker(engine, expire_on_commit=False)
+ try:
+ async with session_maker() as session:
+ product = await _make_product(session, name="auto-serie-dup")
+
+ def _series():
+ return FiscalSeries(
+ product_id=product.id, tenant_ref="t1", branch_ref="b1",
+ document_model="55", serie=1, next_number=1,
+ )
+
+ session.add(_series())
+ await session.commit()
+
+ session.add(_series())
+ with pytest.raises(IntegrityError):
+ await session.commit()
+ await session.rollback()
+ finally:
+ await engine.dispose()
diff --git a/tests/shared/__init__.py b/tests/shared/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/shared/test_for_update_populate_existing.py b/tests/shared/test_for_update_populate_existing.py
new file mode 100644
index 0000000..db18802
--- /dev/null
+++ b/tests/shared/test_for_update_populate_existing.py
@@ -0,0 +1,189 @@
+"""Ported verbatim (mechanism unchanged, only the scanned directory) from
+`auto/backend/tests/shared/test_for_update_populate_existing.py` -- the
+static AST guard for the exact bug `documents.service.allocate_fiscal_number`
+exists to avoid: `SELECT ... FOR UPDATE` locks the row in Postgres, but if
+the calling session already has that row's object in its identity map (an
+earlier unlocked fetch of the same object, same session), SQLAlchemy's
+default behavior hands back the CACHED object instead of repopulating it
+from the freshly-(re)locked row. The Postgres lock is real; the in-memory
+object the code decides against is a stale snapshot.
+`.execution_options(populate_existing=True)` is the fix -- see
+`fiscal_svc.documents.service.allocate_fiscal_number`'s docstring for the
+full write-up (ported from the auto's own `tenants.service.
+allocate_fiscal_number`).
+
+Scans every `.py` file under `src/fiscal_svc/` (never `tests/`) with the
+`ast` module -- deliberately not regex over the source text, because a
+fluent call chain like
+`select(...).where(...).with_for_update().execution_options(...)` routinely
+breaks across several lines/parens, and a text-based check would either
+miss those or need to reimplement a parser badly. Requires that every
+`.with_for_update()` call has a `.execution_options(populate_existing=
+True)` call somewhere in the SAME chain of method calls."""
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+SRC_DIR = Path(__file__).resolve().parents[2] / "src" / "fiscal_svc"
+REPO_DIR = SRC_DIR.parents[1]
+
+_FIX_EXPLANATION = """
+WHY: `.with_for_update()` alone locks the row in Postgres, but if this
+session already has an object for that row in its identity map (e.g. an
+earlier unlocked lookup of the same row, same session, done for a 404/
+access check before calling into the locking code), SQLAlchemy's default
+identity-map behavior returns that CACHED object instead of repopulating it
+from the row `FOR UPDATE` just (re)read. The lock becomes decorative: the
+code holds a real Postgres lock on a row it never actually re-reads, and
+makes its decision against a stale in-memory snapshot instead.
+
+HOW TO FIX: chain `.execution_options(populate_existing=True)` onto the
+SAME query as the `.with_for_update()`, e.g.:
+
+ query = query.with_for_update().execution_options(populate_existing=True)
+
+or, split across lines/parens, as long as it's the same chain:
+
+ result = await session.execute(
+ select(Model)
+ .where(Model.id == model_id)
+ .with_for_update()
+ .execution_options(populate_existing=True)
+ )
+
+If you have found a `FOR UPDATE` that must NOT populate_existing (none exist
+today -- this would be unusual), do not just delete this check or special-
+case your file/line here. Talk to the team about adding an explicit,
+commented opt-out marker to this test first, so the next reader still gets
+an explanation instead of a silently-shrinking guard.
+""".strip()
+
+
+def _is_populate_existing_true_kwarg(call: ast.Call) -> bool:
+ """True if `call` is `.execution_options(...)` carrying a literal
+ `populate_existing=True` keyword argument."""
+ if not (isinstance(call.func, ast.Attribute) and call.func.attr == "execution_options"):
+ return False
+ return any(
+ kw.arg == "populate_existing" and isinstance(kw.value, ast.Constant) and kw.value.value is True
+ for kw in call.keywords
+ )
+
+
+def _build_parent_map(tree: ast.AST) -> dict[int, ast.AST]:
+ parents: dict[int, ast.AST] = {}
+ for parent in ast.walk(tree):
+ for child in ast.iter_child_nodes(parent):
+ parents[id(child)] = parent
+ return parents
+
+
+def _chain_root(node: ast.AST, parents: dict[int, ast.AST]) -> ast.AST:
+ """Walk up from `node` while still inside the same fluent method-call
+ chain -- i.e. while each successive parent is itself a `Call` or
+ `Attribute` node, the two node shapes chaining (`a.b().c().d()`) is
+ built from in the AST. This is what lets a chain split across many
+ lines/parens (still a single expression to the parser) be treated as
+ one unit, and stops at the first non-chain boundary (assignment,
+ statement, argument to an unrelated call, ...).
+ """
+ current = node
+ while True:
+ parent = parents.get(id(current))
+ if parent is None or not isinstance(parent, (ast.Call, ast.Attribute)):
+ return current
+ current = parent
+
+
+def _find_violations(src_dir: Path) -> list[str]:
+ """Returns one `path:line` string per `.with_for_update()` call site
+ under `src_dir` that does not have a `.execution_options(
+ populate_existing=True)` call in the same method-call chain."""
+ violations: list[str] = []
+ for path in sorted(src_dir.rglob("*.py")):
+ source = path.read_text()
+ try:
+ tree = ast.parse(source, filename=str(path))
+ except SyntaxError:
+ continue
+ parents = _build_parent_map(tree)
+ for node in ast.walk(tree):
+ if not (
+ isinstance(node, ast.Call)
+ and isinstance(node.func, ast.Attribute)
+ and node.func.attr == "with_for_update"
+ ):
+ continue
+ root = _chain_root(node, parents)
+ has_populate_existing = any(
+ isinstance(candidate, ast.Call) and _is_populate_existing_true_kwarg(candidate)
+ for candidate in ast.walk(root)
+ )
+ if not has_populate_existing:
+ # `path.relative_to(REPO_DIR)` when `src_dir` is actually
+ # inside this repo (the real guard below); falls back to the
+ # absolute path when it isn't (the synthetic-`tmp_path`
+ # self-tests further down, which scan a throwaway directory
+ # outside the repo entirely).
+ try:
+ rel = path.relative_to(REPO_DIR)
+ except ValueError:
+ rel = path
+ # `end_lineno`, not `lineno`: for a multi-line chain, a
+ # Call/Attribute node's `lineno` is inherited from where the
+ # WHOLE expression starts (e.g. the `select(...)` at the top
+ # of the chain), while `end_lineno` lands on the line the
+ # `.with_for_update()` text itself is on -- the line the
+ # next developer actually needs to look at.
+ violations.append(f"{rel}:{node.end_lineno}")
+ return violations
+
+
+def test_every_with_for_update_call_site_chains_populate_existing():
+ violations = _find_violations(SRC_DIR)
+ assert not violations, (
+ "Found `.with_for_update()` call site(s) missing "
+ "`.execution_options(populate_existing=True)` in the same call "
+ "chain:\n " + "\n ".join(violations) + "\n\n" + _FIX_EXPLANATION
+ )
+
+
+# --- Detection proven in BOTH directions (Global Constraints) --------------
+#
+# The test above only proves the guard is currently GREEN against this
+# repo's real code -- on its own that's equally consistent with "the guard
+# actually detects the bug" and "the guard is a no-op that always passes".
+# These two exercise `_find_violations` directly against synthetic files in
+# a throwaway directory, so the guard's OWN detection logic is proven both
+# ways: it flags the missing fix, and it does not false-positive on the
+# fixed shape (including a chain split across lines, the exact shape
+# `allocate_fiscal_number`'s real call site uses).
+
+
+def test_find_violations_flags_with_for_update_missing_populate_existing(tmp_path):
+ (tmp_path / "offender.py").write_text(
+ "async def f(session):\n"
+ " return await session.execute(\n"
+ " select(Model).where(Model.id == x).with_for_update()\n"
+ " )\n"
+ )
+
+ violations = _find_violations(tmp_path)
+
+ assert len(violations) == 1
+ assert "offender.py" in violations[0]
+
+
+def test_find_violations_does_not_flag_a_correctly_fixed_multiline_chain(tmp_path):
+ (tmp_path / "fixed.py").write_text(
+ "async def f(session):\n"
+ " return await session.execute(\n"
+ " select(Model)\n"
+ " .where(Model.id == x)\n"
+ " .with_for_update()\n"
+ " .execution_options(populate_existing=True)\n"
+ " )\n"
+ )
+
+ assert _find_violations(tmp_path) == []