tests.rs 14 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437
  1. //! Web module tests: bus semantics, token gate, and a full server boot on an
  2. //! ephemeral port with a fake (unreachable) node.
  3. use crate::config::{Cluster, Config, Node};
  4. use crate::web::events::{Event, build_snapshot};
  5. use crate::web::{AppState, router};
  6. use std::collections::HashMap;
  7. use std::net::SocketAddr;
  8. pub(crate) async fn call(addr: SocketAddr, path: &str) -> reqwest::Response {
  9. reqwest::Client::new()
  10. .get(format!("http://{addr}{path}"))
  11. .send()
  12. .await
  13. .unwrap()
  14. }
  15. /// Config with one node at 127.0.0.1:2222 — connection refused instantly,
  16. /// machine-independent (no real SSH, no hardware).
  17. pub(crate) fn fake_config() -> Config {
  18. Config {
  19. ssh_user: "u".into(),
  20. ssh_key_path: "/nonexistent/key".into(),
  21. clusters: vec![Cluster {
  22. name: "fake".into(),
  23. master_ib_ip: None,
  24. nodes: vec![Node {
  25. name: "n0".into(),
  26. ssh_ip: "127.0.0.1:2222".into(),
  27. ib_ip: None,
  28. rank: 0,
  29. host_key: None,
  30. entity: None,
  31. }],
  32. extra_entities: vec![],
  33. }],
  34. nccl: HashMap::new(),
  35. hf_cache_dir: None,
  36. init_port: 25000,
  37. before_serve: HashMap::new(),
  38. exec_aliases: HashMap::new(),
  39. home_assistant: None,
  40. mikrotik: None,
  41. global_mods_dir: None,
  42. drop_caches: true,
  43. mount_aliases: HashMap::new(),
  44. recipe_sources: Vec::new(),
  45. recipe_repos: Vec::new(),
  46. docker_image_sources: Vec::new(),
  47. cluster_sources: Vec::new(),
  48. registry: None,
  49. }
  50. }
  51. #[path = "tests_auth.rs"]
  52. mod auth;
  53. pub(crate) fn app_state(token: Option<&str>) -> AppState {
  54. app_state_with(fake_config(), token)
  55. }
  56. pub(crate) fn app_state_with(config: Config, token: Option<&str>) -> AppState {
  57. let (tx, _) = tokio::sync::broadcast::channel(16);
  58. AppState {
  59. config: std::sync::Arc::new(config),
  60. bus: tx,
  61. token: token.map(String::from),
  62. sessions: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
  63. last: std::sync::Arc::new(std::sync::Mutex::new(None)),
  64. jobs: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
  65. log_attaches: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
  66. }
  67. }
  68. pub(crate) async fn boot() -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
  69. boot_with(fake_config()).await
  70. }
  71. pub(crate) async fn boot_with(
  72. config: Config,
  73. ) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
  74. boot_app(app_state_with(config, None)).await
  75. }
  76. async fn boot_app(app: AppState) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
  77. let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
  78. let addr = listener.local_addr().unwrap();
  79. let handle = tokio::spawn(async move {
  80. axum::serve(listener, router(app)).await.unwrap();
  81. });
  82. (addr, std::sync::Arc::new(handle))
  83. }
  84. #[tokio::test]
  85. async fn api_state_reports_unreachable_node() {
  86. let (addr, handle) = boot().await;
  87. let body: serde_json::Value = reqwest::get(format!("http://{addr}/api/state"))
  88. .await
  89. .unwrap()
  90. .json()
  91. .await
  92. .unwrap();
  93. assert_eq!(body["clusters"].as_array().unwrap().len(), 1);
  94. assert_eq!(body["clusters"][0]["name"], "fake");
  95. assert_eq!(body["clusters"][0]["nodes"][0]["reachable"], false);
  96. assert_eq!(body["clusters"][0]["free"], 0);
  97. handle.abort();
  98. }
  99. /// The recipes panel is mounted under /recipes: the page, its own SSE
  100. /// stream and the /r/ viewer all live behind the prefix.
  101. #[tokio::test]
  102. async fn recipes_panel_is_mounted_under_prefix() {
  103. let (addr, handle) = boot().await;
  104. let page = reqwest::get(format!("http://{addr}/recipes"))
  105. .await
  106. .unwrap()
  107. .text()
  108. .await
  109. .unwrap();
  110. // its own live stream resolves behind the prefix (headers arrive at once)
  111. let sse = reqwest::get(format!("http://{addr}/recipes/events"))
  112. .await
  113. .unwrap();
  114. assert_eq!(sse.status(), 200);
  115. assert!(
  116. page.contains("auth(\"/recipes/events\")"),
  117. "prefix not applied"
  118. );
  119. assert!(!page.contains("auth(\"/events\")"));
  120. // the dashboard's own routes still resolve
  121. let ui = reqwest::get(format!("http://{addr}/"))
  122. .await
  123. .unwrap()
  124. .text()
  125. .await
  126. .unwrap();
  127. assert!(ui.contains("recwin") || ui.contains("Recipes"));
  128. // Recipe logs open a dashboard WINDOW (postMessage bridge), not an inline
  129. // stream: the report page posts {sm121:"openLogs"}, the dashboard listens
  130. // and opens its logs window. No inline /api/logs fetch in the report page.
  131. assert!(
  132. page.contains("sm121: \"openLogs\"") && !page.contains("fetch(\"/api/logs\""),
  133. "report page should postMessage for logs, not stream inline"
  134. );
  135. assert!(
  136. ui.contains("openLogsForRecipe") && ui.contains("sm121 === \"openLogs\""),
  137. "dashboard should handle the recipes iframe logs message"
  138. );
  139. handle.abort();
  140. }
  141. #[test]
  142. fn jobs_persist_and_mark_interrupted() {
  143. use crate::web::jobs;
  144. let app = app_state(None);
  145. let dir = std::env::temp_dir().join(format!("sm121-jobs-test-{}", std::process::id()));
  146. std::fs::create_dir_all(&dir).unwrap();
  147. let path = dir.join("jobs.json");
  148. // a registry snapshot with a running job: the process died mid-flight
  149. std::fs::write(
  150. &path,
  151. r#"[{"id":"j9","kind":"serve","cluster":"c","recipe":"r","status":"running","message":"started","started":1},
  152. {"id":"j8","kind":"power","cluster":"c","recipe":null,"status":"ok","message":"on","started":2}]"#,
  153. )
  154. .unwrap();
  155. jobs::load_from(Some(&path), &app);
  156. let jobs = app.jobs.lock().unwrap().clone();
  157. std::fs::remove_dir_all(&dir).unwrap();
  158. assert_eq!(jobs.len(), 2);
  159. let running = jobs.iter().find(|j| j.id == "j9").unwrap();
  160. assert_eq!(
  161. running.status,
  162. jobs::JobStatus::Failed,
  163. "running -> failed on boot"
  164. );
  165. assert!(
  166. running.message.contains("runlock"),
  167. "the ledger hint: {}",
  168. running.message
  169. );
  170. let done = jobs.iter().find(|j| j.id == "j8").unwrap();
  171. assert_eq!(done.status, jobs::JobStatus::Ok);
  172. }
  173. #[test]
  174. fn stale_from_snapshot_picks_unreachable_nodes() {
  175. let app = app_state(None);
  176. use crate::web::api::stale_from_snapshot;
  177. let node = |name: &str, ip: &str, reachable: bool| crate::web::events::NodeSnap {
  178. name: name.into(),
  179. ssh_ip: ip.into(),
  180. reachable,
  181. occupied: false,
  182. containers: Vec::new(),
  183. rank: 0,
  184. ha: None,
  185. };
  186. let snap = crate::web::events::Snapshot {
  187. ts: 1,
  188. clusters: vec![crate::web::events::ClusterSnap {
  189. name: "c".into(),
  190. up: false,
  191. switch_off: false,
  192. soft_off: Vec::new(),
  193. free: 0,
  194. nodes: vec![node("a", "1.1.1.1", true), node("b", "1.1.1.2", false)],
  195. }],
  196. jobs: Vec::new(),
  197. topo: Default::default(),
  198. attaches: Vec::new(),
  199. };
  200. *app.last.lock().unwrap() = Some(snap);
  201. assert_eq!(stale_from_snapshot(&app, "c"), vec!["b".to_string()]);
  202. assert!(stale_from_snapshot(&app, "other").is_empty());
  203. }
  204. #[tokio::test]
  205. async fn token_gate_covers_recipes_panel() {
  206. let app = app_state(Some("s3cret"));
  207. *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
  208. ts: 1,
  209. clusters: Vec::new(),
  210. jobs: Vec::new(),
  211. topo: Default::default(),
  212. attaches: Vec::new(),
  213. });
  214. let (addr, handle) = boot_app(app).await;
  215. let base = format!("http://{addr}");
  216. // bare requests are rejected on every panel route...
  217. for path in ["/recipes", "/recipes/events"] {
  218. let st = reqwest::get(format!("{base}{path}"))
  219. .await
  220. .unwrap()
  221. .status();
  222. assert_eq!(st, 401, "{path} must be gated");
  223. }
  224. // ...and the session cookie (issued by POST /api/login) opens them —
  225. // the iframe and the report page's SSE ride the cookie, not the token.
  226. let client = reqwest::Client::new();
  227. let res = client
  228. .post(format!("{base}/api/login"))
  229. .header("authorization", "Bearer s3cret")
  230. .send()
  231. .await
  232. .unwrap();
  233. assert_eq!(res.status(), 204);
  234. let cookie = res
  235. .headers()
  236. .get("set-cookie")
  237. .unwrap()
  238. .to_str()
  239. .unwrap()
  240. .split(';')
  241. .next()
  242. .unwrap()
  243. .to_string();
  244. let page = client
  245. .get(format!("{base}/recipes"))
  246. .header("cookie", &cookie)
  247. .send()
  248. .await
  249. .unwrap();
  250. assert_eq!(page.status(), 200, "the recipes page opens with the cookie");
  251. let sse = client
  252. .get(format!("{base}/recipes/events"))
  253. .header("cookie", &cookie)
  254. .send()
  255. .await
  256. .unwrap();
  257. assert_eq!(sse.status(), 200);
  258. drop(sse);
  259. handle.abort();
  260. }
  261. #[tokio::test]
  262. async fn sse_replays_cached_snapshot_on_connect() {
  263. use tokio::io::{AsyncReadExt, AsyncWriteExt};
  264. // Nothing is broadcast after this point: the frame must come from the
  265. // cached snapshot, otherwise a fresh client waits a whole poll tick.
  266. let app = app_state(None);
  267. *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
  268. ts: 1,
  269. clusters: Vec::new(),
  270. jobs: Vec::new(),
  271. topo: Default::default(),
  272. attaches: Vec::new(),
  273. });
  274. let (addr, handle) = boot_app(app).await;
  275. let mut sock = tokio::net::TcpStream::connect(addr).await.unwrap();
  276. sock.write_all(b"GET /events HTTP/1.1\r\nHost: x\r\n\r\n")
  277. .await
  278. .unwrap();
  279. let mut buf = vec![0u8; 4096];
  280. let n = sock.read(&mut buf).await.unwrap();
  281. let text = String::from_utf8_lossy(&buf[..n]);
  282. assert!(
  283. text.contains("\"type\":\"state\""),
  284. "no replayed frame: {text}"
  285. );
  286. handle.abort();
  287. }
  288. #[tokio::test]
  289. async fn bus_delivers_state_events() {
  290. let (tx, mut rx) = tokio::sync::broadcast::channel::<Event>(8);
  291. let snap = crate::web::events::Snapshot {
  292. ts: 1,
  293. clusters: Vec::new(),
  294. jobs: Vec::new(),
  295. topo: Default::default(),
  296. attaches: Vec::new(),
  297. };
  298. tx.send(Event::State {
  299. ts: snap.ts,
  300. snapshot: snap.clone(),
  301. })
  302. .unwrap();
  303. let got = rx.recv().await.unwrap();
  304. match got {
  305. Event::State { ts, snapshot } => {
  306. assert_eq!(ts, 1);
  307. assert!(snapshot.clusters.is_empty());
  308. }
  309. Event::Job { .. } => panic!("unexpected job event"),
  310. Event::Log { .. } => panic!("unexpected log event"),
  311. }
  312. }
  313. #[tokio::test]
  314. async fn snapshot_serializes_with_type_tag() {
  315. let ev = Event::State {
  316. ts: 7,
  317. snapshot: build_snapshot(
  318. &fake_config(),
  319. &crate::web::probe::Fleet {
  320. nodes: std::collections::HashMap::new(),
  321. switch_off: std::collections::HashMap::new(),
  322. soft_off: std::collections::HashMap::new(),
  323. ha: std::collections::HashMap::new(),
  324. },
  325. &[],
  326. &[],
  327. &std::collections::HashMap::new(),
  328. ),
  329. };
  330. let json = serde_json::to_value(&ev).unwrap();
  331. assert_eq!(json["type"], "state");
  332. assert_eq!(json["ts"], 7);
  333. assert!(json["snapshot"].is_object());
  334. }
  335. // (P2/P3/P4 API round-trip tests moved to api_tests.rs — 500-line cap)
  336. /// Click-through UI tests: boots the real server on an ephemeral port and
  337. /// drives the embedded ui.html with Playwright (deno, npm-free).
  338. /// Skipped when deno or the chromium binary is unavailable.
  339. #[tokio::test]
  340. async fn ui_playwright_checks() {
  341. if std::process::Command::new("deno")
  342. .arg("--version")
  343. .output()
  344. .is_err()
  345. {
  346. eprintln!("SKIP ui_playwright_checks: deno not available");
  347. return;
  348. }
  349. // The recipes panel needs at least one row: a fixture recipe source
  350. // (the launcher repo no longer carries a local `recipes/`).
  351. let fx = std::env::temp_dir().join(format!("sm121-ui-fixture-{}", std::process::id()));
  352. std::fs::create_dir_all(fx.join("ztest-ui-fixture")).unwrap();
  353. std::fs::write(
  354. fx.join("ztest-ui-fixture/recipe.toml"),
  355. "name = \"ztest-ui-fixture\"\nengine = \"vllm\"\n",
  356. )
  357. .unwrap();
  358. let (addr, _h) = boot_with(playwright_config(Some(&fx))).await;
  359. // Serialize with the report viewer's Playwright suite (shared browser).
  360. // The lock + deno run live in spawn_blocking: the std guard must not
  361. // be held across an await, and std::process::output must not block
  362. // this runtime's only thread (the axum server runs on it).
  363. let url = format!("http://{addr}");
  364. let out = tokio::task::spawn_blocking(move || {
  365. let _pw = crate::ui_test_lock::playwright_session();
  366. std::process::Command::new("deno")
  367. .arg("run")
  368. .arg("-A")
  369. .arg(concat!(
  370. env!("CARGO_MANIFEST_DIR"),
  371. "/src/web/ui.playwright.mjs"
  372. ))
  373. .env("SM121_WEB_URL", url)
  374. .output()
  375. })
  376. .await
  377. .unwrap()
  378. .unwrap();
  379. let stdout = String::from_utf8_lossy(&out.stdout);
  380. let stderr = String::from_utf8_lossy(&out.stderr);
  381. if stdout.contains("SKIP") {
  382. eprintln!("SKIP ui_playwright_checks:\n{stdout}");
  383. return;
  384. }
  385. assert!(
  386. out.status.success(),
  387. "playwright UI checks failed:\n{stdout}\n{stderr}"
  388. );
  389. let _ = std::fs::remove_dir_all(&fx);
  390. eprintln!("{stdout}");
  391. }
  392. /// Two offline clusters (both nodes on refused loopback ports) — the
  393. /// fleet the Playwright UI checks render: chip strip with two chips,
  394. /// down-cluster action bar, type-the-name power flow. `recipe_dir` (a dir
  395. /// holding `<name>/recipe.toml` fixtures) feeds the recipes panel — the
  396. /// launcher repo no longer carries a local `recipes/`, so the test seeds
  397. /// its own.
  398. fn playwright_config(recipe_dir: Option<&std::path::Path>) -> Config {
  399. let mut cfg = fake_config();
  400. if let Some(d) = recipe_dir {
  401. cfg.recipe_sources.push(d.display().to_string());
  402. }
  403. cfg.clusters.push(Cluster {
  404. name: "fake2".into(),
  405. master_ib_ip: None,
  406. nodes: vec![Node {
  407. name: "n1".into(),
  408. ssh_ip: "127.0.0.1:2223".into(),
  409. ib_ip: None,
  410. rank: 0,
  411. host_key: None,
  412. entity: None,
  413. }],
  414. extra_entities: vec![],
  415. });
  416. cfg
  417. }