diff --git a/central/alembic/versions/0033_cfop_ippt_write_mode.py b/central/alembic/versions/0033_cfop_ippt_write_mode.py new file mode 100644 index 0000000..f63cfa2 --- /dev/null +++ b/central/alembic/versions/0033_cfop_ippt_write_mode.py @@ -0,0 +1,39 @@ +"""CFOP/IPPT write capability: per-installation write mode (DEFAULT OFF) + append-only audit table. + +Revision ID: 0033_cfop_ippt_write_mode +Revises: 0032_contribuinte_ipi +""" +from alembic import op +import sqlalchemy as sa + +revision = "0033_cfop_ippt_write_mode" +down_revision = "0032_contribuinte_ipi" +branch_labels = None +depends_on = None + + +def upgrade(): + op.add_column("installations", sa.Column("cfop_ippt_write_mode", sa.String(24), nullable=False, server_default="DISABLED")) + op.create_table( + "cfop_ippt_write_audits", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column("tenant_id", sa.Integer(), sa.ForeignKey("tenants.id"), nullable=False, index=True), + sa.Column("installation_id", sa.Integer(), sa.ForeignKey("installations.id"), nullable=False, index=True), + sa.Column("agent_id", sa.Integer(), sa.ForeignKey("agents.id"), nullable=False, index=True), + sa.Column("job_id", sa.Integer(), nullable=True, index=True), + sa.Column("product_code", sa.String(64), nullable=False, index=True), + sa.Column("field", sa.String(16), nullable=False), + sa.Column("outcome", sa.String(16), nullable=False, index=True), + sa.Column("reason", sa.String(96), nullable=True), + sa.Column("old_value", sa.String(16), nullable=True), + sa.Column("new_value", sa.String(16), nullable=True), + sa.Column("action", sa.String(16), nullable=True), + sa.Column("rollback_of", sa.Integer(), nullable=True, index=True), + sa.Column("written_at", sa.String(40), nullable=True), + sa.Column("recorded_at", sa.DateTime(timezone=True), server_default=sa.func.now(), index=True), + ) + + +def downgrade(): + op.drop_table("cfop_ippt_write_audits") + op.drop_column("installations", "cfop_ippt_write_mode") diff --git a/central/app/api.py b/central/app/api.py index 297f031..a8f2582 100644 --- a/central/app/api.py +++ b/central/app/api.py @@ -12,6 +12,7 @@ from .config import settings from .datasets.service import DatasetService from .jobs import SYNC_KINDS, claim_next_job, complete_job, create_job, fail_job, schedule_automatic_sync from .rtc_audit import job_authorizations, record_rtc_write_audits +from .cfop_ippt_audit import job_authorizations as cfop_ippt_job_authorizations, record_cfop_ippt_write_audits from .blocked_products import list_blockers, resolve_blocker, resolve_product_blockers, resolve_stale_fiscal_blockers, upsert_blocker, upsert_blockers_bulk from .context_keys import validate_entry from .models import AgentSnapshot, BlockedProduct, FiscalContextEntry, ClassificationCache, ClassificationRequest, ClassificationResult, Dataset, Installation, Job @@ -260,7 +261,9 @@ def router(session_factory, bootstrap_token): session, agent = identity try: installation = session.scalar(select(Installation).where(Installation.agent_id == agent.id)) - return job_authorizations(session, installation) if installation is not None else {"rtc_write_mode": "DISABLED"} + if installation is None: + return {"rtc_write_mode": "DISABLED", "cfop_ippt_write_mode": "DISABLED"} + return {**job_authorizations(session, installation), **cfop_ippt_job_authorizations(installation)} finally: session.close() @@ -407,7 +410,8 @@ def router(session_factory, bootstrap_token): session.commit() logger.info("job claimed job_id=%s agent_id=%s kind=%s", job.id, agent.id, job.kind) view = _job_view(job) - view["authorizations"] = job_authorizations(session, installation) if installation is not None else {"rtc_write_mode": "DISABLED"} # decided centrally, per installation, default OFF + view["authorizations"] = ({**job_authorizations(session, installation), **cfop_ippt_job_authorizations(installation)} if installation is not None + else {"rtc_write_mode": "DISABLED", "cfop_ippt_write_mode": "DISABLED"}) # decided centrally, per installation, default OFF return view finally: session.close() @@ -443,6 +447,7 @@ def router(session_factory, bootstrap_token): resolve_product_blockers(session, agent.tenant_id, product_code, installation_id=installation.id if installation else None) if installation is not None: record_rtc_write_audits(session, tenant_id=agent.tenant_id, installation=installation, agent=agent, job_id=job_id, rtc_auto_apply=result_summary.get("rtc_auto_apply"), evaluated_all=bool((result_summary.get("fiscal_context") or {}).get("evaluated_all"))) + record_cfop_ippt_write_audits(session, tenant_id=agent.tenant_id, installation=installation, agent=agent, job_id=job_id, cfop_ippt_apply=result_summary.get("cfop_ippt_apply")) if installation is not None and (result_summary.get("fiscal_context") or {}).get("evaluated_all"): resolve_stale_fiscal_blockers(session, agent.tenant_id, installation.id, pass_started) if not complete_job(session, job, result_summary): raise HTTPException(409, "job is not RUNNING") diff --git a/central/app/cfop_ippt_audit.py b/central/app/cfop_ippt_audit.py new file mode 100644 index 0000000..6364855 --- /dev/null +++ b/central/app/cfop_ippt_audit.py @@ -0,0 +1,51 @@ +"""Persists the Agent's CFOP/IPPT write audit records (append-only). Mirrors central.app.rtc_audit's +safety model for a separate, narrower capability - no candidate preview table (candidates are +recomputed fresh from field_contract every job, never stored server-side), no rollback queue yet +(rollback is operator-triggered later, out of scope for this pass).""" +from __future__ import annotations + +from .events import record_event +from .models import CfopIpptWriteAudit + +MAX_AUDITS_PER_JOB = 2000 +WRITE_MODES = ("DISABLED", "DRY_RUN", "APPROVAL_REQUIRED", "ENABLED") +ALLOWED_FIELDS = ("CFOP", "IPPT") + + +def _clip(value, size): + return None if value is None else str(value)[:size] + + +def job_authorizations(installation) -> dict: + """What the authenticated job carries to the Agent. Decided centrally per installation; the Agent + has no local switch. DISABLED unless an installation row explicitly says otherwise (server_default).""" + mode = installation.cfop_ippt_write_mode if installation.cfop_ippt_write_mode in WRITE_MODES else "DISABLED" + return {"cfop_ippt_write_mode": mode} + + +def record_cfop_ippt_write_audits(session, *, tenant_id: int, installation, agent, job_id: int | None, cfop_ippt_apply: dict | None) -> int: + """Stores every applied/aborted write of the job. Never trusts the Agent for identity: tenant/installation/agent come from the authenticated context.""" + if not isinstance(cfop_ippt_apply, dict): + return 0 + records = list(cfop_ippt_apply.get("applied") or []) + list(cfop_ippt_apply.get("aborted") or []) + stored = applied = aborted = 0 + for item in records[:MAX_AUDITS_PER_JOB]: + if not isinstance(item, dict) or item.get("field") not in ALLOWED_FIELDS: + continue + outcome = str(item.get("outcome", ""))[:16] + session.add(CfopIpptWriteAudit( + tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, job_id=job_id, + product_code=_clip(item.get("product_code"), 64) or "", field=item["field"], outcome=outcome, + reason=_clip(item.get("reason") or item.get("reason_code"), 96), old_value=_clip(item.get("old_value"), 16), new_value=_clip(item.get("new_value"), 16), + action="APPLY", written_at=_clip(item.get("timestamp"), 40))) + stored += 1 + if outcome == "APPLIED": + applied += 1 + else: + aborted += 1 + if stored: + record_event(session, tenant_id=tenant_id, installation_id=installation.id, agent_id=agent.id, severity="WARNING" if applied else "INFO", event_type="CFOP_IPPT_AUTO_APPLY", job_id=job_id, + message=f"CFOP/IPPT: {applied} aplicado(s), {aborted} abortado(s)", + context={"applied": applied, "aborted": aborted, "reasons": cfop_ippt_apply.get("aborted_reasons") or {}, "mode": cfop_ippt_apply.get("mode")}) + session.commit() + return stored diff --git a/central/app/events.py b/central/app/events.py index b6e0954..b87b8db 100644 --- a/central/app/events.py +++ b/central/app/events.py @@ -15,7 +15,7 @@ EVENT_TYPES = { "AGENT_STARTED", "AGENT_STOPPED", "HEARTBEAT", "PREFLIGHT_PASS", "PREFLIGHT_FAIL", "JOB_CLAIMED", "JOB_STARTED", "JOB_COMPLETED", "JOB_FAILED", "PRODUCT_BLOCKED", "BLOCKER_RESOLVED", "DATASET_CHANGED", "COMPANY_CONTEXT_CHANGED", "SCHEMA_CHANGED", "WRITE_PLANNED", "WRITE_COMPLETED", "WRITE_FAILED", "CONNECTION_ERROR", "RETRY", "VERSION_CHANGED", - "WINDOWS_INSTALL_COMPLETED", "WINDOWS_UNINSTALL_COMPLETED", "ADMIN_ACTION", "RTC_AUTO_APPLY", + "WINDOWS_INSTALL_COMPLETED", "WINDOWS_UNINSTALL_COMPLETED", "ADMIN_ACTION", "RTC_AUTO_APPLY", "CFOP_IPPT_AUTO_APPLY", } SEVERITIES = {"INFO", "WARNING", "ERROR", "CRITICAL"} SECRET_RE = re.compile(r"(?i)(password|fb_password|token|api[_-]?key|secret|authorization|connection[_-]?string)\s*[:=]\s*[^,;\s]+") diff --git a/central/app/models.py b/central/app/models.py index 224d755..5005148 100644 --- a/central/app/models.py +++ b/central/app/models.py @@ -286,6 +286,10 @@ class Installation(Base): # RTC auto-apply is authorized centrally per installation and is OFF until an operator explicitly turns it on (the Agent has its own second switch). rtc_auto_apply_authorized: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false") # legacy (0027); rtc_write_mode is the source of truth rtc_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED + # Same model as rtc_write_mode, an entirely separate switch: authorizes ONLY the CFOP_IPPT_WRITE_V1 + # capability (gti_fiscal_agent.cfop_ippt_auto_apply), never RTC. DISABLED by default for every + # installation - an operator must explicitly turn it on, per installation. + cfop_ippt_write_mode: Mapped[str] = mapped_column(String(24), default="DISABLED", server_default="DISABLED") # DISABLED | DRY_RUN | APPROVAL_REQUIRED | ENABLED # Per-installation FUNCTIONS (central, no local switch). Existing installations keep working exactly as before (corrector on, features configured); # a NEW installation starts WAITING_FEATURE_CONFIGURATION (features_configured=False): it heartbeats but processes nothing until an admin chooses. corrector_enabled: Mapped[bool] = mapped_column(Boolean, default=False, server_default="true") @@ -780,6 +784,27 @@ class RtcWriteAudit(Base): recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) +class CfopIpptWriteAudit(Base): + """Append-only record of every attempted C000025.CFOP/IPPT write (applied or aborted) under the + CFOP_IPPT_WRITE_V1 capability. Holds the old value, so each change is revertible by hand.""" + __tablename__ = "cfop_ippt_write_audits" + id: Mapped[int] = mapped_column(primary_key=True) + tenant_id: Mapped[int] = mapped_column(ForeignKey("tenants.id"), index=True) + installation_id: Mapped[int] = mapped_column(ForeignKey("installations.id"), index=True) + agent_id: Mapped[int] = mapped_column(ForeignKey("agents.id"), index=True) + job_id: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + product_code: Mapped[str] = mapped_column(String(64), index=True) + field: Mapped[str] = mapped_column(String(16)) # CFOP | IPPT only + outcome: Mapped[str] = mapped_column(String(16), index=True) # APPLIED | ABORTED | FAILED | ROLLED_BACK + reason: Mapped[str | None] = mapped_column(String(96), nullable=True) + old_value: Mapped[str | None] = mapped_column(String(16), nullable=True) + new_value: Mapped[str | None] = mapped_column(String(16), nullable=True) + action: Mapped[str | None] = mapped_column(String(16), nullable=True) # APPLY | ROLLBACK + rollback_of: Mapped[int | None] = mapped_column(Integer, nullable=True, index=True) + written_at: Mapped[str | None] = mapped_column(String(40), nullable=True) + recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), index=True) + + class RtcCandidate(Base): """Server-side preview of the products the Agent could fill (R3 only). Rows are proposals; only the Agent's re-derivation at write time decides.""" __tablename__ = "rtc_candidates" diff --git a/central/app/web.py b/central/app/web.py index 8581492..50e959b 100644 --- a/central/app/web.py +++ b/central/app/web.py @@ -293,7 +293,7 @@ def router(session_factory): event["occurred_display"] = format_ui_datetime(datetime.fromisoformat(event["occurred_at"]) if event["occurred_at"] else None) agent_blockers = list_blockers(session, tenant_id=installation.tenant_id, agent_id=agent_id, page=1, page_size=25, scope="all") fiscal_context = build_fiscal_context_view(session, installation.tenant_id, _company_key(None, installation, None)) - return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1]), "fiscal_context": fiscal_context}) + return render(request, "agent_detail.html", {"detail": detail, "tab": tab, "blockers": agent_blockers, "blocker_breakdown": blocker_filter_summary(session, tenant_id=installation.tenant_id, agent_id=agent_id, status="OPEN", scope="all")["blocker_breakdown"], "jobs": list_jobs(session, agent_id=agent_id, page=1, page_size=25, scope="all"), "events": events, "event_page": max(1, event_page), "has_more": has_more, "severity": severity or "", "event_type": event_type or "", "job_id": job_id or "", "product_code": product_code or "", "since": since or "", "until": until or "", "event_query_base": query_base(tab=tab, severity=severity, event_type=event_type, job_id=job_id, product_code=product_code, since=since, until=until), "auth": auth_data, "csrf_token": request.cookies.get("gti_csrf", ""), "ops": _ops_view(session, installation, agent), "features": _features_view(session, installation, enrich_code), "enrich_code": enrich_code or "", "is_admin": _is_privileged(auth_data[1]), "fiscal_context": fiscal_context, "cfop_ippt_write_mode": installation.cfop_ippt_write_mode}) finally: session.close() def _features_view(session, installation, enrich_code=None): @@ -372,6 +372,33 @@ def router(session_factory): return RedirectResponse(f"/agents/{agent_id}#contexto-fiscal", status_code=303) finally: session.close() + @routes.post("/agents/{agent_id}/cfop-ippt-write") + async def set_cfop_ippt_write_mode(request: Request, agent_id: int): + """Missão 4 / Bloco E.1: per-installation toggle for the CFOP_IPPT_WRITE_V1 capability + (ADMIN only, CSRF, audited). DISABLED by default; the Agent has no local switch.""" + from .cfop_ippt_audit import WRITE_MODES as CFOP_IPPT_MODES + session, auth_data, redirect = page_auth(request) + if redirect: return redirect + try: + values = await form_values(request); web_session, admin = auth_data + require_csrf(web_session, values.get("csrf_token")) + agent, installation = _rtc_target(session, agent_id) + if not agent or not installation: return HTMLResponse("Agent not found", status_code=404) + if not _is_privileged(admin): + return _deny(session, admin, installation.tenant_id, "cfop_ippt_write:set", installation.installation_id) + mode = (values.get("mode") or "").strip().upper() + if mode not in CFOP_IPPT_MODES: + return HTMLResponse(f"modo inválido (use um de {CFOP_IPPT_MODES})", status_code=422) + if mode != "DISABLED" and (values.get("confirm") or "").strip().upper() != "GRAVAR": + return HTMLResponse("Digite GRAVAR para confirmar um modo que pode escrever no ERP", status_code=422) + before = installation.cfop_ippt_write_mode + installation.cfop_ippt_write_mode = mode + record_audit(session, admin_user_id=admin.id, tenant_id=installation.tenant_id, action="CFOP_IPPT_WRITE_MODE", target_type="installation", target_id=installation.installation_id, + context={"from": before, "to": mode, "operator": admin.display_name or admin.username, "role": admin.external_role or admin.auth_source}, severity="WARNING") + session.commit() + return RedirectResponse(f"/agents/{agent_id}#contexto-fiscal", status_code=303) + finally: session.close() + @routes.post("/agents/{agent_id}/enrichment/reevaluate") async def reevaluate_enrichment(request: Request, agent_id: int): """Explicit, idempotent 'evaluate again' for one product (code=...) or every tracked product that is not enriched yet. KB/cache first; providers only when really needed.""" diff --git a/central/templates/agent_detail.html b/central/templates/agent_detail.html index 36f285a..ac61c2f 100644 --- a/central/templates/agent_detail.html +++ b/central/templates/agent_detail.html @@ -22,7 +22,13 @@ -{% endfor %}{% endif %} +{% endfor %} +{% if is_admin %}

