| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437 |
- //! Web module tests: bus semantics, token gate, and a full server boot on an
- //! ephemeral port with a fake (unreachable) node.
- use crate::config::{Cluster, Config, Node};
- use crate::web::events::{Event, build_snapshot};
- use crate::web::{AppState, router};
- use std::collections::HashMap;
- use std::net::SocketAddr;
- pub(crate) async fn call(addr: SocketAddr, path: &str) -> reqwest::Response {
- reqwest::Client::new()
- .get(format!("http://{addr}{path}"))
- .send()
- .await
- .unwrap()
- }
- /// Config with one node at 127.0.0.1:2222 — connection refused instantly,
- /// machine-independent (no real SSH, no hardware).
- pub(crate) fn fake_config() -> Config {
- Config {
- ssh_user: "u".into(),
- ssh_key_path: "/nonexistent/key".into(),
- clusters: vec![Cluster {
- name: "fake".into(),
- master_ib_ip: None,
- nodes: vec![Node {
- name: "n0".into(),
- ssh_ip: "127.0.0.1:2222".into(),
- ib_ip: None,
- rank: 0,
- host_key: None,
- entity: None,
- }],
- extra_entities: vec![],
- }],
- nccl: HashMap::new(),
- hf_cache_dir: None,
- init_port: 25000,
- before_serve: HashMap::new(),
- exec_aliases: HashMap::new(),
- home_assistant: None,
- mikrotik: None,
- global_mods_dir: None,
- drop_caches: true,
- mount_aliases: HashMap::new(),
- recipe_sources: Vec::new(),
- recipe_repos: Vec::new(),
- docker_image_sources: Vec::new(),
- cluster_sources: Vec::new(),
- registry: None,
- }
- }
- #[path = "tests_auth.rs"]
- mod auth;
- pub(crate) fn app_state(token: Option<&str>) -> AppState {
- app_state_with(fake_config(), token)
- }
- pub(crate) fn app_state_with(config: Config, token: Option<&str>) -> AppState {
- let (tx, _) = tokio::sync::broadcast::channel(16);
- AppState {
- config: std::sync::Arc::new(config),
- bus: tx,
- token: token.map(String::from),
- sessions: std::sync::Arc::new(std::sync::Mutex::new(std::collections::HashMap::new())),
- last: std::sync::Arc::new(std::sync::Mutex::new(None)),
- jobs: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
- log_attaches: std::sync::Arc::new(std::sync::Mutex::new(Vec::new())),
- }
- }
- pub(crate) async fn boot() -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
- boot_with(fake_config()).await
- }
- pub(crate) async fn boot_with(
- config: Config,
- ) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
- boot_app(app_state_with(config, None)).await
- }
- async fn boot_app(app: AppState) -> (SocketAddr, std::sync::Arc<tokio::task::JoinHandle<()>>) {
- let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
- let addr = listener.local_addr().unwrap();
- let handle = tokio::spawn(async move {
- axum::serve(listener, router(app)).await.unwrap();
- });
- (addr, std::sync::Arc::new(handle))
- }
- #[tokio::test]
- async fn api_state_reports_unreachable_node() {
- let (addr, handle) = boot().await;
- let body: serde_json::Value = reqwest::get(format!("http://{addr}/api/state"))
- .await
- .unwrap()
- .json()
- .await
- .unwrap();
- assert_eq!(body["clusters"].as_array().unwrap().len(), 1);
- assert_eq!(body["clusters"][0]["name"], "fake");
- assert_eq!(body["clusters"][0]["nodes"][0]["reachable"], false);
- assert_eq!(body["clusters"][0]["free"], 0);
- handle.abort();
- }
- /// The recipes panel is mounted under /recipes: the page, its own SSE
- /// stream and the /r/ viewer all live behind the prefix.
- #[tokio::test]
- async fn recipes_panel_is_mounted_under_prefix() {
- let (addr, handle) = boot().await;
- let page = reqwest::get(format!("http://{addr}/recipes"))
- .await
- .unwrap()
- .text()
- .await
- .unwrap();
- // its own live stream resolves behind the prefix (headers arrive at once)
- let sse = reqwest::get(format!("http://{addr}/recipes/events"))
- .await
- .unwrap();
- assert_eq!(sse.status(), 200);
- assert!(
- page.contains("auth(\"/recipes/events\")"),
- "prefix not applied"
- );
- assert!(!page.contains("auth(\"/events\")"));
- // the dashboard's own routes still resolve
- let ui = reqwest::get(format!("http://{addr}/"))
- .await
- .unwrap()
- .text()
- .await
- .unwrap();
- assert!(ui.contains("recwin") || ui.contains("Recipes"));
- // Recipe logs open a dashboard WINDOW (postMessage bridge), not an inline
- // stream: the report page posts {sm121:"openLogs"}, the dashboard listens
- // and opens its logs window. No inline /api/logs fetch in the report page.
- assert!(
- page.contains("sm121: \"openLogs\"") && !page.contains("fetch(\"/api/logs\""),
- "report page should postMessage for logs, not stream inline"
- );
- assert!(
- ui.contains("openLogsForRecipe") && ui.contains("sm121 === \"openLogs\""),
- "dashboard should handle the recipes iframe logs message"
- );
- handle.abort();
- }
- #[test]
- fn jobs_persist_and_mark_interrupted() {
- use crate::web::jobs;
- let app = app_state(None);
- let dir = std::env::temp_dir().join(format!("sm121-jobs-test-{}", std::process::id()));
- std::fs::create_dir_all(&dir).unwrap();
- let path = dir.join("jobs.json");
- // a registry snapshot with a running job: the process died mid-flight
- std::fs::write(
- &path,
- r#"[{"id":"j9","kind":"serve","cluster":"c","recipe":"r","status":"running","message":"started","started":1},
- {"id":"j8","kind":"power","cluster":"c","recipe":null,"status":"ok","message":"on","started":2}]"#,
- )
- .unwrap();
- jobs::load_from(Some(&path), &app);
- let jobs = app.jobs.lock().unwrap().clone();
- std::fs::remove_dir_all(&dir).unwrap();
- assert_eq!(jobs.len(), 2);
- let running = jobs.iter().find(|j| j.id == "j9").unwrap();
- assert_eq!(
- running.status,
- jobs::JobStatus::Failed,
- "running -> failed on boot"
- );
- assert!(
- running.message.contains("runlock"),
- "the ledger hint: {}",
- running.message
- );
- let done = jobs.iter().find(|j| j.id == "j8").unwrap();
- assert_eq!(done.status, jobs::JobStatus::Ok);
- }
- #[test]
- fn stale_from_snapshot_picks_unreachable_nodes() {
- let app = app_state(None);
- use crate::web::api::stale_from_snapshot;
- let node = |name: &str, ip: &str, reachable: bool| crate::web::events::NodeSnap {
- name: name.into(),
- ssh_ip: ip.into(),
- reachable,
- occupied: false,
- containers: Vec::new(),
- rank: 0,
- ha: None,
- };
- let snap = crate::web::events::Snapshot {
- ts: 1,
- clusters: vec![crate::web::events::ClusterSnap {
- name: "c".into(),
- up: false,
- switch_off: false,
- soft_off: Vec::new(),
- free: 0,
- nodes: vec![node("a", "1.1.1.1", true), node("b", "1.1.1.2", false)],
- }],
- jobs: Vec::new(),
- topo: Default::default(),
- attaches: Vec::new(),
- };
- *app.last.lock().unwrap() = Some(snap);
- assert_eq!(stale_from_snapshot(&app, "c"), vec!["b".to_string()]);
- assert!(stale_from_snapshot(&app, "other").is_empty());
- }
- #[tokio::test]
- async fn token_gate_covers_recipes_panel() {
- let app = app_state(Some("s3cret"));
- *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
- ts: 1,
- clusters: Vec::new(),
- jobs: Vec::new(),
- topo: Default::default(),
- attaches: Vec::new(),
- });
- let (addr, handle) = boot_app(app).await;
- let base = format!("http://{addr}");
- // bare requests are rejected on every panel route...
- for path in ["/recipes", "/recipes/events"] {
- let st = reqwest::get(format!("{base}{path}"))
- .await
- .unwrap()
- .status();
- assert_eq!(st, 401, "{path} must be gated");
- }
- // ...and the session cookie (issued by POST /api/login) opens them —
- // the iframe and the report page's SSE ride the cookie, not the token.
- let client = reqwest::Client::new();
- let res = client
- .post(format!("{base}/api/login"))
- .header("authorization", "Bearer s3cret")
- .send()
- .await
- .unwrap();
- assert_eq!(res.status(), 204);
- let cookie = res
- .headers()
- .get("set-cookie")
- .unwrap()
- .to_str()
- .unwrap()
- .split(';')
- .next()
- .unwrap()
- .to_string();
- let page = client
- .get(format!("{base}/recipes"))
- .header("cookie", &cookie)
- .send()
- .await
- .unwrap();
- assert_eq!(page.status(), 200, "the recipes page opens with the cookie");
- let sse = client
- .get(format!("{base}/recipes/events"))
- .header("cookie", &cookie)
- .send()
- .await
- .unwrap();
- assert_eq!(sse.status(), 200);
- drop(sse);
- handle.abort();
- }
- #[tokio::test]
- async fn sse_replays_cached_snapshot_on_connect() {
- use tokio::io::{AsyncReadExt, AsyncWriteExt};
- // Nothing is broadcast after this point: the frame must come from the
- // cached snapshot, otherwise a fresh client waits a whole poll tick.
- let app = app_state(None);
- *app.last.lock().unwrap() = Some(crate::web::events::Snapshot {
- ts: 1,
- clusters: Vec::new(),
- jobs: Vec::new(),
- topo: Default::default(),
- attaches: Vec::new(),
- });
- let (addr, handle) = boot_app(app).await;
- let mut sock = tokio::net::TcpStream::connect(addr).await.unwrap();
- sock.write_all(b"GET /events HTTP/1.1\r\nHost: x\r\n\r\n")
- .await
- .unwrap();
- let mut buf = vec![0u8; 4096];
- let n = sock.read(&mut buf).await.unwrap();
- let text = String::from_utf8_lossy(&buf[..n]);
- assert!(
- text.contains("\"type\":\"state\""),
- "no replayed frame: {text}"
- );
- handle.abort();
- }
- #[tokio::test]
- async fn bus_delivers_state_events() {
- let (tx, mut rx) = tokio::sync::broadcast::channel::<Event>(8);
- let snap = crate::web::events::Snapshot {
- ts: 1,
- clusters: Vec::new(),
- jobs: Vec::new(),
- topo: Default::default(),
- attaches: Vec::new(),
- };
- tx.send(Event::State {
- ts: snap.ts,
- snapshot: snap.clone(),
- })
- .unwrap();
- let got = rx.recv().await.unwrap();
- match got {
- Event::State { ts, snapshot } => {
- assert_eq!(ts, 1);
- assert!(snapshot.clusters.is_empty());
- }
- Event::Job { .. } => panic!("unexpected job event"),
- Event::Log { .. } => panic!("unexpected log event"),
- }
- }
- #[tokio::test]
- async fn snapshot_serializes_with_type_tag() {
- let ev = Event::State {
- ts: 7,
- snapshot: build_snapshot(
- &fake_config(),
- &crate::web::probe::Fleet {
- nodes: std::collections::HashMap::new(),
- switch_off: std::collections::HashMap::new(),
- soft_off: std::collections::HashMap::new(),
- ha: std::collections::HashMap::new(),
- },
- &[],
- &[],
- &std::collections::HashMap::new(),
- ),
- };
- let json = serde_json::to_value(&ev).unwrap();
- assert_eq!(json["type"], "state");
- assert_eq!(json["ts"], 7);
- assert!(json["snapshot"].is_object());
- }
- // (P2/P3/P4 API round-trip tests moved to api_tests.rs — 500-line cap)
- /// Click-through UI tests: boots the real server on an ephemeral port and
- /// drives the embedded ui.html with Playwright (deno, npm-free).
- /// Skipped when deno or the chromium binary is unavailable.
- #[tokio::test]
- async fn ui_playwright_checks() {
- if std::process::Command::new("deno")
- .arg("--version")
- .output()
- .is_err()
- {
- eprintln!("SKIP ui_playwright_checks: deno not available");
- return;
- }
- // The recipes panel needs at least one row: a fixture recipe source
- // (the launcher repo no longer carries a local `recipes/`).
- let fx = std::env::temp_dir().join(format!("sm121-ui-fixture-{}", std::process::id()));
- std::fs::create_dir_all(fx.join("ztest-ui-fixture")).unwrap();
- std::fs::write(
- fx.join("ztest-ui-fixture/recipe.toml"),
- "name = \"ztest-ui-fixture\"\nengine = \"vllm\"\n",
- )
- .unwrap();
- let (addr, _h) = boot_with(playwright_config(Some(&fx))).await;
- // Serialize with the report viewer's Playwright suite (shared browser).
- // The lock + deno run live in spawn_blocking: the std guard must not
- // be held across an await, and std::process::output must not block
- // this runtime's only thread (the axum server runs on it).
- let url = format!("http://{addr}");
- let out = tokio::task::spawn_blocking(move || {
- let _pw = crate::ui_test_lock::playwright_session();
- std::process::Command::new("deno")
- .arg("run")
- .arg("-A")
- .arg(concat!(
- env!("CARGO_MANIFEST_DIR"),
- "/src/web/ui.playwright.mjs"
- ))
- .env("SM121_WEB_URL", url)
- .output()
- })
- .await
- .unwrap()
- .unwrap();
- let stdout = String::from_utf8_lossy(&out.stdout);
- let stderr = String::from_utf8_lossy(&out.stderr);
- if stdout.contains("SKIP") {
- eprintln!("SKIP ui_playwright_checks:\n{stdout}");
- return;
- }
- assert!(
- out.status.success(),
- "playwright UI checks failed:\n{stdout}\n{stderr}"
- );
- let _ = std::fs::remove_dir_all(&fx);
- eprintln!("{stdout}");
- }
- /// Two offline clusters (both nodes on refused loopback ports) — the
- /// fleet the Playwright UI checks render: chip strip with two chips,
- /// down-cluster action bar, type-the-name power flow. `recipe_dir` (a dir
- /// holding `<name>/recipe.toml` fixtures) feeds the recipes panel — the
- /// launcher repo no longer carries a local `recipes/`, so the test seeds
- /// its own.
- fn playwright_config(recipe_dir: Option<&std::path::Path>) -> Config {
- let mut cfg = fake_config();
- if let Some(d) = recipe_dir {
- cfg.recipe_sources.push(d.display().to_string());
- }
- cfg.clusters.push(Cluster {
- name: "fake2".into(),
- master_ib_ip: None,
- nodes: vec![Node {
- name: "n1".into(),
- ssh_ip: "127.0.0.1:2223".into(),
- ib_ip: None,
- rank: 0,
- host_key: None,
- entity: None,
- }],
- extra_entities: vec![],
- });
- cfg
- }
|