mod common;

use std::io::{BufRead, BufReader, Read, Write};
use std::net::{Shutdown, TcpListener, TcpStream};
use std::process::Command;
use std::time::Duration;

struct Request {
    method: String,
    path: String,
    headers: String,
    body: Vec<u8>,
}

fn request(socket: &mut TcpStream) -> Request {
    socket
        .set_read_timeout(Some(Duration::from_secs(5)))
        .unwrap();
    let mut reader = BufReader::new(socket);
    let mut line = String::new();
    reader.read_line(&mut line).unwrap();
    let mut words = line.split_whitespace();
    let method = words.next().unwrap().to_owned();
    let path = words.next().unwrap().to_owned();
    let mut headers = String::new();
    loop {
        line.clear();
        reader.read_line(&mut line).unwrap();
        if line == "\r\n" {
            break;
        }
        headers.push_str(&line.to_ascii_lowercase());
    }
    let mut body = Vec::new();
    if headers.contains("transfer-encoding: chunked") {
        loop {
            line.clear();
            reader.read_line(&mut line).unwrap();
            let len = usize::from_str_radix(line.trim().split(';').next().unwrap(), 16).unwrap();
            if len == 0 {
                break;
            }
            let start = body.len();
            body.resize(start + len, 0);
            reader.read_exact(&mut body[start..]).unwrap();
            let mut newline = [0; 2];
            reader.read_exact(&mut newline).unwrap();
        }
    } else if let Some(line) = headers
        .lines()
        .find(|line| line.starts_with("content-length:"))
    {
        let len = line.split_once(':').unwrap().1.trim().parse().unwrap();
        body.resize(len, 0);
        reader.read_exact(&mut body).unwrap();
    }
    Request { method, path, headers, body }
}

fn server(
    count: usize,
    handler: impl Fn(Request) -> Vec<u8> + Send + 'static,
) -> (String, std::thread::JoinHandle<()>) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let thread = std::thread::spawn(move || {
        for _ in 0..count {
            let (mut socket, _) = listener.accept().unwrap();
            let req = request(&mut socket);
            let _ = socket.write_all(&handler(req));
        }
    });
    (url, thread)
}

#[test]
fn fetch_ignores_guest_promise_hooks_and_internal_array_iterators() {
    let (url, thread) = server(1, |request| {
        assert_eq!(request.method, "POST");
        assert_eq!(request.body, b"upload");
        response("200 OK", "X-Test: captured\r\n", b"download")
    });
    let source = format!(
        r#"
        const then = Promise.prototype.then, originalPromise = Promise;
        function observe(promise) {{
            Object.defineProperty(promise, 'constructor', {{ value: undefined }});
            const result = Reflect.apply(then, promise, [value => value, error => {{ throw error; }}]);
            Object.defineProperty(result, 'constructor', {{ value: originalPromise }});
            return result;
        }}
        const chunk = new TextEncoder().encode('upload');
        const poison = () => {{ throw Error('Guest hook invoked'); }};
        Object.defineProperty(Promise.prototype, 'constructor', {{ get: poison, configurable: true }});
        Object.defineProperty(Promise, Symbol.species, {{ get: poison }});
        Promise.prototype.then = poison;
        Array.prototype[Symbol.iterator] = poison;
        const body = new ReadableStream({{ start(controller) {{ controller.enqueue(chunk); controller.close(); }} }});
        const result = await observe(fetch('{url}', {{ method: 'POST', body, duplex: 'half' }}));
        if (result.status !== 200 || result.headers.get('x-test') !== 'captured' || await observe(result.text()) !== 'download') throw Error('Fetch behavior');
        console.log('fetch primordials safe');
        "#
    );
    assert_eq!(
        common::run(&source, &[]).stdout,
        b"fetch primordials safe\n"
    );
    thread.join().unwrap();
}

fn response(status: &str, headers: &str, bytes: &[u8]) -> Vec<u8> {
    let mut result = format!(
        "HTTP/1.1 {status}\r\nConnection: close\r\nContent-Length: {}\r\n{headers}\r\n",
        bytes.len()
    )
    .into_bytes();
    result.extend_from_slice(bytes);
    result
}

#[test]
fn headers_body_helpers_and_explicit_stubs() {
    let out = common::run(
        r#"
        const h=new Headers([['Z',' a '],['x','1'],['X','2'],['set-cookie','a=1'],['set-cookie','b=2']]);
        if(h.get('X')!=='1, 2'||h.getSetCookie().length!==2||[...h.keys()].join('|')!=='set-cookie|set-cookie|x|z') throw Error('Headers');
        for(const bad of [['a b','x'],['x','a\r\nb'],['x','\u0100']]) {try {new Headers([bad]);throw Error('Accepted invalid header');}catch(e){if(!(e instanceof TypeError))throw e;}}
        const a=new Response(new Uint8Array([0,255,3])), b=a.clone();
        if(a.bodyUsed||b.bodyUsed||!(a.body instanceof ReadableStream))throw Error('Body flags');
        console.log([...await a.bytes()].join(','),[...new Uint8Array(await b.arrayBuffer())].join(','));
        try {await a.text();throw Error('Consumed body');}catch(e){if(!(e instanceof TypeError))throw e;}
        const j=Response.json({x:42});if((await j.json()).x!==42||j.headers.get('content-type')!=='application/json')throw Error('JSON');
        if(await new Response('\ufeff\ufeffx').text()!=='\ufeffx')throw Error('BOM');
        const params=new URLSearchParams('q=a+b');const req=new Request('https://example.com',{method:'POST',body:params});
        if(await req.text()!=='q=a+b')throw Error('Params');
        if(new Response(null,{status:200.9}).status!==200)throw Error('status conversion');
        for(const operation of [()=>fetch('data:text/plain,x'),()=>new Request('https://example.com',{cache:'reload'})]) {
            try {await operation();throw Error('Expected stub');}catch(e){if(!e.message.includes('not yet implemented'))throw e;}
        }
        const controller=new AbortController(), log=[];
        controller.signal.addEventListener('abort',()=>log.push(1));controller.signal.onabort=e=>{if(e.target!==controller.signal)throw Error('Event target');log.push(2);};controller.signal.addEventListener('abort',()=>log.push(3));controller.abort();
        if(log.join(',')!=='1,2,3'||controller.signal.reason.name!=='AbortError')throw Error('Abort events');
        const combined=AbortSignal.any([AbortSignal.abort('reason')]);try{combined.throwIfAborted();throw Error('Expected abort');}catch(e){if(e!=='reason')throw e;}
        console.log('helpers');
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"0,255,3 0,255,3\nhelpers\n");
}

