fix: persist and recover the Agent's real installation_id (heartbeat activation)
POST /v1/agents/enroll now returns installation_id; new installs persist it to agent.env immediately. Runtime recovery precedence: existing SQLite installation_identity -> configured INSTALLATION_ID -> authenticated GET /v1/agents/me/identity. Recovery fails closed on missing/malformed/ mismatched identity rather than inventing a UUID (which the Hub would then reject as a conflicting Installation on an already-enrolled Agent). An explicit AGENT_JOBS_ENABLED gate stops the service from claiming a job right after the first heartbeat. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,34 @@
|
||||
import httpx
|
||||
|
||||
from gti_fiscal_agent.client import HubClient
|
||||
|
||||
|
||||
def test_identity_requests_authenticated_identity_only(monkeypatch):
|
||||
calls = {}
|
||||
payload = {
|
||||
"agent_id": 2,
|
||||
"installation_id": "00000000-0000-4000-8000-000000000222",
|
||||
"hostname": "DESKTOP-B0HL02I",
|
||||
"version": "1.1.1",
|
||||
"status": "OFFLINE",
|
||||
}
|
||||
|
||||
class Response:
|
||||
status_code = 200
|
||||
|
||||
def raise_for_status(self):
|
||||
return None
|
||||
|
||||
def json(self):
|
||||
return payload
|
||||
|
||||
def fake_get(url, *, headers, timeout):
|
||||
calls.update(url=url, headers=headers, timeout=timeout)
|
||||
return Response()
|
||||
|
||||
monkeypatch.setattr(httpx, "get", fake_get)
|
||||
client = HubClient("https://hub.example", "test-agent-token")
|
||||
assert client.identity() == payload
|
||||
|
||||
assert calls["url"] == "https://hub.example/v1/agents/me/identity"
|
||||
assert calls["headers"] == {"Authorization": "Bearer test-agent-token"}
|
||||
@@ -22,6 +22,19 @@ service = load_module()
|
||||
class _FakeState:
|
||||
def __init__(self):
|
||||
self.logged = []
|
||||
self.identity = None
|
||||
|
||||
def stored_installation_id(self):
|
||||
return self.identity
|
||||
|
||||
def installation_id(self, candidate=None):
|
||||
if self.identity and candidate and self.identity != candidate:
|
||||
raise ValueError("configured installation_id conflicts with persisted identity")
|
||||
if self.identity is None:
|
||||
if not candidate:
|
||||
raise ValueError("installation_id is required")
|
||||
self.identity = candidate
|
||||
return self.identity
|
||||
|
||||
def log_ipi_write(self, *args):
|
||||
self.logged.append(args)
|
||||
@@ -61,7 +74,7 @@ def _config(**overrides):
|
||||
base = dict(
|
||||
hub_url="http://localhost:8000", agent_id="a1", agent_token="tok", company_id="c1", company_uf="BA",
|
||||
database_path=AUTHORIZED, firebird_writes=False, poll_interval_seconds=0.01,
|
||||
agent_version="0.1.0", schema_fingerprint="fp", state_db=":memory:",
|
||||
agent_version="0.1.0", schema_fingerprint="fp", state_db=":memory:", agent_jobs_enabled=True,
|
||||
authorized_write_database_path=None, expected_schema_fingerprint="fp",
|
||||
)
|
||||
base.update(overrides)
|
||||
@@ -74,6 +87,13 @@ def test_config_defaults_from_env():
|
||||
assert config.firebird_writes is False
|
||||
assert config.hub_url == "http://localhost:8000"
|
||||
assert config.poll_interval_seconds == 5.0
|
||||
assert config.agent_jobs_enabled is True
|
||||
|
||||
|
||||
def test_config_can_disable_agent_job_polling():
|
||||
config = service.Config({"FIREBIRD_DATABASE_PATH": AUTHORIZED, "AGENT_JOBS_ENABLED": "0"})
|
||||
|
||||
assert config.agent_jobs_enabled is False
|
||||
|
||||
|
||||
def test_config_reads_gti_hub_url_and_writes_flag():
|
||||
@@ -83,6 +103,54 @@ def test_config_reads_gti_hub_url_and_writes_flag():
|
||||
assert config.firebird_writes is True
|
||||
|
||||
|
||||
def test_identity_recovery_persists_existing_hub_installation_id(tmp_path):
|
||||
config = _config(agent_id="2")
|
||||
state = service.AgentState(tmp_path / "agent.db")
|
||||
installation_id = "00000000-0000-4000-8000-000000000222"
|
||||
|
||||
class FakeIdentityClient:
|
||||
calls = 0
|
||||
|
||||
def identity(self):
|
||||
self.calls += 1
|
||||
return {"agent_id": 2, "installation_id": installation_id}
|
||||
|
||||
client = FakeIdentityClient()
|
||||
recovered = service.resolve_installation_id(config, {}, state, client)
|
||||
|
||||
assert recovered == installation_id
|
||||
assert state.installation_id() == installation_id
|
||||
assert client.calls == 1
|
||||
|
||||
restarted = service.resolve_installation_id(config, {}, state, client)
|
||||
assert restarted == installation_id
|
||||
assert client.calls == 1
|
||||
|
||||
|
||||
def test_identity_recovery_fails_closed_for_another_agent_or_missing_id(tmp_path):
|
||||
config = _config(agent_id="2")
|
||||
|
||||
class FakeIdentityClient:
|
||||
def __init__(self, payload):
|
||||
self.payload = payload
|
||||
def identity(self):
|
||||
return self.payload
|
||||
|
||||
state = service.AgentState(tmp_path / "agent.db")
|
||||
with pytest.raises(RuntimeError, match="identity does not match configured Agent"):
|
||||
service.resolve_installation_id(
|
||||
config,
|
||||
{},
|
||||
state,
|
||||
FakeIdentityClient({"agent_id": 3, "installation_id": "00000000-0000-4000-8000-000000000333"}),
|
||||
)
|
||||
assert state.connection.execute("SELECT COUNT(*) FROM installation_identity").fetchone()[0] == 0
|
||||
|
||||
with pytest.raises(RuntimeError, match="identity response has no installation_id"):
|
||||
service.resolve_installation_id(config, {}, state, FakeIdentityClient({"agent_id": 2}))
|
||||
assert state.connection.execute("SELECT COUNT(*) FROM installation_identity").fetchone()[0] == 0
|
||||
|
||||
|
||||
def test_path_guard_fails_closed_before_any_write(monkeypatch):
|
||||
config = _config(firebird_writes=True, database_path=r"D:\PRODUCAO\ERP.FDB")
|
||||
executor = _FakeExecutor(path_guard_ok=False)
|
||||
@@ -226,6 +294,57 @@ class _ExternalStopFlag:
|
||||
self.stop = False
|
||||
|
||||
|
||||
def test_run_sends_heartbeat_but_never_polls_when_jobs_disabled(monkeypatch, tmp_path):
|
||||
stopper = _ExternalStopFlag()
|
||||
fake_hub = _FakeHubForLoop()
|
||||
installation_id = "00000000-0000-4000-8000-000000000222"
|
||||
calls = {"identity": 0, "next_job": 0, "heartbeat": 0, "heartbeat_installation_id": None}
|
||||
|
||||
def identity():
|
||||
calls["identity"] += 1
|
||||
return {"agent_id": 2, "installation_id": installation_id}
|
||||
|
||||
def heartbeat_metadata(payload):
|
||||
calls["heartbeat"] += 1
|
||||
calls["heartbeat_installation_id"] = payload["installation_id"]
|
||||
stopper.stop = True
|
||||
return {"agent_id": 2}
|
||||
|
||||
def forbidden_next_job():
|
||||
calls["next_job"] += 1
|
||||
raise AssertionError("jobs are disabled; next_job must not be called")
|
||||
|
||||
fake_hub.identity = identity
|
||||
fake_hub.heartbeat_metadata = heartbeat_metadata
|
||||
fake_hub.next_job = forbidden_next_job
|
||||
monkeypatch.setattr(service, "HubClient", lambda *a, **k: fake_hub)
|
||||
monkeypatch.setattr(service, "load_executor", lambda: _FakeExecutor())
|
||||
monkeypatch.setattr(service.time, "sleep", lambda _seconds: None)
|
||||
|
||||
config = _config(
|
||||
agent_id="2",
|
||||
agent_jobs_enabled=False,
|
||||
poll_interval_seconds=0,
|
||||
state_db=str(tmp_path / "agent.db"),
|
||||
)
|
||||
env = {
|
||||
"AGENT_ID": "2",
|
||||
"FIREBIRD_DATABASE_PATH": AUTHORIZED,
|
||||
"FIREBIRD_WRITES": "0",
|
||||
"FISCAL_MODE": "DRY_RUN",
|
||||
}
|
||||
|
||||
service.run(config, env, stopper=stopper)
|
||||
|
||||
assert calls == {
|
||||
"identity": 1,
|
||||
"next_job": 0,
|
||||
"heartbeat": 1,
|
||||
"heartbeat_installation_id": installation_id,
|
||||
}
|
||||
assert service.AgentState(config.state_db).stored_installation_id() == installation_id
|
||||
|
||||
|
||||
def test_run_honors_an_externally_supplied_stopper(monkeypatch):
|
||||
# This is exactly the pattern the Windows Service wrapper uses: it never
|
||||
# constructs GracefulStop() (which would crash off the main thread), it
|
||||
@@ -248,7 +367,7 @@ def test_run_honors_an_externally_supplied_stopper(monkeypatch):
|
||||
stopper.stop = True
|
||||
|
||||
monkeypatch.setattr(service.time, "sleep", counting_sleep)
|
||||
service.run(config, {}, stopper=stopper) # no max_iterations — only the external flag can stop it
|
||||
service.run(config, {"AGENT_ID": "2", "INSTALLATION_ID": "00000000-0000-4000-8000-000000000222"}, stopper=stopper) # no max_iterations — only the external flag can stop it
|
||||
assert stopper.stop is True
|
||||
assert calls["n"] >= 3
|
||||
|
||||
@@ -264,7 +383,7 @@ def test_run_survives_hub_down_then_processes_one_job_and_stops(monkeypatch):
|
||||
monkeypatch.setattr(service.time, "sleep", lambda seconds: None)
|
||||
|
||||
# first heartbeat call raises -> backoff branch; second call onward succeeds.
|
||||
service.run(config, {}, max_iterations=4)
|
||||
service.run(config, {"AGENT_ID": "2", "INSTALLATION_ID": "00000000-0000-4000-8000-000000000222"}, max_iterations=4)
|
||||
|
||||
assert fake_hub.heartbeat_calls >= 2
|
||||
assert len(fake_hub.completed) == 1
|
||||
|
||||
@@ -1,3 +1,5 @@
|
||||
import pytest
|
||||
|
||||
from gti_fiscal_agent.state import AgentState
|
||||
|
||||
|
||||
@@ -18,3 +20,37 @@ def test_ipi_write_log_records_per_product_audit(tmp_path):
|
||||
assert log[0]["write_status"] == "SUCCEEDED"
|
||||
assert log[0]["before_value"] is None
|
||||
assert log[0]["proposed_value"] == "3.25"
|
||||
|
||||
|
||||
def test_installation_identity_requires_an_existing_or_explicit_id(tmp_path):
|
||||
state = AgentState(tmp_path / "agent.db")
|
||||
|
||||
assert state.stored_installation_id() is None
|
||||
with pytest.raises(ValueError, match="installation_id is required"):
|
||||
state.installation_id()
|
||||
|
||||
assert state.connection.execute("SELECT COUNT(*) FROM installation_identity").fetchone()[0] == 0
|
||||
|
||||
|
||||
def test_installation_identity_persists_and_reuses_explicit_id(tmp_path):
|
||||
db_path = tmp_path / "agent.db"
|
||||
state = AgentState(db_path)
|
||||
existing_id = "00000000-0000-4000-8000-000000000222"
|
||||
|
||||
assert state.installation_id(existing_id) == existing_id
|
||||
assert state.installation_id() == existing_id
|
||||
assert state.stored_installation_id() == existing_id
|
||||
|
||||
restarted_state = AgentState(db_path)
|
||||
assert restarted_state.installation_id() == existing_id
|
||||
|
||||
|
||||
def test_installation_identity_rejects_conflicting_or_malformed_ids(tmp_path):
|
||||
state = AgentState(tmp_path / "agent.db")
|
||||
existing_id = "00000000-0000-4000-8000-000000000222"
|
||||
state.installation_id(existing_id)
|
||||
|
||||
with pytest.raises(ValueError, match="conflicts"):
|
||||
state.installation_id("00000000-0000-4000-8000-000000000223")
|
||||
with pytest.raises(ValueError, match="UUID"):
|
||||
state.installation_id("not-a-uuid")
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
import importlib.util
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
|
||||
SERVICE_WRAPPER = Path(__file__).parents[1] / "service" / "gti_agent_windows_service.py"
|
||||
|
||||
|
||||
def load_service_wrapper():
|
||||
spec = importlib.util.spec_from_file_location("gti_agent_windows_service_test", SERVICE_WRAPPER)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[spec.name] = module
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def test_main_enters_scm_dispatcher_when_started_without_command(monkeypatch):
|
||||
service = load_service_wrapper()
|
||||
calls = []
|
||||
|
||||
class FakeServiceManager:
|
||||
def Initialize(self):
|
||||
calls.append("initialize")
|
||||
|
||||
def PrepareToHostSingle(self, service_class):
|
||||
calls.append(("prepare", service_class))
|
||||
|
||||
def StartServiceCtrlDispatcher(self):
|
||||
calls.append("dispatcher")
|
||||
|
||||
def forbidden_handle_command_line(service_class):
|
||||
raise AssertionError("SCM start must not be parsed as a CLI command")
|
||||
|
||||
monkeypatch.setattr(service, "servicemanager", FakeServiceManager())
|
||||
monkeypatch.setattr(service, "win32serviceutil", SimpleNamespace(HandleCommandLine=forbidden_handle_command_line))
|
||||
monkeypatch.setattr(service.sys, "argv", [str(SERVICE_WRAPPER)])
|
||||
|
||||
service.main()
|
||||
|
||||
assert calls == ["initialize", ("prepare", service.GTIFiscalAgentService), "dispatcher"]
|
||||
|
||||
|
||||
def test_main_preserves_pywin32_cli_commands(monkeypatch):
|
||||
service = load_service_wrapper()
|
||||
called = []
|
||||
|
||||
class ForbiddenServiceManager:
|
||||
def __getattr__(self, name):
|
||||
raise AssertionError("CLI commands must not enter the SCM dispatcher")
|
||||
|
||||
monkeypatch.setattr(service, "servicemanager", ForbiddenServiceManager())
|
||||
monkeypatch.setattr(service, "win32serviceutil", SimpleNamespace(HandleCommandLine=lambda cls: called.append(cls)))
|
||||
monkeypatch.setattr(service.sys, "argv", [str(SERVICE_WRAPPER), "start"])
|
||||
|
||||
service.main()
|
||||
|
||||
assert called == [service.GTIFiscalAgentService]
|
||||
Reference in New Issue
Block a user