Something went wrong. Try again.
A lexicon-driven AppView for ATProto. happyview.dev
backfill firehose jetstream atproto appview oauth lexicon
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474//! One run: load the project's interpreters, build its snapshot with the//! tests in it, and run each test file on its own database.
use std::collections::BTreeSet;use std::io::Write;use std::path::{Path, PathBuf};use std::sync::Arc;use std::time::Instant;
use futures_util::stream::{self, StreamExt};use serde_json::Value;
use crate::AppState;use crate::cli::TestArgs;use crate::config::Config;use crate::config::project::ConfigMode;use crate::plugin::{ExecutionError, PluginRegistry, ScriptExecuteOutput, WasmRuntime, loader};use crate::project::build::{BuildEnv, BuildOptions, Diagnostic, build_snapshot_with, install};use crate::project::snapshot::{ProjectScript, ProjectSnapshot, ScriptBody};use crate::script::{DispatchError, Invocation, ScriptOrigin, Trigger, dispatch};
use super::env::TestEnv;use super::report::{FileError, FileResult, exit_code, render};
/// The first SDK whose interpreters accept a test file as input.const TEST_RUN_SDK: &str = "0.3.0";
/// Held for as long as a `Runner` lives. Preparing one installs its project's/// settings as the process-wide configuration layer, which is read through/// for every later build and run, so a second runner in the same process/// would read the first's.static RUN: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
pub struct Runner { pub env: TestEnv, pub root: PathBuf, _run: tokio::sync::MutexGuard<'static, ()>,}
/// The configuration a test run starts from. The defaults fill only what/// neither the environment nor `happyview.toml` sets.fn load_config(root: &Path) -> Result<Config, String> { let env_file = root.join(".env.test").to_string_lossy().into_owned(); Config::load_from_with(Some(root), ConfigMode::Serve, |_| { vec![ // `Config::from_env` requires one; the run opens its own template // and never this URL. ("DATABASE_URL".to_string(), "sqlite::memory:".to_string()), ( "PUBLIC_URL".to_string(), format!("https://{}", super::TEST_HOST), ), // A test's `ctx.env` comes from here, never from the developer's // `.env`, which holds real secrets. ("HAPPYVIEW_SCRIPT_ENV_FILE".to_string(), env_file), ] }) .map_err(|e| e.to_string())}
/// A run that never produced an outcome. An interpreter built on an SDK from/// before test runs cannot decode a test file's input: its `BAD_INPUT` names/// the unknown `test` kind, and reads as the file's fault unless it is named/// for what it is. Any other `BAD_INPUT` is the plugin's to explain.fn dispatch_failure(error: DispatchError, interpreter: Option<&str>) -> FileError { let message = match error { DispatchError::NoInterpreter { language } => { crate::script::no_interpreter_message(&language) } DispatchError::Execution(ExecutionError::PluginError { code, message, .. }) if code == "BAD_INPUT" && message.contains("`test`") => { format!( "the '{}' interpreter predates test runs and refused this file ({message}); \ it needs happyview-plugin-sdk {TEST_RUN_SDK} or later", interpreter.unwrap_or("unknown") ) } // The display text names no budget. A test file is always source, // never a compiled module, so what stopped it is its own code running // past its wall-clock budget. DispatchError::Execution(ExecutionError::Timeout) => format!( "the file ran past its {}-second limit for its own code", crate::plugin::executor::TEST_FILE_GUEST_SECONDS ), other => other.to_string(), }; FileError { message, file: None, line: None, }}
impl Runner { pub async fn prepare(args: &TestArgs) -> Result<Runner, String> { let run = RUN.lock().await; let root = args.root.clone(); let config = load_config(&root)?; let dirs = if args.plugin_dirs.is_empty() { crate::project::check::default_plugin_dirs(&root) } else { args.plugin_dirs.clone() }; let cache_dir = config .plugin_cache_dir .clone() .unwrap_or_else(|| root.join(".happyview").join("plugin-cache")); let runtime = Arc::new( WasmRuntime::with_cache_dir(Some(cache_dir)) .map_err(|e| format!("could not start the wasm runtime: {e}"))?, ); let registry = Arc::new(PluginRegistry::new().with_runtime(runtime.clone())); let mut plugins = Vec::new(); for dir in &dirs { plugins.push( loader::load_from_file(dir) .await .map_err(|e| format!("{}: could not load the plugin: {e}", dir.display()))?, ); } if let Some((id, e)) = registry.register_all(plugins).await.into_iter().next() { return Err(format!("plugin {id}: {e}")); } let env = TestEnv::new(registry, runtime, Some(config)) .await .map_err(|e| format!("could not create the test database: {e}"))?; Ok(Runner { env, root, _run: run, }) }
/// Load the project's configuration again, after a watch saw /// `happyview.toml` or `.env.test` change. A refused load keeps the /// current one. pub async fn reload_config(&mut self) -> Result<(), String> { let config = load_config(&self.root)?; self.env.set_config(config).await; Ok(()) }
/// Build with the tests included and install the result on the base /// state, which every file's state shares. Answers the installed /// snapshot, pinned, and the build's warnings. pub async fn build(&self) -> Result<(Arc<ProjectSnapshot>, Vec<Diagnostic>), Vec<Diagnostic>> { // The base state's executor is what prepares each test file with // `ChunkRole::Test`. let report = build_snapshot_with( &BuildEnv::from_state(&self.env.base), &self.root, BuildOptions { include_tests: true, }, ) .await?; install(&self.env.base, report.snapshot).await; Ok((self.env.base.project.load_full(), report.warnings)) }
pub async fn run( &self, snapshot: &Arc<ProjectSnapshot>, only: Option<&BTreeSet<String>>, filter: Option<&str>, ) -> Vec<FileResult> { let files: Vec<(&String, &ProjectScript)> = snapshot .tests .iter() .filter(|(key, _)| only.is_none_or(|only| only.contains(*key))) .filter(|(key, _)| filter.is_none_or(|f| key.contains(f))) .collect(); let jobs = std::thread::available_parallelism().map_or(4, |n| n.get()); // Built before they reach the stream: a closure inside the stream // leaves the run's future unprovably `Send`, and it could not be // spawned. let runs: Vec<_> = files .into_iter() .map(|(key, script)| self.run_file(snapshot, key, script)) .collect(); let mut results: Vec<FileResult> = stream::iter(runs).buffer_unordered(jobs).collect().await; results.sort_by(|a, b| a.path.cmp(&b.path)); results }
async fn run_file( &self, snapshot: &Arc<ProjectSnapshot>, key: &str, script: &ProjectScript, ) -> FileResult { let started = Instant::now(); let fail = |message: String| FileResult { path: key.to_string(), reports: Vec::new(), error: Some(FileError { message, file: None, line: None, }), duration: started.elapsed(), }; let file = match self.env.file_state(key).await { Ok(file) => file, Err(e) => return fail(format!("could not create the test database: {e}")), }; // A task of its own: a guest runs on the thread that polls it until // its deadline, so a file that spins would otherwise hold up every // file sharing its task, and a host panic is caught at the join // rather than ending the run. let outcome = AbortOnDrop(tokio::spawn(execute( file.state.clone(), snapshot.clone(), key.to_string(), script.clone(), ))) .await; let reports = file.harness.reports(); let error = match outcome { Err(join) => Some(panic_error(join)), Ok(Ok(ScriptExecuteOutput::Returned { .. })) if reports.is_empty() => Some(FileError { message: "the file reported no tests; define them with describe and it".into(), file: None, line: None, }), Ok(Ok(ScriptExecuteOutput::Returned { .. })) => None, Ok(Ok(ScriptExecuteOutput::Error { message, line, chunk, .. })) => Some(FileError { message, file: chunk, line, }), Ok(Err(e)) => { let interpreter = file .state .plugin_registry .get_interpreter_by_language(&script.language_id) .await .map(|p| p.info.id.clone()); Some(dispatch_failure(e, interpreter.as_deref())) } }; file.finish().await; FileResult { path: key.to_string(), reports, error, duration: started.elapsed(), } }}
/// Run one test file on its own state.async fn execute( state: AppState, snapshot: Arc<ProjectSnapshot>, key: String, script: ProjectScript,) -> Result<ScriptExecuteOutput, DispatchError> { // A test file is always source; dispatch runs a compiled module as // itself, reading no source, should one ever arrive here. let source = match &script.body { ScriptBody::Source { source, .. } => source.as_ref(), ScriptBody::Module { .. } => "", }; let trigger_id = format!("test:{key}"); // A project origin: dispatch names the chunk by the file's path, sends its // prepared artifact, and resolves its imports against this snapshot for // the whole run. dispatch( &state, &Invocation { trigger_id: &trigger_id, trigger: Trigger::Test, language: &script.language_id, source, input: &Value::Null, caller_did: None, has_pds_auth: false, method: None, collection: None, params: None, delegate_did: None, space: None, origin: ScriptOrigin::Project { snapshot: &snapshot, script: &script, }, }, None, ) .await}
/// A file's task, aborted when the run that awaits it is dropped. Otherwise/// its guest would carry on writing to a database in a directory the/// dropped run's `TestEnv` deletes.struct AbortOnDrop<T>(tokio::task::JoinHandle<T>);
impl<T> Drop for AbortOnDrop<T> { fn drop(&mut self) { self.0.abort(); }}
impl<T> std::future::Future for AbortOnDrop<T> { type Output = Result<T, tokio::task::JoinError>;
fn poll( mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, ) -> std::task::Poll<Self::Output> { std::pin::Pin::new(&mut self.0).poll(cx) }}
/// A file whose task never finished. A panic is a bug in the host, and says/// so, so it is not read as the test's; a cancelled task was stopped from/// outside and is no one's bug.fn panic_error(join: tokio::task::JoinError) -> FileError { let message = match join.try_into_panic() { Ok(panic) => { let reason = panic .downcast_ref::<&str>() .map(|s| s.to_string()) .or_else(|| panic.downcast_ref::<String>().cloned()) .unwrap_or_else(|| "no message".into()); format!( "happyview panicked running this file ({reason}); this is a bug in happyview, not in the test" ) } Err(join) if join.is_cancelled() => { "the run was cancelled before this file finished".into() } Err(join) => join.to_string(), }; FileError { message, file: None, line: None, }}
pub fn print_diagnostics(diagnostics: &[Diagnostic], out: &mut dyn Write) { for diagnostic in diagnostics { let _ = writeln!(out, "{diagnostic}"); }}
/// Runs one at a time in a process, since the runner it prepares holds the/// process-wide configuration.pub async fn run(args: &TestArgs, out: &mut (dyn Write + Send)) -> u8 { // Before anything resolves: a test run never asks PLC or DNS. super::identity::forge_identities(); let mut runner = match Runner::prepare(args).await { Ok(runner) => runner, Err(e) => { let _ = writeln!(out, "{e}"); return 2; } }; let snapshot = match runner.build().await { Ok((snapshot, warnings)) => { print_diagnostics(&warnings, out); snapshot } Err(diagnostics) => { print_diagnostics(&diagnostics, out); return 2; } }; let started = Instant::now(); let results = runner.run(&snapshot, None, args.filter.as_deref()).await; let code = if results.is_empty() { let _ = match &args.filter { Some(filter) => writeln!(out, "no test file matches '{filter}'"), None => writeln!(out, "no test files: name one *.test.<ext> under src/"), }; 1 } else { match render(&results, started.elapsed(), out) { Ok(summary) => exit_code(&summary), Err(_) => 1, } }; if args.watch { // The watch reuses this runner, and with it the lock it holds. return super::watch::watch(&mut runner, snapshot, code, args.filter.as_deref(), out).await; } code}
#[cfg(test)]mod tests { use super::*;
#[test] fn an_interpreter_refusing_a_test_file_is_named_with_the_sdk_it_needs() { let error = dispatch_failure( DispatchError::Execution(ExecutionError::PluginError { code: "BAD_INPUT".into(), message: "unknown variant `test`".into(), retryable: false, }), Some("lua"), ); assert!( error.message.contains("'lua' interpreter"), "{}", error.message ); assert!( error .message .contains("happyview-plugin-sdk 0.3.0 or later"), "{}", error.message ); let other = dispatch_failure( DispatchError::Execution(ExecutionError::PluginError { code: "HOST_ERROR".into(), message: "boom".into(), retryable: false, }), Some("lua"), ); assert!(!other.message.contains("predates"), "{}", other.message); let malformed = dispatch_failure( DispatchError::Execution(ExecutionError::PluginError { code: "BAD_INPUT".into(), message: "missing field `source`".into(), retryable: false, }), Some("lua"), ); assert!( !malformed.message.contains("predates"), "{}", malformed.message ); assert!( malformed.message.contains("missing field `source`"), "{}", malformed.message ); let timeout = dispatch_failure(DispatchError::Execution(ExecutionError::Timeout), None); assert_eq!( timeout.message, "the file ran past its 60-second limit for its own code" ); }
#[tokio::test] async fn a_panicking_file_becomes_its_own_error() { let join = tokio::spawn(async { panic!("boom") }).await.unwrap_err(); let message = panic_error(join).message; assert!(message.contains("panicked"), "{message}"); assert!(message.contains("boom"), "{message}");
let handle = tokio::spawn(std::future::pending::<()>()); handle.abort(); assert_eq!( panic_error(handle.await.unwrap_err()).message, "the run was cancelled before this file finished" ); }}