semora-cli 1.1.0

//! A real server on a random port plus a helper that runs the `semora` binary.
#![allow(dead_code)]

use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::Duration;

use semora_proto::Scope;
use semora_server::app::{self, AppState};
use semora_server::auth::{apikey, users};
use semora_server::config::Config;

/// How long a step of a test may take before the test calls it hung. Every step ends on a real
/// signal (a frame, a line of output, a row in the database), so a passing run never waits this
/// long; the bound only has to outlast a machine that is starved of CPU.
pub const HUNG: Duration = Duration::from_secs(60);

pub struct Server {
    pub url: String,
    pub key: String,
    pub user_id: i64,
    pub state: AppState,
    _dir: tempfile::TempDir,
}

impl Server {
    pub async fn start() -> Server {
        let dir = tempfile::tempdir().unwrap();
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let port = listener.local_addr().unwrap().port();
        let url = format!("http://127.0.0.1:{port}");
        let mut cfg = Config::for_tests(dir.path());
        cfg.public_url = url.parse().unwrap();
        let state = app::build_state(cfg).await.unwrap();
        let user = users::create(
            &state.db,
            "admin@example.com",
            "Admin",
            "correct horse battery",
            true,
        )
        .await
        .unwrap();
        semora_server::routes::setup::mark_complete(&state, "Semora", false)
            .await
            .unwrap();
        let (_, key) = apikey::create(&state, user.id, "cli-test", &Scope::all(), None, "web")
            .await
            .unwrap();
        let router = app::router(state.clone());
        tokio::spawn(async move {
            axum::serve(
                listener,
                router.into_make_service_with_connect_info::<SocketAddr>(),
            )
            .await
            .unwrap();
        });
        Server {
            url,
            key,
            user_id: user.id,
            state,
            _dir: dir,
        }
    }
}

pub struct Out {
    pub code: i32,
    pub stdout: String,
    pub stderr: String,
}

impl Out {
    pub fn json(&self) -> serde_json::Value {
        serde_json::from_str(&self.stdout).unwrap_or_else(|e| {
            panic!(
                "stdout is not JSON ({e}):\n{}\nstderr:\n{}",
                self.stdout, self.stderr
            )
        })
    }
}

/// A working directory and an isolated config directory for the CLI.
pub struct Cli {
    pub work: tempfile::TempDir,
    pub config: tempfile::TempDir,
    pub url: String,
    pub key: Option<String>,
    pub extra_env: Vec<(String, String)>,
}

impl Cli {
    pub fn new(server: &Server) -> Cli {
        Cli {
            work: tempfile::tempdir().unwrap(),
            config: tempfile::tempdir().unwrap(),
            url: server.url.clone(),
            key: Some(server.key.clone()),
            extra_env: Vec::new(),
        }
    }

    pub fn path(&self, rel: &str) -> PathBuf {
        self.work.path().join(rel)
    }

    pub fn write(&self, rel: &str, content: impl AsRef<[u8]>) -> PathBuf {
        let p = self.path(rel);
        std::fs::create_dir_all(p.parent().unwrap()).unwrap();
        std::fs::write(&p, content).unwrap();
        p
    }

    /// Lines of the CLI's `document.cache`, each split into path, server and id.
    pub fn cache(&self) -> Vec<Vec<String>> {
        let text =
            std::fs::read_to_string(self.config.path().join("document.cache")).unwrap_or_default();
        text.lines()
            .map(|l| l.splitn(3, '\t').map(str::to_string).collect())
            .collect()
    }

    pub fn command(&self, args: &[&str]) -> Command {
        let mut c = Command::new(env!("CARGO_BIN_EXE_semora"));
        c.args(args)
            .current_dir(self.work.path())
            .env("SEMORA_URL", &self.url)
            .env("SEMORA_CONFIG_DIR", self.config.path())
            .env("SEMORA_NO_KEYRING", "1")
            .env("SEMORA_NO_BROWSER", "1")
            .env_remove("SEMORA_DEFAULT_TAGS")
            .env_remove("SEMORA_WAIT_TIMEOUT");
        match &self.key {
            Some(k) => c.env("SEMORA_API_KEY", k),
            None => c.env_remove("SEMORA_API_KEY"),
        };
        for (k, v) in &self.extra_env {
            c.env(k, v);
        }
        c
    }