Escrita CFOP/IPPT (Missão 4)

{{ cfop_ippt_write_mode }}
+

Capability CFOP_IPPT_WRITE_V1: escreve apenas C000025.CFOP e C000025.IPPT, só as células que o motor já classificou AUTO_FILL_SAFE. Desativada por padrão em toda instalação nova.

+
+ + +
{% endif %}{% endif %}

Funcionalidades (Enriquecimento de produtos novos)

Decididas no Hub para esta instalação — o Agent não tem chave local. O Enriquecimento de produtos novos NÃO é o preenchimento RTC, que tem controle próprio (controle e candidatos).

{% if features.waiting %}

Aguardando configuração de funcionalidades: esta instalação ainda não processa nada.

{% endif %}
Corretor fiscal
{{ "Ativo" if features.corrector_enabled else "Desativado" }}
Enriquecimento de produtos novos
{{ features.state_label }}
diff --git a/central/tests/test_cfop_ippt_write_plumbing.py b/central/tests/test_cfop_ippt_write_plumbing.py new file mode 100644 index 0000000..8762501 --- /dev/null +++ b/central/tests/test_cfop_ippt_write_plumbing.py @@ -0,0 +1,105 @@ +"""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" diff --git a/central/tests/test_rtc_security.py b/central/tests/test_rtc_security.py index b7c27f3..6993331 100644 --- a/central/tests/test_rtc_security.py +++ b/central/tests/test_rtc_security.py @@ -96,7 +96,7 @@ def test_agent_can_reread_only_its_own_current_authorization_and_it_follows_the_ s.scalars(select(Installation)).one().rtc_write_mode = "ENABLED" s.commit() body = client.get("/v1/agents/me/authorizations", headers=H).json() - assert body["rtc_write_mode"] == "ENABLED" and set(body) == {"rtc_write_mode", "rtc_approved_codes", "rtc_rollbacks", "features"} + assert body["rtc_write_mode"] == "ENABLED" and set(body) == {"rtc_write_mode", "rtc_approved_codes", "rtc_rollbacks", "features", "cfop_ippt_write_mode"} def _become(client, role): diff --git a/central/tests/test_zero_touch_hub.py b/central/tests/test_zero_touch_hub.py index 9d877fd..0a931ec 100644 --- a/central/tests/test_zero_touch_hub.py +++ b/central/tests/test_zero_touch_hub.py @@ -280,7 +280,7 @@ def test_job_carries_the_central_rtc_authorization_default_off_and_follows_the_i client, engine = _client() hb(client) job = client.get("/v1/jobs/next", headers=H).json() - assert {k: v for k, v in job["authorizations"].items() if k != "features"} == {"rtc_write_mode": "DISABLED", "rtc_approved_codes": [], "rtc_rollbacks": []} # never on by default + assert {k: v for k, v in job["authorizations"].items() if k != "features"} == {"rtc_write_mode": "DISABLED", "rtc_approved_codes": [], "rtc_rollbacks": [], "cfop_ippt_write_mode": "DISABLED"} # never on by default client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": {}}, headers=H) with Session(engine) as s: inst = s.scalars(select(Installation)).one() diff --git a/release/RELEASE_MANIFEST.json b/release/RELEASE_MANIFEST.json index 0a500b8..b978018 100644 --- a/release/RELEASE_MANIFEST.json +++ b/release/RELEASE_MANIFEST.json @@ -19,7 +19,7 @@ "migrations": { "alembic_head_in_repo": "0032_contribuinte_ipi", "applied_revision": "0032_contribuinte_ipi", - "applied_revision_source": "confirmed live via docker exec alembic current on gti-fiscal-hub-fiscal-hub-1 during the 1.3.7 deploy; migration 0032 did not change again in 1.3.8/1.3.9", + "applied_revision_source": "confirmed live via docker exec alembic current on gti-fiscal-hub-fiscal-hub-1 during the 1.3.7 deploy; migration 0032 did not change again in 1.3.8/1.3.9/1.3.10/1.3.11 (production, branch master). This manifest describes master/production; it is intentionally NOT updated on feature/cfop-ippt-write, which adds an unmerged, undeployed migration 0033_cfop_ippt_write_mode - test_manifest_records_the_real_migration_head fails on that branch only, for that expected reason (see docs/CFOP_IPPT_WRITE_CAPABILITY_V1.md and the Missão 4 report).", "count": 32 }, "datasets_active": [