gtiandClaude Sonnet 5
bb663a979c
Hub: installations.processing_enabled / sync_interval_seconds / last_sync_* (migration 0026). The first heartbeat of an installation without coverage queues exactly one INITIAL_SYNC; afterwards one SYNC_INCREMENTAL per interval (idempotent, backs off on failure). One fiscal job at a time per Agent (FIFO, stale RUNNING recovery). Job completion now updates the coverage snapshot, blockers and JOB_STARTED/COMPLETED/FAILED events. Panel: Sincronizar agora / Reprocessar tudo / Pausar / Retomar + status; pause is enforced Hub-side. Agent: local AGENT_JOBS_ENABLED is ignored (central control); SYNC_INCREMENTAL re-classifies only new/changed products via a per-product fingerprint+decision cache and rebuilds coverage from it; results carry operational summary and blockers; Firebird credentials come from a DPAPI firebird.bin. Setup: read-only SGV Firebird discovery (auto-select only when unambiguous, wizard chooser otherwise), starts the service, waits for the first heartbeat, upgrade-in-place mode. All jobs stay read-only (FIREBIRD_WRITES=0). Version 1.1.3. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
119 lines
7.1 KiB
Python
119 lines
7.1 KiB
Python
"""Zero-touch Setup: discovery -> enroll -> secrets in DPAPI stores -> service registered AND started
|
|
-> first heartbeat awaited; upgrade preserves identity and drops the legacy local toggle.
|
|
The Windows/SCM/Hub edges are faked; the orchestration is real."""
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).parents[1]
|
|
for extra in (ROOT, ROOT / "installer", ROOT / "agent" / "src"):
|
|
sys.path.insert(0, str(extra))
|
|
|
|
setup_main = importlib.import_module("setup_main")
|
|
common = setup_main.common
|
|
fbd = setup_main.fbd
|
|
|
|
ENROLL = {"agent_id": 7, "agent_token": "tok-never-in-env", "hub_url": "https://gtifiscal.gtigrupo.com.br", "agent_version": "1.1.3",
|
|
"installation_id": "00000000-0000-4000-8000-0000000000bb"}
|
|
|
|
|
|
@pytest.fixture
|
|
def world(tmp_path, monkeypatch):
|
|
calls = []
|
|
monkeypatch.setenv("GTI_INSTALL_DIR", str(tmp_path / "inst"))
|
|
monkeypatch.setattr(setup_main, "MEIPASS", tmp_path / "payload")
|
|
payload = tmp_path / "payload"
|
|
(payload / "GTIFiscalAgentService").mkdir(parents=True); (payload / "GTIFiscalAgentService" / "GTIFiscalAgentService.exe").write_bytes(b"x")
|
|
(payload / "Uninstall.exe").write_bytes(b"x"); (payload / "scripts").mkdir(); (payload / "scripts" / "firebird_write_executor.py").write_text("# x")
|
|
(payload / "VERSION").write_text("1.1.3\n")
|
|
monkeypatch.setattr(common, "enroll", lambda code, hub=None: calls.append(("enroll", code)) or dict(ENROLL))
|
|
monkeypatch.setattr(common, "write_credential", lambda d, t: calls.append(("hub_cred", t)))
|
|
monkeypatch.setattr(common, "write_firebird_credential", lambda d, u, p: calls.append(("fb_cred", u)))
|
|
monkeypatch.setattr(common, "install_service", lambda d: calls.append(("install_service",)))
|
|
monkeypatch.setattr(common, "write_arp_uninstall", lambda d: None)
|
|
monkeypatch.setattr(common, "service_running", lambda: False)
|
|
monkeypatch.setattr(common, "service_installed", lambda: True)
|
|
monkeypatch.setattr(common, "start_service", lambda: calls.append(("start",)))
|
|
monkeypatch.setattr(common, "stop_service", lambda: calls.append(("stop",)))
|
|
monkeypatch.setattr(common, "wait_service_state", lambda wanted, timeout=45: True)
|
|
monkeypatch.setattr(common, "wait_for_heartbeat", lambda token, hub, timeout=120: calls.append(("heartbeat_wait",)) or True)
|
|
monkeypatch.setattr(common, "audit_event", lambda *a, **k: None)
|
|
return calls, tmp_path / "inst"
|
|
|
|
|
|
def one_db(monkeypatch):
|
|
cand = fbd.Candidate(r"D:\SGV\loja.fdb", "OK", "SYSDBA", "masterkey")
|
|
monkeypatch.setattr(fbd, "discover", lambda **k: fbd.Discovery("SELECTED", [cand]))
|
|
|
|
|
|
def test_zero_touch_install_ends_with_service_started_and_heartbeat_confirmed(world, monkeypatch):
|
|
calls, inst = world
|
|
one_db(monkeypatch)
|
|
result = setup_main.install_flow("AAAA-BBBB-CCCC-DDDD")
|
|
names = [c[0] for c in calls]
|
|
assert names == ["enroll", "hub_cred", "fb_cred", "install_service", "start", "heartbeat_wait"]
|
|
assert result["service_running"] is True and result["heartbeat"] is True and result["firebird_path"] == r"D:\SGV\loja.fdb"
|
|
env = (inst / "agent.env").read_text(encoding="utf-8")
|
|
assert r"FIREBIRD_DATABASE_PATH=D:\SGV\loja.fdb" in env and "FIREBIRD_WRITES=0" in env and "FISCAL_MODE=DRY_RUN" in env
|
|
assert "tok-never-in-env" not in env and "masterkey" not in env and "AGENT_JOBS_ENABLED" not in env
|
|
assert "Nenhuma outra ação" in setup_main.final_message(result) # the operator is told nothing else is needed
|
|
|
|
|
|
def test_ambiguous_databases_ask_the_operator_and_nothing_is_enrolled_if_cancelled(world, monkeypatch):
|
|
calls, _ = world
|
|
a, b = fbd.Candidate(r"D:\a.fdb", "OK", "SYSDBA", "masterkey"), fbd.Candidate(r"D:\b.fdb", "OK", "SYSDBA", "masterkey")
|
|
monkeypatch.setattr(fbd, "discover", lambda **k: fbd.Discovery("AMBIGUOUS", [a, b]))
|
|
with pytest.raises(RuntimeError, match="seleção do banco Firebird"):
|
|
setup_main.install_flow("AAAA-BBBB-CCCC-DDDD") # headless, no chooser: refuses to guess
|
|
assert calls == []
|
|
with pytest.raises(RuntimeError, match="cancelada"):
|
|
setup_main.install_flow("AAAA-BBBB-CCCC-DDDD", chooser=lambda found: None)
|
|
assert calls == [] # the install code was NOT consumed by an install that could not complete
|
|
|
|
|
|
def test_operator_choice_is_verified_before_use(world, monkeypatch):
|
|
calls, inst = world
|
|
a, b = fbd.Candidate(r"D:\a.fdb", "OK", "SYSDBA", "masterkey"), fbd.Candidate(r"D:\b.fdb", "OK", "SYSDBA", "masterkey")
|
|
monkeypatch.setattr(fbd, "discover", lambda **k: fbd.Discovery("AMBIGUOUS", [a, b]))
|
|
monkeypatch.setattr(fbd, "probe", lambda path, u, p, **k: fbd.Candidate(path, "OK", u, p))
|
|
result = setup_main.install_flow("AAAA-BBBB-CCCC-DDDD", chooser=lambda found: {"path": r"D:\b.fdb", "user": "SYSDBA", "password": "masterkey"})
|
|
assert result["firebird_path"] == r"D:\b.fdb"
|
|
monkeypatch.setattr(fbd, "probe", lambda path, u, p, **k: fbd.Candidate(path, "AUTH_FAILED"))
|
|
with pytest.raises(RuntimeError, match="não é utilizável"):
|
|
setup_main.install_flow("AAAA-BBBB-CCCC-DDDD", chooser=lambda found: {"path": r"D:\b.fdb", "user": "x", "password": "y"})
|
|
|
|
|
|
def test_upgrade_preserves_identity_drops_local_toggle_and_restarts(world):
|
|
calls, inst = world
|
|
inst.mkdir(parents=True)
|
|
(inst / "credentials.bin").write_bytes(b"blob")
|
|
(inst / "agent.env").write_text("GTI_HUB_URL=https://gtifiscal.gtigrupo.com.br\nAGENT_ID=2\nINSTALLATION_ID=bb50884a-5e78-4f26-8cac-804932dae8de\n"
|
|
"AGENT_JOBS_ENABLED=0\nAGENT_VERSION=1.1.1\nFIREBIRD_DATABASE_PATH=D:/SGV/x.fdb\nFISCAL_MODE=DRY_RUN\nFIREBIRD_WRITES=0\n", encoding="utf-8")
|
|
result = setup_main.upgrade_flow()
|
|
env = (inst / "agent.env").read_text(encoding="utf-8")
|
|
assert "AGENT_JOBS_ENABLED" not in env and "AGENT_VERSION=1.1.3" in env
|
|
assert "AGENT_ID=2" in env and "INSTALLATION_ID=bb50884a-5e78-4f26-8cac-804932dae8de" in env and "FIREBIRD_DATABASE_PATH=D:/SGV/x.fdb" in env
|
|
assert [c[0] for c in calls] == ["stop", "install_service", "start"]
|
|
assert "enroll" not in [c[0] for c in calls] # NEVER re-enrolls: no new Agent, no duplicate installation
|
|
assert result["mode"] == "UPGRADE" and result["installation_id"] == "bb50884a-5e78-4f26-8cac-804932dae8de" and result["service_running"]
|
|
|
|
|
|
def test_second_run_on_enrolled_host_is_an_upgrade_not_a_new_install(world):
|
|
calls, inst = world
|
|
inst.mkdir(parents=True); (inst / "credentials.bin").write_bytes(b"b"); (inst / "agent.env").write_text("AGENT_ID=2\nINSTALLATION_ID=x\n")
|
|
assert common.detect_existing(inst)["agent_id"] == "2"
|
|
assert common.detect_existing(inst / "nope") is None
|
|
|
|
|
|
def test_service_is_registered_delayed_auto_with_restart_recovery():
|
|
text = (ROOT / "installer" / "install_common.py").read_text(encoding="utf-8")
|
|
assert '"start=", "delayed-auto"' in text and 'SERVICE_FAILURE = "restart/60000/restart/60000/restart/60000"' in text
|
|
|
|
|
|
def test_no_write_flags_anywhere_in_zero_touch_installer():
|
|
for name in ("setup_main.py", "firebird_discovery.py"):
|
|
text = (ROOT / "installer" / name).read_text(encoding="utf-8")
|
|
assert "FIREBIRD_WRITES=1" not in text and "AUTO_APPLY" not in text and "UPDATE " not in text.upper().replace("UPDATE_", "")
|