Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

test(e2e): isolate worker server runtime environments

Browse Source
This commit is contained in:
Hare
2026-09-22 22:26:02 +09:00
parent 043cb7bf22
commit 0869ef0cc6
16 changed files with 1568 additions and 134 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+2
View File
@@ -6286,8 +6286,10 @@ name = "yoi-e2e"
version = "0.0.0"
dependencies = [
"libc",
"reqwest",
"serde",
"serde_json",
"sha2 0.11.0",
"tempfile",
]
crates/tui/src/console/mod.rs
+12
View File
@@ -623,6 +623,16 @@ fn read_terminal_events(stop: Arc<AtomicBool>, tx: mpsc::UnboundedSender<Termina
}
}
#[cfg(feature = "e2e-test")]
pub(crate) async fn run_e2e_rewind(worker_name: String) -> Result<(), Box<dyn std::error::Error>> {
let mut terminal = enter_fullscreen()?;
let result = run_e2e_rewind_fixture(&mut terminal, worker_name).await;
let leave_result = leave_fullscreen(&mut terminal);
result?;
leave_result?;
Ok(())
}
#[cfg(feature = "e2e-test")]
async fn run_e2e_rewind_fixture(
terminal: &mut ConsoleTerminal,
@@ -637,6 +647,8 @@ async fn run_e2e_rewind_fixture(
entries: Vec::new(),
},
state: WorkerStatus::Idle.into(),
in_flight: protocol::InFlightSnapshot::default(),
internal_workers: Vec::new(),
greeting: Greeting {
worker_name: worker_name.clone(),
cwd: workspace_root.display().to_string(),
crates/tui/src/lib.rs
+47 -26
View File
@@ -137,7 +137,51 @@ pub async fn launch(options: LaunchOptions) -> ExitCode {
}
let mut terminal_mode = TerminalModeGuard::new();
let result = match mode {
#[cfg(feature = "e2e-test")]
let result = if std::env::var_os("YOI_TUI_TEST_REWIND_FIXTURE").is_some() {
let worker_name = match &mode {
LaunchMode::Spawn { worker_name, .. } => worker_name
.clone()
.unwrap_or_else(|| "e2e-rewind".to_string()),
_ => "e2e-rewind".to_string(),
};
console::run_e2e_rewind(worker_name).await
} else {
launch_mode(target.as_ref(), mode, &workspace_root).await
};
#[cfg(not(feature = "e2e-test"))]
let result = launch_mode(target.as_ref(), mode, &workspace_root).await;
// Always restore the terminal first so any pending eprintln below
// shows up cleanly in scrollback rather than inside an active
// alternate-screen buffer.
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "terminal_cleanup_started", serde_json::json!({}));
let _ = terminal_mode.restore();
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "terminal_cleanup_finished", serde_json::json!({}));
match result {
Ok(()) => {
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "exit", serde_json::json!({ "status": "success" }));
ExitCode::SUCCESS
}
Err(e) => {
eprintln!("yoi: {e}");
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "exit", serde_json::json!({ "status": "failure" }));
ExitCode::FAILURE
}
}
}
async fn launch_mode(
target: &dyn Target,
mode: LaunchMode,
workspace_root: &std::path::Path,
) -> Result<(), Box<dyn std::error::Error>> {
match mode {
LaunchMode::Spawn {
worker_name,
profile,
@@ -145,7 +189,7 @@ pub async fn launch(options: LaunchOptions) -> ExitCode {
Ok(Some(selection)) => match target.spawn_worker() {
Ok(spawn) => {
console::run_standalone(
workspace_root.clone(),
workspace_root.to_path_buf(),
spawn.state_dir,
Some(selection.worker_name),
Some(selection.profile),
@@ -158,7 +202,7 @@ pub async fn launch(options: LaunchOptions) -> ExitCode {
Err(error) => Err(Box::new(error) as Box<dyn std::error::Error>),
},
LaunchMode::StandaloneResume { include_all } => {
match standalone_picker::pick(target.as_ref(), include_all) {
match standalone_picker::pick(target, include_all) {
Ok(Some(intent)) => console::run_standalone_restore(intent).await,
Ok(None) => Ok(()),
Err(error) => Err(Box::new(error) as Box<dyn std::error::Error>),
@@ -195,28 +239,5 @@ pub async fn launch(options: LaunchOptions) -> ExitCode {
}
Err(e) => Err(Box::new(e) as Box<dyn std::error::Error>),
},
};
// Always restore the terminal first so any pending eprintln below
// shows up cleanly in scrollback rather than inside an active
// alternate-screen buffer.
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "terminal_cleanup_started", serde_json::json!({}));
let _ = terminal_mode.restore();
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "terminal_cleanup_finished", serde_json::json!({}));
match result {
Ok(()) => {
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "exit", serde_json::json!({ "status": "success" }));
ExitCode::SUCCESS
}
Err(e) => {
eprintln!("yoi: {e}");
#[cfg(feature = "e2e-test")]
e2e_observer::emit("tui", "exit", serde_json::json!({ "status": "failure" }));
ExitCode::FAILURE
}
}
}
crates/worker-runtime/Cargo.toml
+1 -1
View File
@@ -41,7 +41,7 @@ reqwest = { version = "0.13", optional = true, default-features = false, feature
ring.workspace = true
tar.workspace = true
thiserror = { workspace = true }
tokio = { workspace = true, features = ["net", "process", "rt", "sync", "time"] }
tokio = { workspace = true, features = ["net", "process", "rt", "signal", "sync", "time"] }
tracing.workspace = true
tracing-subscriber.workspace = true
toml.workspace = true
crates/worker-runtime/src/fs_store.rs
+57 -7
View File
@@ -123,8 +123,8 @@ impl FsRuntimeStore {
})?;
let owner_lock = acquire_runtime_store_owner_lock(&root)?;
recover_runtime_store_migration(&root)?;
let existed = root.exists();
if existed && !root.is_dir() {
let root_existed = root.exists();
if root_existed && !root.is_dir() {
return Err(RuntimeError::StoreCorrupt {
operation: "open runtime store",
path: root,
@@ -132,17 +132,36 @@ impl FsRuntimeStore {
});
}
if !existed {
if !root_existed {
fs::create_dir_all(&root).map_err(|source| RuntimeError::StoreIo {
operation: "create runtime store root",
path: root.clone(),
source,
})?;
}
let snapshot_exists = root.join(RUNTIME_FILE).is_file();
let workers_dir = root.join(WORKERS_DIR);
if !snapshot_exists
&& workers_dir.is_dir()
&& fs::read_dir(&workers_dir)
.map_err(|source| RuntimeError::StoreIo {
operation: "inspect uninitialized runtime store",
path: workers_dir.clone(),
source,
})?
.next()
.is_some()
{
return Err(RuntimeError::StoreCorrupt {
operation: "open runtime store",
path: root,
message: "Worker records exist without a Runtime snapshot".to_string(),
});
}
fs::create_dir_all(root.join(WORKERS_DIR)).map_err(|source| RuntimeError::StoreIo {
fs::create_dir_all(&workers_dir).map_err(|source| RuntimeError::StoreIo {
operation: "create runtime store",
path: root.join(WORKERS_DIR),
path: workers_dir,
source,
})?;
let legacy_events = root.join("events.jsonl");
@@ -158,14 +177,14 @@ impl FsRuntimeStore {
}
}
if existed {
if snapshot_exists {
migrate_runtime_store(&root, runtime_id)?;
}
let store = Self {
root,
_owner_lock: Some(owner_lock),
};
let state = if existed {
let state = if snapshot_exists {
Some(store.load_runtime_state()?)
} else {
None
@@ -2137,6 +2156,37 @@ mod tests {
document["schema_version"].as_u64().unwrap()
}
#[test]
fn auth_initialized_directory_without_snapshot_is_a_fresh_store() {
let parent = tempfile::tempdir().unwrap();
let root = parent.path().join("runtime");
fs::create_dir_all(&root).unwrap();
fs::write(root.join("auth.toml"), b"fixture auth material").unwrap();
let opened = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap();
assert!(opened.state.is_none());
assert!(root.join(WORKERS_DIR).is_dir());
assert_eq!(
fs::read(root.join("auth.toml")).unwrap(),
b"fixture auth material"
);
}
#[test]
fn worker_records_without_runtime_snapshot_are_rejected() {
let parent = tempfile::tempdir().unwrap();
let root = parent.path().join("runtime");
fs::create_dir_all(root.join(WORKERS_DIR).join("orphan")).unwrap();
let error = FsRuntimeStore::open_or_create(root.clone(), "runtime-test").unwrap_err();
assert!(matches!(
error,
RuntimeError::StoreCorrupt { path, .. } if path == root
));
}
#[test]
fn runtime_store_owner_lock_child_probe() {
let Some(root) = std::env::var_os("YOI_TEST_RUNTIME_STORE_LOCK_ROOT") else {
crates/worker-runtime/src/http_server.rs
+35
View File
@@ -139,6 +139,22 @@ pub enum RuntimeHttpStoreSelection {
},
}
pub async fn serve_runtime_http_with_shutdown<F>(
runtime: Runtime,
listener: TcpListener,
local_token: Option<String>,
shutdown: F,
) -> Result<(), RuntimeHttpServerError>
where
F: std::future::Future<Output = ()> + Send + 'static,
{
let local_token = local_token.ok_or(RuntimeHttpServerError::AuthRequired)?;
axum::serve(listener, runtime_http_router(runtime, local_token))
.with_graceful_shutdown(shutdown)
.await?;
Ok(())
}
/// Serve an existing Runtime on a pre-bound listener.
pub async fn serve_runtime_http(
runtime: Runtime,
@@ -150,6 +166,25 @@ pub async fn serve_runtime_http(
Ok(())
}
pub async fn serve_runtime_http_with_workspace_auth_shutdown<F>(
runtime: Runtime,
listener: TcpListener,
local_token: Option<String>,
workspace_auth: WorkspaceRuntimeHttpAuth,
shutdown: F,
) -> Result<(), RuntimeHttpServerError>
where
F: std::future::Future<Output = ()> + Send + 'static,
{
axum::serve(
listener,
runtime_http_router_with_optional_auth(runtime, local_token, Some(workspace_auth)),
)
.with_graceful_shutdown(shutdown)
.await?;
Ok(())
}
pub async fn serve_runtime_http_with_workspace_auth(
runtime: Runtime,
listener: TcpListener,
crates/worker-runtime/src/main.rs
+21 -2
View File
@@ -75,6 +75,23 @@ fn init_serve_tracing() {
.try_init();
}
async fn shutdown_signal() {
#[cfg(unix)]
{
let mut terminate =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("install SIGTERM handler");
tokio::select! {
_ = tokio::signal::ctrl_c() => {},
_ = terminate.recv() => {},
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
}
}
fn run() -> Result<(), ProcessError> {
let args = env::args().skip(1).collect::<Vec<_>>();
if matches!(args.first().map(String::as_str), Some("migrate")) {
@@ -104,11 +121,12 @@ fn run() -> Result<(), ProcessError> {
"yoi-runtime listening on {local_addr}; intended client is a trusted backend/proxy, not a browser"
);
let server = if let Some(workspace_auth) = workspace_http_auth {
worker_runtime::http_server::serve_runtime_http_with_workspace_auth(
worker_runtime::http_server::serve_runtime_http_with_workspace_auth_shutdown(
worker_runtime,
listener,
config.http.local_token,
workspace_auth,
shutdown_signal(),
)
.await
} else {
@@ -117,10 +135,11 @@ fn run() -> Result<(), ProcessError> {
"Runtime HTTP server requires Workspace issuer auth or --local-token".to_owned(),
)
})?;
worker_runtime::http_server::serve_runtime_http(
worker_runtime::http_server::serve_runtime_http_with_shutdown(
worker_runtime,
listener,
Some(local_token),
shutdown_signal(),
)
.await
};
crates/workspace-server/Cargo.toml
+1 -1
View File
@@ -38,7 +38,7 @@ thiserror.workspace = true
ticket.workspace = true
memory.workspace = true
merge-request.workspace = true
tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "sync", "time"] }
tokio = { workspace = true, features = ["fs", "macros", "net", "rt-multi-thread", "signal", "sync", "time"] }
tower.workspace = true
tokio-tungstenite = { workspace = true, features = ["rustls-tls-webpki-roots"] }
worker.workspace = true
crates/workspace-server/src/lib.rs
+1
View File
@@ -48,6 +48,7 @@ pub use repositories::{ConfiguredRepository, RepositoryLogRead, RepositoryRegist
pub use server::{
AuthConfig, ServerConfig, WorkspaceApi, WorkspaceServerApi, build_router,
build_workspace_server_router, serve, serve_workspace_catalog,
serve_workspace_catalog_with_shutdown,
};
pub use server_api::{
GitCommitSummary, GitRemoteSummary, GitRepositorySummary, RepositorySummary,
crates/workspace-server/src/main.rs
+184 -7
View File
@@ -9,18 +9,19 @@ use yoi_workspace_server::hosts::{
EMBEDDED_RUNTIME_ID, RemoteRuntimeConfig, is_loopback_runtime_origin,
};
use yoi_workspace_server::store::{
SqliteWorkspaceStore, WorkspaceRuntimeAuthenticationMode, WorkspaceRuntimeBinding,
WorkspaceRuntimeBindingState,
AccountRecord, ApiTokenRecord, SqliteWorkspaceStore, UserRecord,
WorkspaceRuntimeAuthenticationMode, WorkspaceRuntimeBinding, WorkspaceRuntimeBindingState,
};
use yoi_workspace_server::{
ControlPlaneStore, ResolvedWorkspaceBackendConfig, ServerConfig, ServerHostConfigFile,
WorkspaceRecord, serve_workspace_catalog,
WorkspaceRecord, serve_workspace_catalog_with_shutdown,
};
#[derive(Debug)]
enum Command {
Serve(ServeOptions),
Migrate(MigrateOptions),
BootstrapAuth(BootstrapAuthOptions),
Skills(SkillsCommand),
Help,
}
@@ -38,6 +39,13 @@ struct MigrateOptions {
help: bool,
}
#[derive(Debug)]
struct BootstrapAuthOptions {
handle: String,
display_name: Option<String>,
label: Option<String>,
}
#[derive(Debug)]
struct SkillWorkspaceOptions {
workspace_id: String,
@@ -77,6 +85,7 @@ async fn run() -> Result<(), Box<dyn std::error::Error>> {
match parse_command(&args)? {
Command::Serve(options) => run_serve(options).await,
Command::Migrate(options) => run_migrate(options),
Command::BootstrapAuth(options) => run_bootstrap_auth(options),
Command::Skills(command) => run_skills(command),
Command::Help => Ok(()),
}
@@ -90,6 +99,7 @@ fn parse_command(args: &[String]) -> Result<Command, CliError> {
match command.as_str() {
"migrate" => parse_migrate_options(rest).map(Command::Migrate),
"bootstrap-auth" => parse_bootstrap_auth_options(rest).map(Command::BootstrapAuth),
"skills" => parse_skills_command(rest),
"serve" => {
if rest.iter().any(|arg| arg == "--help" || arg == "-h") {
@@ -103,7 +113,7 @@ fn parse_command(args: &[String]) -> Result<Command, CliError> {
Ok(Command::Help)
}
other => Err(CliError(format!(
"unknown command `{other}`; expected `migrate`, `skills`, or `serve`"
"unknown command `{other}`; expected `bootstrap-auth`, `migrate`, `skills`, or `serve`"
))),
}
}
@@ -214,6 +224,131 @@ fn run_migrate(options: MigrateOptions) -> Result<(), Box<dyn std::error::Error>
Ok(())
}
fn run_bootstrap_auth(options: BootstrapAuthOptions) -> Result<(), Box<dyn std::error::Error>> {
let database_path = ServerConfig::default_server_database_path();
if let Some(parent) = database_path.parent() {
std::fs::create_dir_all(parent)?;
}
let store = SqliteWorkspaceStore::open(&database_path)?;
if !store.list_workspaces()?.is_empty() {
return Err(Box::new(CliError(
"bootstrap-auth is available only before the first Workspace is created".to_string(),
)));
}
let handle = yoi_workspace_server::auth::normalize_handle(&options.handle)?;
let display_name = options
.display_name
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.unwrap_or(&handle)
.to_string();
let now = yoi_workspace_server::auth::now_rfc3339();
let user = match store.any_user()? {
Some(user) if user.handle == handle => user,
Some(_) => {
return Err(Box::new(CliError(
"bootstrap-auth refuses to replace an existing bootstrap user".to_string(),
)));
}
None => {
let account_id = yoi_workspace_server::auth::new_id("acct-user");
store.upsert_account(&AccountRecord {
account_id: account_id.clone(),
kind: "user".to_string(),
handle: handle.clone(),
display_name: display_name.clone(),
created_at: now.clone(),
updated_at: now.clone(),
})?;
let user = UserRecord {
user_id: yoi_workspace_server::auth::new_id("user"),
account_id,
handle,
display_name,
created_at: now.clone(),
updated_at: now.clone(),
};
store.upsert_user(&user)?;
user
}
};
let access_token = yoi_workspace_server::auth::mint_secret("yoi_api");
store.create_api_token(&ApiTokenRecord {
token_id: yoi_workspace_server::auth::new_id("api-token"),
token_hash: yoi_workspace_server::auth::token_hash(&access_token),
user_id: user.user_id.clone(),
label: options
.label
.unwrap_or_else(|| "initial host setup".to_string()),
created_at: now,
expires_at: None,
revoked_at: None,
last_used_at: None,
})?;
println!(
"{}",
serde_json::to_string(&serde_json::json!({
"access_token": access_token,
"token_type": "bearer",
"user": {
"user_id": user.user_id,
"account_id": user.account_id,
"handle": user.handle,
"display_name": user.display_name,
}
}))?
);
Ok(())
}
fn parse_bootstrap_auth_options(args: &[String]) -> Result<BootstrapAuthOptions, CliError> {
let mut handle = None;
let mut display_name = None;
let mut label = None;
let mut index = 0;
while index < args.len() {
let arg = &args[index];
let (field, target) = match arg.as_str() {
"--handle" => ("--handle", &mut handle),
"--display-name" => ("--display-name", &mut display_name),
"--label" => ("--label", &mut label),
_ if arg.starts_with("--handle=") => {
handle = Some(value_after_equals(arg, "--handle")?.to_string());
index += 1;
continue;
}
_ if arg.starts_with("--display-name=") => {
display_name = Some(value_after_equals(arg, "--display-name")?.to_string());
index += 1;
continue;
}
_ if arg.starts_with("--label=") => {
label = Some(value_after_equals(arg, "--label")?.to_string());
index += 1;
continue;
}
other => return Err(CliError(format!("unknown bootstrap-auth option `{other}`"))),
};
index += 1;
let value = args
.get(index)
.ok_or_else(|| CliError(format!("{field} requires a value")))?;
if target.replace(value.clone()).is_some() {
return Err(CliError(format!("{field} must be provided only once")));
}
index += 1;
}
let handle = handle.ok_or_else(|| CliError("bootstrap-auth requires --handle".to_string()))?;
Ok(BootstrapAuthOptions {
handle,
display_name,
label,
})
}
fn init_serve_tracing() {
let filter = tracing_subscriber::EnvFilter::try_from_default_env()
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info"));
@@ -226,6 +361,23 @@ fn init_serve_tracing() {
.try_init();
}
async fn shutdown_signal() {
#[cfg(unix)]
{
let mut terminate =
tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
.expect("install SIGTERM handler");
tokio::select! {
_ = tokio::signal::ctrl_c() => {},
_ = terminate.recv() => {},
}
}
#[cfg(not(unix))]
{
let _ = tokio::signal::ctrl_c().await;
}
}
async fn run_serve(options: ServeOptions) -> Result<(), Box<dyn std::error::Error>> {
init_serve_tracing();
let database_path = ServerConfig::default_server_database_path();
@@ -284,7 +436,8 @@ async fn run_serve(options: ServeOptions) -> Result<(), Box<dyn std::error::Erro
database_path.display(),
local_addr
);
serve_workspace_catalog(resolved.server, store, listener).await?;
serve_workspace_catalog_with_shutdown(resolved.server, store, listener, shutdown_signal())
.await?;
Ok(())
}
@@ -490,7 +643,7 @@ fn parse_listen(value: &str) -> Result<SocketAddr, CliError> {
fn print_help() {
println!(
"yoi-server\n\nUsage:\n yoi-server migrate [--dry-run] [--database <PATH>]\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help"
"yoi-server\n\nUsage:\n yoi-server bootstrap-auth --handle <HANDLE> [--display-name <NAME>] [--label <LABEL>]\n yoi-server migrate [--dry-run] [--database <PATH>]\n yoi-server skills <COMMAND> [OPTIONS]\n yoi-server serve [OPTIONS]\n\nOptions:\n -h, --help Print help"
);
}
@@ -527,6 +680,28 @@ mod tests {
}
}
#[test]
fn parse_bootstrap_auth_requires_handle_and_accepts_host_setup_fields() {
let command = parse_command(&[
"bootstrap-auth".to_string(),
"--handle=e2e-owner".to_string(),
"--display-name".to_string(),
"E2E Owner".to_string(),
"--label".to_string(),
"e2e setup".to_string(),
])
.unwrap();
let Command::BootstrapAuth(options) = command else {
panic!("expected bootstrap-auth command");
};
assert_eq!(options.handle, "e2e-owner");
assert_eq!(options.display_name.as_deref(), Some("E2E Owner"));
assert_eq!(options.label.as_deref(), Some("e2e setup"));
let error = parse_bootstrap_auth_options(&[]).unwrap_err();
assert_eq!(error.to_string(), "bootstrap-auth requires --handle");
}
#[test]
fn parse_migrate_uses_the_shared_schema_path() {
let command = parse_command(&[
@@ -662,7 +837,9 @@ mod tests {
let error = parse_command(&[command.to_owned()]).unwrap_err();
assert_eq!(
error.to_string(),
format!("unknown command `{command}`; expected `migrate`, `skills`, or `serve`")
format!(
"unknown command `{command}`; expected `bootstrap-auth`, `migrate`, `skills`, or `serve`"
)
);
}
}
crates/workspace-server/src/server.rs
+21
View File
@@ -1995,6 +1995,10 @@ struct WorkspaceListQuery {
limit: Option<usize>,
}
async fn server_health() -> Json<serde_json::Value> {
Json(serde_json::json!({ "status": "ok" }))
}
async fn list_server_workspaces(
State(api): State<WorkspaceServerApi>,
headers: HeaderMap,
@@ -2663,6 +2667,7 @@ pub async fn build_workspace_server_router(
api.preload().await?;
api.recover_workspace_deletions().await?;
let catalog = Router::new()
.route("/health", get(server_health))
.route(
"/api/workspaces",
get(list_server_workspaces).post(create_server_workspace),
@@ -4565,6 +4570,22 @@ struct ApiFailureLogEvent<'a> {
diagnostics: Option<&'a [RuntimeDiagnostic]>,
}
pub async fn serve_workspace_catalog_with_shutdown<F>(
template: ServerConfig,
store: Arc<dyn ControlPlaneStore>,
listener: TcpListener,
shutdown: F,
) -> Result<()>
where
F: std::future::Future<Output = ()> + Send + 'static,
{
let router = build_workspace_server_router(template, store).await?;
axum::serve(listener, router)
.with_graceful_shutdown(shutdown)
.await?;
Ok(())
}
pub async fn serve_workspace_catalog(
template: ServerConfig,
store: Arc<dyn ControlPlaneStore>,
tests/e2e/Cargo.toml
+7
View File
@@ -12,8 +12,10 @@ e2e = []
[dependencies]
libc.workspace = true
reqwest = { version = "0.13", default-features = false, features = ["blocking", "json", "rustls"] }
serde = { workspace = true, features = ["derive"] }
serde_json.workspace = true
sha2.workspace = true
tempfile.workspace = true
[[test]]
@@ -25,3 +27,8 @@ required-features = ["e2e"]
name = "rewind"
path = "tests/rewind.rs"
required-features = ["e2e"]
[[test]]
name = "worker_environment"
path = "tests/worker_environment.rs"
required-features = ["e2e"]
tests/e2e/src/lib.rs
+128 -89
View File
@@ -1267,6 +1267,65 @@ impl Drop for PanelHarness {
}
}
mod worker_environment;
pub use worker_environment::{
WorkerBinaryProviderInfo, WorkerE2eEnvironment, WorkerEnvironmentCleanupReport,
worker_binary_provider_info,
};
#[derive(Debug)]
struct FixtureAllocation {
temp_root: TempDir,
root: PathBuf,
home: PathBuf,
xdg_data_home: PathBuf,
xdg_state_home: PathBuf,
xdg_config_home: PathBuf,
xdg_runtime_dir: PathBuf,
artifacts_dir: PathBuf,
}
fn allocate_fixture(kind: &str) -> Result<FixtureAllocation> {
let workspace_root = workspace_root()?;
let target_dir = workspace_root.join("target");
let temp_parent = target_dir.join("e2e-tmp");
let artifact_parent = target_dir.join("e2e-artifacts");
fs::create_dir_all(&temp_parent)?;
fs::create_dir_all(&artifact_parent)?;
let fixture_id = format!(
"{kind}-{}-{}-{}",
std::process::id(),
now_ms(),
FIXTURE_COUNTER.fetch_add(1, Ordering::Relaxed)
);
let temp_root = tempfile::Builder::new()
.prefix(&format!("yoi-e2e-{fixture_id}-"))
.tempdir_in(&temp_parent)?;
let root = temp_root.path().to_path_buf();
let allocation = FixtureAllocation {
temp_root,
home: root.join("home"),
xdg_data_home: root.join("data"),
xdg_state_home: root.join("state"),
xdg_config_home: root.join("config"),
xdg_runtime_dir: root.join("run"),
artifacts_dir: artifact_parent.join(fixture_id),
root,
};
for dir in [
&allocation.home,
&allocation.xdg_data_home,
&allocation.xdg_state_home,
&allocation.xdg_config_home,
&allocation.xdg_runtime_dir,
&allocation.artifacts_dir,
] {
fs::create_dir_all(dir)?;
}
Ok(allocation)
}
#[derive(Debug, Clone, Serialize)]
pub struct FixtureCleanupReport {
pub fixture_root: PathBuf,
@@ -1299,107 +1358,87 @@ pub struct FixtureWorkspace {
impl FixtureWorkspace {
pub fn new(binary: &Path) -> Result<Self> {
let workspace_root = workspace_root()?;
let target_dir = workspace_root.join("target");
let temp_parent = target_dir.join("e2e-tmp");
let artifact_parent = target_dir.join("e2e-artifacts");
fs::create_dir_all(&temp_parent)?;
fs::create_dir_all(&artifact_parent)?;
Self::new_inner(binary, true)
}
let fixture_id = format!(
"{}-{}-{}",
std::process::id(),
now_ms(),
FIXTURE_COUNTER.fetch_add(1, Ordering::Relaxed)
);
let temp_root = tempfile::Builder::new()
.prefix(&format!("yoi-e2e-{fixture_id}-"))
.tempdir_in(&temp_parent)?;
let root = temp_root.path().to_path_buf();
let artifacts_dir = artifact_parent.join(fixture_id);
pub fn new_rewind(binary: &Path) -> Result<Self> {
Self::new_inner(binary, false)
}
fn new_inner(binary: &Path, setup_dashboard: bool) -> Result<Self> {
let allocation = allocate_fixture("tui")?;
let root = allocation.root;
let artifacts_dir = allocation.artifacts_dir;
let workspace = root.join("workspace");
let home = root.join("home");
let xdg_data_home = root.join("data");
let xdg_state_home = root.join("state");
let xdg_config_home = root.join("config");
let xdg_runtime_dir = root.join("runtime");
for dir in [
&workspace,
&home,
&xdg_data_home,
&xdg_state_home,
&xdg_config_home,
&xdg_runtime_dir,
&artifacts_dir,
] {
fs::create_dir_all(dir)?;
}
fs::create_dir_all(&workspace)?;
let mut fixture = Self {
temp_root: Some(temp_root),
temp_root: Some(allocation.temp_root),
root,
workspace,
home,
xdg_data_home,
xdg_state_home,
xdg_config_home,
xdg_runtime_dir,
home: allocation.home,
xdg_data_home: allocation.xdg_data_home,
xdg_state_home: allocation.xdg_state_home,
xdg_config_home: allocation.xdg_config_home,
xdg_runtime_dir: allocation.xdg_runtime_dir,
artifacts_dir,
ready_ticket_id: String::new(),
planning_ticket_id: String::new(),
};
fixture.write_fixture_metadata("created", None)?;
let worker_metadata_root = active_worker_metadata_root(&fixture.home);
write_blocking_worker_metadata(&worker_metadata_root, "workspace")?;
write_blocking_worker_metadata(&worker_metadata_root, "workspace-orchestrator")?;
run_yoi(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
&["ticket", "init"],
)?;
let first = create_ticket(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
READY_FIXTURE_TICKET_TITLE,
)?;
run_yoi(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
&["ticket", "state", &first, "ready"],
)?;
let second = create_ticket(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
PLANNING_FIXTURE_TICKET_TITLE,
)?;
fixture.ready_ticket_id = first;
fixture.planning_ticket_id = second;
fixture.setup_orchestration_overlay(binary)?;
if setup_dashboard {
let worker_metadata_root = active_worker_metadata_root(&fixture.home);
write_blocking_worker_metadata(&worker_metadata_root, "workspace")?;
write_blocking_worker_metadata(&worker_metadata_root, "workspace-orchestrator")?;
run_yoi(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
&["ticket", "init"],
)?;
let first = create_ticket(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
READY_FIXTURE_TICKET_TITLE,
)?;
run_yoi(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
&["ticket", "state", &first, "ready"],
)?;
let second = create_ticket(
binary,
&fixture.workspace,
&fixture.home,
&fixture.xdg_data_home,
&fixture.xdg_state_home,
&fixture.xdg_config_home,
&fixture.xdg_runtime_dir,
&fixture.artifacts_dir,
PLANNING_FIXTURE_TICKET_TITLE,
)?;
fixture.ready_ticket_id = first;
fixture.planning_ticket_id = second;
fixture.setup_orchestration_overlay(binary)?;
}
fixture.write_fixture_metadata("ready", None)?;
Ok(fixture)
}
tests/e2e/src/worker_environment.rs
+978
View File
@@ -0,0 +1,978 @@
use std::fs::{self, File, OpenOptions};
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, ExitStatus, Stdio};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::thread::{self, JoinHandle};
use std::time::{Duration, Instant};
use reqwest::blocking::{Client, RequestBuilder};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use super::{
FixtureAllocation, HarnessError, Result, allocate_fixture, command_display,
current_target_profile_dir, now_ms, test_profile, workspace_root, yoi_binary_info,
};
const STARTUP_WAIT: Duration = Duration::from_secs(15);
const EXIT_WAIT: Duration = Duration::from_secs(5);
const HTTP_WAIT_STEP: Duration = Duration::from_millis(50);
const MAX_PROCESS_LOG_BYTES: usize = 256 * 1024;
const MAX_COMMAND_OUTPUT_BYTES: usize = 64 * 1024;
static WORKER_ENV_COUNTER: AtomicU64 = AtomicU64::new(0);
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerBinaryRecord {
pub name: String,
pub path: PathBuf,
pub sha256: String,
pub provider: String,
pub build_command: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkerBinaryProviderInfo {
pub workspace_root: PathBuf,
pub profile: String,
pub binaries: Vec<WorkerBinaryRecord>,
}
impl WorkerBinaryProviderInfo {
pub fn binary(&self, name: &str) -> Result<PathBuf> {
self.binaries
.iter()
.find(|record| record.name == name)
.map(|record| record.path.clone())
.ok_or_else(|| HarnessError::Protocol(format!("binary provider omitted {name}")))
}
}
pub fn worker_binary_provider_info() -> Result<WorkerBinaryProviderInfo> {
static INFO: OnceLock<std::result::Result<WorkerBinaryProviderInfo, String>> = OnceLock::new();
match INFO.get_or_init(|| resolve_worker_binaries().map_err(|error| error.to_string())) {
Ok(info) => Ok(info.clone()),
Err(message) => Err(HarnessError::Protocol(message.clone())),
}
}
fn resolve_worker_binaries() -> Result<WorkerBinaryProviderInfo> {
let workspace_root = workspace_root()?;
let profile = test_profile();
let yoi = yoi_binary_info()?;
let mut records = vec![binary_record(
"yoi",
yoi.binary,
yoi.provider,
yoi.build_command,
)?];
let server_override = resolve_override("YOI_E2E_SERVER_BIN", &workspace_root)?;
let runtime_override = resolve_override("YOI_E2E_RUNTIME_BIN", &workspace_root)?;
let mut build_command = None;
if server_override.is_none() || runtime_override.is_none() {
let cargo = PathBuf::from(std::env::var_os("CARGO").unwrap_or_else(|| "cargo".into()));
let mut args = vec![
"build".to_string(),
"-p".to_string(),
"yoi-workspace-server".to_string(),
"--bin".to_string(),
"yoi-server".to_string(),
"-p".to_string(),
"worker-runtime".to_string(),
"--bin".to_string(),
"yoi-runtime".to_string(),
];
if profile == "release" {
args.push("--release".to_string());
}
let rendered = command_display(&cargo, &args);
eprintln!("yoi-e2e worker binary provider=cargo-build command={rendered}");
let output = Command::new(&cargo)
.args(&args)
.current_dir(&workspace_root)
.output()?;
if !output.status.success() {
return Err(HarnessError::CommandFailed {
program: cargo,
args,
status: output.status,
stdout: bounded_text(&output.stdout, MAX_COMMAND_OUTPUT_BYTES),
stderr: bounded_text(&output.stderr, MAX_COMMAND_OUTPUT_BYTES),
});
}
build_command = Some(rendered);
}
let profile_dir = current_target_profile_dir()?;
let (server_path, server_provider, server_build) = match server_override {
Some(path) => (path, "YOI_E2E_SERVER_BIN".to_string(), None),
None => (
profile_dir.join(executable_name("yoi-server")),
"cargo-build".to_string(),
build_command.clone(),
),
};
let (runtime_path, runtime_provider, runtime_build) = match runtime_override {
Some(path) => (path, "YOI_E2E_RUNTIME_BIN".to_string(), None),
None => (
profile_dir.join(executable_name("yoi-runtime")),
"cargo-build".to_string(),
build_command,
),
};
records.push(binary_record(
"yoi-server",
server_path,
server_provider,
server_build,
)?);
records.push(binary_record(
"yoi-runtime",
runtime_path,
runtime_provider,
runtime_build,
)?);
let info = WorkerBinaryProviderInfo {
workspace_root: workspace_root.clone(),
profile,
binaries: records,
};
let artifact_dir = workspace_root.join("target/e2e-artifacts");
fs::create_dir_all(&artifact_dir)?;
fs::write(
artifact_dir.join("worker-binary-provider.json"),
serde_json::to_vec_pretty(&info)?,
)?;
Ok(info)
}
fn resolve_override(name: &str, workspace_root: &Path) -> Result<Option<PathBuf>> {
let Some(path) = std::env::var_os(name) else {
return Ok(None);
};
let path = PathBuf::from(path);
let path = if path.is_absolute() {
path
} else {
workspace_root.join(path)
};
if !path.is_file() {
return Err(HarnessError::MissingBinary(path));
}
Ok(Some(path))
}
fn binary_record(
name: &str,
path: PathBuf,
provider: String,
build_command: Option<String>,
) -> Result<WorkerBinaryRecord> {
if !path.is_file() {
return Err(HarnessError::MissingBinary(path));
}
let bytes = fs::read(&path)?;
let digest = Sha256::digest(bytes);
Ok(WorkerBinaryRecord {
name: name.to_string(),
path,
sha256: format!(
"sha256:{}",
digest
.iter()
.map(|byte| format!("{byte:02x}"))
.collect::<String>()
),
provider,
build_command,
})
}
fn executable_name(name: &str) -> String {
format!("{name}{}", std::env::consts::EXE_SUFFIX)
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkerEnvironmentCleanupReport {
pub fixture_root: PathBuf,
pub artifacts_dir: PathBuf,
pub cleanup_attempted: bool,
pub cleanup_success: bool,
pub fixture_root_exists_after: bool,
pub cleanup_error: Option<String>,
pub report_path: PathBuf,
}
#[derive(Debug)]
pub struct WorkerE2eEnvironment {
temp_root: Option<tempfile::TempDir>,
pub root: PathBuf,
pub home: PathBuf,
pub xdg_data_home: PathBuf,
pub xdg_state_home: PathBuf,
pub xdg_config_home: PathBuf,
pub xdg_runtime_dir: PathBuf,
pub server_data_dir: PathBuf,
pub runtime_data_dir: PathBuf,
pub server_config: PathBuf,
pub artifacts_dir: PathBuf,
pub workspace_id: Option<String>,
pub runtime_id: String,
pub account_handle: String,
pub server_url: Option<String>,
pub runtime_url: Option<String>,
server: Option<ManagedProcess>,
runtime: Option<ManagedProcess>,
access_token: Option<String>,
client: Client,
binaries: WorkerBinaryProviderInfo,
}
impl WorkerE2eEnvironment {
pub fn new() -> Result<Self> {
let allocation = allocate_fixture("worker")?;
Self::from_allocation(allocation)
}
fn from_allocation(allocation: FixtureAllocation) -> Result<Self> {
let unique = format!(
"{}-{}-{}",
std::process::id(),
now_ms(),
WORKER_ENV_COUNTER.fetch_add(1, Ordering::Relaxed)
);
let server_data_dir = allocation.xdg_data_home.join("yoi/server");
let runtime_data_dir = allocation.root.join("runtime-data");
assert_absent_or_empty(&server_data_dir, "Server data directory")?;
assert_absent_or_empty(&runtime_data_dir, "Runtime data directory")?;
let server_config = allocation.xdg_config_home.join("yoi/server.toml");
if let Some(parent) = server_config.parent() {
fs::create_dir_all(parent)?;
}
fs::write(
&server_config,
"[browser]\npublic_url = \"http://localhost:5173\"\n",
)?;
let binaries = worker_binary_provider_info()?;
fs::write(
allocation.artifacts_dir.join("binary-provider.json"),
serde_json::to_vec_pretty(&binaries)?,
)?;
let client = Client::builder()
.connect_timeout(Duration::from_secs(1))
.timeout(Duration::from_secs(3))
.build()
.map_err(|error| HarnessError::Protocol(format!("build E2E HTTP client: {error}")))?;
let environment = Self {
temp_root: Some(allocation.temp_root),
root: allocation.root,
home: allocation.home,
xdg_data_home: allocation.xdg_data_home,
xdg_state_home: allocation.xdg_state_home,
xdg_config_home: allocation.xdg_config_home,
xdg_runtime_dir: allocation.xdg_runtime_dir,
server_data_dir,
runtime_data_dir,
server_config,
artifacts_dir: allocation.artifacts_dir,
workspace_id: None,
runtime_id: format!("runtime-{unique}"),
account_handle: format!("e2e-{}-{}", std::process::id(), unique.replace('-', "")),
server_url: None,
runtime_url: None,
server: None,
runtime: None,
access_token: None,
client,
binaries,
};
environment.write_metadata("created")?;
Ok(environment)
}
pub fn start(&mut self) -> Result<()> {
assert_absent_or_empty(&self.server_data_dir, "Server data directory")?;
assert_absent_or_empty(&self.runtime_data_dir, "Runtime data directory")?;
let server_binary = self.binaries.binary("yoi-server")?;
let mut command = self.isolated_command(&server_binary);
command.args([
"serve",
"--listen",
"127.0.0.1:0",
"--config",
self.server_config.to_string_lossy().as_ref(),
]);
let server = ManagedProcess::spawn(
"server",
command,
&self.artifacts_dir.join("server.stdout.log"),
&self.artifacts_dir.join("server.stderr.log"),
)?;
let server_addr = server.wait_for_address("on http://", STARTUP_WAIT)?;
let server_url = format!("http://{server_addr}");
self.server = Some(server);
self.server_url = Some(server_url.clone());
self.wait_for_server_health(&server_url)?;
let bootstrap = self.run_command(
&server_binary,
&[
"bootstrap-auth",
"--handle",
&self.account_handle,
"--display-name",
"Yoi Worker E2E",
"--label",
"worker-e2e-bootstrap",
],
true,
)?;
let bootstrap: Value = serde_json::from_slice(&bootstrap.stdout)?;
self.access_token = bootstrap
.get("access_token")
.and_then(Value::as_str)
.map(str::to_owned);
if self.access_token.is_none() {
return Err(HarnessError::Protocol(
"bootstrap-auth omitted access_token".to_string(),
));
}
let created = self.request_json(
self.client
.post(format!("{server_url}/api/workspaces"))
.json(&serde_json::json!({
"operation_key": format!("worker-e2e-{}", self.runtime_id),
"display_name": format!("Worker E2E {}", self.runtime_id),
"repository": {
"repository_key": "fixture",
"uri": "https://example.invalid/yoi-e2e.git",
"default_ref": "develop"
}
})),
"create Workspace",
)?;
let workspace_id = json_string(&created, "/workspace/workspace_id")?;
self.workspace_id = Some(workspace_id.clone());
let identity = self.request_json(
self.client.get(format!(
"{server_url}/api/w/{workspace_id}/settings/signing-identity"
)),
"read Workspace signing identity",
)?;
let public_bundle = identity.get("public_bundle").cloned().ok_or_else(|| {
HarnessError::Protocol("Workspace public bundle is absent".to_string())
})?;
let bundle_path = self.root.join("workspace-public-identity.json");
fs::write(&bundle_path, serde_json::to_vec_pretty(&public_bundle)?)?;
let runtime_binary = self.binaries.binary("yoi-runtime")?;
self.run_command(
&runtime_binary,
&[
"identity",
"init",
"--runtime-id",
&self.runtime_id,
"--fs-root",
self.runtime_data_dir.to_string_lossy().as_ref(),
],
false,
)?;
let runtime_identity = self.run_command(
&runtime_binary,
&[
"identity",
"show",
"--json",
"--fs-root",
self.runtime_data_dir.to_string_lossy().as_ref(),
],
false,
)?;
let runtime_identity: Value = serde_json::from_slice(&runtime_identity.stdout)?;
self.run_command(
&runtime_binary,
&[
"trust-workspace",
"add",
"--bundle",
bundle_path.to_string_lossy().as_ref(),
"--fs-root",
self.runtime_data_dir.to_string_lossy().as_ref(),
],
false,
)?;
let mut command = self.isolated_command(&runtime_binary);
command.args([
"--bind",
"127.0.0.1:0",
"--display-name",
&format!("E2E {}", self.runtime_id),
"--store",
"fs",
"--fs-root",
self.runtime_data_dir.to_string_lossy().as_ref(),
]);
let runtime = ManagedProcess::spawn(
"runtime",
command,
&self.artifacts_dir.join("runtime.stdout.log"),
&self.artifacts_dir.join("runtime.stderr.log"),
)?;
let runtime_addr = runtime.wait_for_address("listening on ", STARTUP_WAIT)?;
let runtime_url = format!("http://{runtime_addr}");
self.runtime = Some(runtime);
self.runtime_url = Some(runtime_url.clone());
self.request_json(
self.client
.post(format!("{server_url}/api/w/{workspace_id}/runtimes"))
.json(&serde_json::json!({
"public_bundle": runtime_identity,
"display_name": format!("E2E {}", self.runtime_id),
"endpoint": runtime_url,
"expected_revision": null
})),
"register Runtime",
)?;
let connection = self.request_json(
self.client.post(format!(
"{server_url}/api/w/{workspace_id}/runtimes/{}/connection-tests",
self.runtime_id
)),
"verify Runtime connection",
)?;
if connection.get("status").and_then(Value::as_str) != Some("compatible") {
return Err(HarnessError::Protocol(format!(
"Runtime connection was not compatible: {connection}"
)));
}
self.assert_connected_and_empty()?;
self.write_metadata("running")?;
Ok(())
}
pub fn assert_connected_and_empty(&self) -> Result<()> {
let server_url = self.required_server_url()?;
let workspace_id = self.required_workspace_id()?;
let runtimes = self.request_json(
self.client
.get(format!("{server_url}/api/w/{workspace_id}/runtimes")),
"list Runtimes",
)?;
let connected = runtimes
.get("items")
.and_then(Value::as_array)
.is_some_and(|items| {
items.iter().any(|item| {
item.get("runtime_id").and_then(Value::as_str) == Some(self.runtime_id.as_str())
&& item
.pointer("/management/binding/connection_state")
.and_then(Value::as_str)
== Some("verified")
})
});
if !connected {
return Err(HarnessError::Protocol(format!(
"Runtime {} is not verified in {runtimes}",
self.runtime_id
)));
}
let workers = self.request_json(
self.client
.get(format!("{server_url}/api/w/{workspace_id}/workers")),
"list Workers",
)?;
if workers
.get("items")
.and_then(Value::as_array)
.is_none_or(|items| !items.is_empty())
{
return Err(HarnessError::Protocol(format!(
"fresh Worker list is not empty: {workers}"
)));
}
Ok(())
}
pub fn server_port(&self) -> Result<u16> {
url_port(self.required_server_url()?)
}
pub fn runtime_port(&self) -> Result<u16> {
url_port(
self.runtime_url
.as_deref()
.ok_or_else(|| HarnessError::Protocol("Runtime is not started".to_string()))?,
)
}
pub fn server_pid(&self) -> Option<u32> {
self.server.as_ref().map(ManagedProcess::id)
}
pub fn runtime_pid(&self) -> Option<u32> {
self.runtime.as_ref().map(ManagedProcess::id)
}
pub fn shutdown(&mut self) -> Result<()> {
if let Some(mut runtime) = self.runtime.take() {
runtime.shutdown(EXIT_WAIT)?;
}
if let Some(mut server) = self.server.take() {
server.shutdown(EXIT_WAIT)?;
}
self.write_metadata("stopped")?;
Ok(())
}
pub fn cleanup(mut self) -> Result<WorkerEnvironmentCleanupReport> {
self.shutdown()?;
self.cleanup_inner(true)
}
fn cleanup_inner(&mut self, strict: bool) -> Result<WorkerEnvironmentCleanupReport> {
let mut cleanup_error = None;
if let Some(temp_root) = self.temp_root.take()
&& let Err(error) = temp_root.close()
{
cleanup_error = Some(error.to_string());
}
let fixture_root_exists_after = self.root.exists();
let cleanup_success = cleanup_error.is_none() && !fixture_root_exists_after;
let report = WorkerEnvironmentCleanupReport {
fixture_root: self.root.clone(),
artifacts_dir: self.artifacts_dir.clone(),
cleanup_attempted: true,
cleanup_success,
fixture_root_exists_after,
cleanup_error,
report_path: self.artifacts_dir.join("cleanup.json"),
};
fs::write(&report.report_path, serde_json::to_vec_pretty(&report)?)?;
if strict && !report.cleanup_success {
return Err(HarnessError::Protocol(format!(
"Worker E2E cleanup failed; see {}",
report.report_path.display()
)));
}
Ok(report)
}
fn wait_for_server_health(&self, server_url: &str) -> Result<()> {
let deadline = Instant::now() + STARTUP_WAIT;
loop {
let last = match self.client.get(format!("{server_url}/health")).send() {
Ok(response) if response.status().is_success() => return Ok(()),
Ok(response) => format!("HTTP {}", response.status()),
Err(error) => error.to_string(),
};
if Instant::now() >= deadline {
return Err(HarnessError::Protocol(format!(
"Server health did not become ready: {last}"
)));
}
if let Some(server) = self.server.as_ref()
&& !server.is_running()
{
return Err(HarnessError::Protocol(
"Server exited before health became ready".to_string(),
));
}
thread::sleep(HTTP_WAIT_STEP);
}
}
fn request_json(&self, request: RequestBuilder, operation: &str) -> Result<Value> {
let request = if let Some(token) = self.access_token.as_deref() {
request.bearer_auth(token)
} else {
request
};
let response = request
.send()
.map_err(|error| HarnessError::Protocol(format!("{operation}: {error}")))?;
let status = response.status();
let body = response
.bytes()
.map_err(|error| HarnessError::Protocol(format!("{operation}: {error}")))?;
self.append_operation(operation, status.as_u16())?;
if !status.is_success() {
return Err(HarnessError::Protocol(format!(
"{operation} returned {status}: {}",
bounded_text(&body, MAX_COMMAND_OUTPUT_BYTES)
)));
}
serde_json::from_slice(&body).map_err(HarnessError::from)
}
fn append_operation(&self, operation: &str, status: u16) -> Result<()> {
let mut file = OpenOptions::new()
.create(true)
.append(true)
.open(self.artifacts_dir.join("operations.jsonl"))?;
serde_json::to_writer(
&mut file,
&serde_json::json!({
"ts_ms": now_ms(),
"operation": operation,
"status": status,
}),
)?;
writeln!(file)?;
Ok(())
}
fn run_command(
&self,
binary: &Path,
args: &[&str],
redact_stdout: bool,
) -> Result<std::process::Output> {
let args_owned = args
.iter()
.map(|arg| (*arg).to_string())
.collect::<Vec<_>>();
let mut command = self.isolated_command(binary);
let output = command.args(args).output()?;
let mut log = OpenOptions::new()
.create(true)
.append(true)
.open(self.artifacts_dir.join("setup-commands.jsonl"))?;
serde_json::to_writer(
&mut log,
&serde_json::json!({
"ts_ms": now_ms(),
"binary": binary,
"args": args,
"status": output.status.code(),
"stdout": if redact_stdout { "<redacted>" } else { "captured-not-persisted" },
"env_clear": true,
}),
)?;
writeln!(log)?;
if !output.status.success() {
return Err(HarnessError::CommandFailed {
program: binary.to_path_buf(),
args: args_owned,
status: output.status,
stdout: if redact_stdout {
"<redacted>".to_string()
} else {
bounded_text(&output.stdout, MAX_COMMAND_OUTPUT_BYTES)
},
stderr: bounded_text(&output.stderr, MAX_COMMAND_OUTPUT_BYTES),
});
}
Ok(output)
}
fn isolated_command(&self, binary: &Path) -> Command {
let mut command = Command::new(binary);
command
.current_dir(&self.root)
.env_clear()
.env("HOME", &self.home)
.env("XDG_DATA_HOME", &self.xdg_data_home)
.env("XDG_STATE_HOME", &self.xdg_state_home)
.env("XDG_CONFIG_HOME", &self.xdg_config_home)
.env("XDG_RUNTIME_DIR", &self.xdg_runtime_dir);
command
}
fn required_server_url(&self) -> Result<&str> {
self.server_url
.as_deref()
.ok_or_else(|| HarnessError::Protocol("Server is not started".to_string()))
}
fn required_workspace_id(&self) -> Result<&str> {
self.workspace_id
.as_deref()
.ok_or_else(|| HarnessError::Protocol("Workspace is not created".to_string()))
}
fn write_metadata(&self, phase: &str) -> Result<()> {
fs::write(
self.artifacts_dir.join("worker-environment.json"),
serde_json::to_vec_pretty(&serde_json::json!({
"phase": phase,
"fixture_root": self.root,
"server_data_dir": self.server_data_dir,
"runtime_data_dir": self.runtime_data_dir,
"home": self.home,
"xdg_data_home": self.xdg_data_home,
"xdg_state_home": self.xdg_state_home,
"xdg_config_home": self.xdg_config_home,
"xdg_runtime_dir": self.xdg_runtime_dir,
"server_config": self.server_config,
"artifacts_dir": self.artifacts_dir,
"workspace_id": self.workspace_id,
"runtime_id": self.runtime_id,
"account_handle": self.account_handle,
"server_url": self.server_url,
"runtime_url": self.runtime_url,
"server_pid": self.server_pid(),
"runtime_pid": self.runtime_pid(),
"env_clear": true,
"credentials_persisted_to_artifacts": false,
}))?,
)?;
Ok(())
}
}
impl Drop for WorkerE2eEnvironment {
fn drop(&mut self) {
if let Some(mut runtime) = self.runtime.take() {
let _ = runtime.shutdown(EXIT_WAIT);
}
if let Some(mut server) = self.server.take() {
let _ = server.shutdown(EXIT_WAIT);
}
if self.temp_root.is_some() {
let _ = self.cleanup_inner(false);
}
}
}
#[derive(Debug)]
struct ManagedProcess {
name: String,
child: Child,
stdout: Arc<Mutex<Vec<u8>>>,
stderr: Arc<Mutex<Vec<u8>>>,
readers: Vec<JoinHandle<()>>,
}
impl ManagedProcess {
fn spawn(
name: &str,
mut command: Command,
stdout_path: &Path,
stderr_path: &Path,
) -> Result<Self> {
command.stdout(Stdio::piped()).stderr(Stdio::piped());
let mut child = command.spawn()?;
let stdout_pipe = child
.stdout
.take()
.ok_or_else(|| HarnessError::Protocol(format!("{name} stdout was not piped")))?;
let stderr_pipe = child
.stderr
.take()
.ok_or_else(|| HarnessError::Protocol(format!("{name} stderr was not piped")))?;
let stdout = Arc::new(Mutex::new(Vec::new()));
let stderr = Arc::new(Mutex::new(Vec::new()));
let readers = vec![
capture_stream(stdout_pipe, stdout_path.to_path_buf(), Arc::clone(&stdout)),
capture_stream(stderr_pipe, stderr_path.to_path_buf(), Arc::clone(&stderr)),
];
Ok(Self {
name: name.to_string(),
child,
stdout,
stderr,
readers,
})
}
fn id(&self) -> u32 {
self.child.id()
}
fn is_running(&self) -> bool {
unsafe { libc::kill(self.child.id() as i32, 0) == 0 }
}
fn wait_for_address(&self, marker: &str, timeout: Duration) -> Result<String> {
let start = Instant::now();
loop {
let output = self.output();
if let Some(address) = parse_address_after(&output, marker) {
return Ok(address);
}
if start.elapsed() >= timeout {
return Err(HarnessError::Protocol(format!(
"timed out waiting for {} address after {marker:?}; bounded output: {}",
self.name,
bounded_text(output.as_bytes(), 8 * 1024)
)));
}
let rc = unsafe { libc::kill(self.child.id() as i32, 0) };
if rc != 0 {
return Err(HarnessError::Protocol(format!(
"{} exited before publishing its address; bounded output: {}",
self.name,
bounded_text(output.as_bytes(), 8 * 1024)
)));
}
thread::sleep(Duration::from_millis(20));
}
}
fn output(&self) -> String {
let stdout = self
.stdout
.lock()
.map(|value| value.clone())
.unwrap_or_default();
let stderr = self
.stderr
.lock()
.map(|value| value.clone())
.unwrap_or_default();
format!(
"{}\n{}",
String::from_utf8_lossy(&stdout),
String::from_utf8_lossy(&stderr)
)
}
fn shutdown(&mut self, timeout: Duration) -> Result<ExitStatus> {
if let Some(status) = self.child.try_wait()? {
self.join_readers();
return Ok(status);
}
let signal_result = unsafe { libc::kill(self.child.id() as i32, libc::SIGTERM) };
if signal_result != 0 {
return Err(io_error("send SIGTERM", std::io::Error::last_os_error()));
}
let start = Instant::now();
loop {
if let Some(status) = self.child.try_wait()? {
self.join_readers();
if status.success() {
return Ok(status);
}
return Err(HarnessError::Protocol(format!(
"{} did not complete graceful shutdown: {status}",
self.name
)));
}
if start.elapsed() >= timeout {
let _ = self.child.kill();
let _ = self.child.wait();
self.join_readers();
return Err(HarnessError::Protocol(format!(
"{} did not exit within {timeout:?}",
self.name
)));
}
thread::sleep(Duration::from_millis(20));
}
}
fn join_readers(&mut self) {
for reader in self.readers.drain(..) {
let _ = reader.join();
}
}
}
impl Drop for ManagedProcess {
fn drop(&mut self) {
if self.child.try_wait().ok().flatten().is_none() {
let _ = self.child.kill();
let _ = self.child.wait();
}
self.join_readers();
}
}
fn capture_stream(
mut stream: impl Read + Send + 'static,
path: PathBuf,
shared: Arc<Mutex<Vec<u8>>>,
) -> JoinHandle<()> {
thread::spawn(move || {
let mut file = File::create(path).ok();
let mut written = 0usize;
let mut truncated = false;
let mut buffer = [0u8; 4096];
loop {
let count = match stream.read(&mut buffer) {
Ok(0) => break,
Ok(count) => count,
Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
Err(_) => break,
};
if let Ok(mut captured) = shared.lock()
&& captured.len() < MAX_PROCESS_LOG_BYTES
{
let remaining = MAX_PROCESS_LOG_BYTES - captured.len();
captured.extend_from_slice(&buffer[..count.min(remaining)]);
}
if let Some(file) = file.as_mut() {
let remaining = MAX_PROCESS_LOG_BYTES.saturating_sub(written);
let keep = count.min(remaining);
if keep > 0 {
let _ = file.write_all(&buffer[..keep]);
written += keep;
}
if keep < count && !truncated {
let _ = file.write_all(b"\n<bounded log truncated>\n");
truncated = true;
}
}
}
})
}
fn parse_address_after(output: &str, marker: &str) -> Option<String> {
let rest = output.split(marker).nth(1)?;
let address = rest
.trim_start()
.split(|character: char| character.is_whitespace() || character == ';' || character == '`')
.next()?
.trim_start_matches("http://")
.trim_end_matches('/');
(!address.is_empty()).then(|| address.to_string())
}
fn assert_absent_or_empty(path: &Path, label: &str) -> Result<()> {
if !path.exists() {
return Ok(());
}
if path.is_dir() && fs::read_dir(path)?.next().is_none() {
return Ok(());
}
Err(HarnessError::Protocol(format!(
"{label} {} must start empty",
path.display()
)))
}
fn json_string(value: &Value, pointer: &str) -> Result<String> {
value
.pointer(pointer)
.and_then(Value::as_str)
.map(str::to_owned)
.ok_or_else(|| HarnessError::Protocol(format!("JSON response omitted {pointer}: {value}")))
}
fn url_port(url: &str) -> Result<u16> {
reqwest::Url::parse(url)
.ok()
.and_then(|url| url.port())
.ok_or_else(|| HarnessError::Protocol(format!("URL has no explicit port: {url}")))
}
fn bounded_text(bytes: &[u8], limit: usize) -> String {
let keep = bytes.len().min(limit);
let mut text = String::from_utf8_lossy(&bytes[..keep]).into_owned();
if bytes.len() > keep {
text.push_str("\n<bounded output truncated>");
}
text
}
fn io_error(operation: &str, error: std::io::Error) -> HarnessError {
HarnessError::Protocol(format!("{operation}: {error}"))
}
tests/e2e/tests/rewind.rs
+1 -1
View File
@@ -6,7 +6,7 @@ use yoi_e2e::{FixtureWorkspace, KeyPress, PanelHarness, yoi_binary};
fn single_pod_rewind_picker_applies_without_escape_and_suppresses_duplicate_enter()
-> yoi_e2e::Result<()> {
let binary = yoi_binary()?;
let fixture = FixtureWorkspace::new(&binary)?;
let fixture = FixtureWorkspace::new_rewind(&binary)?;
let mut tui = PanelHarness::spawn(fixture.rewind_fixture_config(binary))?;
tui.expect_mouse_capture_enabled()?;
tests/e2e/tests/worker_environment.rs
+72
View File
@@ -0,0 +1,72 @@
use std::fs;
use yoi_e2e::{Result, WorkerE2eEnvironment, WorkerEnvironmentCleanupReport};
#[test]
fn starts_server_and_runtime_from_empty_storage() -> Result<()> {
let mut environment = WorkerE2eEnvironment::new()?;
assert!(!environment.server_data_dir.exists());
assert!(!environment.runtime_data_dir.exists());
environment.start()?;
assert!(environment.server_data_dir.join("server.db").is_file());
assert!(
environment
.runtime_data_dir
.join("runtime/runtime.json")
.is_file()
);
environment.assert_connected_and_empty()?;
let artifacts_dir = environment.artifacts_dir.clone();
let report = environment.cleanup()?;
assert!(report.cleanup_success);
let persisted: WorkerEnvironmentCleanupReport =
serde_json::from_slice(&fs::read(artifacts_dir.join("cleanup.json"))?)?;
assert!(persisted.cleanup_success);
assert!(!persisted.fixture_root_exists_after);
Ok(())
}
#[test]
fn concurrent_environments_are_isolated_and_cleanup_is_local() -> Result<()> {
let mut first = WorkerE2eEnvironment::new()?;
let mut second = WorkerE2eEnvironment::new()?;
assert_ne!(first.root, second.root);
assert_ne!(first.server_data_dir, second.server_data_dir);
assert_ne!(first.runtime_data_dir, second.runtime_data_dir);
assert_ne!(first.runtime_id, second.runtime_id);
assert_ne!(first.account_handle, second.account_handle);
assert_ne!(first.artifacts_dir, second.artifacts_dir);
first.start()?;
second.start()?;
assert_ne!(first.server_port()?, second.server_port()?);
assert_ne!(first.runtime_port()?, second.runtime_port()?);
assert_ne!(first.server_pid(), second.server_pid());
assert_ne!(first.runtime_pid(), second.runtime_pid());
assert_ne!(first.workspace_id, second.workspace_id);
first.assert_connected_and_empty()?;
second.assert_connected_and_empty()?;
let second_root = second.root.clone();
let second_server_data = second.server_data_dir.clone();
let second_runtime_data = second.runtime_data_dir.clone();
let second_artifacts = second.artifacts_dir.clone();
let first_artifacts = first.artifacts_dir.clone();
let first_report = first.cleanup()?;
assert!(first_report.cleanup_success);
assert!(first_artifacts.is_dir());
assert!(second_root.is_dir());
assert!(second_server_data.is_dir());
assert!(second_runtime_data.is_dir());
assert!(second_artifacts.is_dir());
second.assert_connected_and_empty()?;
let second_report = second.cleanup()?;
assert!(second_report.cleanup_success);
assert!(second_artifacts.join("cleanup.json").is_file());
Ok(())
}