GTI_GERENCIADOR_FISCAL/central/tests/test_cfop_ippt_write_plumbing.py

"""Missão 4 / Bloco E.1-E.2: Hub-side plumbing for the CFOP_IPPT_WRITE_V1 capability - per-installation
mode (ADMIN-only, CSRF, audited, DEFAULT OFF), delivered inside the authenticated job, and the
append-only audit trail recorded when the job completes. Mirrors central/tests/test_rtc_security.py's
security posture for a separate, narrower capability."""
from sqlalchemy import select
from sqlalchemy.orm import Session

from central.app.models import AuditEvent, CfopIpptWriteAudit, Installation
from central.tests.test_control_panel_pages import _panel_client
from central.tests.test_rtc_security import _become, _post
from central.tests.test_zero_touch_hub import H, OPERATIONAL, _client, finish, hb


def test_default_mode_is_disabled_everywhere():
    client, engine = _client()
    hb(client)
    with Session(engine) as s:
        assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED"
    job = client.get("/v1/jobs/next", headers=H).json()
    assert job["authorizations"]["cfop_ippt_write_mode"] == "DISABLED"
    body = client.get("/v1/agents/me/authorizations", headers=H).json()
    assert body["cfop_ippt_write_mode"] == "DISABLED"


def test_a_forged_completion_payload_cannot_change_the_mode_or_fabricate_an_audit_row():
    """Flag OFF => zero jobs carry write permission, and nothing the Agent sends back can forge one -
    the mode is read fresh from the Installation row, never from the job payload."""
    client, engine = _client()
    hb(client)
    job = client.get("/v1/jobs/next", headers=H).json()
    forged = {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [],
             "authorizations": {"cfop_ippt_write_mode": "ENABLED"},
             "cfop_ippt_apply": {"applied": [{"product_code": "X", "field": "CFOP", "outcome": "APPLIED", "old_value": None, "new_value": "5102"}]}}
    client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": forged}, headers=H)
    with Session(engine) as s:
        assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED"
        # the audit row IS stored (it's just evidence of what the Agent reported for that job) but the
        # mode itself never moved - a forged "ENABLED" in the payload changes nothing authoritative.
        row = s.scalars(select(CfopIpptWriteAudit)).one()
        assert row.outcome == "APPLIED" and row.field == "CFOP" and row.new_value == "5102"


def test_write_mode_is_restricted_to_gerenciador_admins_and_every_denial_is_audited():
    client, agent_id, _ = _panel_client()
    _become(client, "PADRAO")
    assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR").status_code == 403
    with client._session_factory() as s:
        acts = [e.action for e in s.scalars(select(AuditEvent))]
        assert "CFOP_IPPT_WRITE_MODE" not in acts


def test_admin_can_confirm_a_write_mode_and_it_is_audited():
    client, agent_id, _ = _panel_client()
    _become(client, "ADMIN")
    assert client.post(f"/agents/{agent_id}/cfop-ippt-write", data={"csrf_token": "bad", "mode": "ENABLED", "confirm": "GRAVAR"}, follow_redirects=False).status_code in (400, 403)
    resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR")
    assert resp.status_code == 303
    with client._session_factory() as s:
        inst = s.scalars(select(Installation)).one()
        assert inst.cfop_ippt_write_mode == "ENABLED"
        acts = [e.action for e in s.scalars(select(AuditEvent))]
        assert "CFOP_IPPT_WRITE_MODE" in acts


def test_enabling_without_typing_gravar_is_refused():
    client, agent_id, _ = _panel_client()
    _become(client, "ADMIN")
    resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED")
    assert resp.status_code == 422
    with client._session_factory() as s:
        assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED"


def test_disabling_never_requires_the_gravar_confirmation():
    client, agent_id, _ = _panel_client()
    _become(client, "ADMIN")
    assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="ENABLED", confirm="GRAVAR").status_code == 303
    assert _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="DISABLED").status_code == 303
    with client._session_factory() as s:
        assert s.scalars(select(Installation)).one().cfop_ippt_write_mode == "DISABLED"


def test_invalid_mode_is_rejected():
    client, agent_id, _ = _panel_client()
    _become(client, "ADMIN")
    resp = _post(client, f"/agents/{agent_id}/cfop-ippt-write", mode="NOT_A_REAL_MODE", confirm="GRAVAR")
    assert resp.status_code == 422


def test_record_cfop_ippt_write_audits_only_accepts_cfop_or_ippt_and_ignores_other_fields():
    from central.app.cfop_ippt_audit import record_cfop_ippt_write_audits
    client, agent_id, _ = _panel_client()
    with client._session_factory() as s:
        installation = s.scalars(select(Installation)).one()
        from central.app.models import Agent
        agent = s.scalars(select(Agent)).one()
        n = record_cfop_ippt_write_audits(s, tenant_id=installation.tenant_id, installation=installation, agent=agent, job_id=1, cfop_ippt_apply={
            "applied": [{"product_code": "A1", "field": "CFOP", "outcome": "APPLIED", "old_value": None, "new_value": "5102"}],
            "aborted": [{"product_code": "A2", "field": "IPPT", "outcome": "ABORTED", "reason": "STALE_VALUE_CHANGED_SINCE_DECISION"},
                       {"product_code": "A3", "field": "CCLASSTRIB", "outcome": "APPLIED", "new_value": "999999"}],  # not CFOP/IPPT: must be ignored
        })
        assert n == 2
        rows = {r.product_code: r for r in s.scalars(select(CfopIpptWriteAudit))}
        assert set(rows) == {"A1", "A2"}
        assert rows["A1"].outcome == "APPLIED" and rows["A2"].reason == "STALE_VALUE_CHANGED_SINCE_DECISION"