#[test]
fn request_response_clone_and_body_disturbance_follow_stream_state() {
    let output = common::run(
        r#"
        const backing = new Uint8Array([0, 1, 2, 3]).buffer;
        const request = new Request('https://example.com', {
            method: 'POST', body: new DataView(backing, 1, 2),
        });
        new Uint8Array(backing).fill(9);
        const requestCopy = request.clone();
        if (request.bodyUsed || requestCopy.bodyUsed) throw Error('Clone disturbed body');
        const [first, second] = await Promise.all([request.bytes(), requestCopy.bytes()]);
        if (first.join() !== '1,2' || second.join() !== '1,2') throw Error('Request snapshot');
        if (!request.bodyUsed || !requestCopy.bodyUsed) throw Error('Body consumption');

        const source = new Request('https://example.com', {method:'POST', body:'source'});
        const transferred = new Request(source);
        if (!source.bodyUsed || await transferred.text() !== 'source') throw Error('Request transfer');

        const form = new FormData();
        form.append('field', 'before');
        const formRequest = new Request('https://example.com', {method:'POST', body:form});
        const formCopy = formRequest.clone();
        form.set('field', 'after');
        const formText = await formRequest.text(), formCopyText = await formCopy.text();
        if (!formText.includes('before') || !formCopyText.includes('before') ||
            formCopyText.includes('after')) throw Error('FormData snapshot');

        const locked = new Response('locked'), reader = locked.body.getReader();
        if (locked.bodyUsed) throw Error('Lock is not disturbance');
        for (const operation of [() => locked.clone(), () => locked.text()]) {
            try {await operation();throw Error('Locked body accepted');}
            catch (error) {if (!(error instanceof TypeError)) throw error;}
        }
        await reader.cancel();
        const disturbed = new Response('disturbed'), disturbedReader = disturbed.body.getReader();
        await disturbedReader.read();
        if (!disturbed.bodyUsed) throw Error('Direct stream read was not reflected');
        try {disturbed.clone();throw Error('Disturbed body cloned');}
        catch (error) {if (!(error instanceof TypeError)) throw error;}

        const reason = {streamError:true};
        const errored = new Response(new ReadableStream({start(controller) {controller.error(reason);}}));
        const erroredCopy = errored.clone();
        for (const body of [errored, erroredCopy]) {
            try {await body.text();throw Error('Stream error was lost');}
            catch (error) {if (error !== reason || !body.bodyUsed) throw error;}
        }
        console.log('clones');
    "#,
        &[],
    );
    assert_eq!(output.stdout, b"clones\n");
}

#[test]
fn timeout_abort_rejects_pending_fetch_response_body_reads_with_reason() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let thread = std::thread::spawn(move || {
        let (mut socket, _) = listener.accept().unwrap();
        request(&mut socket);
        socket
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nhello")
            .unwrap();
        socket
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let mut byte = [0];
        let _ = socket.read(&mut byte);
    });
    let source = r#"
        const signal = AbortSignal.timeout(100);
        const response = await fetch('URL', {signal});
        const reader = response.body.getReader();
        if ((await reader.read()).value.length !== 5) throw Error('Initial response chunk');
        try {await reader.read();throw Error('Pending response read survived timeout');}
        catch (error) {
            if (error !== signal.reason || error.name !== 'TimeoutError' || error.code !== 23) throw error;
        }
        console.log('timeout abort');
    "#
    .replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"timeout abort\n");
    thread.join().unwrap();
}

#[test]
fn navigator_user_agent_sets_fetch_defaults_without_overriding_explicit_headers() {
    let (url, thread) = server(8, |req| {
        let ua = req
            .headers
            .lines()
            .find_map(|line| line.strip_prefix("user-agent:"))
            .expect("Missing User-Agent")
            .trim();
        response("200 OK", "", ua.as_bytes())
    });
    let source = r#"
        const url = 'URL';
        const original = navigator.userAgent;
        if ((await (await fetch(url)).text()) !== original.toLowerCase()) throw Error('Initial default');
        const request = new Request(url);
        navigator.userAgent = 'first/1.0';
        const pending = fetch(request);
        navigator.userAgent = 'second/2.0';
        if ((await (await pending).text()) !== 'first/1.0') throw Error('In-flight UA changed');
        if ((await (await fetch(request)).text()) !== 'second/2.0') throw Error('Default not updated');
        if (request.headers.has('user-agent')) throw Error('Original request mutated');
        if ((await (await fetch(url, {headers: {'uSeR-aGeNt': 'explicit'}})).text()) !== 'explicit') throw Error('Explicit UA lost');
        if ((await (await fetch(url, {headers: {'User-Agent': ''}})).text()) !== '') throw Error('Explicit empty UA lost');
        navigator.userAgent = '';
        if ((await (await fetch(url)).text()) !== '') throw Error('Empty default lost');
        for (const invalid of ['bad\r\ninjected: yes', 'bad\0value', '\u0100']) {
            navigator.userAgent = invalid;
            try { await fetch(url); throw Error('Invalid UA accepted'); }
            catch (error) { if (!(error instanceof TypeError)) throw error; }
        }
        if ((await (await fetch(url, {headers: {'User-Agent': 'override'}})).text()) !== 'override') throw Error('Invalid default affected explicit UA');
        navigator.userAgent = 'hardened/1.0';
        Object.defineProperty(navigator, 'userAgent', {get() {throw Error('Guest getter');}});
        Headers.prototype.get = Headers.prototype.set = () => {throw Error('Guest header method');};
        if ((await (await fetch(url)).text()) !== 'hardened/1.0') throw Error('Internal default lost');
        console.log('fetch UA');
    "#
    .replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"fetch UA\n");
    thread.join().unwrap();
}

