"""python3 test_tiny_beacon.py — pins the Linux beacon to docs/ble-mesh/fixtures.json (no radio)."""
import json, os, subprocess, sys
HERE = os.path.dirname(os.path.abspath(__file__))
FX = os.environ.get("TINY_BLE_FIXTURES", os.path.expanduser("~/tinyai-id/docs/ble-mesh/fixtures.json"))
fx = json.load(open(FX))["sticky_full"]
r = subprocess.run([sys.executable, os.path.join(HERE, "tiny_beacon.py"), "--encode", "--kind", "sticky", "--flags", "catalog,battery,agent",
                    "--caps", "speaker,display,imu,wifi,env,touch,eink", "--tools", "7", "--session", "14879", "--rssi1m", "-61"],
                   capture_output=True, text=True, check=True)
ad = json.loads(r.stdout)["ad"]
assert ad[:4] == "0eff", ad                  # AD len 14, type manufacturer data
assert ad[4:] == fx["mfg_hex"], (ad[4:], fx["mfg_hex"])
# social name rides only with --social; and never past 31 bytes
r = subprocess.run([sys.executable, os.path.join(HERE, "tiny_beacon.py"), "--encode", "--kind", "reachy", "--name", "reachy-mini", "--tools", "9"], capture_output=True, text=True, check=True)
assert json.loads(r.stdout)["len"] == 18  # FULL + 3-byte mine tail (default flags carry mine)
r = subprocess.run([sys.executable, os.path.join(HERE, "tiny_beacon.py"), "--encode", "--kind", "reachy", "--name", "reachy-mini", "--tools", "9", "--social"], capture_output=True, text=True, check=True)
o = json.loads(r.stdout); assert o["len"] == 28 and o["ad"].endswith(b"reachy-m".hex()), o
print("ok: linux beacon == fixtures.json sticky_full; name only when social; 31-byte budget kept")
# ROLLOUT §2.5: the account secret's wire shape (43-char base64url of 32 bytes) and the bluetoothctl
# interval line (§4.2, 250–400 ms) — decode_secret rejects anything else, fetch stays fail-open.
sys.path.insert(0, HERE)
import tiny_beacon as tb
import base64
raw = bytes(range(32))
b64 = base64.urlsafe_b64encode(raw).decode().rstrip("=")
assert len(b64) == 43 and tb.decode_secret(b64) == raw
assert tb.decode_secret(b64 + "A") is None and tb.decode_secret("not base64url!" + "x" * 29) is None and tb.decode_secret(None) is None
assert tb.fetch_account_secret(device_json="/nonexistent/device.json") is None
assert "tiny-beacon/" in tb.USER_AGENT  # bare urllib UA → 403 at the edge
script = tb.bluetoothctl_script(b"\x01\x02", "reachy", False, 0)
assert "interval 250 400" in script and "name off" in script, script
assert tb.mine_tag(raw, 0x1092) != tb.mine_tag(raw, 0x1093) and len(tb.mine_tag(raw, 1)) == 4
print("ok: account secret decode + fail-open fetch + 250–400 ms interval")
# the bluetoothctl payload is exactly 'TN' + header, with or without the social name AD behind it
class _A: pass
a = _A(); a.kind = "reachy"; a.flags = ""; a.caps = ""; a.tools = 9; a.rssi1m = -63; a.social = True; a.name = "reachy-mini"
ad = tb.build(a, 4242)
pl = tb.mfg_payload_of(ad)
assert pl[:2] == b"TN" and len(pl) == 14, pl.hex()  # TN + header + mine tail
a.social = False
pl2 = tb.mfg_payload_of(tb.build(a, 4242))
assert len(pl2) == 14 and pl2[:4] == pl[:4] and pl2[5:] == pl[5:]  # only the flags byte (TN·ver·kind·FLAGS) differs
print("ok: mfg payload slice excludes the name AD")
# §2.4 mine tail: with flags.mine the AD is 18 B and ends with the tag's own 3 bytes (fixture qbrain_full_mine_tail)
fxq = json.load(open(FX)).get("qbrain_full_mine_tail")
if fxq:
    h = fxq["header"]
    ad = tb.mfg_ad(h["kind"], h["flags"], h["caps"], h["tools"], h["session"], h["rssi1m"], bytes([1, 2, 3]))
    assert ad.hex() == "11ff" + fxq["mfg_hex"], ad.hex()
    ad_nomine = tb.mfg_ad(h["kind"], h["flags"] & ~tb.FLAGS["mine"], h["caps"], h["tools"], h["session"], h["rssi1m"], bytes([1, 2, 3]))
    assert len(ad_nomine) == 15  # tail dropped without flags.mine
    assert tb.mine_tag(raw, 4242) == base64.urlsafe_b64encode(tb.mine_tail(raw, 4242)).decode().rstrip("=")
    a.social = True; a.flags = ""
    assert len(tb.build(a, 4242)) <= 31
    print("ok: mine tail = 18 B AD, tag bytes, 31 B budget kept with --social")

