from __future__ import annotations import hashlib import importlib.util import json import os import sys from pathlib import Path from types import SimpleNamespace import pytest HERE = Path(__file__).parent spec = importlib.util.spec_from_file_location( "baseline", HERE / "baseline_controller.py" ) b = importlib.util.module_from_spec(spec) spec.loader.exec_module(b) RUNTIME = {"active": "inactive", "sub": "dead", "main_pid": 0, "profile_processes": []} REG = { "customers": [], "diagnostic_session_digest": None, "owner": {"chat_id": "8693203710", "topic_id": "0", "user_id": "8693203710"}, "registry_mode": "ordinary_v1", "version": 1, } def private(p: Path, raw: bytes): p.parent.mkdir(parents=True, exist_ok=True, mode=0o700) p.parent.chmod(0o700) p.write_bytes(raw) p.chmod(0o600) def fixture(tmp: Path): profile = tmp / "profile" other = tmp / "other" profile.mkdir(mode=0o700) other.mkdir(mode=0o700) private(other / "marker", b"other") live = json.loads( Path( "/home/cube/.hermes/profiles/dualcoachtest/data/onboarding/telegram-customer-bootstrap-v1/ledger.json" ).read_text() ) private( profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json", b.canonical(live), ) private(profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.lock", b"") private(profile / "gateway.lock", b"stale") (profile / "sessions").mkdir(mode=0o700) (profile / "cron/output").mkdir(parents=True, mode=0o700) archive = profile / "data/profile-reset-archives/task26-live-reset-2e0894ea" archive.mkdir(parents=True, mode=0o700) private(archive / "payload/customers/registry.json", b.canonical(REG)) manifest = { "evidence_digest": "e" * 64, "entries": [ { "mode": "0600", "path": "customers/registry.json", "sha256": b.sha(b.canonical(REG)), "size": len(b.canonical(REG)), } ], } mraw = b.canonical(manifest) private(archive / "manifest.json", mraw) receipt = { "evidence_digest": "e" * 64, "manifest_sha256": b.sha(mraw), "status": "PASS", "post_reset_empty_baseline": True, } rraw = b.canonical(receipt) private(archive / "receipt.json", rraw) c = { "archive_run_id": archive.name, "archive_manifest_sha256": b.sha(mraw), "archive_receipt_sha256": b.sha(rraw), "archive_evidence_digest": "e" * 64, "archive_tree_sha256": b.tree_digest(archive), "canonical_registry": REG, "canonical_registry_sha256": b.sha(b.canonical(REG)), "session_id": "cb_v4olwxbpSQatMtVLR4QLmw", "sid_hash": "63bdc993abc7b4f0025e69b10246faf8b5f4da3b7c040f02987a77bef3fa7c59", "invite_ledger_sha256": b.sha(b.canonical(live)), "other_profile_sha256": b.tree_digest(other), } return profile, other, archive, c def test_complete_and_candidate_gateway_loaders_accept_zero_customers(tmp_path): profile, other, archive, c = fixture(tmp_path) before = ( profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" ).read_bytes() b.preflight(profile, other, archive, c, RUNTIME) b.publish(profile, b.canonical(REG)) out = b.verify_baseline(profile, other, archive, c, RUNTIME) assert ( out["customers"] == 0 and (profile / "customers").stat().st_mode & 0o777 == 0o700 and (profile / "customers/registry.json").stat().st_mode & 0o777 == 0o600 ) assert ( profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" ).read_bytes() == before sys.path[:0] = [ "/home/cube/.hermes/profiles/dualcoachtest/workspace/checkin_cli", "/home/cube/projects/richard/hermes-agent", ] from gateway.platforms.nutrition_coaching import load_committed_customer_registry from gateway.platforms.telegram import TelegramAdapter registry, path = load_committed_customer_registry(profile) assert len(registry.customers) == 0 and path == profile / "customers/registry.json" root, resolved = TelegramAdapter._configured_nutrition_registry( SimpleNamespace(registry_path="customers/registry.json"), SimpleNamespace(profile_root=profile, extra={}), ) assert root == profile.resolve() and resolved == path @pytest.mark.parametrize( "mutation", ["schema", "symlink", "mode", "service", "pending"] ) def test_preflight_negative_authorities(tmp_path, mutation): profile, other, archive, c = fixture(tmp_path) runtime = dict(RUNTIME) if mutation == "schema": p = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" v = json.loads(p.read_text()) v["unknown"] = 1 private(p, b.canonical(v)) elif mutation == "symlink": (profile / "customers").symlink_to(other, target_is_directory=True) elif mutation == "mode": (profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json").chmod( 0o644 ) elif mutation == "service": runtime = { "active": "active", "sub": "running", "main_pid": 1, "profile_processes": [1], } else: private(profile / "data/owner-actions/pending.json", b"{}\n") with pytest.raises(b.BaselineError): b.preflight(profile, other, archive, c, runtime) def test_verify_rejects_hardlinked_registry(tmp_path): profile, other, archive, c = fixture(tmp_path) (profile / "customers").mkdir(mode=0o700) private(profile / "source", b.canonical(REG)) os.link(profile / "source", profile / "customers/registry.json") with pytest.raises(b.BaselineError, match="single-link|unsafe private"): b.verify_baseline(profile, other, archive, c, RUNTIME) def test_archive_manifest_drift_fails(tmp_path): profile, other, archive, c = fixture(tmp_path) private(archive / "manifest.json", b"{}\n") with pytest.raises(b.BaselineError, match="archive pin"): b.preflight(profile, other, archive, c, RUNTIME) def test_invite_claim_fails(tmp_path): profile, other, archive, c = fixture(tmp_path) p = profile / "data/onboarding/telegram-customer-bootstrap-v1/ledger.json" v = json.loads(p.read_text()) v["sessions"][0]["state"] = "CLAIMED" v["digest"] = b.sha( b.canonical({"schema": v["schema"], "sessions": v["sessions"]})[:-1] ) private(p, b.canonical(v)) c["invite_ledger_sha256"] = b.sha(b.canonical(v)) with pytest.raises(b.BaselineError, match="unclaimed PREPARED"): b.preflight(profile, other, archive, c, RUNTIME) def test_publication_failure_rolls_back_before_visibility(tmp_path, monkeypatch): profile, _, _, _ = fixture(tmp_path) real = os.rename def fail(*a, **k): raise OSError("injected") monkeypatch.setattr(os, "rename", fail) with pytest.raises(OSError): b.publish(profile, b.canonical(REG)) assert not (profile / "customers").exists() and not list( profile.glob(".customers-baseline-*") ) monkeypatch.setattr(os, "rename", real) def test_canonical_bytes_are_archive_schema_not_copy(): raw = b.canonical(REG) assert ( hashlib.sha256(raw).hexdigest() == "9eb1b5ae0e511dbafbed6b683ec7baa2ffb15518637d8ad63200567c4009a518" and json.loads(raw)["customers"] == [] )