File size: 6,970 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 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 | """Preflight + dream-corpus endpoints — the Coach UI's first two gates
before kicking off a production training run.
Both are pure read endpoints; tests verify the response shape and that the
required-missing logic is correctly partitioned from optional vars.
"""
from __future__ import annotations
import json
from pathlib import Path
from fastapi.testclient import TestClient
from mindxtrain.operator.app import app
client = TestClient(app)
# ---- preflight -----------------------------------------------------------
_REQUIRED = ("AMD_DEV_CLOUD_TOKEN", "AMD_DEV_CLOUD_SSH_KEY_ID", "HF_TOKEN", "HF_HUB_USERNAME")
_OPTIONAL = ("MINDXTRAIN_API_KEY", "MINDXTRAIN_MINDX_HOME", "LIGHTHOUSE_API_KEY")
def _strip_preflight_env(monkeypatch):
"""Clear every env var the preflight endpoint inspects."""
for name in _REQUIRED + _OPTIONAL:
monkeypatch.delenv(name, raising=False)
def test_preflight_reports_required_missing(monkeypatch):
_strip_preflight_env(monkeypatch)
r = client.get("/coach/api/preflight")
assert r.status_code == 200
data = r.json()
assert data["ready"] is False
assert set(data["required_missing"]) == set(_REQUIRED)
# All vars present as keys, all False
assert all(data["vars"][n] is False for n in _REQUIRED + _OPTIONAL)
def test_preflight_ready_when_required_set(monkeypatch):
_strip_preflight_env(monkeypatch)
for name in _REQUIRED:
monkeypatch.setenv(name, "fake-value")
r = client.get("/coach/api/preflight")
assert r.status_code == 200
data = r.json()
assert data["ready"] is True
assert data["required_missing"] == []
assert all(data["vars"][n] is True for n in _REQUIRED)
# Optional still unset — should not gate readiness
assert all(data["vars"][n] is False for n in _OPTIONAL)
def test_preflight_empty_string_treated_as_unset(monkeypatch):
"""Whitespace-only / empty values must not count as 'set'."""
_strip_preflight_env(monkeypatch)
for name in _REQUIRED:
monkeypatch.setenv(name, " ") # whitespace only
r = client.get("/coach/api/preflight")
assert r.status_code == 200
assert r.json()["ready"] is False
def test_preflight_does_not_echo_values(monkeypatch):
"""Endpoint must never leak the actual secret values."""
_strip_preflight_env(monkeypatch)
monkeypatch.setenv("HF_TOKEN", "hf_super_secret_xyz")
body = client.get("/coach/api/preflight").text
assert "hf_super_secret_xyz" not in body
# ---- dream-corpus ----------------------------------------------------------
def _seed_corpus(tmp_path: Path, n_examples: int, n_evolutions: int = 0) -> Path:
"""Build a minimal LTM tree shaped like mindX/data/memory.
Writes `n_examples` consolidation rows and (optionally) `n_evolutions`
proposal rows so we can exercise the two-bucket response shape.
"""
root = tmp_path / "data" / "memory"
ltm = root / "ltm" / "agent_a"
ltm.mkdir(parents=True)
train = ltm / "20260514_010000_training.jsonl"
with train.open("w") as fh:
for i in range(n_examples):
fh.write(
json.dumps({
"messages": [
{"role": "system", "content": "test"},
{"role": "user", "content": f"q{i}"},
{"role": "assistant", "content": f"a{i}"},
]
})
+ "\n"
)
if n_evolutions:
evo = ltm / "20260514_010001_evolutions.jsonl"
with evo.open("w") as fh:
for i in range(n_evolutions):
fh.write(
json.dumps({
"messages": [
{"role": "system", "content": "evo"},
{"role": "user", "content": f"insight{i}"},
{"role": "assistant", "content": f"proposal{i}"},
]
})
+ "\n"
)
return root
def test_dream_corpus_uses_env_root(monkeypatch, tmp_path):
root = _seed_corpus(tmp_path, n_examples=5)
monkeypatch.setenv("MINDXTRAIN_MINDX_HOME", str(tmp_path))
r = client.get("/coach/api/dream-corpus")
assert r.status_code == 200
data = r.json()
assert data["exists"] is True
assert data["consolidation"]["unique_rows"] == 5
assert data["evolutions"]["unique_rows"] == 0
assert data["ready"] is True
assert data["note"] is None
assert data["root"] == str(root)
def test_dream_corpus_reports_both_buckets(monkeypatch, tmp_path):
"""Two-bucket shape: consolidation + evolutions independently."""
_seed_corpus(tmp_path, n_examples=5, n_evolutions=3)
monkeypatch.setenv("MINDXTRAIN_MINDX_HOME", str(tmp_path))
r = client.get("/coach/api/dream-corpus")
assert r.status_code == 200
data = r.json()
assert data["consolidation"]["unique_rows"] == 5
assert data["consolidation"]["files"] == 1
assert data["evolutions"]["unique_rows"] == 3
assert data["evolutions"]["files"] == 1
assert data["ready"] is True
def test_dream_corpus_ready_with_evolutions_only(monkeypatch, tmp_path):
"""Backward compat-adjacent: ready=True if EITHER bucket has rows."""
_seed_corpus(tmp_path, n_examples=0, n_evolutions=4)
monkeypatch.setenv("MINDXTRAIN_MINDX_HOME", str(tmp_path))
r = client.get("/coach/api/dream-corpus")
data = r.json()
assert data["consolidation"]["unique_rows"] == 0
assert data["evolutions"]["unique_rows"] == 4
assert data["ready"] is True
def test_dream_corpus_explicit_root_arg(monkeypatch, tmp_path):
root = _seed_corpus(tmp_path, n_examples=3)
monkeypatch.delenv("MINDXTRAIN_MINDX_HOME", raising=False)
r = client.get(f"/coach/api/dream-corpus?root={root}")
assert r.status_code == 200
data = r.json()
assert data["root"] == str(root)
assert data["consolidation"]["unique_rows"] == 3
assert data["ready"] is True
def test_dream_corpus_missing_root_returns_friendly_note(monkeypatch, tmp_path):
monkeypatch.setenv("MINDXTRAIN_MINDX_HOME", str(tmp_path / "nowhere"))
r = client.get("/coach/api/dream-corpus")
assert r.status_code == 200
data = r.json()
assert data["exists"] is False
assert data["ready"] is False
assert data["note"] is not None
assert "MINDXTRAIN_MINDX_HOME" in data["note"]
def test_dream_corpus_empty_tree_reports_not_ready(monkeypatch, tmp_path):
(tmp_path / "data" / "memory" / "ltm").mkdir(parents=True)
monkeypatch.setenv("MINDXTRAIN_MINDX_HOME", str(tmp_path))
r = client.get("/coach/api/dream-corpus")
assert r.status_code == 200
data = r.json()
assert data["exists"] is True
assert data["consolidation"]["unique_rows"] == 0
assert data["evolutions"]["unique_rows"] == 0
assert data["ready"] is False
assert data["note"] is not None
assert "dream cycle" in data["note"].lower()
|