    /// Runs the CLI on a blocking thread so the in-process server keeps serving.
    pub async fn run(&self, args: &[&str]) -> Out {
        let mut cmd = self.command(args);
        tokio::task::spawn_blocking(move || {
            let o = cmd.output().expect("run semora");
            Out {
                code: o.status.code().unwrap_or(-1),
                stdout: String::from_utf8_lossy(&o.stdout).into_owned(),
                stderr: String::from_utf8_lossy(&o.stderr).into_owned(),
            }
        })
        .await
        .unwrap()
    }
}

pub fn exists(p: &Path) -> bool {
    p.exists()
}

// ---------------------------------------------------------------------------
// A simulated viewer: the browser side of a document's websocket
// ---------------------------------------------------------------------------

pub type Ws =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

pub struct Browser {
    pub ws: Ws,
    pub cookie: String,
    pub csrf: String,
}

impl Server {
    /// A browser session for the admin: `(cookie header, csrf token)`.
    pub async fn session(&self) -> (String, String) {
        let token = semora_server::auth::session::create(
            &self.state.db,
            self.user_id,
            Some("test"),
            "127.0.0.1",
        )
        .await
        .unwrap();
        let csrf = semora_server::auth::csrf::token_for(
            &self.state.keys,
            &semora_server::crypto::hash_token(&token),
        );
        (format!("__Host-semora_session={token}"), csrf)
    }

    /// Opens the document channel the way the viewer shell does (cookie + Origin).
    pub async fn browser(&self, doc: &str) -> Browser {
        use tokio_tungstenite::tungstenite::client::IntoClientRequest;
        let token = semora_server::auth::session::create(
            &self.state.db,
            self.user_id,
            Some("test"),
            "127.0.0.1",
        )
        .await
        .unwrap();
        let cookie = format!("__Host-semora_session={token}");
        let csrf = semora_server::auth::csrf::token_for(
            &self.state.keys,
            &semora_server::crypto::hash_token(&token),
        );
        let ws_url = self.url.replacen("http://", "ws://", 1);
        let mut req = format!("{ws_url}/ws/d/{doc}")
            .into_client_request()
            .unwrap();
        req.headers_mut().insert("cookie", cookie.parse().unwrap());
        req.headers_mut()
            .insert("origin", self.url.parse().unwrap());
        req.headers_mut()
            .insert("sec-websocket-protocol", "semora.v1".parse().unwrap());
        let (ws, _) = tokio_tungstenite::connect_async(req)
            .await
            .expect("browser connects");
        Browser { ws, cookie, csrf }
    }
}

impl Browser {
    pub async fn send(&mut self, v: serde_json::Value) {
        use futures_util::SinkExt;
        self.ws
            .send(tokio_tungstenite::tungstenite::Message::text(v.to_string()))
            .await
            .unwrap();
    }

    /// Next message of `kind` within `ms`.
    pub async fn next_of(&mut self, kind: &str, ms: u64) -> Option<serde_json::Value> {
        use futures_util::StreamExt;
        let deadline = tokio::time::Instant::now() + std::time::Duration::from_millis(ms);
        loop {
            let left = deadline.saturating_duration_since(tokio::time::Instant::now());
            match tokio::time::timeout(left, self.ws.next()).await {
                Ok(Some(Ok(tokio_tungstenite::tungstenite::Message::Text(t)))) => {
                    let v: serde_json::Value = serde_json::from_str(t.as_str()).unwrap();
                    if v["type"] == kind {
                        return Some(v);
                    }
                }
                Ok(Some(Ok(_))) => continue,
                _ => return None,
            }
        }
    }

    /// Waits until the channel reports at least one agent (in `hello` or `presence`). An agent is
    /// a CLI process that has to start, and on a machine starved of CPU that takes seconds: the
    /// wait ends on the signal, and `HUNG` only says when to give up.
    pub async fn wait_for_agent(&mut self) {
        use futures_util::StreamExt;
        let deadline = tokio::time::Instant::now() + HUNG;
        loop {
            let left = deadline.saturating_duration_since(tokio::time::Instant::now());
            let msg = tokio::time::timeout(left, self.ws.next())
                .await
                .expect("an agent joins");
            if let Some(Ok(tokio_tungstenite::tungstenite::Message::Text(t))) = msg {
                let v: serde_json::Value = serde_json::from_str(t.as_str()).unwrap();
                if (v["type"] == "hello" || v["type"] == "presence")
                    && v["agents"].as_u64().unwrap_or(0) > 0
                {
                    return;
                }
            }
        }
    }
}