use bonsai_db::AnalyzerDb;
use bonsai_lang_api::{FlowEvent, LanguageRegistry};
use bonsai_vfs::Vfs;
use std::sync::Arc;
fn db_with(source: &str) -> AnalyzerDb {
let vfs = Arc::new(Vfs::new());
vfs.write("a.py".to_string(), Arc::<str>::from(source));
let registry = Arc::new(LanguageRegistry::new());
registry.register(Arc::new(bonsai_lang_python::PythonAdapter::new()));
let db = AnalyzerDb::new(vfs, registry);
for f in db.vfs().all_files() {
let _ = db.decl_index(f);
}
db
}
fn calls_in(db: &AnalyzerDb, fn_name: &str) -> Vec<(String, Vec<String>)> {
let g = db.global_index();
let mut out = Vec::new();
for f in g.all_files() {
for decl in g.decls_in(f) {
if decl.name != fn_name {
continue;
}
walk(&decl.flow_events, &mut out);
}
}
out
}
fn walk(events: &[FlowEvent], out: &mut Vec<(String, Vec<String>)>) {
for event in events {
match event {
FlowEvent::Call { name, args, .. } => {
out.push((
name.clone(),
args.iter().map(|arg| arg.value_text.clone()).collect(),
));
}
FlowEvent::Branch {
then_events,
else_events,
..
} => {
walk(then_events, out);
walk(else_events, out);
}
FlowEvent::Loop { body, .. } | FlowEvent::Defer { body, .. } | FlowEvent::Using { body, .. } => {
walk(body, out);
}
FlowEvent::Try {
body,
catch_events,
finally_events,
..
} => {
walk(body, out);
walk(catch_events, out);
walk(finally_events, out);
}
_ => {}
}
}
}
#[test]
fn asyncio_to_thread_remains_an_exact_runtime_call() {
let db = db_with(
r#"
import asyncio
async def load_asset(path):
return await asyncio.to_thread(_read_bytes, path)
def _read_bytes(p):
return open(p).read()
"#,
);
let calls = calls_in(&db, "load_asset");
assert!(
calls.iter().any(|(name, args)| name == "asyncio.to_thread"
&& args == &vec!["_read_bytes".to_string(), "path".to_string()]),
"expected the compiler-lowered asyncio.to_thread call, got {calls:?}"
);
assert!(
!calls.iter().any(|(name, _)| name == "_read_bytes"),
"the adapter must not invent an _read_bytes call from library semantics: {calls:?}"
);
}