149 lines
8.0 KiB
Python
149 lines
8.0 KiB
Python
"""Agent side of the zero-touch model: delta sync via fingerprint cache, read-only guarantee,
|
|
blockers/operational summary in the job result, central pause, Firebird credential resolution."""
|
|
import json
|
|
import time
|
|
|
|
import pytest
|
|
|
|
from test_run_agent_service import AUTHORIZED, _FakeExecutor, _config, service
|
|
|
|
|
|
class FakeHub:
|
|
def __init__(self):
|
|
self.classified = []
|
|
|
|
def classify_batch(self, products):
|
|
self.classified.append([p["product_id"] for p in products])
|
|
return [{"status": "EXACT_NCM" if p["current_ncm"] else "INVALID_NCM", "ipi_status": "IPI_RESOLVED", "classification": {}} for p in products]
|
|
|
|
|
|
def product(code, ncm="03055100", name="X"):
|
|
return {"CODIGO": code, "PRODUTO": name, "EAN_GTIN": None, "CLASSIFICACAO_FISCAL": ncm, "NCM": None, "CFOP": "5102", "CST": None, "ALIQUOTA": None, "CEST": None, "CSOSN": "102", "IPI": None}
|
|
|
|
|
|
@pytest.fixture
|
|
def env_setup(monkeypatch, tmp_path):
|
|
hub = FakeHub()
|
|
monkeypatch.setattr(service, "HubClient", lambda *a, **k: hub)
|
|
monkeypatch.setattr(service, "build_job_coverage", lambda env, client, products, decisions: {"total_products": len(products)})
|
|
state = service.AgentState(tmp_path / "s.db")
|
|
return hub, state
|
|
|
|
|
|
def run_job(kind, products, state, monkeypatch, config=None):
|
|
monkeypatch.setattr(service, "read_products", lambda env, codigos=None: products)
|
|
return service.execute_job({"job_id": 1, "kind": kind, "agent_id": 2, "payload": {}}, config or _config(), _FakeExecutor(), state, {})
|
|
|
|
|
|
def test_initial_sync_processes_everything_and_reports_operational_summary(env_setup, monkeypatch):
|
|
hub, state = env_setup
|
|
products = [product("1"), product("2"), product("3", ncm="")]
|
|
result, error = run_job("INITIAL_SYNC", products, state, monkeypatch)
|
|
assert error is None
|
|
assert result["products_total"] == 3 and result["products_changed"] == 3 and hub.classified == [["1", "2", "3"]]
|
|
op = result["operational"]
|
|
assert op["TOTAL_PRODUCTS"] == 3 and op["COMPLETE"] == 2 and op["SAFE_BLOCKED"] == 1
|
|
assert op["COMPLETE"] + op["CORRECTED"] + op["SAFE_BLOCKED"] + op["FAILED_TECHNICAL"] == op["TOTAL_PRODUCTS"] # invariant closes
|
|
assert [b["product_code"] for b in result["blocked_products"]] == ["3"] and result["resolved_products"] == ["1", "2"]
|
|
|
|
|
|
def test_incremental_sync_reclassifies_only_new_and_changed_products(env_setup, monkeypatch):
|
|
hub, state = env_setup
|
|
base = [product("1"), product("2"), product("3")]
|
|
run_job("INITIAL_SYNC", base, state, monkeypatch)
|
|
hub.classified.clear()
|
|
|
|
result, error = run_job("SYNC_INCREMENTAL", base, state, monkeypatch) # nothing changed
|
|
assert error is None and result["products_changed"] == 0 and hub.classified == [[]] or hub.classified == []
|
|
assert result["operational"]["TOTAL_PRODUCTS"] == 3 # coverage still complete, rebuilt from the cache
|
|
|
|
hub.classified.clear()
|
|
changed = [product("1"), product("2", ncm="", name="broken now"), product("3"), product("4")] # 2 changed, 4 new
|
|
result, _ = run_job("SYNC_INCREMENTAL", changed, state, monkeypatch)
|
|
assert sorted(hub.classified[0]) == ["2", "4"] and result["products_changed"] == 2
|
|
assert result["operational"]["TOTAL_PRODUCTS"] == 4 and result["operational"]["SAFE_BLOCKED"] == 1
|
|
|
|
|
|
def test_removed_product_leaves_local_cache_but_is_not_reported_as_resolved(env_setup, monkeypatch):
|
|
hub, state = env_setup
|
|
run_job("INITIAL_SYNC", [product("1"), product("2")], state, monkeypatch)
|
|
result, _ = run_job("SYNC_INCREMENTAL", [product("1")], state, monkeypatch)
|
|
assert state.cached_decision("2") is None and result["operational"]["TOTAL_PRODUCTS"] == 1
|
|
assert result["resolved_products"] == [] # no history is auto-resolved/deleted for a product that merely disappeared
|
|
|
|
|
|
def test_every_sync_kind_is_read_only(env_setup, monkeypatch):
|
|
hub, state = env_setup
|
|
for kind in ("INITIAL_SYNC", "SYNC_INCREMENTAL", "REPROCESS_ALL"):
|
|
result, error = run_job(kind, [product("1")], state, monkeypatch)
|
|
assert error is None and result["writes_attempted"] == 0 and result["writes_succeeded"] == 0 and result["firebird_writes_enabled"] is False
|
|
with pytest.raises(RuntimeError, match="writes are disabled"):
|
|
service.open_write_connection({}, AUTHORIZED, firebird_writes=False) # no write-capable connection can even be opened
|
|
|
|
|
|
def test_hub_offline_during_job_fails_the_job_cleanly_no_crash(env_setup, monkeypatch):
|
|
hub, state = env_setup
|
|
def boom(products): raise ConnectionError("hub down")
|
|
hub.classify_batch = boom
|
|
result, error = run_job("SYNC_INCREMENTAL", [product("9")], state, monkeypatch)
|
|
assert result is None and "ConnectionError" in error # reported as a job failure, the service loop keeps running
|
|
|
|
|
|
def test_firebird_offline_during_job_fails_the_job_cleanly(env_setup, monkeypatch):
|
|
hub, state = env_setup
|
|
def down(env, codigos=None): raise OSError("Firebird service not running")
|
|
monkeypatch.setattr(service, "read_products", down)
|
|
result, error = service.execute_job({"job_id": 1, "kind": "SYNC_INCREMENTAL", "agent_id": 2, "payload": {}}, _config(), _FakeExecutor(), state, {})
|
|
assert result is None and "OSError" in error
|
|
|
|
|
|
def test_firebird_credentials_come_from_the_dpapi_file_not_agent_env(monkeypatch, tmp_path):
|
|
from gti_fiscal_agent import credential_store
|
|
path = tmp_path / "firebird.bin"
|
|
monkeypatch.setattr(credential_store, "read_credential", lambda p, env=None: json.dumps({"user": "SYSDBA", "password": "s3cret"}) if str(p) == str(path) else (_ for _ in ()).throw(FileNotFoundError()))
|
|
env = {"FIREBIRD_DATABASE_PATH": AUTHORIZED, "GTI_FIREBIRD_CREDENTIAL_FILE": str(path)}
|
|
service.Config(env)
|
|
assert env["FB_USER"] == "SYSDBA" and env["FB_PASSWORD"] == "s3cret"
|
|
explicit = {"FIREBIRD_DATABASE_PATH": AUTHORIZED, "FB_PASSWORD": "dev", "FB_USER": "u", "GTI_FIREBIRD_CREDENTIAL_FILE": str(path)}
|
|
service.Config(explicit)
|
|
assert explicit["FB_PASSWORD"] == "dev" # an explicit env value still wins (dev override)
|
|
|
|
|
|
def test_central_resume_after_pause_makes_agent_poll_again(monkeypatch, tmp_path):
|
|
class Stopper: stop = False
|
|
stopper = Stopper()
|
|
states = iter([False, True]) # paused, then resumed
|
|
calls = {"next_job": 0}
|
|
|
|
class Hub:
|
|
def identity(self): return {"agent_id": 2, "installation_id": "00000000-0000-4000-8000-000000000222"}
|
|
def heartbeat_metadata(self, payload): return {"processing_enabled": next(states)}
|
|
def next_job(self):
|
|
calls["next_job"] += 1; stopper.stop = True; return None
|
|
|
|
monkeypatch.setattr(service, "HubClient", lambda *a, **k: Hub())
|
|
monkeypatch.setattr(service, "load_executor", lambda: _FakeExecutor())
|
|
monkeypatch.setattr(service.time, "sleep", lambda s: None)
|
|
service.run(_config(agent_id="2", poll_interval_seconds=0, state_db=str(tmp_path / "a.db")), {"AGENT_ID": "2"}, stopper=stopper)
|
|
assert calls["next_job"] == 1 # first cycle paused (no poll), second cycle resumed (polled)
|
|
|
|
|
|
def test_heartbeats_keep_flowing_while_a_long_job_runs(monkeypatch, tmp_path):
|
|
beats = []
|
|
|
|
class Hub:
|
|
def identity(self): return {"agent_id": 2, "installation_id": "00000000-0000-4000-8000-000000000222"}
|
|
def heartbeat_metadata(self, payload): beats.append(payload["status"]); return {"processing_enabled": True}
|
|
def next_job(self): return {"job_id": 5, "kind": "INITIAL_SYNC", "agent_id": 2, "payload": {}}
|
|
def complete_job(self, job_id, summary): stopper.stop = True
|
|
def fail_job(self, job_id, error): stopper.stop = True
|
|
|
|
class Stopper: stop = False
|
|
stopper = Stopper()
|
|
monkeypatch.setattr(service, "BUSY_HEARTBEAT_SECONDS", 0.05)
|
|
monkeypatch.setattr(service, "HubClient", lambda *a, **k: Hub())
|
|
monkeypatch.setattr(service, "load_executor", lambda: _FakeExecutor())
|
|
monkeypatch.setattr(service, "execute_job", lambda *a, **k: (time.sleep(0.4), ({"ok": 1}, None))[1])
|
|
service.run(_config(agent_id="2", poll_interval_seconds=0, state_db=str(tmp_path / "a.db")), {"AGENT_ID": "2"}, stopper=stopper)
|
|
assert beats.count("BUSY") >= 3 # heartbeats during the job, not only before it
|