# ── v0.2: Catalog GATT + daemon roster (lane THOR-CATALOG 2026-09-21) ────────────────────────────
import hashlib, struct, tempfile, time as _time
# Fixture linux_catalog_mine_tail = the bytes Thor put on air (btmon + Mac scan, proof/thor-sighting.json):
# kind 7 linux · flags provisioned|catalog|mine|agent = 0x65 · caps wifi|relay = 0x4100 · tools 34 · session 58904 · rssi1m -63
LINUX_CATALOG = {"header": {"kind": 7, "flags": 0x65, "caps": 0x4100, "tools": 34, "session": 58904, "rssi1m": -63},
                 "mfg_hex_no_tail": "ffff544e10076500412218e6c1", "on_air": "11ffffff544e10076500412218e6c13a9585"}
h = LINUX_CATALOG["header"]
ad = tb.mfg_ad(h["kind"], h["flags"], h["caps"], h["tools"], h["session"], h["rssi1m"], bytes.fromhex("3a9585"))
assert ad.hex() == LINUX_CATALOG["on_air"], ad.hex()
assert ad[:15].hex() == "11ff" + LINUX_CATALOG["mfg_hex_no_tail"]
assert h["flags"] & tb.FLAGS["catalog"] and h["flags"] & tb.FLAGS["mine"] and not (h["flags"] & tb.FLAGS["social"])
a = _A(); a.kind = "linux"; a.flags = ""; a.caps = ""; a.tools = 0; a.rssi1m = -63; a.social = False; a.name = "thor"
assert tb.build_flags(a, catalog=True) == 0x65 and tb.build_flags(a, catalog=False) == 0x61 and tb.build_caps(a) == 0x4100
ad2 = tb.build(a, 58904, catalog=True, tools=34)
assert ad2[:15] == ad[:15] and len(ad2) == 18  # same header from the CLI path; tail differs (test secret)
print("ok: linux catalog AD == Thor's on-air bytes (flags 0x65, tools 34, 18 B)")

# roster: the freshest LIVE registry row wins; stale rows and dead pids are skipped; names sorted+deduped
now = int(_time.time() * 1000)
reg = {"agents": {
    "old": {"id": "old", "type": "zenoh", "pid": os.getpid(), "last_seen": now - 60_000, "metadata": {"tools": ["zzz"], "tool_count": 1}},
    "dead": {"id": "dead", "type": "zenoh", "pid": 2_000_000_000, "last_seen": now, "metadata": {"tools": ["dead_tool"], "tool_count": 1}},
    "live": {"id": "live", "type": "zenoh", "pid": os.getpid(), "last_seen": now - 1000,
             "metadata": {"tools": ["use_ble", "bash", "bash", "ask_tiny"], "tool_count": 34, "model": "bedrock:x"}},
}}
with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as f:
    json.dump(reg, f); regp = f.name
r = tb.roster_from_registry(regp)
assert r["id"] == "live" and r["names"] == ["ask_tiny", "bash", "use_ble"] and r["count"] == 34, r
assert tb.roster_from_registry(regp, require_pid=False)["id"] == "dead"  # only the pid check kept it out
assert tb.roster_from_registry("/nonexistent.json") is None
ro = tb.Roster(7, "daemon", regp)
assert ro.count == 34 and ro.names[0] == "ask_tiny" and ro.daemon == "live"
assert tb.Roster(7, "static").count == 7 and tb.Roster(7, "static").names == []
os.environ["TMPDIR"] = "/var/folders/x"; assert tb.registry_path() == "/var/folders/x/tiny/mesh_registry.json"; del os.environ["TMPDIR"]
assert tb.registry_path() == "/tmp/tiny/mesh_registry.json"
print("ok: roster = freshest live daemon row (stale + dead-pid rows skipped), registry path = os.tmpdir()/tiny")

# tools pages: every page ≤ 500 B, p/n consistent, union == names, order kept
names34 = sorted(["ask_tiny", "bash", "fileEditor", "mesh_broadcast", "mesh_peers", "mesh_send", "my_gh_activity", "my_tello", "my_time",
                  "my_units", "my_weather", "tiny_events", "tiny_learn", "tiny_messages", "tiny_pay_quote", "tiny_recall", "tiny_schedule",
                  "tiny_search", "tiny_send_message", "tiny_unlearn", "tiny_wallet", "tiny_whoami", "use_device", "use_github", "use_image",
                  "use_integrations", "use_mcp", "use_memory", "use_npm", "use_openapi", "use_ring", "use_ble", "use_tools", "use_pypi"])
