File size: 3,610 Bytes
dfb775d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 | """dcoach page routes + decentralized panel + prompt-eval + streamed proof loop."""
from __future__ import annotations
import json
from fastapi.testclient import TestClient
from mindxtrain.operator.app import app
client = TestClient(app)
def test_dcoach_and_prompts_pages_serve():
for path in ("/coach/dcoach", "/coach/prompts"):
r = client.get(path)
assert r.status_code == 200, r.text
assert "text/html" in r.headers["content-type"]
def test_decentralized_panel_data():
r = client.get("/coach/api/decentralized")
assert r.status_code == 200, r.text
data = r.json()
assert data["thesis"]
names = {n["name"] for n in data["networks"]}
# every 2026 network the deep-dive covers is represented
assert {"Templar · Bittensor SN3", "Gensyn", "Pluralis · Node0"} <= names
for n in data["networks"]:
assert {"name", "what", "hardware", "token", "fit"} <= set(n)
assert any("Verde" in f["maps_to"] for f in data["fit"])
def test_eval_prompt_similarity_only():
# no judge → only the free semantic-similarity evaluator runs
r = client.post("/coach/api/eval/prompt", json={
"query": "who are you?",
"response": "i am codephreak, augmentic intelligence.",
"reference": "i am codephreak, augmentic intelligence.",
"use_judge": False,
})
assert r.status_code == 200, r.text
body = r.json()
assert "semantic_similarity" in body["scores"]
assert "correctness" not in body["scores"]
assert body["scores"]["semantic_similarity"]["score"] > 0.9
assert body["overall"] > 0.9 and body["advantageous"] is True
def test_dcoach_run_streams_phases(monkeypatch):
"""Patch the heavy loop with a fast fake; assert the SSE bridge streams phases
+ the terminal result + [DONE]."""
from mindxtrain.governance import classroom as _cr
from mindxtrain.governance import proof_loop as _pl
report = _cr.ClassroomReport(
inquiries=["who?"], before=["an AI"], after=["codephreak"],
before_recall=0.1, recall=0.6, imprint_delta=0.5,
pairwise_after_better=1.0, persona_maintained=True, passed=True,
)
def _fake(*, run_id, on_event=None, **kw):
if on_event:
on_event("dataset", "authored 5 rows")
on_event("train", "imprinting…")
return _pl.ProofResult(
run_id=run_id, dataset_path="/tmp/s.jsonl", rows=5,
train_params={"epochs": 12, "grad_accum": 1, "per_device": 1},
classroom=report, boardroom_outcome="approved",
boardroom_rationale="approved 3-0", passed=True,
next_params={"epochs": 12, "grad_accum": 1, "per_device": 1},
)
monkeypatch.setattr(_pl, "run_proof_loop", _fake)
with client.stream("POST", "/coach/api/dcoach/run",
json={"persona": "codephreak", "run_id": "t1"}) as resp:
assert resp.status_code == 200
events = []
for line in resp.iter_lines():
if line.startswith("data:"):
payload = line[len("data:"):].strip()
if payload == "[DONE]":
events.append({"phase": "done"})
else:
events.append(json.loads(payload))
phases = [e["phase"] for e in events]
assert phases[0] == "start"
assert "dataset" in phases and "train" in phases
assert "result" in phases and phases[-1] == "done"
result = next(e["result"] for e in events if e["phase"] == "result")
assert result["passed"] is True
assert result["boardroom_outcome"] == "approved"
|