jonatanritter
c903a9ce0e
Port certificate lifecycle (parse/encrypt/upload/deactivate) and FiscalSeries CRUD from the auto, adapted to (product_id, tenant_ref, branch_ref) tenancy. Closes the "guard retroativo" PATCH /v1/series next_number regression check that Task 3 deferred to this task. Routes: POST/GET/DELETE /v1/certificados, POST/GET/PATCH /v1/series.
264 lines
10 KiB
Python
264 lines
10 KiB
Python
"""Task 4: `POST/GET /v1/series`, `PATCH /v1/series/{id}` -- ported from the
|
|
auto's `tests/modules/tenants/test_fiscal_series.py` CRUD half (allocation
|
|
itself, and its concurrency proofs, are Task 3's
|
|
`tests/documents/test_allocate_fiscal_number.py`). Covers: CRUD happy path,
|
|
duplicate tuple -> 409, anti-oracle 404 (cross-product), missing API key ->
|
|
401, and the "guard retroativo" this Task closes: `PATCH .../next_number`
|
|
regressing below the highest `numero` a series has already emitted -> 409."""
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import select
|
|
|
|
from fiscal_svc.core.db import get_session
|
|
from fiscal_svc.documents.models import FiscalDocument, FiscalSeries
|
|
from fiscal_svc.main import app
|
|
from fiscal_svc.tenancy.service import create_product
|
|
|
|
|
|
@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()
|
|
|
|
|
|
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
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_fiscal_series_returns_201(db_session):
|
|
_, key = await _product_and_key(db_session)
|
|
payload = {"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1014}
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.post("/v1/series", json=payload, headers=_headers(key))
|
|
|
|
assert response.status_code == 201, response.text
|
|
body = response.json()
|
|
assert body["tenant_ref"] == "t1"
|
|
assert body["branch_ref"] == "b1"
|
|
assert body["document_model"] == "55"
|
|
assert body["serie"] == 1
|
|
assert body["next_number"] == 1014
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_fiscal_series_returns_created_series(db_session):
|
|
_, key = await _product_and_key(db_session)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 2, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
response = await client.get(
|
|
"/v1/series", params={"tenant_ref": "t1", "branch_ref": "b1"}, headers=_headers(key)
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
series_list = response.json()
|
|
assert len(series_list) == 2
|
|
assert {s["serie"] for s in series_list} == {1, 2}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_fiscal_series_updates_next_number(db_session):
|
|
_, key = await _product_and_key(db_session)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
create_response = await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
series_id = create_response.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/v1/series/{series_id}", json={"next_number": 500}, headers=_headers(key)
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["next_number"] == 500
|
|
|
|
|
|
async def _series_with_documents(db_session, product, *, max_numero: int, next_number: int) -> FiscalSeries:
|
|
"""Fabrica uma série + um `FiscalDocument` VIVO com `numero=max_numero`
|
|
-- exatamente como o guard vai ler (`numero`/`series_id`/`deleted_at IS
|
|
NULL`), sem passar pelo fluxo real de emissão (Task 5)."""
|
|
series = FiscalSeries(
|
|
product_id=product.id, tenant_ref="t1", branch_ref="b1",
|
|
document_model="55", serie=1, next_number=next_number,
|
|
)
|
|
db_session.add(series)
|
|
await db_session.flush()
|
|
|
|
document = FiscalDocument(
|
|
product_id=product.id, tenant_ref="t1", branch_ref="b1", series_id=series.id,
|
|
document_model="55", serie=1, numero=max_numero,
|
|
chave_acesso=str(uuid.uuid4().int)[:44].zfill(44), codigo_numerico="12345678",
|
|
status="ASSINADO", ambiente="homologacao", xml_assinado="<NFe/>",
|
|
)
|
|
db_session.add(document)
|
|
await db_session.commit()
|
|
await db_session.refresh(series)
|
|
return series
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_fiscal_series_next_number_regression_is_409(db_session):
|
|
product, key = await _product_and_key(db_session)
|
|
series = await _series_with_documents(db_session, product, max_numero=1014, next_number=1015)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
regressed = await client.patch(
|
|
f"/v1/series/{series.id}", json={"next_number": 1000}, headers=_headers(key)
|
|
)
|
|
equal_to_max = await client.patch(
|
|
f"/v1/series/{series.id}", json={"next_number": 1014}, headers=_headers(key)
|
|
)
|
|
allowed = await client.patch(
|
|
f"/v1/series/{series.id}", json={"next_number": 1015}, headers=_headers(key)
|
|
)
|
|
|
|
assert regressed.status_code == 409, regressed.text
|
|
assert regressed.json()["detail"]["code"] == "fiscal_series_number_regression"
|
|
assert equal_to_max.status_code == 409, equal_to_max.text
|
|
assert allowed.status_code == 200, allowed.text
|
|
assert allowed.json()["next_number"] == 1015
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_fiscal_series_next_number_ignores_soft_deleted_documents(db_session):
|
|
"""O guard só olha documentos VIVOS -- um `FiscalDocument`
|
|
soft-deletado não deve travar o `next_number` para sempre."""
|
|
product, key = await _product_and_key(db_session)
|
|
series = await _series_with_documents(db_session, product, max_numero=1014, next_number=1015)
|
|
|
|
result = await db_session.execute(select(FiscalDocument).where(FiscalDocument.series_id == series.id))
|
|
document = result.scalar_one()
|
|
document.deleted_at = datetime.now(timezone.utc)
|
|
await db_session.commit()
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.patch(
|
|
f"/v1/series/{series.id}", json={"next_number": 1}, headers=_headers(key)
|
|
)
|
|
|
|
assert response.status_code == 200, response.text
|
|
assert response.json()["next_number"] == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_fiscal_series_rejects_explicit_null(db_session):
|
|
_, key = await _product_and_key(db_session)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
create_response = await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
series_id = create_response.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/v1/series/{series_id}", json={"next_number": None}, headers=_headers(key)
|
|
)
|
|
|
|
assert response.status_code == 422, response.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_create_duplicate_fiscal_series_is_409(db_session):
|
|
_, key = await _product_and_key(db_session)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
response = await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 999},
|
|
headers=_headers(key),
|
|
)
|
|
|
|
assert response.status_code == 409, response.text
|
|
assert response.json()["detail"]["code"] == "duplicate_fiscal_series"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_different_serie_same_model_is_ok(db_session):
|
|
_, key = await _product_and_key(db_session)
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
response = await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 2, "next_number": 1},
|
|
headers=_headers(key),
|
|
)
|
|
|
|
assert response.status_code == 201, response.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_patch_fiscal_series_other_product_series_is_404(db_session):
|
|
product_a, key_a = await _product_and_key(db_session, name="auto")
|
|
_, key_b = await _product_and_key(db_session, name="crm")
|
|
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
create_response = await client.post(
|
|
"/v1/series",
|
|
json={"tenant_ref": "t1", "branch_ref": "b1", "document_model": "55", "serie": 1, "next_number": 1},
|
|
headers=_headers(key_a),
|
|
)
|
|
series_id = create_response.json()["id"]
|
|
|
|
response = await client.patch(
|
|
f"/v1/series/{series_id}", json={"next_number": 42}, headers=_headers(key_b)
|
|
)
|
|
|
|
assert response.status_code == 404, response.text
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_fiscal_series_missing_api_key_is_401(db_session):
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
response = await client.get("/v1/series", params={"tenant_ref": "t1", "branch_ref": "b1"})
|
|
|
|
assert response.status_code == 401, response.text
|