gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
f34f001ed5676e033c6541c2b5bf921725374198
GTI_GERENCIADOR_FISCAL/central/tests/test_zero_touch_hub.py
T
gti 7e83d17533 feat: wire field_contract into product enrichment; auto-discover CRT/CONTRIBUINTE_IPI at heartbeat; targeted per-product ERP evidence 
field_contract.resolve_product was only exercised by its own unit tests. /v1/products/resolve now calls it for every product whose NCM
resolves, merging CFOP/CST/CSOSN/PIS-COFINS/CST_IPI/SITUACAO_TRIBUTARIA/IAT/IPPT decisions alongside CLASSIFICACAO_FISCAL/CEST/IPI - reusing
the same engine, the same FiscalContextEntry table the blocker-resolution workflow already writes to, no duplicated rules.

Closes three real transport/discovery gaps found while wiring it in:
- Agent heartbeat never sent company_crt (env-var UF only) or company_contribuinte_ipi, so Installation.company_crt stayed permanently empty
  even though company_context.resolve() already proves both from C000004/EMISSOR for every coverage job. Resolved once, cached, retried only
  while unresolved. Migration 0032 adds Installation.company_contribuinte_ipi (NULL=unknown, never coerced to "N").
- The Agent's enrichment payload only ever reported 3 old fields (CLASSIFICACAO_FISCAL/CEST/IPI) as "current", so the Hub could never see the
  product's own CFOP/CST/CSOSN/... nor the classification columns (TIPO/PROD_COZINHA/PRODUTO_SERVICO/CODGRUPO) infer_role needs - separated
  into a read-only CURRENT_REPORT_FIELDS list, the write allowlist (PRODUCT_ENRICHMENT_WRITE_V1) untouched.
- Added a targeted, product_code-scoped evidence query (firebird_reader.read_product_evidence: recipes/composed/purchases/stock, indexed WHERE,
  never a full-table scan) so st_withheld/product_role can be proven for a single ad-hoc resolve, not only during a full sync job.

Verified read-only against the real database (D:\PROJETOS\GTISISTEMAS\Banco\GTI.FDB): CRT=1, CONTRIBUINTE_IPI=N, company_context.resolve()
reaches HIGH confidence; infer_role correctly resolved RESALE/OWN_PRODUCTION for 5/8 sample real products with zero operator questions.

