diff --git a/flake.nix b/flake.nix index 0c7e5dd..c031df8 100644 --- a/flake.nix +++ b/flake.nix @@ -135,9 +135,10 @@ installPhaseCommand = '' mkdir -p $out cp -a target/userland/bundle/. $out/ - test -f $out/bin/chat.wasm - test -f $out/etc/systemd/system/chat.service - test -f $out/etc/systemd/system/multi-user.target.wants/chat.service + test -f $out/bin/web.wasm + test -f $out/srv/www/index.html + test -f $out/etc/systemd/system/web.service + test -f $out/etc/systemd/system/multi-user.target.wants/web.service ''; doInstallCargoArtifacts = false; # The upstream browser fixture links to an optional, unbuilt artifact. diff --git a/host/tests/it/chat.rs b/host/tests/it/chat.rs index e556db3..a30802d 100644 --- a/host/tests/it/chat.rs +++ b/host/tests/it/chat.rs @@ -283,6 +283,24 @@ async fn seeded_chat_case(open_connection: bool) -> anyhow::Result<()> { anyhow::ensure!(std::time::Instant::now() < deadline, "Reply did not appear"); tokio::task::yield_now().await; } + let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", port)).await?; + stream.write_all(format!("GET /api/harness/conversations/{root}/stream?after=0 HTTP/1.1\r\nHost: localhost:{port}\r\nConnection: close\r\n\r\n").as_bytes()).await?; + let mut streamed = Vec::new(); + tokio::time::timeout(Duration::from_secs(10), async { + loop { + let mut chunk = [0; 4096]; + let count = stream.read(&mut chunk).await?; + anyhow::ensure!(count > 0, "Stream closed before replay"); + streamed.extend_from_slice(&chunk[..count]); + if String::from_utf8_lossy(&streamed).contains("Chat test reply") { + break; + } + } + Ok::<_, anyhow::Error>(()) + }) + .await??; + assert!(String::from_utf8_lossy(&streamed).contains("application/x-ndjson")); + drop(stream); let (status, body) = request(port, "/api/harness/credential-requests", None).await?; assert_eq!(status, 200); let value: serde_json::Value = serde_json::from_slice(&body)?; diff --git a/host/tests/it/p1.rs b/host/tests/it/p1.rs index 194852a..69cf297 100644 --- a/host/tests/it/p1.rs +++ b/host/tests/it/p1.rs @@ -13,8 +13,8 @@ fn bundle() -> anyhow::Result { || !bundle.join("etc/sausage/config.yaml").exists() || std::fs::read_to_string(&stamp).ok().as_deref() != Some(key.as_str()) { - if stamp.exists() { - std::fs::remove_file(&stamp)?; + if bundle.exists() { + std::fs::remove_dir_all(&bundle)?; } assert!( std::process::Command::new("nu") diff --git a/host/tests/it/p2.rs b/host/tests/it/p2.rs index 2272608..be8d339 100644 --- a/host/tests/it/p2.rs +++ b/host/tests/it/p2.rs @@ -390,7 +390,7 @@ async fn request_approve_fetch_and_ssh_push() -> Result<()> { ensure!( browser .raw("GET", "/register", Value::Null)? - .contains("/app.js"), + .contains("src=\"app.js\""), "registration page missing" ); let start: Challenge = serde_json::from_value(browser.json( diff --git a/userland/web/src/main.rs b/userland/web/src/main.rs index 628deeb..e574f8b 100644 --- a/userland/web/src/main.rs +++ b/userland/web/src/main.rs @@ -54,16 +54,34 @@ fn write(stdin: &InputStream, stream: &OutputStream, mut bytes: &[u8]) -> io::Re } } } +fn wait_response(stdin: &InputStream, event: &Pollable, client: &InputStream) -> io::Result<()> { + let stop = stdin.subscribe(); + let disconnect = client.subscribe(); + loop { + let ready = poll(&[&stop, event, &disconnect]); + match stdin.read(65536) { + Err(StreamError::Closed) => std::process::exit(0), + Err(e) => return Err(error(e)), + Ok(_) => {} + } + if ready.contains(&2) { + client.read(65536).map_err(error)?; + } + if ready.contains(&1) { + return Ok(()); + } + } +} fn proxy( stdin: &InputStream, + client: &InputStream, method: &str, path: &str, - content_type: Option<&str>, - origin: Option<&str>, - host: Option<&str>, + headers: (Option<&str>, Option<&str>, Option<&str>), body: &[u8], output: &OutputStream, ) -> io::Result<()> { + let (content_type, origin, host) = headers; let headers = Fields::new(); headers .append("content-length", &body.len().to_string().into_bytes()) @@ -102,7 +120,7 @@ fn proxy( write(stdin, &stream, body)?; } OutgoingBody::finish(outgoing, None).map_err(error)?; - wait(stdin, &future.subscribe())?; + wait_response(stdin, &future.subscribe(), client)?; let response = future .get() .ok_or_else(|| error("missing HTTP response"))? @@ -119,7 +137,7 @@ fn proxy( let stream = incoming.stream().map_err(error)?; write(stdin, output, format!("HTTP/1.1 {status} Response\r\nContent-Type: {content_type}\r\nConnection: close\r\n\r\n").as_bytes())?; loop { - wait(stdin, &stream.subscribe())?; + wait_response(stdin, &stream.subscribe(), client)?; match stream.read(65536) { Ok(chunk) => write(stdin, output, &chunk)?, Err(StreamError::Closed) => break, @@ -165,11 +183,10 @@ fn serve( if let Some(path) = path.strip_prefix("/api/") { return proxy( stdin, + input, method, path, - header("content-type"), - header("origin"), - header("host"), + (header("content-type"), header("origin"), header("host")), &bytes[end..end + length], output, );