gustavosantos/GTI_GERENCIADOR_FISCAL
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: re-check central RTC authorization right before every write (GET /v1/agents/me/authorizations), human labels on the cClassTrib control page, final corpus/scale evidence; 1.2.1

Browse Source 
Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
gti
2026-09-25 08:57:45 -03:00
co-authored by Claude Sonnet 5
parent 42dc172763
commit 7063c8c161
39 changed files with 4231 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent/tests/test_rtc_auto_apply.py
+38
View File
@@ -254,6 +254,16 @@ def test_catalog_hash_is_recorded_for_audit():
# ---------------------------------------------------------------- job integration (gates)
import pytest # noqa: E402
@pytest.fixture(autouse=True)
def _hub_echoes_the_job_authorization(monkeypatch):
"""By default the Hub still says what the job said; individual tests change that to prove the pre-write re-check."""
from test_run_agent_service import service
monkeypatch.setattr(service, "_fetch_authorizations", lambda config, job: dict(job.get("authorizations") or {}))
def _service():
from test_run_agent_service import _config, service
return service, _config
@@ -399,3 +409,31 @@ def test_connection_is_closed_even_when_a_product_fails_mid_batch(monkeypatch):
monkeypatch.setattr(service, "open_write_connection", lambda env, path: conn)
summary = service._rtc_auto_apply({}, _config(), {"job_id": 1, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
assert closed == [True] and summary["write_successes"] == 2 and summary["aborted_reasons"]
def test_authorization_revoked_after_the_job_was_claimed_means_zero_writes(monkeypatch):
service, _config = _service()
db = FakeFirebird(PRODUCTS)
fiscal = _fiscal_for(db)
monkeypatch.setattr(service, "open_write_connection", lambda env, path: _Closeable(db))
monkeypatch.setattr(service, "_fetch_authorizations", lambda config, job: {"rtc_write_mode": "DISABLED"}) # operator switched it off while the job was running
summary = service._rtc_auto_apply({}, _config(), {"job_id": 5, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
assert summary["write_successes"] == 0 and summary["aborted_reasons"] == {"AUTHORIZATION_CHANGED": 3}
assert [db.products[c]["CCLASSTRIB"] for c in ("000001", "000002", "000003")] == [None, None, None]
def test_unreachable_hub_is_not_authorization_and_approval_removal_is_honoured(monkeypatch):
service, _config = _service()
db = FakeFirebird(PRODUCTS)
fiscal = _fiscal_for(db)
monkeypatch.setattr(service, "open_write_connection", lambda env, path: _Closeable(db))
def down(config, job):
raise ConnectionError("hub down")
monkeypatch.setattr(service, "_fetch_authorizations", down)
summary = service._rtc_auto_apply({}, _config(), {"job_id": 6, "authorizations": {"rtc_write_mode": "ENABLED"}}, _products_of(db), fiscal, NORMAL)
assert summary["write_successes"] == 0 and summary["aborted_reasons"] == {"AUTHORIZATION_UNVERIFIABLE": 3}
monkeypatch.setattr(service, "_fetch_authorizations", lambda config, job: {"rtc_write_mode": "APPROVAL_REQUIRED", "rtc_approved_codes": ["000001"]}) # 000002 was un-approved meanwhile
summary = service._rtc_auto_apply({}, _config(), {"job_id": 7, "authorizations": {"rtc_write_mode": "APPROVAL_REQUIRED", "rtc_approved_codes": ["000001", "000002"]}}, _products_of(db), fiscal, NORMAL)
assert summary["write_successes"] == 1 and db.products["000001"]["CCLASSTRIB"] == "200003" and db.products["000002"]["CCLASSTRIB"] is None