"""hive — the swarm address book service (see hive/app.py).
The app module reads env config at import time, so each test reimports it with
fresh env into its own key file / database.
"""
import importlib
import sqlite3
import pytest
from fastapi.testclient import TestClient
@pytest.fixture()
def hive(tmp_path, monkeypatch):
monkeypatch.setenv("HIVE_KEY_FILE", str(tmp_path / ".swarm-key"))
monkeypatch.setenv("HIVE_DB", str(tmp_path / "hive.db"))
monkeypatch.setenv("HIVE_TTL_SEC", "300")
# Deterministic admin key: no ambient .hive-admin-key from the checkout.
monkeypatch.setenv("HIVE_ADMIN_KEY", "admin-secret")
monkeypatch.setenv("HIVE_ADMIN_KEY_FILE", str(tmp_path / ".hive-admin-key"))
(tmp_path / ".swarm-key").write_text("secret-key-123")
import hive.app as hive_app
module = importlib.reload(hive_app)
yield module
module._db.close()
def _headers(key: str = "secret-key-123") -> dict:
return {"X-Swarm-Key": key}
def _announce(client, name="quiet-otter", instance_id="11111111-2222-3333-4444-555555555555", port=8099):
return client.post(
"/announce",
json={"name": name, "instance_id": instance_id, "version": "0.1.0", "port": port,
"meta": {"hostname": "gpu-box", "os": "Linux 6.1", "cpu_cores": 16, "ram_mb": 65536}},
headers=_headers(),
)
def test_health_open_without_key(hive):
with TestClient(hive.app) as client:
resp = client.get("/health")
assert resp.status_code == 200
assert resp.json()["hive"] is True
def test_announce_requires_key(hive):
with TestClient(hive.app) as client:
assert client.post("/announce", json={"name": "x", "instance_id": "12345678", "port": 8099}).status_code == 401
assert client.post(
"/announce",
json={"name": "x", "instance_id": "12345678", "port": 8099},
headers={"X-Swarm-Key": "wrong"},
).status_code == 403
def test_peers_requires_key(hive):
with TestClient(hive.app) as client:
assert client.get("/peers").status_code == 401
assert client.get("/peers", headers={"X-Swarm-Key": "wrong"}).status_code == 403
def test_announce_upsert_and_peers_listing(hive):
with TestClient(hive.app) as client:
assert _announce(client).status_code == 200
# same instance_id again = upsert (renamed, different port), not a second row
assert _announce(client, name="renamed-otter", port=8123).status_code == 200
# a second machine
assert _announce(client, name="amber-falcon", instance_id="99999999-2222-3333-4444-555555555555").status_code == 200
resp = client.get("/peers", headers=_headers())
assert resp.status_code == 200
body = resp.json()
peers = body["peers"]
assert len(peers) == 2
assert [p["name"] for p in peers] == ["amber-falcon", "renamed-otter"] # sorted by name
otter = peers[1]
assert otter["port"] == 8123
assert otter["address"] == f"{otter['host']}:8123"
assert otter["online"] is True
assert otter["meta"]["hostname"] == "gpu-box"
assert otter["meta"]["ram_mb"] == 65536
assert otter["version"] == "0.1.0"
assert body["ttl_sec"] == 300
def test_offline_after_ttl(hive, tmp_path):
with TestClient(hive.app) as client:
_announce(client)
# backdate the last announcement beyond the TTL — machine went quiet
conn = sqlite3.connect(tmp_path / "hive.db")
conn.execute("UPDATE machines SET last_seen = last_seen - 400")
conn.commit()
conn.close()
peers = client.get("/peers", headers=_headers()).json()["peers"]
assert peers[0]["online"] is False
def test_previous_key_still_accepted(hive, tmp_path):
# rotation window: .swarm-key.previous keeps working
(tmp_path / ".swarm-key.previous").write_text("old-key-42")
import hive.app as module
importlib.reload(module)
with TestClient(module.app) as client:
resp = client.post(
"/announce",
json={"name": "x", "instance_id": "12345678", "port": 8099},
headers={"X-Swarm-Key": "old-key-42"},
)
assert resp.status_code == 200
module._db.close()
def test_hive_without_key_file_refuses(hive, tmp_path, monkeypatch):
monkeypatch.setenv("HIVE_KEY_FILE", str(tmp_path / ".no-key"))
monkeypatch.setenv("HIVE_DB", str(tmp_path / "hive2.db"))
import hive.app as module
importlib.reload(module)
with TestClient(module.app) as client:
resp = client.post(
"/announce",
json={"name": "x", "instance_id": "12345678", "port": 8099},
headers={"X-Swarm-Key": "whatever"},
)
assert resp.status_code == 503
module._db.close()
# ── config documents ────────────────────────────────────────────────────────
def _admin(key: str = "admin-secret") -> dict:
return {"X-Admin-Key": key}
def _publish(client, layers, note="", force=False, key="admin-secret"):
return client.put(
"/config",
json={"layers": layers, "note": note, "force": force},
headers=_admin(key),
)
def _global(**settings):
return {"name": "global", "config": {"settings": settings}}
def test_publish_requires_admin_key(hive):
with TestClient(hive.app) as client:
resp = client.put("/config", json={"layers": [_global(llm_complete_timeout=30)]})
assert resp.status_code == 403
bad = client.put(
"/config",
json={"layers": [_global(llm_complete_timeout=30)]},
headers=_admin("wrong"),
)
assert bad.status_code == 403
def test_fetch_returns_404_before_anything_is_published(hive):
with TestClient(hive.app) as client:
assert client.get("/config", headers=_headers()).status_code == 404
def test_publish_then_fetch_global_layer(hive):
with TestClient(hive.app) as client:
published = _publish(client, [_global(llm_complete_timeout=30)]).json()
assert published["status"] == "published"
assert published["version"] == 1
body = client.get("/config", headers=_headers()).json()
assert body["version"] == 1
assert body["layers"] == [
{"name": "global", "config": {"settings": {"llm_complete_timeout": 30}}}
]
def test_layers_target_by_tag_and_name(hive):
layers = [
{"name": "global", "config": {"settings": {"llm_complete_timeout": 30}}},
{"name": "lab", "match": {"tags": ["lab"]}, "config": {"settings": {"llm_complete_retries": 1}}},
{"name": "yuki", "match": {"names": ["yuki"]}, "config": {"settings": {"navi_push_cooldown_sec": 5}}},
]
with TestClient(hive.app) as client:
_publish(client, layers)
plain = client.get("/config", params={"name": "other"}, headers=_headers()).json()
assert [layer["name"] for layer in plain["layers"]] == ["global"]
lab = client.get(
"/config", params={"name": "other", "tags": "beta,lab"}, headers=_headers()
).json()
assert [layer["name"] for layer in lab["layers"]] == ["global", "lab"]
named = client.get("/config", params={"name": "yuki"}, headers=_headers()).json()
assert [layer["name"] for layer in named["layers"]] == ["global", "yuki"]
# The hash is over the layers that instance receives, not the document.
assert plain["hash"] != named["hash"]
def test_identical_publish_is_a_no_op(hive):
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)])
again = _publish(client, [_global(llm_complete_timeout=30)]).json()
assert again["status"] == "unchanged"
assert again["version"] == 1
forced = _publish(client, [_global(llm_complete_timeout=30)], force=True).json()
assert forced["status"] == "renotified"
assert forced["version"] == 1
assert len(client.get("/config/history", headers=_admin()).json()["versions"]) == 1
def test_empty_document_clears_the_overlay(hive):
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)])
cleared = _publish(client, [], note="back to local").json()
assert cleared["status"] == "published"
assert cleared["version"] == 2
body = client.get("/config", headers=_headers()).json()
assert body["layers"] == []
def test_invalid_document_is_rejected_with_a_reason(hive):
with TestClient(hive.app) as client:
resp = _publish(client, [{"name": "x"}])
assert resp.status_code == 400
assert "config" in resp.json()["detail"]
resp = _publish(client, [{"name": "x", "config": {}, "match": {"nope": ["a"]}}])
assert resp.status_code == 400
def test_announce_hint_tells_a_behind_instance_to_fetch(hive):
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)])
# An older build that reports no config: unknown, not behind.
old = _announce(client).json()
assert old["config"] is None
# A current build with an empty hash is behind.
behind = client.post(
"/announce",
json={
"name": "quiet-otter",
"instance_id": "11111111-2222-3333-4444-555555555555",
"port": 8099,
"meta": {"config": {"hash": "", "status": "applied"}},
},
headers=_headers(),
).json()
assert behind["config"]["fetch"] is True
assert behind["config"]["version"] == 1
# One that reports the exact effective hash is up to date.
effective = client.get("/config", headers=_headers()).json()["hash"]
current = client.post(
"/announce",
json={
"name": "quiet-otter",
"instance_id": "11111111-2222-3333-4444-555555555555",
"port": 8099,
"meta": {"config": {"hash": effective, "status": "applied"}},
},
headers=_headers(),
).json()
assert current["config"] is None
def test_ack_is_recorded_and_shown_in_status(hive):
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)])
_announce(client)
resp = client.post(
"/config/ack",
json={
"instance_id": "11111111-2222-3333-4444-555555555555",
"version": 1,
"hash": "sha256:x",
"status": "staged",
"restart_required": ["openai_api_key"],
},
headers=_headers(),
)
assert resp.status_code == 200
status = client.get("/config/status", headers=_admin()).json()
row = status["instances"][0]
assert row["ack"]["status"] == "staged"
assert row["ack"]["restart_required"] == ["openai_api_key"]
def test_status_marks_behind_instances(hive):
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)])
effective = client.get("/config", headers=_headers()).json()["hash"]
client.post(
"/announce",
json={
"name": "quiet-otter",
"instance_id": "11111111-2222-3333-4444-555555555555",
"port": 8099,
"meta": {"config": {"hash": "sha256:old", "status": "applied"}},
},
headers=_headers(),
)
status = client.get("/config/status", headers=_admin()).json()
assert status["behind"] == ["quiet-otter"]
assert status["instances"][0]["effective_hash"] == effective
client.post(
"/announce",
json={
"name": "quiet-otter",
"instance_id": "11111111-2222-3333-4444-555555555555",
"port": 8099,
"meta": {"config": {"hash": effective, "status": "applied"}},
},
headers=_headers(),
)
assert client.get("/config/status", headers=_admin()).json()["behind"] == []
def test_history_version_and_rollback(hive):
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)], note="first")
_publish(client, [_global(llm_complete_timeout=45)], note="second")
history = client.get("/config/history", headers=_admin()).json()["versions"]
assert [v["id"] for v in history] == [2, 1]
assert history[1]["note"] == "first"
version = client.get("/config/version/1", headers=_admin()).json()
assert version["layers"][0]["config"]["settings"]["llm_complete_timeout"] == 30
rolled = client.post("/config/rollback/1", headers=_admin()).json()
assert rolled["status"] == "rolled_back"
assert rolled["version"] == 3
assert client.get("/config", headers=_headers()).json()["layers"][0]["config"] == {
"settings": {"llm_complete_timeout": 30}
}
# History keeps the bad version rather than rewriting it.
assert len(client.get("/config/history", headers=_admin()).json()["versions"]) == 3
async def test_notify_all_pings_every_registered_instance(hive, monkeypatch):
with TestClient(hive.app) as client:
_announce(client, name="one", instance_id="aaaaaaaa-0000-0000-0000-000000000001")
_announce(client, name="two", instance_id="aaaaaaaa-0000-0000-0000-000000000002")
calls = []
async def fake(peer, version, doc_hash):
calls.append((peer["name"], version, doc_hash))
return True
monkeypatch.setattr(hive, "_notify_one", fake)
await hive._notify_all(4, "sha256:x")
assert sorted(name for name, _, _ in calls) == ["one", "two"]
async def test_retry_sweep_targets_only_the_instance_that_is_behind(hive, monkeypatch):
with TestClient(hive.app) as client:
_announce(client, name="one", instance_id="aaaaaaaa-0000-0000-0000-000000000001")
_announce(client, name="two", instance_id="aaaaaaaa-0000-0000-0000-000000000002")
_publish(client, [_global(llm_complete_timeout=30)])
effective = client.get("/config", headers=_headers()).json()["hash"]
client.post(
"/announce",
json={
"name": "one",
"instance_id": "aaaaaaaa-0000-0000-0000-000000000001",
"port": 8099,
"meta": {"config": {"hash": effective, "status": "applied"}},
},
headers=_headers(),
)
client.post(
"/announce",
json={
"name": "two",
"instance_id": "aaaaaaaa-0000-0000-0000-000000000002",
"port": 8099,
"meta": {"config": {"hash": "sha256:old", "status": "applied"}},
},
headers=_headers(),
)
calls = []
async def fake(peer, version, doc_hash):
calls.append(peer["name"])
return True
monkeypatch.setattr(hive, "_notify_one", fake)
await hive._notify_behind()
assert calls == ["two"]
def test_a_fresh_ack_beats_a_stale_announce(hive):
"""The announce may lag an interval behind; an ack for the current hash is proof."""
iid = "11111111-2222-3333-4444-555555555555"
with TestClient(hive.app) as client:
_publish(client, [_global(llm_complete_timeout=30)])
effective = client.get("/config", headers=_headers()).json()["hash"]
client.post(
"/announce",
json={
"name": "quiet-otter",
"instance_id": iid,
"port": 8099,
"meta": {"config": {"hash": "sha256:old", "status": "applied"}},
},
headers=_headers(),
)
assert client.get("/config/status", headers=_admin()).json()["behind"] == [
"quiet-otter"
]
client.post(
"/config/ack",
json={"instance_id": iid, "version": 1, "hash": effective, "status": "applied"},
headers=_headers(),
)
assert client.get("/config/status", headers=_admin()).json()["behind"] == []
