| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299 |
- //! 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(),
- docker_image_sources: Vec::new(),
- cluster_sources: Vec::new(),
- registry: None,
- }
- }
- pub(crate) fn app_state(token: Option<&str>) -> AppState {
- app_state_with(fake_config(), token)
- }
- 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),
- 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
- }
- 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"));
- handle.abort();
- }
- #[tokio::test]
- async fn token_gate_enforces_when_set() {
- let app = app_state(Some("s3cret"));
- let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
- let addr = listener.local_addr().unwrap();
- tokio::spawn(async move {
- axum::serve(listener, router(app)).await.unwrap();
- });
- let base = format!("http://{addr}/api/state");
- // No token → 401.
- assert_eq!(reqwest::get(&base).await.unwrap().status(), 401);
- // Wrong token → 401.
- assert_eq!(
- reqwest::get(format!("{base}?token=wrong"))
- .await
- .unwrap()
- .status(),
- 401
- );
- // Right token (query form — what SSE clients use) → 200.
- assert_eq!(
- reqwest::get(format!("{base}?token=s3cret"))
- .await
- .unwrap()
- .status(),
- 200
- );
- }
- #[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;
- }
- let (addr, _h) = boot_with(playwright_config()).await;
- // tokio::process (not std::process::output — that would block this
- // runtime's only thread while the axum server runs on it → deadlock).
- let out = tokio::process::Command::new("deno")
- .arg("run")
- .arg("-A")
- .arg("src/web/ui.playwright.mjs")
- .env("SM121_WEB_URL", format!("http://{addr}"))
- .output()
- .await
- .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}"
- );
- 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.
- fn playwright_config() -> Config {
- let mut cfg = fake_config();
- 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
- }
|