char/avril

(wip) why'd you have to go and make things so complicaaaaated?

git clone https://git.t4t.associates/char/avril

Charlotte Somdeslop: consolidate public type and registry test utils7325e8c

main
branchesmainall refs →
16.0 KiB466 linesraw
1#![allow(dead_code)]
2// Shared across test targets; each uses only part of this module.
3use std::collections::{BTreeMap, BTreeSet};
4use std::io::{Read, Write};
5use std::net::TcpListener;
6use std::path::Path;
7use std::process::{Command, Output};
8use std::sync::{
9    Arc, Mutex,
10    atomic::{AtomicBool, AtomicUsize, Ordering},
11};
12use std::thread::JoinHandle;
13use std::time::Duration;
14
15use avril_runtime::registry::{Options, PackageStore, checksum, npm::resolution::NpmResolver};
16use serde_json::json;
17use url::Url;
18
19/// The directory's `avril.lock` as data: locks are fJSON, so plain JSON parsing misses
20/// implicit root braces and other fJSON spellings.
21pub fn read_lock(directory: &std::path::Path) -> serde_json::Value {
22    let path = directory.join("avril.lock");
23    let source = std::fs::read_to_string(&path).unwrap();
24    serde_fjson::parse(&source).unwrap()
25}
26
27// Downloads must remain driven while tests wait outside Tokio.
28pub fn runtime_handle() -> tokio::runtime::Handle {
29    static RUNTIME: std::sync::OnceLock<tokio::runtime::Runtime> = std::sync::OnceLock::new();
30    RUNTIME
31        .get_or_init(|| {
32            tokio::runtime::Builder::new_multi_thread()
33                .worker_threads(2)
34                .enable_all()
35                .build()
36                .unwrap()
37        })
38        .handle()
39        .clone()
40}
41
42pub fn manifest(name: &str, version: &str, extra: serde_json::Value) -> String {
43    let mut package = json!({"name": name, "version": version, "type": "module"});
44    package
45        .as_object_mut()
46        .unwrap()
47        .extend(extra.as_object().unwrap().clone());
48    package.to_string()
49}
50
51pub fn resolver(options: Options) -> NpmResolver {
52    NpmResolver::new(Arc::new(
53        PackageStore::new(runtime_handle(), options).unwrap(),
54    ))
55}
56
57pub fn files(dir: &Path, entries: &[(&str, &str)]) {
58    for (name, source) in entries {
59        write(dir, name, source);
60    }
61}
62
63pub fn write(dir: &Path, name: &str, source: &str) {
64    let path = dir.join(name);
65    std::fs::create_dir_all(path.parent().unwrap()).unwrap();
66    std::fs::write(path, source).unwrap();
67}
68
69pub fn success(output: &Output) -> String {
70    assert!(
71        output.status.success(),
72        "{}",
73        String::from_utf8_lossy(&output.stderr)
74    );
75    String::from_utf8(output.stdout.clone()).unwrap()
76}
77
78pub fn failure(output: &Output) -> String {
79    assert!(
80        !output.status.success(),
81        "{}",
82        String::from_utf8_lossy(&output.stdout)
83    );
84    String::from_utf8_lossy(&output.stderr).into_owned()
85}
86
87struct RegistryState {
88    files: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
89    abbreviated: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
90    requests: Arc<Mutex<Vec<String>>>,
91    responses: Arc<Mutex<Vec<String>>>,
92    stop: Arc<AtomicBool>,
93    hold_tarballs: Arc<AtomicBool>,
94    hold_paths: Arc<Mutex<BTreeSet<String>>>,
95    active: Arc<AtomicUsize>,
96    max: Arc<AtomicUsize>,
97}
98
99struct ActiveGuard {
100    state: Arc<RegistryState>,
101}
102
103impl Drop for ActiveGuard {
104    fn drop(&mut self) {
105        self.state.active.fetch_sub(1, Ordering::SeqCst);
106    }
107}
108
109pub struct Registry {
110    pub url: Url,
111    pub files: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
112    pub requests: Arc<Mutex<Vec<String>>>,
113    /// When set, tarball responses stall until cleared, so guest code can
114    /// revoke authority mid-acquisition deterministically.
115    pub hold_tarballs: Arc<AtomicBool>,
116    /// Opt-in concurrent serving for overlap/bound tests; default off keeps
117    /// the original serial accept loop.
118    pub concurrent: Arc<AtomicBool>,
119    /// Tarball paths to stall while present; stall loops also watch stop.
120    pub hold_paths: Arc<Mutex<BTreeSet<String>>>,
121    pub active_tarballs: Arc<AtomicUsize>,
122    pub max_tarballs: Arc<AtomicUsize>,
123    pub responses: Arc<Mutex<Vec<String>>>,
124    abbreviated: Arc<Mutex<BTreeMap<String, Vec<u8>>>>,
125    stop: Arc<AtomicBool>,
126    worker: Option<JoinHandle<()>>,
127    handlers: Arc<Mutex<Vec<JoinHandle<()>>>>,
128}
129
130impl Registry {
131    pub fn new() -> Self {
132        let listener = TcpListener::bind("127.0.0.1:0").unwrap();
133        let url = Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
134        listener.set_nonblocking(true).unwrap();
135        let files = Arc::new(Mutex::new(BTreeMap::<String, Vec<u8>>::new()));
136        let requests = Arc::new(Mutex::new(Vec::new()));
137        let stop = Arc::new(AtomicBool::new(false));
138        let hold_tarballs = Arc::new(AtomicBool::new(false));
139        let concurrent = Arc::new(AtomicBool::new(false));
140        let hold_paths = Arc::new(Mutex::new(BTreeSet::<String>::new()));
141        let active = Arc::new(AtomicUsize::new(0));
142        let max = Arc::new(AtomicUsize::new(0));
143        let responses = Arc::new(Mutex::new(Vec::<String>::new()));
144        let abbreviated = Arc::new(Mutex::new(BTreeMap::<String, Vec<u8>>::new()));
145        let handlers = Arc::new(Mutex::new(Vec::<JoinHandle<()>>::new()));
146        let state = Arc::new(RegistryState {
147            files: files.clone(),
148            abbreviated: abbreviated.clone(),
149            requests: requests.clone(),
150            responses: responses.clone(),
151            stop: stop.clone(),
152            hold_tarballs: hold_tarballs.clone(),
153            hold_paths: hold_paths.clone(),
154            active: active.clone(),
155            max: max.clone(),
156        });
157        let accept_concurrent = concurrent.clone();
158        let accept_handlers = handlers.clone();
159        let worker = std::thread::spawn(move || {
160            while !state.stop.load(Ordering::Relaxed) {
161                let stream = match listener.accept() {
162                    Ok((stream, _)) => stream,
163                    Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
164                        std::thread::sleep(Duration::from_millis(1));
165                        continue;
166                    }
167                    Err(error) => panic!("{error}"),
168                };
169                if accept_concurrent.load(Ordering::Relaxed) {
170                    let state = state.clone();
171                    accept_handlers
172                        .lock()
173                        .unwrap()
174                        .push(std::thread::spawn(move || {
175                            serve_stream(state, stream);
176                        }));
177                    continue;
178                }
179                serve_stream(state.clone(), stream);
180            }
181        });
182        Self {
183            url,
184            files,
185            abbreviated,
186            requests,
187            hold_tarballs,
188            concurrent,
189            hold_paths,
190            active_tarballs: active,
191            max_tarballs: max,
192            responses,
193            stop,
194            worker: Some(worker),
195            handlers,
196        }
197    }
198
199    pub fn package(
200        &self,
201        name: &str,
202        version: &str,
203        files: &[(&str, &str)],
204        exports: &[(&str, &str)],
205    ) {
206        let mut artifacts = self.files.lock().unwrap();
207        let index_path = format!("/{name}/meta.json");
208        let mut index = artifacts
209            .get(&index_path)
210            .map(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).unwrap())
211            .unwrap_or_else(|| json!({"versions": {}}));
212        index["versions"][version] = json!({});
213        artifacts.insert(index_path, serde_json::to_vec(&index).unwrap());
214        let manifest: BTreeMap<_, _> = files
215            .iter()
216            .map(|(path, source)| {
217                artifacts.insert(
218                    format!("/{name}/{version}/{path}"),
219                    source.as_bytes().to_vec(),
220                );
221                (
222                    format!("/{path}"),
223                    json!({"size": source.len(), "checksum": checksum(source.as_bytes())}),
224                )
225            })
226            .collect();
227        let exports: BTreeMap<_, _> = exports.iter().copied().collect();
228        artifacts.insert(
229            format!("/{name}/{version}_meta.json"),
230            serde_json::to_vec(
231                &json!({"manifest": manifest, "exports": exports, "moduleGraph2": {}}),
232            )
233            .unwrap(),
234        );
235    }
236
237    pub fn tarball_bytes(files: &[(&str, &str)]) -> Vec<u8> {
238        use flate2::{Compression, write::GzEncoder};
239        let mut tar_data = Vec::new();
240        {
241            let mut tar = tar::Builder::new(&mut tar_data);
242            for (path, source) in files {
243                let full = format!("package/{path}");
244                let mut header = tar::Header::new_gnu();
245                header.set_mode(0o644);
246                header.set_size(source.len() as u64);
247                header.set_cksum();
248                tar.append_data(&mut header, &full, source.as_bytes())
249                    .unwrap();
250            }
251            tar.finish().unwrap();
252        }
253        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
254        use std::io::Write;
255        encoder.write_all(&tar_data).unwrap();
256        encoder.finish().unwrap()
257    }
258
259    pub fn tarball_integrity(bytes: &[u8]) -> String {
260        use base64::Engine;
261        use sha2::Digest;
262        let digest = sha2::Sha512::digest(bytes);
263        format!(
264            "sha512-{}",
265            base64::engine::general_purpose::STANDARD.encode(digest)
266        )
267    }
268
269    fn tarball_path(name: &str, version: &str) -> String {
270        let leaf = name.rsplit('/').next().unwrap();
271        format!("/{name}/-/{leaf}-{version}.tgz")
272    }
273
274    pub fn publish_package(
275        &self,
276        name: &str,
277        version: &str,
278        extra: serde_json::Value,
279        sources: &[(&str, &str)],
280    ) {
281        let package = manifest(name, version, extra);
282        let mut files = vec![("package.json", package.as_str())];
283        if !sources.iter().any(|(path, _)| path.starts_with("index.")) {
284            files.push(("index.js", "export const value = 1;\nexport default 1;"));
285        }
286        files.extend_from_slice(sources);
287        self.publish_npm(name, version, &files);
288    }
289
290    pub fn resolver(&self, dir: &Path) -> NpmResolver {
291        resolver(self.options(dir))
292    }
293
294    pub fn frozen_resolver(&self, dir: &Path) -> NpmResolver {
295        let mut options = self.options(dir);
296        options.offline = true;
297        options.frozen = true;
298        resolver(options)
299    }
300
301    pub fn command(&self, dir: &Path, subcommand: &str) -> Command {
302        // This fixture also builds in CLI unit tests, which never launch commands.
303        let Some(binary) = option_env!("CARGO_BIN_EXE_avril") else {
304            panic!("CLI commands require an integration-test target");
305        };
306        let mut command = Command::new(binary);
307        command
308            .current_dir(dir)
309            .arg(subcommand)
310            .arg(format!("--jsr-registry={}", self.url))
311            .arg(format!("--npm-registry={}", self.url))
312            .arg(format!("--cache-dir={}", dir.join("cache").display()));
313        command
314    }
315
316    pub fn run_npm(&self, dir: &Path, args: &[&str], entry: &str) -> Output {
317        let mut command = self.command(dir, args[0]);
318        command.args(&args[1..]);
319        // `add` and friends take no entry module.
320        if !entry.is_empty() {
321            command.arg(entry);
322        }
323        command.output().unwrap()
324    }
325
326    pub fn publish_npm(&self, name: &str, version: &str, files: &[(&str, &str)]) {
327        let tarball = Self::tarball_bytes(files);
328        let integrity = Self::tarball_integrity(&tarball);
329        self.publish_npm_dist(name, version, tarball, &integrity, None, true);
330    }
331
332    pub fn publish_npm_dist(
333        &self,
334        name: &str,
335        version: &str,
336        tarball: Vec<u8>,
337        integrity: &str,
338        shasum: Option<&str>,
339        latest: bool,
340    ) {
341        let mut artifacts = self.files.lock().unwrap();
342        let tarball_path = Self::tarball_path(name, version);
343        let tarball_url = format!("{}{tarball_path}", self.url.as_str().trim_end_matches('/'));
344        artifacts.insert(tarball_path, tarball);
345        let packument_path = format!("/{name}");
346        let mut packument = artifacts
347            .get(&packument_path)
348            .map(|bytes| serde_json::from_slice::<serde_json::Value>(bytes).unwrap())
349            .unwrap_or_else(|| json!({"name": name, "dist-tags": {}, "versions": {}}));
350        packument["versions"][version] = json!({
351            "name": name,
352            "version": version,
353            "dist": {
354                "tarball": tarball_url,
355                "integrity": integrity,
356            },
357        });
358        if let Some(shasum) = shasum {
359            packument["versions"][version]["dist"]["shasum"] = json!(shasum);
360        }
361        if latest {
362            packument["dist-tags"]["latest"] = json!(version);
363        }
364        artifacts.insert(packument_path, serde_json::to_vec(&packument).unwrap());
365    }
366
367    /// Abbreviated metadata for Accept-negotiating clients; others get the ordinary response.
368    pub fn publish_npm_abbreviated(&self, name: &str, body: Vec<u8>) {
369        self.abbreviated
370            .lock()
371            .unwrap()
372            .insert(format!("/{name}"), body);
373    }
374
375    pub fn options(&self, directory: &std::path::Path) -> Options {
376        Options {
377            jsr_registry: self.url.clone(),
378            npm_registry: self.url.clone(),
379            cache: directory.join("cache"),
380            lock: directory.join("avril.lock"),
381            project_root: directory.to_owned(),
382            offline: false,
383            frozen: false,
384            update: false,
385            ephemeral_lock: false,
386            vendor: None,
387        }
388    }
389}
390
391fn serve_stream(state: Arc<RegistryState>, mut stream: std::net::TcpStream) {
392    stream
393        .set_read_timeout(Some(Duration::from_secs(3)))
394        .unwrap();
395    let mut request = Vec::new();
396    let mut buf = [0; 1024];
397    while !request.windows(4).any(|bytes| bytes == b"\r\n\r\n") {
398        let length = match stream.read(&mut buf) {
399            Ok(0) | Err(_) => break,
400            Ok(length) => length,
401        };
402        request.extend_from_slice(&buf[..length]);
403    }
404    let request = String::from_utf8_lossy(&request);
405    let Some(path) = request
406        .lines()
407        .next()
408        .and_then(|line| line.split_whitespace().nth(1))
409    else {
410        return;
411    };
412    assert!(!request.to_ascii_lowercase().contains("accept: text/html"));
413    let abbreviated = request.lines().any(|line| {
414        line.to_ascii_lowercase().starts_with("accept:")
415            && line.contains("application/vnd.npm.install-v1+json")
416    });
417    let _guard = path.ends_with(".tgz").then(|| {
418        let active = state.active.fetch_add(1, Ordering::SeqCst) + 1;
419        state.max.fetch_max(active, Ordering::SeqCst);
420        ActiveGuard { state: state.clone() }
421    });
422    state.requests.lock().unwrap().push(path.to_owned());
423    while !state.stop.load(Ordering::Relaxed)
424        && ((path.ends_with(".tgz") && state.hold_tarballs.load(Ordering::Relaxed))
425            || state.hold_paths.lock().unwrap().contains(path))
426    {
427        std::thread::sleep(Duration::from_millis(10));
428    }
429    if state.stop.load(Ordering::Relaxed) {
430        return;
431    }
432    let mut bytes = None;
433    if abbreviated {
434        bytes = state.abbreviated.lock().unwrap().get(path).cloned();
435    }
436    let bytes = bytes.or_else(|| state.files.lock().unwrap().get(path).cloned());
437    let (status, bytes) = match bytes {
438        Some(bytes) => ("200 OK", bytes),
439        None => ("404 Not Found", b"not found".to_vec()),
440    };
441    let headers = format!(
442        "HTTP/1.1 {status}\r\nContent-Type: application/octet-stream\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
443        bytes.len()
444    );
445    stream
446        .set_write_timeout(Some(Duration::from_secs(3)))
447        .unwrap();
448    let _ = stream
449        .write_all(headers.as_bytes())
450        .and_then(|_| stream.write_all(&bytes));
451    state.responses.lock().unwrap().push(path.to_owned());
452}
453
454impl Drop for Registry {
455    fn drop(&mut self) {
456        self.stop.store(true, Ordering::Relaxed);
457        self.worker.take().unwrap().join().unwrap();
458        for handler in self.handlers.lock().unwrap().drain(..) {
459            if std::thread::panicking() {
460                let _ = handler.join();
461            } else {
462                handler.join().unwrap();
463            }
464        }
465    }
466}