pages = tb.tools_pages(names34)
assert len(pages) == 2 and all(len(p) <= tb.TOOLS_PAGE_MAX for p in pages), [len(p) for p in pages]
got = [json.loads(p) for p in pages]
assert [g["p"] for g in got] == [0, 1] and all(g["n"] == 2 for g in got)
assert [i["name"] for g in got for i in g["items"]] == names34
big = tb.tools_pages([f"tool_{i:03d}_with_a_rather_long_name" for i in range(120)])
assert len(big) >= 8 and all(len(p) <= 500 for p in big) and json.loads(big[-1])["n"] == len(big)
assert json.loads(tb.tools_pages([])[0]) == {"p": 0, "n": 1, "items": []}
assert json.loads(tb.tools_pages(["a"], descs={"a": "x" * 100})[0])["items"][0]["desc"] == "x" * 64
print("ok: tools pages ≤ 500 B, p/n consistent, 34 names → 2 pages")

# identity / endpoint shapes (SPEC §3): dev = sha256(device id)[:16], handle only with --social, ≤ 512 B
ident = json.loads(tb.identity_json(7, 0x65, 0x4100, 34, 58904, name="thor", handle="cagataycali", social=False, device_id="1c7c8fa9-x"))
assert ident["v"] == 2 and ident["kind"] == 7 and ident["tools"] == 34 and ident["flags"] == 0x65 and ident["name"] == "thor"
assert len(ident["dev"]) == 16 and ident["dev"] == hashlib.sha256(b"1c7c8fa9-x").hexdigest()[:16] and "handle" not in ident
assert ident["fw"].startswith("tiny-beacon ") and "-" in ident["platform"]
assert "handle" in json.loads(tb.identity_json(7, 0x67, 0, 1, 1, name="thor", handle="cagataycali", social=True))
assert len(tb.identity_json(7, 0x65, 0, 255, 65535, name="x" * 600)) <= tb.ATT_VALUE_MAX
ep = json.loads(tb.endpoint_json("dev-1", ip="192.168.1.151", zenoh_port=1))  # port 1 is never listening
assert ep == {"ip": "192.168.1.151", "device": "tinyapp://device/dev-1"}, ep
os.environ["TINY_ZENOH_LOCATOR"] = "tcp/10.0.0.2:7447"
assert json.loads(tb.endpoint_json(None, ip="10.0.0.2"))["zenoh"] == "tcp/10.0.0.2:7447"; del os.environ["TINY_ZENOH_LOCATOR"]
# Catalog state machine without a radio: page select per central, hello ring, hello_first only when social
class _R: names = names34; count = 34
a.handle = None; a.http = ""
tb.emit = lambda o: None  # quiet the event log during the state-machine checks
cat = tb.Catalog(a, _R(), 58904, {"deviceId": "dev-1"})
p0 = json.loads(cat.read_tools({"device": "/org/bluez/hci0/dev_AA"}))
cat.write_tools(b"\x01", {"device": "/org/bluez/hci0/dev_AA"})
p1 = json.loads(cat.read_tools({"device": "/org/bluez/hci0/dev_AA"}))
assert p0["p"] == 0 and p1["p"] == 1 and json.loads(cat.read_tools({"device": "/org/bluez/hci0/dev_BB"}))["p"] == 0
cat.write_tools(b"\x09", {"device": "/org/bluez/hci0/dev_AA"})
assert json.loads(cat.read_tools({"device": "/org/bluez/hci0/dev_AA"}))["p"] == 1  # clamped to the last page
a.social = True
assert json.loads(cat.read_tools({"device": "/org/bluez/hci0/dev_CC"}))["err"] == "hello_first"
cat.write_hello(b'{"v":2,"kind":"mac","name":"cagatay-mac"}', {"device": "/org/bluez/hci0/dev_CC"})
assert "err" not in json.loads(cat.read_tools({"device": "/org/bluez/hci0/dev_CC"})) and cat.hellos[-1]["hello"]["kind"] == "mac"
for i in range(40): cat.write_hello(b"not json", {"device": "x"})
assert len(cat.hellos) == tb.HELLO_RING and cat.hellos[-1]["hello"] == {"raw": "not json"}
a.social = False
assert json.loads(cat.read_identity({"device": "x"}))["session"] == 58904
print("ok: identity/endpoint shapes, per-central paging, hello ring 32, hello_first gate")
# the dbus path is never imported by --encode/tests; radio caps are reported honestly
assert tb.radio_caps().keys() == {"bluez", "scan_rsp_uuid", "interval_honoured"}
assert tb.bluez_experimental("/nonexistent") is False
with tempfile.NamedTemporaryFile("w", suffix=".conf", delete=False) as f:
    f.write("[General]\n# x\nExperimental = true\n"); cp = f.name
assert tb.bluez_experimental(cp) is True
print("ok: radio caps report (bluez version, scan-rsp UUID ≥ 5.80, interval needs Experimental=true)")