1121 tests passed (tests + central/tests + agent/tests), 0 failed. Version 1.3.7 (single GTI_FISCAL_VERSION shared by Hub and Agent per
central/app/version.py's own design - never independently versioned).
2026-09-27 18:26:20 -03:00

378 lines
24 KiB
Python
Raw Blame History

"""Zero-touch control plane: central scheduler (INITIAL_SYNC once, SYNC_INCREMENTAL
per interval), single-job concurrency, central pause/resume, coverage snapshot from
the job result, idempotency. Everything is Hub-side — no local toggle involved."""
from datetime import datetime, timedelta, timezone
from sqlalchemy import select
from sqlalchemy.orm import Session
from central.app.jobs import claim_next_job, create_job, recover_stale_running
from central.app.models import Agent, AgentSnapshot, AuditEvent, Installation, Job
from central.tests.test_jobs import _client
H = {"Authorization": "Bearer test-token"}
INST = "00000000-0000-4000-8000-0000000000aa"
OPERATIONAL = {"COMPLETE": 480, "CORRECTED": 0, "BLOCKED_MISSING_EVIDENCE": 31, "BLOCKED_CONFLICT": 10, "BLOCKED_COMPANY_CONTEXT": 0, "BLOCKED_SCHEMA": 0,
"FAILED_TECHNICAL": 0, "TOTAL_PRODUCTS": 521, "SAFE_BLOCKED": 41}
def hb(client):
return client.post("/v1/agents/heartbeat", json={"installation_id": INST, "status": "IDLE", "agent_version": "1.1.3"}, headers=H)
def jobs(engine, kind=None):
with Session(engine) as s:
q = select(Job).order_by(Job.id)
if kind: q = q.where(Job.kind == kind)
return s.scalars(q).all()
def finish(client, engine, summary=None):
job = client.get("/v1/jobs/next", headers=H).json()
assert job, "expected a claimable job"
body = {"result_summary": summary or {"operational": OPERATIONAL, "products_processed": 521, "products_changed": 521, "blocked_products": [], "resolved_products": []}}
assert client.post(f"/v1/jobs/{job['job_id']}/complete", json=body, headers=H).status_code == 200
return job
def age_last_sync(engine, seconds):
with Session(engine) as s:
for job in s.scalars(select(Job)).all():
job.finished_at = datetime.now(timezone.utc) - timedelta(seconds=seconds)
job.created_at = datetime.now(timezone.utc) - timedelta(seconds=seconds)
s.commit()
def test_first_heartbeat_creates_exactly_one_initial_sync():
client, engine = _client()
body = hb(client).json()
assert body["processing_enabled"] is True and body["sync_interval_seconds"] == 1800
for _ in range(4):
hb(client)
initial = jobs(engine, "INITIAL_SYNC")
assert len(initial) == 1 and initial[0].status == "PENDING" and initial[0].payload["reason"] == "INITIAL_SYNC"
def test_heartbeat_metadata_stores_auto_discovered_company_crt_and_contribuinte_ipi():
"""company_crt and company_contribuinte_ipi were never sent at heartbeat before (a real transport gap); once the Agent reports them
(auto-discovered from C000004, never guessed), the Hub stores them on Installation - fiscal_completion.build_company reads them directly."""
client, engine = _client()
client.post("/v1/agents/heartbeat", json={"installation_id": INST, "status": "IDLE", "agent_version": "1.1.3", "company_uf": "BA", "company_crt": "1", "company_contribuinte_ipi": "S"}, headers=H)
with Session(engine) as s:
inst = s.scalar(select(Installation))
assert inst.company_uf == "BA" and inst.company_crt == "1" and inst.company_contribuinte_ipi == "S"
def test_initial_sync_completion_sets_coverage_and_invariant():
client, engine = _client()
hb(client); finish(client, engine)
with Session(engine) as s:
snap = s.scalar(select(AgentSnapshot)); inst = s.scalar(select(Installation))
assert (snap.total_products, snap.fiscally_complete, snap.safe_blocked, snap.failed_products) == (521, 480, 41, 0)
assert snap.fiscally_complete + snap.safe_blocked + snap.failed_products == snap.total_products # coverage invariant closes
assert snap.blockers == {"BLOCKED_MISSING_EVIDENCE": 31, "BLOCKED_CONFLICT": 10}
assert inst.last_sync_at is not None and inst.last_sync_changed == 521
for _ in range(3):
hb(client)
assert len(jobs(engine)) == 1 # no second initial sync, and not due for an incremental yet
def test_incremental_sync_scheduled_once_per_interval():
client, engine = _client()
hb(client); finish(client, engine)
age_last_sync(engine, 1801)
hb(client); hb(client); hb(client)
inc = jobs(engine, "SYNC_INCREMENTAL")
assert len(inc) == 1 and inc[0].payload["reason"] == "SCHEDULED"
assert not jobs(engine, "REPROCESS_ALL") # full reprocess never runs automatically
def test_central_pause_stops_scheduling_and_claiming_but_heartbeat_continues():
client, engine = _client()
hb(client)
with Session(engine) as s:
s.scalar(select(Installation)).processing_enabled = False; s.commit()
body = hb(client).json()
assert body["processing_enabled"] is False and body["enabled"] is True # heartbeat keeps working
assert client.get("/v1/jobs/next", headers=H).json() is None # the PENDING initial sync is NOT handed out while paused
with Session(engine) as s:
s.scalar(select(Installation)).processing_enabled = True; s.commit()
assert client.get("/v1/jobs/next", headers=H).json()["kind"] == "INITIAL_SYNC" # resume: jobs flow again
def test_only_one_job_runs_at_a_time_and_queue_is_fifo():
client, engine = _client()
hb(client)
with Session(engine) as s:
agent = s.scalar(select(Agent))
create_job(s, agent, "REPROCESS_PRODUCTS", {"codigos": ["1"]})
first = client.get("/v1/jobs/next", headers=H).json()
assert first["kind"] == "INITIAL_SYNC"
assert client.get("/v1/jobs/next", headers=H).json() is None # second job waits: one fiscal job per Firebird
client.post(f"/v1/jobs/{first['job_id']}/complete", json={"result_summary": {}}, headers=H)
assert client.get("/v1/jobs/next", headers=H).json()["kind"] == "REPROCESS_PRODUCTS"
def test_reprocess_all_and_sync_now_are_not_duplicated():
client, engine = _client()
hb(client); finish(client, engine)
with Session(engine) as s:
agent = s.scalar(select(Agent))
a = create_job(s, agent, "SYNC_INCREMENTAL", {"reason": "MANUAL_SYNC_NOW"}); b = create_job(s, agent, "SYNC_INCREMENTAL")
assert a.id == b.id # double click = one job
full1 = create_job(s, agent, "REPROCESS_ALL"); full2 = create_job(s, agent, "REPROCESS_ALL")
assert full1.id == full2.id and full1.id != a.id # a full reprocess still queues behind the incremental, never twice
assert create_job(s, agent, "SYNC_INCREMENTAL").id == a.id
def test_stale_running_job_does_not_block_the_queue_forever():
client, engine = _client()
hb(client)
job = client.get("/v1/jobs/next", headers=H).json()
with Session(engine) as s:
row = s.get(Job, job["job_id"]); row.started_at = datetime.now(timezone.utc) - timedelta(hours=2); s.commit()
create_job(s, s.scalar(select(Agent)), "REPROCESS_PRODUCTS", {"codigos": ["1"]})
assert client.get("/v1/jobs/next", headers=H).json()["kind"] == "REPROCESS_PRODUCTS"
with Session(engine) as s:
assert s.get(Job, job["job_id"]).status == "FAILED"
def test_job_events_recorded_without_secrets():
client, engine = _client()
hb(client); finish(client, engine)
with Session(engine) as s:
from central.app.models import AgentEvent
types = {e.event_type for e in s.scalars(select(AgentEvent)).all()}
assert {"HEARTBEAT", "JOB_STARTED", "JOB_COMPLETED"} <= types
def test_failed_job_records_failure_and_does_not_reschedule_immediately():
client, engine = _client()
hb(client)
job = client.get("/v1/jobs/next", headers=H).json()
client.post(f"/v1/jobs/{job['job_id']}/fail", json={"error": "Firebird offline"}, headers=H)
for _ in range(3):
hb(client)
assert len(jobs(engine)) == 1 # backoff = the interval; no error loop against a broken Firebird
with Session(engine) as s:
assert s.scalar(select(AgentSnapshot)).last_job_status == "FAILED"
# ---------- panel: central control buttons + status ----------
def _panel():
from central.tests.test_control_panel_pages import _panel_client
client, agent_id, _ = _panel_client()
return client, agent_id, client._session_factory
def test_panel_shows_processing_status_and_the_operational_buttons():
client, agent_id, _ = _panel()
body = client.get(f"/agents/{agent_id}").text
for label in ("Sincronizar agora", "Reprocessar tudo", "Pausar processamento", "Última sincronização", "Sincronização agendada"):
assert label in body
assert "Ativo" in body and "a cada 30 min" in body and "Retomar processamento" not in body
def test_panel_sync_now_pause_resume_are_csrf_protected_audited_and_idempotent():
client, agent_id, factory = _panel()
csrf = client.cookies.get("gti_csrf")
assert client.post(f"/agents/{agent_id}/sync-now", data={"csrf_token": "wrong"}).status_code == 403
assert client.post(f"/agents/{agent_id}/sync-now", data={"csrf_token": csrf}, follow_redirects=False).status_code == 303
assert client.post(f"/agents/{agent_id}/sync-now", data={"csrf_token": csrf}, follow_redirects=False).status_code == 303 # double click
with factory() as s:
incs = s.scalars(select(Job).where(Job.kind == "SYNC_INCREMENTAL")).all()
assert len(incs) == 1 and incs[0].payload["reason"] == "MANUAL_SYNC_NOW"
client.post(f"/agents/{agent_id}/pause", data={"csrf_token": csrf})
with factory() as s:
assert s.scalar(select(Installation).where(Installation.agent_id == agent_id)).processing_enabled is False
paused = client.get(f"/agents/{agent_id}").text
assert "Pausado" in paused and "Retomar processamento" in paused and "disabled" in paused # Sync-now disabled while paused
client.post(f"/agents/{agent_id}/resume", data={"csrf_token": csrf})
with factory() as s:
assert s.scalar(select(Installation).where(Installation.agent_id == agent_id)).processing_enabled is True
actions = [a.action for a in s.scalars(select(AuditEvent)).all()]
assert {"SYNC_NOW", "PROCESSING_PAUSE", "PROCESSING_RESUME"} <= set(actions)
def test_panel_controls_require_login():
from fastapi.testclient import TestClient
from central.app.main import create_app
from central.tests.test_control_plane import seed, setup_db
factory = setup_db(); agent_id, _ = seed(factory)
anon = TestClient(create_app(database_probe=lambda: True, session_factory=factory, bootstrap_token=None))
for action in ("sync-now", "pause", "resume"):
assert anon.post(f"/agents/{agent_id}/{action}", data={}, follow_redirects=False).status_code == 303 # bounced to login, nothing executed
def _fiscal_blocker(code, field, reason, fp):
return {"company": "", "product_code": code, "gtin": None, "description": "P" + code, "blocker_type": "BLOCKED_FISCAL_DATA", "field": field, "reason_code": reason, "evidence_summary": {}, "input_fingerprint": fp}
def test_fiscal_blockers_share_the_single_queue_and_are_resolved_by_the_next_full_pass():
"""PIS/COFINS/RTC blockers are ordinary blockers; when a later FULL pass no longer reports them (data fixed / RTC switched off) they close automatically."""
from central.app.models import BlockedProduct
client, engine = _client()
hb(client)
operational = {**OPERATIONAL, "COMPLETE": 519, "BLOCKED_MISSING_EVIDENCE": 0, "BLOCKED_CONFLICT": 0, "BLOCKED_FISCAL_DATA": 2, "SAFE_BLOCKED": 2, "TOTAL_PRODUCTS": 521}
first = {"operational": operational, "blocked_products": [_fiscal_blocker("10", "PIS", "PIS_CST_INVALID", "a"), _fiscal_blocker("11", "RTC", "CCLASSTRIB_MISSING", "b")], "resolved_products": [],
"fiscal_context": {"rtc_enabled": True, "evaluated_all": True}}
finish(client, engine, first)
with Session(engine) as s:
rows = s.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_(None))).all()
assert sorted((r.product_code, r.field) for r in rows) == [("10", "PIS"), ("11", "RTC")]
snap = s.scalars(select(AgentSnapshot)).one()
assert (snap.fiscally_complete, snap.safe_blocked, snap.total_products) == (519, 2, 521)
age_last_sync(engine, 4000)
hb(client)
second = {"operational": {**OPERATIONAL, "COMPLETE": 520, "BLOCKED_MISSING_EVIDENCE": 0, "BLOCKED_CONFLICT": 0, "BLOCKED_FISCAL_DATA": 1, "SAFE_BLOCKED": 1},
"blocked_products": [_fiscal_blocker("10", "PIS", "PIS_CST_INVALID", "a")], "resolved_products": [], "fiscal_context": {"rtc_enabled": False, "evaluated_all": True}}
finish(client, engine, second)
with Session(engine) as s:
open_rows = s.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_(None))).all()
assert [(r.product_code, r.field) for r in open_rows] == [("10", "PIS")] # RTC blocker closed because RTC is off now
closed = s.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_not(None))).all()
assert [(r.product_code, r.resolution_source) for r in closed] == [("11", "AUTO_REEVALUATION")]
def test_partial_reprocess_never_closes_fiscal_blockers_of_other_products():
from central.app.models import BlockedProduct
client, engine = _client()
hb(client)
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [_fiscal_blocker("10", "PIS", "PIS_CST_INVALID", "a")], "resolved_products": [], "fiscal_context": {"evaluated_all": True}})
age_last_sync(engine, 4000)
hb(client)
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "fiscal_context": {"evaluated_all": False}})
with Session(engine) as s:
assert len(s.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_(None))).all()) == 1
def test_rtc_on_then_off_and_pis_fix_leave_no_stale_open_blockers_and_never_duplicate():
from central.app.models import BlockedProduct
client, engine = _client()
hb(client)
rtc_missing = _fiscal_blocker("20", "RTC", "CCLASSTRIB_MISSING", "r1")
pis_invalid = _fiscal_blocker("21", "PIS", "PIS_CST_INVALID", "p1")
op = {**OPERATIONAL, "BLOCKED_MISSING_EVIDENCE": 0, "BLOCKED_CONFLICT": 0}
finish(client, engine, {"operational": {**op, "COMPLETE": 519, "BLOCKED_FISCAL_DATA": 2, "SAFE_BLOCKED": 2}, "blocked_products": [rtc_missing, pis_invalid, dict(rtc_missing)], "resolved_products": [],
"fiscal_context": {"rtc_enabled": True, "evaluated_all": True}})
with Session(engine) as s:
assert len(s.scalars(select(BlockedProduct)).all()) == 2 # the repeated blocker is one row
age_last_sync(engine, 4000)
hb(client)
# RTC switched off in the ERP AND the PIS value was normalised/fixed: the next full pass reports nothing
finish(client, engine, {"operational": {**op, "COMPLETE": 521}, "blocked_products": [], "resolved_products": [], "fiscal_context": {"rtc_enabled": False, "evaluated_all": True}})
with Session(engine) as s:
rows = s.scalars(select(BlockedProduct)).all()
assert [r.resolved_at is not None for r in rows] == [True, True] and {r.resolution_source for r in rows} == {"AUTO_REEVALUATION"}
assert s.scalars(select(BlockedProduct).where(BlockedProduct.resolved_at.is_(None))).all() == []
age_last_sync(engine, 4000)
hb(client)
# RTC comes back on: the SAME blocker (same fingerprint) reopens instead of being duplicated
finish(client, engine, {"operational": {**op, "COMPLETE": 520, "BLOCKED_FISCAL_DATA": 1, "SAFE_BLOCKED": 1}, "blocked_products": [rtc_missing], "resolved_products": [], "fiscal_context": {"rtc_enabled": True, "evaluated_all": True}})
with Session(engine) as s:
rows = s.scalars(select(BlockedProduct)).all()
assert len(rows) == 2 and [(r.product_code, r.resolved_at is None) for r in rows] == [("20", True), ("21", False)]
def test_job_carries_the_central_rtc_authorization_default_off_and_follows_the_installation():
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
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()
inst.rtc_write_mode = "ENABLED"
s.commit()
age_last_sync(engine, 4000)
hb(client)
assert client.get("/v1/jobs/next", headers=H).json()["authorizations"]["rtc_write_mode"] == "ENABLED"
def test_rtc_write_audits_are_persisted_with_server_side_identity_and_never_duplicated_per_job():
from central.app.models import AgentEvent, RtcWriteAudit
client, engine = _client()
hb(client)
applied = {"outcome": "APPLIED", "product_code": "10", "field": "CCLASSTRIB", "old_value": None, "new_value": "200003", "rule_id": "RTC-R3", "evidence": {"reason": "ANNEX_I_CLEAN_ITEM_UNIQUE_DEFAULT_CLASS"},
"official_source_version": {"cclasstrib_sha256": "ab" * 32}, "local_catalog_hash": "cd" * 32, "timestamp": "2026-09-24T20:00:00+00:00", "job_id": 999, "tenant_id": 12345, "agent_id": 777}
aborted = {"outcome": "ABORTED", "product_code": "11", "field": "CCLASSTRIB", "reason": "RTC_NO_LONGER_ENABLED", "rule_id": "RTC-R4", "timestamp": "2026-09-24T20:00:01+00:00"}
other_field = {"outcome": "APPLIED", "product_code": "12", "field": "IPI", "new_value": "1.0"} # anything but CCLASSTRIB is refused
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_auto_apply": {"mode": "AUTO_APPLY", "applied": [applied, other_field], "aborted": [aborted], "aborted_reasons": {"RTC_NO_LONGER_ENABLED": 1}}})
with Session(engine) as s:
rows = s.scalars(select(RtcWriteAudit).order_by(RtcWriteAudit.id)).all()
assert [(r.product_code, r.outcome, r.old_value, r.new_value, r.rule_id) for r in rows] == [("10", "APPLIED", None, "200003", "RTC-R3"), ("11", "ABORTED", None, None, "RTC-R4")]
inst = s.scalars(select(Installation)).one()
assert all(r.installation_id == inst.id and r.agent_id == inst.agent_id and r.tenant_id == inst.tenant_id for r in rows) # identity comes from the authenticated context, not the payload
assert rows[0].local_catalog_hash == "cd" * 32 and rows[0].official_source_version["cclasstrib_sha256"] == "ab" * 32 and rows[0].job_id != 999
events = [e for e in s.scalars(select(AgentEvent)).all() if e.event_type == "RTC_AUTO_APPLY"]
assert len(events) == 1 and events[0].structured_context["applied"] == 1 and events[0].severity == "WARNING"
def test_report_only_jobs_leave_no_audit_rows():
from central.app.models import RtcWriteAudit
client, engine = _client()
hb(client)
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_auto_apply": {"mode": "REPORT_ONLY", "applied": [], "aborted": [], "ready_total": 280}})
with Session(engine) as s:
assert s.scalars(select(RtcWriteAudit)).all() == []
def _set_mode(engine, mode):
with Session(engine) as s:
s.scalars(select(Installation)).one().rtc_write_mode = mode
s.commit()
def test_preview_is_stored_only_when_the_central_mode_wants_it_and_approvals_travel_in_the_next_job():
from central.app.models import RtcCandidate
client, engine = _client()
hb(client)
preview = [{"product_code": f"P{i}", "description": f"ITEM {i}", "ncm": "10063021", "current": "", "proposed": "200003", "cst": "200", "rule_id": "RTC-R3", "basis": "RTC-R3:ANNEX"} for i in range(3)]
ctx = {"rtc_enabled": True, "evaluated_all": True}
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "fiscal_context": ctx, "rtc_auto_apply": {"mode": "REPORT_ONLY", "preview": preview, "preview_truncated": False}})
with Session(engine) as s:
assert s.scalars(select(RtcCandidate)).all() == [] # DISABLED: nothing is stored
_set_mode(engine, "APPROVAL_REQUIRED")
age_last_sync(engine, 4000); hb(client)
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "fiscal_context": ctx, "rtc_auto_apply": {"mode": "APPROVAL_REQUIRED", "preview": preview, "preview_truncated": False}})
with Session(engine) as s:
rows = s.scalars(select(RtcCandidate).order_by(RtcCandidate.product_code)).all()
assert [(r.product_code, r.status, r.proposed) for r in rows] == [("P0", "PENDING", "200003"), ("P1", "PENDING", "200003"), ("P2", "PENDING", "200003")]
rows[1].status = "APPROVED"
s.commit()
age_last_sync(engine, 4000); hb(client)
job = client.get("/v1/jobs/next", headers=H).json()
assert job["authorizations"]["rtc_approved_codes"] == ["P1"] # only what a human approved
client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "fiscal_context": ctx,
"rtc_auto_apply": {"mode": "APPROVAL_REQUIRED", "preview": [p for p in preview if p["product_code"] != "P2"], "preview_truncated": False,
"applied": [{"outcome": "APPLIED", "product_code": "P1", "field": "CCLASSTRIB", "old_value": None, "new_value": "200003", "rule_id": "RTC-R3", "action": "APPLY", "rule_version": "rtc-rules-v2",
"company_crt": "3", "authorization_origin": "HUB_MODE:APPROVAL_REQUIRED", "description_snapshot": "ITEM 1", "rtc_status": "ENABLED", "local_catalog_state": "MATCH_SAFE"}]}}}, headers=H)
with Session(engine) as s:
status = {r.product_code: r.status for r in s.scalars(select(RtcCandidate)).all()}
assert status == {"P0": "PENDING", "P1": "APPLIED", "P2": "STALE"} # applied is recorded; a product no longer eligible on a full evaluation becomes STALE
from central.app.models import RtcWriteAudit
audit = s.scalars(select(RtcWriteAudit)).one()
assert (audit.rule_version, audit.company_crt, audit.authorization_origin, audit.description_snapshot, audit.action, audit.local_catalog_state) == ("rtc-rules-v2", "3", "HUB_MODE:APPROVAL_REQUIRED", "ITEM 1", "APPLY", "MATCH_SAFE")
assert audit.installation_uuid
def test_rollback_request_travels_in_the_job_and_its_result_closes_the_original_audit():
from central.app.models import RtcWriteAudit
client, engine = _client()
hb(client)
applied = {"outcome": "APPLIED", "product_code": "P1", "field": "CCLASSTRIB", "old_value": None, "new_value": "200003", "rule_id": "RTC-R3", "action": "APPLY"}
finish(client, engine, {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_auto_apply": {"applied": [applied]}})
with Session(engine) as s:
audit = s.scalars(select(RtcWriteAudit)).one(); audit.rollback_state = "REQUESTED"; audit_id = audit.id
s.commit()
age_last_sync(engine, 4000); hb(client)
job = client.get("/v1/jobs/next", headers=H).json()
assert job["authorizations"]["rtc_rollbacks"] == [{"audit_id": audit_id, "product_code": "P1", "written_value": "200003", "rule_id": "RTC-R3"}]
done = {"outcome": "ROLLED_BACK", "product_code": "P1", "field": "CCLASSTRIB", "old_value": "200003", "new_value": None, "rule_id": "RTC-R3", "action": "ROLLBACK", "rollback_of": audit_id}
client.post(f"/v1/jobs/{job['job_id']}/complete", json={"result_summary": {"operational": OPERATIONAL, "blocked_products": [], "resolved_products": [], "rtc_auto_apply": {"rollbacks": [done]}}}, headers=H)
with Session(engine) as s:
rows = s.scalars(select(RtcWriteAudit).order_by(RtcWriteAudit.id)).all()
assert [(r.action, r.outcome, r.rollback_state, r.rollback_of) for r in rows] == [("APPLY", "APPLIED", "DONE", None), ("ROLLBACK", "ROLLED_BACK", None, audit_id)]
Reference in New Issue View Git Blame Copy Permalink