gti
d21a9d1140
git-subtree-dir: fiscal git-subtree-mainline:d95165fc8fgit-subtree-split:102ba84500
84 lines
4.8 KiB
Python
84 lines
4.8 KiB
Python
"""Real zero-touch runtime E2E against the local dev Hub + a real Firebird test database.
|
|
|
|
No manual step between heartbeat and result: a brand-new installation identity sends its
|
|
first heartbeat, the Hub queues exactly one INITIAL_SYNC, the Agent claims and executes it
|
|
(READ-ONLY), coverage/blockers land on the Hub; then the central scheduler queues a
|
|
SYNC_INCREMENTAL by itself (interval shortened to 1 s for the test) which must reprocess
|
|
only the delta (nothing changed -> 0), and finally a central pause stops job execution while
|
|
heartbeats continue.
|
|
|
|
Usage (dev Hub on :8000, .env with FB credentials): python scripts/e2e_zero_touch_local.py
|
|
"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path[:0] = [str(ROOT), str(ROOT / "agent" / "src"), str(ROOT / "agent" / "scripts")]
|
|
|
|
from dotenv import dotenv_values # noqa: E402
|
|
|
|
env_file = dotenv_values(ROOT / ".env")
|
|
HUB = "http://localhost:8000"
|
|
|
|
|
|
def psql(sql):
|
|
out = subprocess.run(["docker", "exec", "projeto-postgres-1", "psql", "-U", "gti", "-d", "gti_fiscal", "-At", "-F", "|", "-c", sql], capture_output=True, text=True)
|
|
return out.stdout.strip()
|
|
|
|
|
|
def main():
|
|
import run_agent_service as service
|
|
|
|
tmp = Path(tempfile.mkdtemp(prefix="gti_zt_"))
|
|
# A brand-new Agent (own tenant + token), like a freshly enrolled installation — never the shared dev agent.
|
|
reg = subprocess.run(["docker", "exec", "projeto-fiscal-hub-1", "python", "central/scripts/register_agent.py", "--company", f"e2e-zero-touch-{int(time.time())}", "--name", "zt"], capture_output=True, text=True).stdout
|
|
token = next(line.split("=", 1)[1] for line in reg.splitlines() if line.strip().startswith("AGENT_TOKEN="))
|
|
installation = str(uuid.uuid4())
|
|
os.environ.update({
|
|
"GTI_HUB_URL": HUB, "AGENT_TOKEN": token, "AGENT_ID": "dev", "INSTALLATION_ID": installation,
|
|
"FIREBIRD_DATABASE_PATH": env_file["FIREBIRD_DATABASE_PATH"], "FB_USER": env_file.get("FB_USER", "SYSDBA"), "FB_PASSWORD": env_file["FB_PASSWORD"],
|
|
"FIREBIRD_CLIENT_LIBRARY": env_file.get("FIREBIRD_CLIENT_LIBRARY", ""), "FIREBIRD_WRITES": "0", "FISCAL_MODE": "DRY_RUN",
|
|
"AGENT_STATE_DB": str(tmp / "state.sqlite"), "AGENT_LOG_PATH": str(tmp / "agent.log"), "POLL_INTERVAL_SECONDS": "0.2",
|
|
"AGENT_JOBS_ENABLED": "0", # legacy local toggle deliberately set to 0: must be ignored (control is central)
|
|
})
|
|
config = service.Config(os.environ)
|
|
service.setup_logging(config.log_path)
|
|
print("INSTALLATION_ID=", installation, "FIREBIRD_WRITES=", config.firebird_writes)
|
|
|
|
def run(n):
|
|
service.run(config, os.environ, max_iterations=n)
|
|
|
|
# 1) first heartbeat -> INITIAL_SYNC by the Hub -> claimed + executed in the same loop
|
|
run(4)
|
|
jobs = psql(f"select j.id,j.kind,j.status,j.result_summary->>'products_total',j.result_summary->>'products_changed' from jobs j join installations i on i.agent_id=j.agent_id where i.installation_id='{installation}' order by j.id")
|
|
print("AFTER_INITIAL:\n" + jobs)
|
|
snap = psql(f"select s.total_products,s.fiscally_complete,s.safe_blocked,s.failed_products,s.blockers from agent_snapshots s join installations i on i.id=s.installation_id where i.installation_id='{installation}'")
|
|
print("SNAPSHOT=", snap)
|
|
blockers = psql(f"select count(*) from blocked_products b join installations i on i.id=b.installation_id where i.installation_id='{installation}'")
|
|
print("BLOCKERS_TOTAL=", blockers)
|
|
|
|
# 2) central scheduler: shorten the interval, next heartbeat queues one SYNC_INCREMENTAL by itself
|
|
psql(f"update installations set sync_interval_seconds=1 where installation_id='{installation}'")
|
|
time.sleep(2)
|
|
run(4)
|
|
print("AFTER_INCREMENTAL:\n" + psql(f"select j.id,j.kind,j.status,j.payload->>'reason',j.result_summary->>'products_total',j.result_summary->>'products_changed' from jobs j join installations i on i.agent_id=j.agent_id where i.installation_id='{installation}' order by j.id"))
|
|
|
|
# 3) central pause: heartbeat continues, nothing is claimed
|
|
psql(f"update installations set processing_enabled=false where installation_id='{installation}'")
|
|
before = psql("select count(*) from jobs where status='RUNNING'")
|
|
time.sleep(2)
|
|
run(3)
|
|
print("PAUSED_RUNNING_JOBS=", before, "LAST_STATUS=", psql(f"select last_status from installations where installation_id='{installation}'"))
|
|
print("WRITES:", psql(f"select coalesce(sum((result_summary->>'writes_attempted')::int),0), coalesce(sum((result_summary->>'writes_succeeded')::int),0) from jobs j join installations i on i.agent_id=j.agent_id where i.installation_id='{installation}'"))
|
|
print("LOCAL_ipi_write_log_rows=", __import__("sqlite3").connect(tmp / "state.sqlite").execute("select count(*), sum(write_status='DRY_RUN_ONLY') from ipi_write_log").fetchone())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|