#[test]
fn worker_user_agent_defaults_are_isolate_local() {
    let (url, thread) = server(2, |req| {
        let ua = req
            .headers
            .lines()
            .find_map(|line| line.strip_prefix("user-agent:"))
            .unwrap()
            .trim();
        response("200 OK", "", ua.as_bytes())
    });
    let dir = tempfile::tempdir().unwrap();
    std::fs::write(
        dir.path().join("worker.js"),
        r#"
        const original = navigator.userAgent;
        navigator.userAgent = 'worker/1.0';
        const received = await (await fetch('URL')).text();
        postMessage({original, current: navigator.userAgent, received});
        close();
    "#
        .replace("URL", &url),
    )
    .unwrap();
    std::fs::write(
        dir.path().join("main.js"),
        r#"
        const original = navigator.userAgent;
        navigator.userAgent = 'parent/1.0';
        const worker = new Worker('./worker.js', {type: 'module'});
        const result = await new Promise((resolve, reject) => {
            worker.onmessage = event => resolve(event.data);
            worker.onerror = reject;
        });
        if (result.original !== original || result.current !== 'worker/1.0' || result.received !== 'worker/1.0') throw Error('Worker UA');
        if (navigator.userAgent !== 'parent/1.0' || (await (await fetch('URL')).text()) !== 'parent/1.0') throw Error('Parent UA');
        console.log('isolated UA');
    "#
        .replace("URL", &url),
    ).unwrap();
    let output = common::wait(
        Command::new(env!("CARGO_BIN_EXE_avril"))
            .arg("run")
            .arg(dir.path().join("main.js"))
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped())
            .spawn()
            .unwrap(),
    );
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(output.stdout, b"isolated UA\n");
    thread.join().unwrap();
}

#[test]
fn downloads_streams_bytes_and_decompresses_without_blocking_js() {
    let (url, thread) = server(4, |req| {
        assert!(req.headers.contains("user-agent: avril/"));
        match req.path.as_str() {
            "/json" => response(
                "200 OK",
                "Content-Type: application/json\r\nSet-Cookie: a=1\r\nSet-Cookie: b=2\r\n",
                b"{\"answer\":42}",
            ),
            "/head" => response("204 No Content", "", b""),
            "/gzip" => response(
                "200 OK",
                "Content-Encoding: gzip\r\n",
                &[
                    31, 139, 8, 0, 0, 0, 0, 0, 2, 3, 203, 72, 205, 201, 201, 7, 0, 134, 166, 16,
                    54, 5, 0, 0, 0,
                ],
            ),
            _ => response("200 OK", "", &[0, 128, 255, 42]),
        }
    });
    let source=r#"
        const root='URL';
        const response=await fetch(root+'/json');if(!response.ok||response.status!==200||response.bodyUsed||response.headers.getSetCookie().length!==2)throw Error('Metadata');
        try{response.headers.set('x','y');throw Error('Mutable headers');}catch(e){if(!(e instanceof TypeError))throw e;}
        const copy=response.clone();if((await copy.json()).answer!==42||!(await response.bytes() instanceof Uint8Array))throw Error('JSON/bytes');
        const bytes=[];for await(const chunk of (await fetch(root+'/binary')).body)for(const byte of chunk)bytes.push(byte);
        if(bytes.join(',')!=='0,128,255,42')throw Error('Stream');
        if(await(await fetch(root+'/gzip')).text()!=='hello')throw Error('Gzip');
        const head=await fetch(root+'/head');if(head.body!==null||(await head.bytes()).length!==0)throw Error('Null body');
        console.log('download');
    "#.replace("URL",&url);
    assert_eq!(common::run(&source, &[]).stdout, b"download\n");
    thread.join().unwrap();
}

#[test]
fn uploads_buffer_views_strings_and_backpressured_streams() {
    let (url, thread) = server(3, |req| {
        assert_eq!(req.method, "POST");
        response("200 OK", "", &req.body)
    });
    let source=r#"
        const root='URL';
        const view=new DataView(new Uint8Array([9,1,2,3,8]).buffer,1,3);
        if([...await(await fetch(root,{method:'POST',body:view})).bytes()].join(',')!=='1,2,3')throw Error('View');
        if(await(await fetch(root,{method:'POST',body:'😀'})).text()!=='😀')throw Error('UTF8');
        let count=0;const stream=new ReadableStream({pull(c){if(count===4)c.close();else c.enqueue(new Uint8Array([++count]));}});
        const r=await fetch(root,{method:'POST',body:stream,duplex:'half'});if([...await r.bytes()].join(',')!=='1,2,3,4')throw Error('Upload');
        console.log('upload');
    "#.replace("URL",&url);
    assert_eq!(common::run(&source, &[]).stdout, b"upload\n");
    thread.join().unwrap();
}

