gtiandClaude Sonnet 5
e935ed61d5
Missao 2 (fechamento de lacunas fiscais, read-only + UI/engine gated): - field_contract: pis_cofins_profile now also accepts the literal sentinels NOT_APPLICABLE/OPERATION_TIME, resolving to NA only from an audited FiscalContextEntry - never inferred from emission statistics. CFOP_ENTRADA under an already-decided entry_scope=MIXED_OR_DYNAMIC now resolves to NA instead of MANUAL, so it stops blocking product completeness and stops re-asking an already-answered question. Regression tests prove: no ADMIN decision => unchanged old behavior; decision present => gated bucket, no write. - context_keys: validate_entry accepts the two sentinels for pis_cofins_profile. - Contexto Fiscal UI: PIS/COFINS and CFOP_ENTRADA cards show the Bloco 2 cross-database evidence (5 installations, CRT 1 and 3: emission never reads the PIS/COFINS cadastral fields anywhere) and a link to an ADMIN-facing, anonymized dossie-for-the-accountant; entry_scope is now a dropdown. - release/tests: decoupled the Hub's GTI_FISCAL_VERSION from the Agent's own version (test_versions_are_coherent_across_hub_and_packages vs the new test_agent_version_is_internally_coherent) - a Hub-only release must not require touching, rebuilding or republishing the Agent. - New read-only evidence: docs/runs/GTI_dryrun_universe_summary.json (527 products, Agent 2's real database), docs/runs/BLOCO2_PIS_COFINS_CROSS_DB_V1.json (5 installations), docs/fiscal_completeness_matrix.md, docs/dossie_contador.md (internal, real names) + central/static/docs/dossie_contador.md (public, anonymized - no CNPJ/razao social/product names), docs/AGENT2_UPGRADE_RUNBOOK_V1.md. - CLAUDE.md: recorded the permanent fiscal-mission rules (no AI/scraping, no Firebird schema changes, deploy only via deploy_hub.ps1, never reenroll, no remote access to client Agent hosts, no client data in unauthenticated static routes). No schema/migration change. Zero Firebird writes (new scripts never open a write connection). Hub only - Agent code untouched. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
51 lines
3.4 KiB
Python
51 lines
3.4 KiB
Python
from fastapi.testclient import TestClient
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from central.app.context_keys import validate_entry
|
|
from central.app.main import create_app
|
|
from central.app.models import Base
|
|
|
|
H = {"Authorization": "Bearer test-token"}
|
|
|
|
|
|
def client():
|
|
engine = create_engine("sqlite://", connect_args={"check_same_thread": False}, poolclass=StaticPool)
|
|
Base.metadata.create_all(engine)
|
|
return TestClient(create_app(session_factory=lambda: Session(engine), bootstrap_token="test-token"))
|
|
|
|
|
|
def test_validation_rules():
|
|
ok, err = validate_entry({"scope": "group", "scope_key": "000019", "key": "product_role", "value": "resale"})
|
|
assert err is None and ok["value"] == "RESALE" and ok["scope"] == "GROUP"
|
|
for bad in ({"scope": "GROUP", "key": "product_role", "value": "RESALE"}, # group without key
|
|
{"scope": "COMPANY", "key": "product_role", "value": "BOSS"}, # not in enum
|
|
{"scope": "GROUP", "scope_key": "1", "key": "ncm", "value": "22030000"}, # ncm is product scope
|
|
{"scope": "PRODUCT", "scope_key": "1", "key": "ncm", "value": "2203"}, # not 8 digits
|
|
{"scope": "COMPANY", "key": "pis_cofins_profile", "value": "{\"x\": 1}"}, # unknown profile key
|
|
{"scope": "COMPANY", "key": "icms_rate", "value": "abc"}, {"scope": "COMPANY", "key": "nope", "value": "1"}):
|
|
assert validate_entry(bad)[1]
|
|
assert validate_entry({"scope": "COMPANY", "key": "pis_cofins_profile", "value": "{\"pis_out_cst\": \"49\"}"})[1] is None
|
|
# Missão 2 / Bloco 2.3: the two audited sentinels must validate (case-insensitive) alongside the JSON profile shape.
|
|
ok, err = validate_entry({"scope": "COMPANY", "key": "pis_cofins_profile", "value": "not_applicable"})
|
|
assert err is None and ok["value"] == "NOT_APPLICABLE"
|
|
ok, err = validate_entry({"scope": "COMPANY", "key": "pis_cofins_profile", "value": "OPERATION_TIME"})
|
|
assert err is None and ok["value"] == "OPERATION_TIME"
|
|
|
|
|
|
def test_import_is_all_or_nothing_upserting_and_listed():
|
|
c = client()
|
|
body = {"company_key": "abc123", "entries": [{"scope": "GROUP", "scope_key": "000019", "key": "product_role", "value": "RESALE", "reason": "revenda"},
|
|
{"scope": "COMPANY", "key": "entry_scope", "value": "INTRA"}]}
|
|
assert c.post("/v1/context/import", json=body, headers=H).json() == {"created": 2, "updated": 0}
|
|
body["entries"][0]["value"] = "OWN_PRODUCTION"
|
|
assert c.post("/v1/context/import", json=body, headers=H).json() == {"created": 0, "updated": 2}
|
|
listed = c.get("/v1/context", params={"company_key": "abc123"}, headers=H).json()["entries"]
|
|
assert {(e["scope"], e["key"], e["value"]) for e in listed} == {("GROUP", "product_role", "OWN_PRODUCTION"), ("COMPANY", "entry_scope", "INTRA")}
|
|
bad = {"company_key": "abc123", "entries": [{"scope": "COMPANY", "key": "rounding", "value": "A"}, {"scope": "COMPANY", "key": "rounding", "value": "Z"}]}
|
|
r = c.post("/v1/context/import", json=bad, headers=H)
|
|
assert r.status_code == 422
|
|
assert len(c.get("/v1/context", params={"company_key": "abc123"}, headers=H).json()["entries"]) == 2 # nothing partially stored
|
|
assert c.get("/v1/context", params={"company_key": "other"}, headers=H).json()["entries"] == []
|