237 lines
11 KiB
Python
237 lines
11 KiB
Python
from fastapi.testclient import TestClient
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from central.app.blocked_products import upsert_blocker
|
|
from central.app.blocker_resolution import resolve_blocker_with_context, validate_resolution_input
|
|
from central.app.main import create_app
|
|
from central.app.models import AdminUser, Agent, AuditEvent, BlockedProduct, FiscalContextEntry, Job, Installation, Tenant
|
|
from central.tests.test_control_plane import seed, setup_db
|
|
|
|
|
|
def _seed_blocker(session_factory):
|
|
agent_id, admin_id = seed(session_factory)
|
|
with session_factory() as session:
|
|
installation = session.scalar(select(Installation).where(Installation.agent_id == agent_id))
|
|
installation.company_code = "ACME-COMPANY"
|
|
blocker = upsert_blocker(
|
|
session,
|
|
installation.tenant_id,
|
|
{
|
|
"company": "Acme",
|
|
"product_code": "P-001",
|
|
"description": "Produto de teste",
|
|
"blocker_type": "BLOCKED_MISSING_EVIDENCE",
|
|
"field": "CFOP",
|
|
"reason_code": "MISSING_CONTEXT",
|
|
"input_fingerprint": "input-1",
|
|
"evidence_summary": {"source": "test"},
|
|
},
|
|
installation_id=installation.id,
|
|
)
|
|
return blocker.id, admin_id, installation.id
|
|
|
|
|
|
def test_validate_resolution_input_accepts_known_enum_values():
|
|
for key, value in (("entry_scope", "INTER"), ("product_role", "RESALE")):
|
|
normalized, error = validate_resolution_input(
|
|
{"scope": "PRODUCT", "scope_key": "P-001", "key": key, "value": value, "reason": "operator note"}
|
|
)
|
|
assert error is None
|
|
assert normalized == {
|
|
"scope": "PRODUCT",
|
|
"scope_key": "P-001",
|
|
"key": key,
|
|
"value": value,
|
|
"reason": "operator note",
|
|
}
|
|
|
|
|
|
def test_validate_resolution_input_rejects_unsafe_or_incomplete_values():
|
|
invalid_payloads = (
|
|
{"scope": "PRODUCT", "scope_key": "P-001", "key": "entry_scope", "value": "ARBITRARY", "reason": "note"},
|
|
{"scope": "PRODUCT", "scope_key": "P-001", "key": "not-a-key", "value": "x", "reason": "note"},
|
|
{"scope": "NCM", "scope_key": "22030000", "key": "entry_scope", "value": "INTER", "reason": "note"},
|
|
{"scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "RESALE", "reason": ""},
|
|
{"scope": "PRODUCT", "key": "product_role", "value": "RESALE", "reason": "note"},
|
|
{"scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "RESALE", "reason": "note", "context": "free text"},
|
|
)
|
|
|
|
for payload in invalid_payloads:
|
|
normalized, error = validate_resolution_input(payload)
|
|
assert normalized is None
|
|
assert error
|
|
|
|
|
|
def test_resolve_blocker_with_context_persists_one_context_row_job_and_sanitized_audit():
|
|
session_factory = setup_db()
|
|
blocker_id, admin_id, installation_id = _seed_blocker(session_factory)
|
|
|
|
with session_factory() as session:
|
|
admin = session.get(AdminUser, admin_id)
|
|
result = resolve_blocker_with_context(
|
|
session,
|
|
blocker_id,
|
|
admin,
|
|
{
|
|
"scope": "PRODUCT",
|
|
"scope_key": "P-001",
|
|
"key": "product_role",
|
|
"value": "RESALE",
|
|
"reason": "reviewed token=secret-value",
|
|
},
|
|
request_ip="192.0.2.10",
|
|
)
|
|
|
|
entry = session.scalar(select(FiscalContextEntry).where(FiscalContextEntry.key == "product_role"))
|
|
job = session.get(Job, result["job_id"])
|
|
audit = session.get(AuditEvent, result["audit_id"])
|
|
|
|
assert result["blocker_id"] == blocker_id
|
|
assert entry.company_key == "ACME-COMPANY"
|
|
assert entry.scope == "PRODUCT"
|
|
assert entry.scope_key == "P-001"
|
|
assert entry.value == "RESALE"
|
|
assert entry.source == "OPERATOR"
|
|
assert job.agent_id == session.get(Installation, installation_id).agent_id
|
|
assert job.kind == "REPROCESS_PRODUCTS"
|
|
assert job.payload == {"codigos": ["P-001"]}
|
|
assert audit.action == "RESOLVE_BLOCKER_WITH_CONTEXT"
|
|
assert audit.target_type == "blocker"
|
|
assert audit.target_id == str(blocker_id)
|
|
assert audit.context["reprocess_job_id"] == job.id
|
|
assert audit.context["before"] is None
|
|
assert audit.context["after"]["value"] == "RESALE"
|
|
assert audit.context["request_ip"] == "192.0.2.10"
|
|
assert "secret-value" not in str(audit.context)
|
|
assert session.query(FiscalContextEntry).count() == 1
|
|
|
|
|
|
def test_legacy_blocker_without_installation_fails_closed_without_job_or_audit():
|
|
session_factory = setup_db()
|
|
blocker_id, admin_id, _ = _seed_blocker(session_factory)
|
|
|
|
with session_factory() as session:
|
|
blocker = session.get(BlockedProduct, blocker_id)
|
|
blocker.installation_id = None
|
|
session.commit()
|
|
admin = session.get(AdminUser, admin_id)
|
|
|
|
with pytest.raises(LookupError, match="installation"):
|
|
resolve_blocker_with_context(session, blocker_id, admin, {"scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "RESALE", "reason": "reviewed"})
|
|
|
|
assert session.query(FiscalContextEntry).count() == 0
|
|
assert session.query(Job).count() == 0
|
|
assert session.query(AuditEvent).count() == 0
|
|
|
|
|
|
def test_cross_tenant_installation_pointer_fails_before_mutation():
|
|
session_factory = setup_db()
|
|
blocker_id, admin_id, _ = _seed_blocker(session_factory)
|
|
|
|
with session_factory() as session:
|
|
blocker = session.get(BlockedProduct, blocker_id)
|
|
other_tenant = Tenant(name="Other Client")
|
|
session.add(other_tenant)
|
|
session.flush()
|
|
other_agent = Agent(tenant_id=other_tenant.id, name="other-agent", token_hash="other-agent-token".ljust(64, "x"))
|
|
session.add(other_agent)
|
|
session.flush()
|
|
other_installation = Installation(agent_id=other_agent.id, tenant_id=other_tenant.id, installation_id="22222222-2222-4222-8222-222222222222", company_code="OTHER-COMPANY")
|
|
session.add(other_installation)
|
|
session.flush()
|
|
blocker.installation_id = other_installation.id
|
|
session.commit()
|
|
admin = session.get(AdminUser, admin_id)
|
|
|
|
with pytest.raises(LookupError, match="tenant"):
|
|
resolve_blocker_with_context(session, blocker_id, admin, {"scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "RESALE", "reason": "reviewed"})
|
|
|
|
assert session.query(FiscalContextEntry).count() == 0
|
|
assert session.query(Job).count() == 0
|
|
assert session.query(AuditEvent).count() == 0
|
|
|
|
|
|
def test_resolution_input_rejects_oversized_and_non_string_fields_cleanly():
|
|
base = {"scope": "PRODUCT", "scope_key": "P-001", "key": "service_cfop", "value": "5102", "reason": "reviewed"}
|
|
for field, value in (("value", "v" * 513), ("reason", "r" * 2001), ("reason", 123)):
|
|
payload = {**base, field: value}
|
|
normalized, error = validate_resolution_input(payload)
|
|
assert normalized is None
|
|
assert error
|
|
|
|
|
|
def test_boundary_length_audit_keeps_required_fields_instead_of_truncating_context():
|
|
session_factory = setup_db()
|
|
blocker_id, admin_id, _ = _seed_blocker(session_factory)
|
|
|
|
with session_factory() as session:
|
|
admin = session.get(AdminUser, admin_id)
|
|
result = resolve_blocker_with_context(
|
|
session,
|
|
blocker_id,
|
|
admin,
|
|
{"scope": "PRODUCT", "scope_key": "P-001", "key": "service_cfop", "value": "v" * 512, "reason": "r" * 2000},
|
|
)
|
|
audit = session.get(AuditEvent, result["audit_id"])
|
|
required = {"admin_user_id", "tenant_id", "client", "company_key", "installation_id", "product_code", "blocker_id", "key", "scope", "scope_key", "value", "reason", "before", "after", "reprocess_job_id", "request_ip"}
|
|
assert required <= audit.context.keys()
|
|
assert "truncated" not in audit.context
|
|
|
|
|
|
def test_human_resolution_detail_requires_auth_csrf_and_renders_safe_form():
|
|
session_factory = setup_db()
|
|
blocker_id, _, _ = _seed_blocker(session_factory)
|
|
client = TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None))
|
|
|
|
assert client.get(f"/blockers/{blocker_id}", follow_redirects=False).status_code == 303
|
|
client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False)
|
|
detail = client.get(f"/blockers/{blocker_id}")
|
|
assert detail.status_code == 200
|
|
assert "Salvar contexto e reprocessar" in detail.text
|
|
assert "Cliente" in detail.text and "Servidor" in detail.text and "Produto" in detail.text
|
|
assert "Evidência" in detail.text and "Valores aceitos" in detail.text
|
|
assert "data-confirm" not in detail.text
|
|
assert 'name="context"' not in detail.text
|
|
|
|
missing_csrf = client.post(
|
|
f"/blockers/{blocker_id}/resolve-context",
|
|
data={"scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "RESALE", "reason": "reviewed"},
|
|
)
|
|
assert missing_csrf.status_code == 403
|
|
|
|
success = client.post(
|
|
f"/blockers/{blocker_id}/resolve-context",
|
|
data={"csrf_token": client.cookies.get("gti_csrf"), "scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "RESALE", "reason": "reviewed"},
|
|
follow_redirects=False,
|
|
)
|
|
assert success.status_code == 303
|
|
assert success.headers["location"] == f"/blockers/{blocker_id}"
|
|
|
|
|
|
def test_human_and_json_resolution_routes_use_the_same_validated_workflow():
|
|
session_factory = setup_db()
|
|
blocker_id, _, _ = _seed_blocker(session_factory)
|
|
client = TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None))
|
|
client.post("/login", data={"username": "operator", "password": "a-secure-password"}, follow_redirects=False)
|
|
csrf = client.cookies.get("gti_csrf")
|
|
|
|
invalid = client.post(
|
|
f"/v1/blockers/{blocker_id}/resolve-human",
|
|
json={"csrf_token": csrf, "scope": "PRODUCT", "scope_key": "P-001", "key": "product_role", "value": "free text", "reason": "reviewed"},
|
|
)
|
|
assert invalid.status_code == 422
|
|
|
|
api_result = client.post(
|
|
f"/v1/blockers/{blocker_id}/resolve-human",
|
|
json={"csrf_token": csrf, "scope": "PRODUCT", "scope_key": "P-001", "key": "entry_scope", "value": "INTER", "reason": "reviewed"},
|
|
)
|
|
assert api_result.status_code == 200
|
|
assert set(api_result.json()) == {"blocker_id", "job_id", "audit_id"}
|
|
|
|
unauthenticated = TestClient(create_app(database_probe=lambda: True, session_factory=session_factory, bootstrap_token=None)).post(
|
|
f"/v1/blockers/{blocker_id}/resolve-human",
|
|
json={"scope": "PRODUCT", "scope_key": "P-001", "key": "entry_scope", "value": "INTER", "reason": "reviewed"},
|
|
)
|
|
assert unauthenticated.status_code == 401
|