#[test]
fn redirects_check_live_permissions_and_strip_cross_origin_credentials() {
    let (target, target_thread) = server(1, |req| {
        assert!(!req.headers.contains("authorization:"));
        assert!(!req.headers.contains("cookie:"));
        response("200 OK", "", b"target")
    });
    let location = target.clone();
    let (source, source_thread) = server(3, move |_| {
        response("302 Found", &format!("Location: {location}\r\n"), b"")
    });
    let dir = tempfile::tempdir().unwrap();
    let path = dir.path().join("main.js");
    std::fs::write(&path,r#"
        import{pledge}from'avril:std';const source='SOURCE';const target='TARGET';
        const manual=await fetch(source,{redirect:'manual'});if(manual.status!==302||manual.redirected)throw Error('Manual');await manual.body.cancel();
        try{await fetch(source);throw Error('Unauthorized redirect');}catch(e){if(!e.message.includes('PermissionDenied'))throw e;}
        pledge({net:false});try{await fetch(source);throw Error('Pledge ignored');}catch(e){if(!e.message.includes('PermissionDenied'))throw e;}
        console.log('denied');
    "#.replace("SOURCE",&source).replace("TARGET",&target)).unwrap();
    let host = source.strip_prefix("http://").unwrap();
    let output = Command::new(env!("CARGO_BIN_EXE_avril"))
        .arg("run")
        .args(["-0", &format!("--allow-net={host}"), path.to_str().unwrap()])
        .output()
        .unwrap();
    assert!(
        output.status.success(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    assert_eq!(output.stdout, b"denied\n");
    let script = format!(
        "const r=await fetch('{source}',{{headers:{{authorization:'secret',cookie:'secret'}}}});console.log(await r.text(),r.redirected,r.url==='{target}/');"
    );
    assert_eq!(common::run(&script, &[]).stdout, b"target true true\n");
    source_thread.join().unwrap();
    target_thread.join().unwrap();
}

#[test]
fn aborting_downloads_preserves_reasons_and_cancels_native_reads() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let thread = std::thread::spawn(move || {
        let (mut socket, _) = listener.accept().unwrap();
        request(&mut socket);
        socket
            .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nhello")
            .unwrap();
        socket
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let mut byte = [0];
        let _ = socket.read(&mut byte);
    });
    let source=r#"
        const controller=new AbortController(), reason={abort:true};
        const response=await fetch('URL',{signal:controller.signal});const reader=response.body.getReader();const first=await reader.read();if(first.value.length!==5)throw Error('Chunk');
        const pending=reader.read();controller.abort(reason);try{await pending;throw Error('Expected abort');}catch(e){if(e!==reason)throw e;}
        try{await fetch('URL',{signal:AbortSignal.abort(reason)});throw Error('Pre-abort');}catch(e){if(e!==reason)throw e;}
        console.log('abort');
    "#.replace("URL",&url);
    assert_eq!(common::run(&source, &[]).stdout, b"abort\n");
    thread.join().unwrap();
}

#[test]
fn byob_and_external_consumption_track_disturbance_and_close() {
    let (url, thread) = server(1, |_| response("200 OK", "", &[1, 2, 3, 4, 5]));
    let source=r#"
        const response=await fetch('URL');const reader=response.body.getReader({mode:'byob'});
        const bytes=[];while(true){const result=await reader.read(new Uint8Array(2));if(result.done)break;for(const byte of result.value)bytes.push(byte);}
        reader.releaseLock();if(bytes.join(',')!=='1,2,3,4,5'||!response.bodyUsed)throw Error('BYOB');
        try{response.clone();throw Error('Cloned disturbed body');}catch(e){if(!(e instanceof TypeError))throw e;}
        const empty=new Response('');const r=empty.body.getReader({mode:'byob'});if(!(await r.read(new Uint8Array(1))).done)throw Error('Empty BYOB');
        console.log('byob');
    "#.replace("URL",&url);
    assert_eq!(common::run(&source, &[]).stdout, b"byob\n");
    thread.join().unwrap();
}

#[test]
fn forbidden_header_checks_and_private_fetch_bindings_survive_overrides() {
    let (url, thread) = server(1, |req| {
        assert!(req.headers.contains("x-test: yes"));
        response("200 OK", "X-Value: original\r\n", b"ok")
    });
    let source=r#"
        const root='URL';
        for(const name of ['host','HOST','content-length','transfer-encoding','connection','proxy-authorization']) {
            try{await fetch(root,{headers:{[name]:'bad'}});throw Error('Unsafe header');}catch(e){if(!(e instanceof TypeError))throw e;}
        }
        const poison=()=>{throw Error('Poisoned public method');};
        Headers.prototype.append=poison;Headers.prototype.entries=poison;Headers.prototype[Symbol.iterator]=poison;
        JSON.parse=poison;Object.getOwnPropertyDescriptor=poison;AbortSignal.any=poison;
        const r=await fetch(root,{headers:{'X-Test':'yes'}});if(r.headers.get('x-value')!=='original'||await r.text()!=='ok')throw Error('Bindings');
        console.log('protected');
    "#.replace("URL",&url);
    assert_eq!(common::run(&source, &[]).stdout, b"protected\n");
    thread.join().unwrap();
}

#[test]
fn aborting_a_stalled_upload_cancels_its_source() {
    let source = r#"
        const controller=new AbortController();let canceled=false;
        const body=new ReadableStream({pull(){controller.abort('stop');return new Promise(()=>{});},cancel(){canceled=true;}});
        try{await fetch('http://127.0.0.1:1',{method:'POST',body,duplex:'half',signal:controller.signal});throw Error('Expected abort');}catch(e){if(e!=='stop')throw e;}
        if(!canceled)throw Error('Upload source not canceled');console.log('canceled');
    "#;
    assert_eq!(common::run(source, &[]).stdout, b"canceled\n");
}

#[test]
fn aborted_uploads_ignore_cleanup_rejections_and_preserve_the_abort_reason() {
    let source = r#"
        for (const delayed of [false,true]) {
            const controller = new AbortController(), reason = {aborted:true};
            let canceled = 0;
            const body = new ReadableStream({
                pull() {controller.abort(reason); return new Promise(() => {});},
                cancel(error) {
                    if (error !== reason) throw Error('Wrong cancellation reason');
                    canceled++;
                    if (!delayed) throw Error('Cancellation failed');
                    return new Promise((resolve,reject) => queueMicrotask(() => reject(Error('Cancellation failed'))));
                }
            });
            try {
                await fetch('http://127.0.0.1:1', {
                    method:'POST',body,duplex:'half',signal:controller.signal
                });
                throw Error('Expected abort');
            } catch (error) {if (error !== reason) throw error;}
            await new Promise(resolve => setTimeout(resolve));
            if (canceled !== 1) throw Error('Cancellation count');
        }
        console.log('cleanup');
    "#;
    assert_eq!(common::run(source, &[]).stdout, b"cleanup\n");
}

#[test]
fn transferring_cached_request_bodies_cancels_the_old_stream_without_changing_bytes() {
    let (url, thread) = server(1, |req| {
        assert_eq!(req.method, "POST");
        response("200 OK", "", &req.body)
    });
    let source = format!(
        r#"
        const original = new Request('{url}',{{method:'POST',body:new Uint8Array([0,128,255])}});
        const moved = new Request(original);
        if (!original.bodyUsed || moved.bodyUsed) throw Error('Transfer state');
        const reader = original.body.getReader();
        if (!(await reader.read()).done) throw Error('Original stream not canceled');
        reader.releaseLock();
        const response = await fetch(moved);
        if ([...await response.bytes()].join(',') !== '0,128,255') throw Error('Transferred bytes');
        console.log('transfer');
        "#
    );
    assert_eq!(common::run(&source, &[]).stdout, b"transfer\n");
    thread.join().unwrap();
}

#[test]
fn early_redirect_stops_upload_without_canceling_the_response() {
    let (target, target_thread) = server(1, |req| {
        assert_eq!(req.method, "GET");
        assert!(req.body.is_empty());
        response("200 OK", "", b"redirected")
    });
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let source = format!("http://{}", listener.local_addr().unwrap());
    let source_thread = std::thread::spawn(move || {
        let (mut socket, _) = listener.accept().unwrap();
        socket
            .set_read_timeout(Some(Duration::from_secs(5)))
            .unwrap();
        let mut reader = BufReader::new(&mut socket);
        let mut line = String::new();
        loop {
            line.clear();
            reader.read_line(&mut line).unwrap();
            if line == "\r\n" {
                break;
            }
        }
        socket
            .write_all(&response(
                "303 See Other",
                &format!("Location: {target}\r\n"),
                b"",
            ))
            .unwrap();
        // Closing with unread upload bytes can reset TCP before the response arrives.
        socket.shutdown(Shutdown::Write).unwrap();
        let _ = std::io::copy(&mut socket, &mut std::io::sink());
    });
    let script = format!(
        r#"
        let canceled=false;const body=new ReadableStream({{pull(c){{c.enqueue(new Uint8Array(65536));}},cancel(){{canceled=true;}}}},{{highWaterMark:0}});
        const response=await fetch('{source}',{{method:'POST',body,duplex:'half'}});
        if(await response.text()!=='redirected'||!response.redirected)throw Error('Redirect');
        if(!canceled)throw Error('Upload kept running');console.log('early');
    "#
    );
    assert_eq!(common::run(&script, &[]).stdout, b"early\n");
    source_thread.join().unwrap();
    target_thread.join().unwrap();
}

#[test]
fn blob_and_file_uploads_replay_on_redirect_and_download_as_blobs() {
    let (url, thread) = server(2, |req| {
        assert_eq!(req.method, "POST");
        assert_eq!(req.body, b"upload");
        assert!(req.headers.contains("content-type: text/plain\r\n"));
        assert!(req.headers.contains("content-length: 6\r\n"));
        if req.path == "/redirect" {
            response("307 Temporary Redirect", "Location: /final\r\n", b"")
        } else {
            assert_eq!(req.path, "/final");
            response(
                "200 OK",
                "Content-Type: Text/HTML; Charset=GBK\r\n",
                b"download",
            )
        }
    });
    let source = format!(
        r#"
        const file=new File(['upload'],'upload.txt',{{type:'TEXT/PLAIN'}});
        const result=await fetch('{url}/redirect',{{method:'POST',body:file}});
        const blob=await result.blob();
        if(blob.type !== 'text/html;charset=GBK' || await blob.text() !== 'download' || !result.bodyUsed || !result.redirected)throw Error('Network Blob');
        if(await file.text() !== 'upload')throw Error('Upload mutated File');
        console.log('blob network');
    "#
    );
    let output = common::run(&source, &[]);
    assert_eq!(output.stdout, b"blob network\n");
    thread.join().unwrap();
}

#[test]
fn multipart_uploads_are_snapshotted_replayable_and_parse_on_download() {
    let first = std::sync::Arc::new(std::sync::Mutex::new(None::<Vec<u8>>));
    let (url, thread) = server(2, move |req| {
        assert_eq!(req.method, "POST");
        let content_type = req
            .headers
            .lines()
            .find_map(|line| line.strip_prefix("content-type: "))
            .unwrap();
        assert!(content_type.starts_with("multipart/form-data; boundary=----avril"));
        assert!(
            req.headers
                .contains(&format!("content-length: {}\r\n", req.body.len()))
        );
        if req.path == "/start" {
            *first.lock().unwrap() = Some(req.body);
            response("307 Temporary Redirect", "Location: /finish\r\n", b"")
        } else {
            assert_eq!(req.path, "/finish");
            assert_eq!(first.lock().unwrap().as_ref().unwrap(), &req.body);
            response(
                "200 OK",
                &format!("Content-Type: {content_type}\r\n"),
                &req.body,
            )
        }
    });
    let source = format!(
        r#"
        const f=new FormData();f.append('field','value');f.append('file',new File([new Uint8Array([0,255,3])],'data.bin',{{type:'application/octet-stream'}}));
        const captured=new Request('{url}/start',{{method:'POST',body:f}});
        f.set('field','changed');
        const cleared=new Request(captured.clone(),{{headers:{{}}}});
        if(cleared.headers.get('content-type') !== null)throw Error('Inherited body re-added Content-Type');
        const copied=new Request(captured);
        const result=await fetch(copied), data=await result.formData();
        if(!result.redirected || data.get('field') !== 'value' || data.get('file').name !== 'data.bin' || (await data.get('file').bytes()).join() !== '0,255,3')throw Error('Network multipart');
        console.log('multipart network');
    "#
    );
    assert_eq!(common::run(&source, &[]).stdout, b"multipart network\n");
    thread.join().unwrap();
}

#[test]
fn collects_fetch_bodies_and_preserves_stream_state() {
    let (url, thread) = server(5, |req| match req.path.as_str() {
        "/empty" => response("200 OK", "", b""),
        "/text" => response(
            "200 OK",
            "",
            &["\u{feff}hello 😀".as_bytes(), &[0xff]].concat(),
        ),
        "/json" => response("200 OK", "", br#"{"answer":42}"#),
        _ => response(
            "200 OK",
            "",
            &(0..262_177).map(|i| (i % 251) as u8).collect::<Vec<_>>(),
        ),
    });
    let source = r#"
        const root='URL', controller=new AbortController();
        const response=await fetch(root,{signal:controller.signal}), body=response.body;
        const initialReader=body.getReader();
        try{await response.bytes();throw Error('Accepted locked body');}catch(e){if(!(e instanceof TypeError))throw e;}
        if(response.bodyUsed)throw Error('Lock disturbed body');initialReader.releaseLock();
        const pending=response.bytes();
        if(!response.bodyUsed||!body.locked)throw Error('Consumption state');
        try{response.clone();throw Error('Cloned pending body');}catch(e){if(!(e instanceof TypeError))throw e;}
        try{await response.text();throw Error('Consumed pending body');}catch(e){if(!(e instanceof TypeError))throw e;}
        const bytes=await pending;
        if(!(bytes instanceof Uint8Array)||bytes.length!==262177||body.locked)throw Error('Collection');
        for(let i=0;i<bytes.length;i++)if(bytes[i]!==i%251)throw Error('Corrupt bytes');
        controller.abort('after completion');
        const reader=body.getReader();await reader.closed;
        if(!(await reader.read()).done)throw Error('Stream not closed');reader.releaseLock();
        const byob=body.getReader({mode:'byob'});
        if(!(await byob.read(new Uint8Array(1))).done)throw Error('BYOB not closed');byob.releaseLock();
        const buffer=await(await fetch(root)).arrayBuffer();
        if(!(buffer instanceof ArrayBuffer)||buffer.byteLength!==262177)throw Error('ArrayBuffer');
        const view=new Uint8Array(buffer);for(let i=0;i<view.length;i++)if(view[i]!==i%251)throw Error('Corrupt buffer');
        const empty=await fetch(root+'/empty');if(!empty.body||(await empty.bytes()).length!==0)throw Error('Empty');
        const emptyReader=empty.body.getReader();await emptyReader.closed;
        if(!(await emptyReader.read()).done)throw Error('Empty stream not closed');emptyReader.releaseLock();
        const original=await fetch(root+'/text'), wrapped=new Response(original.body);
        if(await wrapped.text()!=='hello 😀�'||!original.bodyUsed||!wrapped.bodyUsed)throw Error('UTF8/wrapped body');
        if((await(await fetch(root+'/json')).json()).answer!==42)throw Error('JSON');
        console.log('collected');
    "#.replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"collected\n");
    thread.join().unwrap();
}

#[test]
fn collection_keeps_cloned_and_custom_streams_working() {
    let (url, thread) = server(1, |_| response("200 OK", "", &vec![123; 131_079]));
    let source = r#"
        const response=await fetch('URL'), copy=response.clone();
        const [a,b]=await Promise.all([response.bytes(),copy.bytes()]);
        for(const bytes of [a,b]){
            if(bytes.length!==131079)throw Error('Clone length');
            for(const byte of bytes)if(byte!==123)throw Error('Clone data');
        }
        if(!response.bodyUsed||!copy.bodyUsed)throw Error('Clone disturbance');
        a[0]=0;if(b[0]!==123)throw Error('Clone shared storage');
        const stream=new ReadableStream({start(c){c.enqueue(new Uint8Array([1,2]));c.enqueue(new Uint8Array([3]));c.close();}});
        if([...await new Response(stream).bytes()].join(',')!=='1,2,3')throw Error('Custom stream');
        const invalid=new Response(new ReadableStream({start(c){c.enqueue('not bytes');c.close();}}));
        try{await invalid.bytes();throw Error('Accepted nonbytes');}catch(e){if(!(e instanceof TypeError))throw e;}
        const reason={custom:true}, failed=new Response(new ReadableStream({start(c){c.error(reason);}}));
        try{await failed.text();throw Error('Accepted failed stream');}catch(e){if(e!==reason)throw e;}
        console.log('fallback');
    "#.replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"fallback\n");
    thread.join().unwrap();
}

#[test]
fn aborting_collection_preserves_reasons_and_releases_locks() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let thread = std::thread::spawn(move || {
        for _ in 0..3 {
            let (mut socket, _) = listener.accept().unwrap();
            request(&mut socket);
            socket
                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\nhello")
                .unwrap();
            socket
                .set_read_timeout(Some(Duration::from_secs(5)))
                .unwrap();
            let mut byte = [0];
            let _ = socket.read(&mut byte);
        }
    });
    let source = r#"
        for(const method of ['bytes','text']){
            const controller=new AbortController(), reason={method};
            const response=await fetch('URL',{signal:controller.signal});
            const pending=response[method]();
            if(!response.bodyUsed||!response.body.locked)throw Error('Pending state');
            setTimeout(()=>controller.abort(reason),0);
            try{await pending;throw Error('Expected abort');}catch(e){if(e!==reason)throw e;}
            if(response.body.locked||!response.bodyUsed)throw Error('Aborted state');
            const reader=response.body.getReader();
            if(await reader.closed.catch(e=>e)!==reason)throw Error('Closed reason');
            try{await reader.read();throw Error('Expected stream error');}catch(e){if(e!==reason)throw e;}
            reader.releaseLock();
        }
        const controller=new AbortController(), reason={before:true};
        const response=await fetch('URL',{signal:controller.signal});controller.abort(reason);
        try{await response.bytes();throw Error('Accepted aborted body');}catch(e){if(e!==reason)throw e;}
        if(response.body.locked||!response.bodyUsed)throw Error('Pre-aborted state');
        console.log('collection aborted');
    "#.replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"collection aborted\n");
    thread.join().unwrap();
}

#[test]
fn collection_rejects_truncated_responses_and_errors_the_stream() {
    let (url, thread) = server(1, |_| {
        b"HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 100000\r\n\r\nshort".to_vec()
    });
    let source = r#"
        const response=await fetch('URL');let failure;
        try{await response.bytes();throw Error('Accepted truncated body');}catch(e){if(!(e instanceof TypeError))throw e;failure=e;}
        if(!response.bodyUsed||response.body.locked)throw Error('Failed state');
        const reader=response.body.getReader();
        if(await reader.closed.catch(e=>e)!==failure)throw Error('Closed error');
        try{await reader.read();throw Error('Accepted errored stream');}catch(e){if(e!==failure)throw e;}
        reader.releaseLock();
        console.log('truncated');
    "#.replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"truncated\n");
    thread.join().unwrap();
}

#[test]
fn body_text_decodes_ascii_unicode_boms_and_malformed_utf8() {
    let (url, thread) = server(12, |req| {
        response(
            "200 OK",
            "Content-Type: text/plain; charset=utf-16\r\n",
            &req.body,
        )
    });
    let source = r#"
        const root='URL';
        const cases=[
            [[], ''],
            [[0,65,127], '\0A\x7f'],
            [[239,187,191], ''],
            [[239,187,191,239,187,191,65], '\ufeffA'],
            [[195,169,226,130,172,240,159,152,128], 'é€😀'],
            [[237,160,128], '���'],
            [[240,128,128,65,255], '���A�'],
            [[226,130], '�'],
            [[255,254,65,0], '��A\0'],
            [[239,187], '�'],
        ];
        const ascii=Array.from({length:128},(_,i)=>i);cases.push([ascii,String.fromCharCode(...ascii)]);
        for(const [bytes,expected] of cases){
            const input=new Uint8Array(bytes);
            const native=await fetch(root,{method:'POST',body:input});
            if(await native.text()!==expected||!native.bodyUsed||native.body.locked)throw Error('Native text');
            if(await new Response(input).text()!==expected)throw Error('Memory text');
            let i=0;const streamed=new Response(new ReadableStream({type:'bytes',pull(c){
                if(i===bytes.length)c.close();else c.enqueue(new Uint8Array([bytes[i++]]));
            }},{highWaterMark:0}));
            const copy=streamed.clone();
            const texts=await Promise.all([streamed.text(),copy.text()]);
            if(texts.some(text=>text!==expected))throw Error('Split/cloned text');
        }
        const jsonBytes=new TextEncoder().encode('\ufeff{"word":"é😀"}');
        const native=await fetch(root,{method:'POST',body:jsonBytes});
        if((await native.json()).word!=='é😀'||(await new Response(jsonBytes).json()).word!=='é😀')throw Error('JSON text');
        const invalid=new Response('{');
        try{await invalid.json();throw Error('Accepted invalid JSON');}catch(e){if(!(e instanceof SyntaxError))throw e;}
        if(!invalid.bodyUsed||invalid.body.locked)throw Error('Invalid JSON state');
        console.log('body text');
    "#.replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"body text\n");
    thread.join().unwrap();
}

#[test]
fn native_collection_converts_blob_and_form_data_before_resolving_private_bytes() {
    let (url, thread) = server(2, |req| {
        if req.path == "/blob" {
            response(
                "200 OK",
                "Content-Type: Text/Plain; Charset=UTF-8\r\n",
                b"immutable",
            )
        } else {
            response("200 OK","Content-Type: multipart/form-data; boundary=B\r\n",b"--B\r\nContent-Disposition: form-data; name=\"file\"; filename=\"data\"\r\n\r\nimmutable\r\n--B--\r\n")
        }
    });
    let source = format!(
        r#"
        const retained=[], U8=Uint8Array, resolutions=new Map();
        Object.defineProperty(Object.prototype,'then',{{configurable:true,get(){{
            if(this instanceof U8)retained.push(this);
            if(this instanceof Blob)resolutions.set(this,(resolutions.get(this)||0)+1);
            const d=Object.getOwnPropertyDescriptor(this,'value');
            if(d && d.value instanceof U8)retained.push(d.value);
            return undefined;
        }}}});
        const response=await fetch('{url}/blob'), blob=await response.blob();
        if(blob.type !== 'text/plain;charset=UTF-8')throw Error('Native MIME conversion');
        if(resolutions.get(blob)!==1)throw Error('Blob promise resolved more than once');
        for(const bytes of retained)if(bytes.byteLength)bytes.fill(0);
        if(await blob.text() !== 'immutable' || !response.bodyUsed || response.body.locked)throw Error('Native Blob aliases a public chunk');
        const form=await(await fetch('{url}/form')).formData(), file=form.get('file');
        for(const bytes of retained)if(bytes.byteLength)bytes.fill(0);
        if(file.name !== 'data' || await file.text() !== 'immutable')throw Error('Native FormData conversion');
        console.log('native conversion');
    "#
    );
    assert_eq!(common::run(&source, &[]).stdout, b"native conversion\n");
    thread.join().unwrap();
}

#[test]
fn fetch_init_dictionaries_reject_primitives_and_preserve_inherited_members() {
    let out = common::run(
        r#"
        for(const init of [2,'bad',true,1n,Symbol()]){
            for(const call of [()=>new Request('https://example.com',init),()=>new Response(null,init),()=>Response.json({},init)]){
                try{call();throw Error('Accepted primitive dictionary');}catch(e){if(!(e instanceof TypeError))throw e;}
            }
        }
        const response=Response.json({ok:true},Object.create({status:201,statusText:'Created'}));
        if(response.status!==201||response.statusText!=='Created')throw Error('Inherited ResponseInit lost');
        console.log('fetch dictionaries');
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"fetch dictionaries\n");
}

#[test]
fn redirect_releases_upload_writes_stalled_on_an_unread_request_body_despite_failing_cleanup() {
    let (target, target_thread) = server(1, |_| response("200 OK", "", b"redirected"));
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let source = format!("http://{}", listener.local_addr().unwrap());
    let (done, finished) = std::sync::mpsc::channel::<()>();
    let source_thread = std::thread::spawn(move || {
        let (mut socket, _) = listener.accept().unwrap();
        let mut reader = BufReader::new(socket.try_clone().unwrap());
        let mut line = String::new();
        while line != "\r\n" {
            line.clear();
            reader.read_line(&mut line).unwrap();
        }
        // Let the upload fill the socket and channel before answering.
        std::thread::sleep(Duration::from_millis(300));
        socket
            .write_all(&response(
                "303 See Other",
                &format!("Location: {target}\r\n"),
                b"",
            ))
            .unwrap();
        // Hold the connection open without reading the body, so the upload stalls.
        let _ = finished.recv_timeout(Duration::from_secs(5));
    });
    let script = format!(
        r#"
        let writes=0;const body=new ReadableStream({{pull(c){{writes++;c.enqueue(new Uint8Array(1<<20));}},cancel(){{throw Error('cleanup failed');}}}},{{highWaterMark:0}});
        const response=await fetch('{source}',{{method:'POST',body,duplex:'half'}});
        console.log(await response.text(), writes > 1);
    "#
    );
    assert_eq!(common::run(&script, &[]).stdout, b"redirected true\n");
    done.send(()).unwrap();
    source_thread.join().unwrap();
    target_thread.join().unwrap();
}

#[test]
fn upload_failures_keep_native_buffer_checks_and_reentrant_abort_precedence() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("http://{}", listener.local_addr().unwrap());
    let output = common::run(
        &format!(
            r#"
        for (const [chunk, expected] of [
            [new Uint8Array(new SharedArrayBuffer(1)), 'buffer'],
            [new Uint8Array(new ArrayBuffer(1, {{maxByteLength: 4}})), 'resizable']
        ]) {{
            const body = new ReadableStream({{start(c) {{c.enqueue(chunk); c.close();}}}});
            try {{
                await fetch('{url}', {{method: 'POST', body, duplex: 'half'}});
                throw Error('Accepted unsupported upload buffer');
            }} catch (error) {{
                if (!(error instanceof TypeError) || !error.message.toLowerCase().includes(expected)) throw error;
            }}
            if (body.locked) throw Error('Failed upload retained its reader');
        }}
        const aborter = new AbortController(), reason = {{}}, order = [];
        const error = {{
            get message() {{
                order.push('message');
                aborter.abort(reason);
                return {{toString() {{order.push('string'); return 'upload failed';}}}};
            }}
        }};
        const body = new ReadableStream({{start(c) {{c.error(error);}}}});
        try {{
            await fetch('{url}', {{method: 'POST', body, duplex: 'half', signal: aborter.signal}});
            throw Error('Accepted errored upload');
        }} catch (error) {{
            if (error !== reason) throw Error('Lost reentrant abort reason');
        }}
        if (order.join('|') !== 'message|string' || body.locked) throw Error('Error conversion or reader release');
        console.log('upload boundary and abort ordering');
        "#
        ),
        &[],
    );
    assert_eq!(output.stdout, b"upload boundary and abort ordering\n");
}