mod common;

use std::io::{Read, Write};
use std::net::TcpListener;
use std::time::{Duration, Instant};

// The test server is shared with no other test target, so it is declared here.
#[path = "common/websocket.rs"]
mod ws_server;

use ws_server::*;

#[test]
fn reflected_construction_live_handlers_and_close_event_conversion() {
    let (url, thread) = server(1, |session| session.echo());
    let source = r#"
        const order = [], poison = function() {throw Error('guest base constructor');};
        class ChildSocket extends WebSocket {}
        class ChildClose extends CloseEvent {}
        Object.setPrototypeOf(WebSocket, poison);
        Object.setPrototypeOf(CloseEvent, poison);
        if (WebSocket.length !== 1 || CloseEvent.length !== 1 ||
            WebSocket.prototype.send.length !== 1 || WebSocket.prototype.close.length !== 0) throw Error('arity');
        const close = new ChildClose({toString() {order.push('type'); return 'close';}}, {
            get code() {order.push('code'); return {valueOf() {order.push('code:number'); return 65537;}};},
            get reason() {order.push('reason'); return {toString() {order.push('reason:string'); return '\ud800';}};},
            get wasClean() {order.push('clean'); return true;}
        });
        if (order.join('|') !== 'type|code|reason|clean|code:number|reason:string' ||
            close.code !== 1 || close.reason !== '\ufffd' || !close.wasClean ||
            Object.getPrototypeOf(close) !== ChildClose.prototype) throw Error('CloseEvent conversion');
        if (Object.getOwnPropertyNames(close).join() !== 'isTrusted') throw Error('emitted event fields');
        const socket = new ChildSocket('URL');
        if (Object.getPrototypeOf(socket) !== ChildSocket.prototype || Object.getOwnPropertyNames(socket).length) throw Error('socket shape');
        const calls = [];
        socket.onmessage = function(event) {
            if (this !== socket) throw Error('handler receiver');
            calls.push('original'); return false;
        };
        socket.addEventListener('message', () => calls.push('listener'));
        const first = new MessageEvent('message', {cancelable:true});
        if (socket.dispatchEvent(first) || !first.defaultPrevented) throw Error('handler cancellation');
        socket.onmessage = function() {calls.push('replacement');};
        socket.dispatchEvent(new MessageEvent('message'));
        if (calls.join('|') !== 'original|listener|replacement|listener') throw Error('live handler order');
        socket.onmessage = null;
        const completed = new Promise((resolve, reject) => {
            socket.onopen = () => socket.send('echo');
            socket.onmessage = event => {
                if (event.data !== 'echo') reject(Error('echo'));
                socket.close(1000, 'done');
            };
            socket.onerror = () => reject(Error('socket error'));
            socket.onclose = event => {
                if (!event.wasClean || event.code !== 1000 || event.reason !== 'done') reject(Error('close'));
                else resolve();
            };
        });
        await completed;
        console.log('reflected sockets and events');
    "#
    .replace("URL", &url);
    let output = common::run(&source, &[]);
    assert_eq!(output.stdout, b"reflected sockets and events\n");
    assert!(
        output.stderr.is_empty(),
        "{}",
        String::from_utf8_lossy(&output.stderr)
    );
    thread.join().unwrap();
}

#[test]
fn handshakes_capture_the_isolate_user_agent_at_construction() {
    let (url, thread) = server(5, |mut session| {
        let ua = session
            .headers
            .lines()
            .find_map(|line| line.strip_prefix("user-agent:"))
            .expect("Missing User-Agent")
            .trim()
            .to_owned();
        session.accept(None)?;
        session.write(TEXT, ua.as_bytes())?;
        let close = session.read()?;
        assert_eq!(close.opcode, CLOSE);
        session.write(CLOSE, &close.payload)
    });
    let source = r#"
        const original = navigator.userAgent;
        function handshake() {
            return new Promise((resolve, reject) => {
                const socket = new WebSocket('URL');
                let received;
                socket.onmessage = event => {received = event.data; socket.close();};
                socket.onerror = () => reject(Error('Handshake failed'));
                socket.onclose = event => {
                    if (!event.wasClean) reject(Error('Unclean close'));
                    else resolve(received);
                };
            });
        }
        if (await handshake() !== original) throw Error('Initial default');
        navigator.userAgent = 'custom/1.0';
        const pending = handshake();
        navigator.userAgent = 'changed/2.0';
        if (await pending !== 'custom/1.0') throw Error('In-flight UA changed');
        if (await handshake() !== 'changed/2.0') throw Error('Default not updated');
        navigator.userAgent = '';
        if (await handshake() !== '') throw Error('Empty UA lost');
        navigator.userAgent = 'hardened/1.0';
        Object.defineProperty(navigator, 'userAgent', {get() {throw Error('Guest getter');}});
        if (await handshake() !== 'hardened/1.0') throw Error('Internal default lost');
        console.log('WebSocket UA');
    "#
    .replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"WebSocket UA\n");
    thread.join().unwrap();
}

#[test]
fn invalid_user_agents_fail_before_opening_a_websocket_transport() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("ws://{}", listener.local_addr().unwrap());
    let source = r#"
        for (const invalid of ['bad\r\nInjected: yes', 'bad\0value', '\u0100']) {
            navigator.userAgent = invalid;
            await new Promise((resolve, reject) => {
                const socket = new WebSocket('URL');
                let errored = false;
                socket.onopen = () => {socket.close(); reject(Error('Invalid UA connected'));};
                socket.onerror = () => {errored = true;};
                socket.onclose = event => {
                    if (!errored || event.wasClean || event.code !== 1006) reject(Error('Failure events'));
                    else resolve();
                };
            });
        }
        console.log('invalid WebSocket UA');
    "#
    .replace("URL", &url);
    assert_eq!(common::run(&source, &[]).stdout, b"invalid WebSocket UA\n");
    listener.set_nonblocking(true).unwrap();
    assert_eq!(
        listener.accept().unwrap_err().kind(),
        std::io::ErrorKind::WouldBlock
    );
}

#[test]
fn echo_text_binary_and_binary_types() {
    let (url, thread) = server(1, |session| session.echo());
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.binaryType = 'arraybuffer';
            const log = [];
            ws.onopen = () => {
                if (ws.url !== 'WSURL/' || ws.readyState !== 1) throw Error('Attributes');
                ws.send('hello');
                ws.send(new Uint8Array([1, 2, 255]));
            };
            ws.onmessage = (event) => {
                if (typeof event.data === 'string') {
                    if (event.data !== 'hello') throw Error('Text');
                    log.push('text');
                } else if (event.data instanceof ArrayBuffer) {
                    if ([...new Uint8Array(event.data)].join(',') !== '1,2,255') throw Error('Binary');
                    log.push('binary');
                    ws.binaryType = 'blob';
                    ws.send(new Blob([new Uint8Array([9, 8]), new Uint8Array([7])], { type: 'application/x-test' }));
                } else {
                    if (!(event.data instanceof Blob) || event.data.size !== 3 || event.data.type !== '') {
                        throw Error('Blob message');
                    }
                    log.push('blob');
                    ws.close(1000, 'done');
                }
            };
            ws.onerror = () => log.push('error');
            ws.onclose = (event) => {
                if (log.join('|') !== 'text|binary|blob') throw Error('Order');
                if (!event.wasClean || event.code !== 1000 || event.reason !== 'done') throw Error('Close');
                console.log('echo ok', event instanceof CloseEvent);
            };
        "#)
        .replace("WSURL", &url),
        &[],
    );
    assert_eq!(out.stdout, b"echo ok true\n");
    thread.join().unwrap();
}

#[test]
fn http_urls_map_to_websocket_scheme() {
    let (url, thread) = server(1, |session| session.echo());
    let http = url.replacen("ws://", "http://", 1);
    let out = common::run(
        &format!(
            r#"
            const ws = new WebSocket('{http}');
            ws.onopen = () => {{
                if (ws.url !== '{url}/') throw Error('Mapped URL');
                console.log('mapped');
                ws.close();
            }};
            ws.onclose = (event) => {{
                if (!event.wasClean || event.code !== 1005) throw Error('Close code');
                console.log('closed');
            }};
        "#
        ),
        &[],
    );
    assert_eq!(out.stdout, b"mapped\nclosed\n");
    thread.join().unwrap();
}

#[test]
fn subprotocol_negotiation() {
    let (url, thread) = server(3, |mut session| {
        match session.path.as_str() {
            "/chosen" => {
                assert_eq!(session.protocol, "chat, v2");
                session.accept(Some("v2"))?;
                let close = session.read()?;
                assert_eq!(close.opcode, CLOSE);
                session.write(CLOSE, &close.payload)?;
            }
            "/unacknowledged" => {
                session.accept(None)?;
                assert_eq!(
                    session.read().err().unwrap().kind(),
                    std::io::ErrorKind::UnexpectedEof
                );
            }
            _ => {
                session.accept(Some("bogus"))?;
                assert_eq!(
                    session.read().err().unwrap().kind(),
                    std::io::ErrorKind::UnexpectedEof
                );
            }
        }
        Ok(())
    });
    let out = common::run(
        &format!(
            r#"
            const log = [];
            await new Promise((resolve) => {{
                const ws = new WebSocket('{url}/chosen', ['chat', 'v2']);
                ws.onopen = () => {{
                    if (ws.protocol !== 'v2') throw Error('Selected protocol');
                    log.push('chosen');
                    ws.close(1000);
                }};
                ws.onclose = () => resolve();
            }});
            await new Promise((resolve) => {{
                const ws = new WebSocket('{url}/unacknowledged', ['chat']);
                ws.onerror = () => log.push('error');
                ws.onclose = (event) => {{
                    if (event.wasClean || event.code !== 1006) throw Error('Failed close');
                    if (ws.readyState !== 3) throw Error('Ready state');
                    log.push('close');
                    resolve();
                }};
            }});
            await new Promise((resolve) => {{
                const ws = new WebSocket('{url}/bogus', ['chat']);
                ws.onclose = (event) => {{
                    if (event.wasClean || event.code !== 1006) throw Error('Unexpected protocol');
                    log.push('close');
                    resolve();
                }};
            }});
            console.log(log.join('|'));
        "#
        ),
        &[],
    );
    assert_eq!(out.stdout, b"chosen|error|close|close\n");
    thread.join().unwrap();
}

#[test]
fn client_close_codes_and_reasons() {
    let (url, thread) = server(3, |mut session| {
        session.accept(None)?;
        let close = session.read()?;
        assert_eq!(close.opcode, CLOSE);
        match session.path.as_str() {
            "/code" => {
                assert_eq!(close.code(), 1000);
                assert_eq!(close.reason(), "bye");
            }
            "/bare" => assert!(close.payload.is_empty()),
            _ => {
                assert_eq!(close.code(), 3001);
                assert_eq!(close.reason(), "later");
            }
        }
        session.write(CLOSE, &close.payload)?;
        Ok(())
    });
    let out = common::run(
        &format!(
            r#"
            const events = [];
            await new Promise((resolve) => {{
                const ws = new WebSocket('{url}/code');
                ws.onopen = () => ws.close(1000, 'bye');
                ws.onclose = (event) => {{
                    if (!event.wasClean || event.code !== 1000 || event.reason !== 'bye') throw Error('Echoed close');
                    if (ws.readyState !== 3 || ws.bufferedAmount !== 0) throw Error('State');
                    events.push(1000);
                    resolve();
                }};
            }});
            await new Promise((resolve) => {{
                const ws = new WebSocket('{url}/bare');
                ws.onopen = () => ws.close();
                ws.onclose = (event) => {{
                    if (!event.wasClean || event.code !== 1005 || event.reason !== '') throw Error('Bare close');
                    events.push(1005);
                    resolve();
                }};
            }});
            await new Promise((resolve) => {{
                const ws = new WebSocket('{url}/later');
                ws.onopen = () => ws.close(3001, 'later');
                ws.onclose = (event) => {{
                    if (event.code !== 3001 || event.reason !== 'later') throw Error('Later close');
                    events.push(3001);
                    resolve();
                }};
            }});
            console.log(events.join(','));
        "#
        ),
        &[],
    );
    assert_eq!(out.stdout, b"1000,1005,3001\n");
    thread.join().unwrap();
}

#[test]
fn server_initiated_close() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        session.write(CLOSE, &Frame::close(3001, "done").payload)?;
        let close = session.read()?;
        assert_eq!(close.code(), 3001);
        assert_eq!(close.reason(), "done");
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.onopen = () => {
                if (ws.readyState !== 1) throw Error('Ready state');
            };
            ws.onclose = (event) => {
                if (!event.wasClean || event.code !== 3001 || event.reason !== 'done') throw Error('Server close');
                if (ws.readyState !== 3) throw Error('Ready state');
                console.log('server closed');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"server closed\n");
    thread.join().unwrap();
}

#[test]
fn handshake_failure_fires_error_then_close() {
    let (url, thread) = server(1, |mut session| session.reject());
    let out = common::run(
        &connect(
            &url,
            r#"
            const order = [];
            ws.onerror = (event) => order.push('error');
            ws.onclose = (event) => {
                if (order.join('|') !== 'error') throw Error('Error before close');
                if (event.wasClean || event.code !== 1006) throw Error('Failed close');
                if (ws.readyState !== 3) throw Error('Ready state');
                console.log('handshake failed');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"handshake failed\n");
    thread.join().unwrap();
}

#[test]
fn refused_connection_fails() {
    let out = common::run(
        r#"
        let errors = 0;
        await new Promise((resolve) => {
            const ws = new WebSocket('ws://127.0.0.1:1/');
            ws.onerror = () => errors++;
            ws.onclose = (event) => {
                if (errors !== 1 || event.wasClean || event.code !== 1006) throw Error('Refused close');
                console.log('refused');
                resolve();
            };
        });
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"refused\n");
}

#[test]
fn permission_denied() {
    let out = common::run(
        r#"
        import { pledge } from 'avril:std';
        pledge({ net: false });
        try {
            new WebSocket('ws://127.0.0.1:1/');
            throw Error('Permission ignored');
        } catch (error) {
            if (!error.message.includes('PermissionDenied')) throw error;
        }
        console.log('denied');
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"denied\n");
}

#[test]
fn constructor_validation() {
    let out = common::run(
        r#"
        const syntax = (operation) => {
            try {
                operation();
                throw Error('Accepted invalid URL');
            } catch (error) {
                if (!(error instanceof DOMException) || error.name !== 'SyntaxError') throw error;
            }
        };
        syntax(() => new WebSocket('ftp://example.com'));
        syntax(() => new WebSocket('/relative'));
        syntax(() => new WebSocket('ws://example.com/#fragment'));
        syntax(() => new WebSocket('ws://example.com/#'));
        syntax(() => new WebSocket('ws://example.com', 'chat\n'));
        syntax(() => new WebSocket('ws://example.com', 'a b'));
        syntax(() => new WebSocket('ws://example.com', ['a', 'a']));
        syntax(() => new WebSocket('ws://example.com', ['a,']));
        syntax(() => new WebSocket('ws://example.com', {}));
        const numeric = new WebSocket('ws://127.0.0.1:1/x', 5);
        if (numeric.protocol !== '') throw Error('Numeric protocols');
        numeric.close();
        numeric.onmessage = () => false;
        const canceled = new MessageEvent('message', { cancelable: true });
        if (numeric.dispatchEvent(canceled) || !canceled.defaultPrevented) {
            throw Error('Event handler return value');
        }
        const invalid = (operation, name) => {
            try {
                operation();
                throw Error('Accepted invalid argument');
            } catch (error) {
                if (error.name !== name) throw error;
            }
        };
        const ws = new WebSocket('ws://127.0.0.1:1/x');
        for (const code of [0, 999, 1001, 2999, 5000]) {
            invalid(() => ws.close(code), 'InvalidAccessError');
        }
        invalid(() => ws.close(1000, 'x'.repeat(124)), 'SyntaxError');
        invalid(() => ws.send('too early'), 'InvalidStateError');
        invalid(() => ws.send(), 'TypeError');
        invalid(() => ws.send(Symbol()), 'TypeError');
        invalid(() => new WebSocket(Symbol()), 'TypeError');
        const marker = {};
        try {
            new WebSocket({ toString() { throw marker; } });
            throw Error('Swallowed coercion failure');
        } catch (error) {
            if (error !== marker) throw error;
        }
        try {
            new WebSocket('invalid URL', [{ toString() { throw marker; } }]);
            throw Error('Protocol conversion skipped');
        } catch (error) {
            if (error !== marker) throw error;
        }
        invalid(() => ws.close(0, Symbol()), 'TypeError');
        invalid(() => ws.send(new ArrayBuffer(1, { maxByteLength: 2 })), 'TypeError');
        ws.binaryType = 'nodebuffer';
        if (ws.binaryType !== 'blob') throw Error('binaryType setter');
        ws.binaryType = 'arraybuffer';
        if (ws.binaryType !== 'arraybuffer') throw Error('binaryType setter');
        if (new CloseEvent('close', { code: -1 }).code !== 65535) throw Error('CloseEvent code');
        if (new CloseEvent('close', { code: 65537 }).code !== 1) throw Error('CloseEvent code wrap');
        if (ws.url !== 'ws://127.0.0.1:1/x' || ws.readyState !== 0 || ws.bufferedAmount !== 0) {
            throw Error('Initial attributes');
        }
        if (ws.extensions !== '' || ws.protocol !== '') throw Error('Initial protocol');
        if (WebSocket.CONNECTING !== 0 || ws.OPEN !== 1 || WebSocket.CLOSING !== 2 || ws.CLOSED !== 3) {
            throw Error('Ready state constants');
        }
        if (ws.onopen !== null || typeof ws.addEventListener !== 'function') throw Error('Event handlers');
        ws.close();
        console.log('validated', ws.readyState);
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"validated 2\n");
}

#[test]
fn constructor_checks_url_before_validating_protocols() {
    let out = common::run(
        r#"
        for (const [url, message] of [
            ['invalid URL', 'URL is invalid'],
            ['ftp://x', 'scheme'],
            ['ws://x/#', 'fragment'],
        ]) {
            for (const protocols of ['a b', ['chat', 'chat']]) {
                try {
                    new WebSocket(url, protocols);
                    throw Error('Accepted invalid constructor arguments');
                } catch (error) {
                    if (error.name !== 'SyntaxError' || !error.message.includes(message)) throw error;
                }
            }
        }
        const marker = {};
        try {
            new WebSocket('ftp://x', {
                get [Symbol.iterator]() { throw marker; }
            });
            throw Error('Protocol conversion skipped');
        } catch (error) {
            if (error !== marker) throw error;
        }
        console.log('constructor order checked');
    "#,
        &[],
    );
    assert_eq!(out.stdout, b"constructor order checked\n");
}

#[test]
fn buffer_sources_send_detached_as_empty_and_reject_shared() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        for expected in [vec![], vec![], vec![], vec![1, 2]] {
            let frame = session.read()?;
            assert_eq!(frame.opcode, BINARY);
            assert_eq!(frame.payload, expected);
        }
        let close = session.read()?;
        assert_eq!(close.opcode, CLOSE);
        session.write(CLOSE, &close.payload)?;
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.onopen = () => {
                const shared = new SharedArrayBuffer(3);
                const resizable = new ArrayBuffer(3, { maxByteLength: 4 });
                for (const data of [
                    shared, new Uint8Array(shared), new DataView(shared),
                    resizable, new Uint8Array(resizable), new DataView(resizable),
                ]) {
                    try {
                        ws.send(data);
                        throw Error('Accepted shared or resizable data');
                    } catch (error) {
                        if (!(error instanceof TypeError)) throw error;
                    }
                }
                const detached = new ArrayBuffer(3);
                const typed = new Uint8Array(detached);
                const view = new DataView(detached);
                detached.transfer();
                ws.send(detached);
                ws.send(typed);
                ws.send(view);
                ws.send(new Uint8Array([0, 1, 2, 3]).subarray(1, 3));
                if (ws.bufferedAmount !== 2) throw Error('Buffered amount');
                ws.close(1000);
            };
            ws.onclose = event => {
                if (!event.wasClean || event.code !== 1000) throw Error('Close');
                console.log('buffer sources checked');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"buffer sources checked\n");
    thread.join().unwrap();
}

#[test]
fn send_converts_other_arguments_to_text() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        for expected in [
            &b"5"[..],
            &b"[object Object]"[..],
            &b"1,2"[..],
            &b"null"[..],
        ] {
            let frame = session.read()?;
            assert_eq!(frame.opcode, TEXT, "must be sent as a text frame");
            assert_eq!(frame.payload, expected);
        }
        let frame = session.read()?;
        assert_eq!(frame.opcode, BINARY);
        assert_eq!(frame.payload, [7]);
        let close = session.read()?;
        session.write(CLOSE, &close.payload)?;
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.onopen = () => {
                ws.send(5);
                ws.send({});
                ws.send([1, 2]);
                ws.send(null);
                ws.send(new Uint8Array([7]));
                ws.close(1000);
            };
            ws.onclose = (event) => {
                if (!event.wasClean || event.code !== 1000) throw Error('Close');
                console.log('converted');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"converted\n");
    thread.join().unwrap();
}

#[test]
fn fragmentation_and_ping() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        write_frame(&mut session.stream, false, TEXT, b"hel")?;
        write_frame(&mut session.stream, true, CONTINUATION, b"lo")?;
        session.write(PING, b"keepalive")?;
        let deadline = Instant::now() + Duration::from_secs(5);
        let mut pong = None;
        let mut reply = None;
        while (pong.is_none() || reply.is_none()) && Instant::now() < deadline {
            let frame = match session.read() {
                Ok(frame) => frame,
                Err(_) => break,
            };
            if frame.opcode == PONG {
                pong = Some(frame.payload);
            } else if frame.opcode == TEXT {
                reply = Some(frame.payload);
            }
        }
        assert_eq!(pong.as_deref(), Some(&b"keepalive"[..]));
        // The fragmented message must arrive as a single text message.
        assert_eq!(reply.as_deref(), Some(&b"reply"[..]));
        session.write(CLOSE, &Frame::close(1000, "").payload)?;
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.onmessage = (event) => {
                if (event.data !== 'hello') throw Error('Fragmented message');
                console.log('fragmented', event.origin);
                ws.send('reply');
            };
        "#,
        ),
        &[],
    );
    let expected = format!("fragmented {}\n", url);
    assert_eq!(out.stdout, expected.as_bytes());
    thread.join().unwrap();
}

#[test]
fn invalid_utf8_text_fails() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        // A text frame carrying invalid UTF-8.
        session.stream.write_all(&[0x81, 0x02, 0xff, 0xfe])?;
        let close = session.read()?;
        assert_eq!(close.opcode, CLOSE);
        assert_eq!(close.code(), 1007);
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            const order = [];
            ws.onmessage = () => order.push('message');
            ws.onerror = (event) => {
                if (!event.message.includes('UTF-8')) throw Error('Missing failure reason');
                order.push('error');
            };
            ws.onclose = (event) => {
                if (order.join('|') !== 'error') throw Error('Error before close');
                if (event.wasClean || event.code !== 1006) throw Error('Failed close');
                console.log('utf8 failed');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"utf8 failed\n");
    thread.join().unwrap();
}

#[test]
fn buffered_amount_only_grows_after_close() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        session.write(CLOSE, &Frame::close(1000, "").payload)?;
        assert_eq!(session.read()?.opcode, CLOSE);
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.onclose = () => {
                ws.send('a');
                ws.send('bb');
                ws.send(new Uint8Array(3));
                if (ws.bufferedAmount !== 6) throw Error('Buffered amount');
                console.log('buffered', ws.bufferedAmount);
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"buffered 6\n");
    thread.join().unwrap();
}

#[test]
fn socket_keeps_process_alive_until_closed() {
    let (close, closing) = std::sync::mpsc::channel();
    let (url, thread) = server(1, move |mut session| {
        session.accept(None)?;
        closing.recv_timeout(Duration::from_secs(5)).unwrap();
        session.write(CLOSE, &Frame::close(1000, "").payload)?;
        assert_eq!(session.read()?.opcode, CLOSE);
        Ok(())
    });
    let (_dir, mut child) = common::spawn(&connect(
        &url,
        r#"
        ws.onopen = () => console.log('open');
        ws.onclose = () => console.log('closed');
    "#,
    ));
    let stdout = child.stdout.take().unwrap();
    let (lines, received) = std::sync::mpsc::channel();
    let output = std::thread::spawn(move || {
        use std::io::BufRead;
        for line in std::io::BufReader::new(stdout).lines() {
            lines.send(line.unwrap()).unwrap();
        }
    });
    assert_eq!(
        received.recv_timeout(Duration::from_secs(5)).unwrap(),
        "open"
    );
    std::thread::sleep(Duration::from_millis(100));
    assert!(
        child.try_wait().unwrap().is_none(),
        "an open socket must keep the process alive"
    );
    close.send(()).unwrap();
    let out = common::wait(child);
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert_eq!(
        received.recv_timeout(Duration::from_secs(5)).unwrap(),
        "closed"
    );
    output.join().unwrap();
    thread.join().unwrap();
}

#[test]
fn ipv6_loopback() {
    let Ok((url, thread)) =
        TcpListener::bind("[::1]:0").map(|_| server_on("::1", 1, |session| session.echo()))
    else {
        // IPv6 loopback is unavailable; nothing to test.
        return;
    };
    let out = common::run(
        &connect(
            &url,
            r#"
            ws.onopen = () => {
                if (ws.url !== 'WSURL/') throw Error('URL');
                console.log('open', ws.readyState);
                ws.send('hello');
            };
            ws.onmessage = (event) => {
                if (event.data !== 'hello') throw Error('Echo');
                if (event.origin !== 'WSURL') throw Error('Origin');
                ws.close(1000);
            };
            ws.onclose = (event) => {
                if (!event.wasClean || event.code !== 1000) throw Error('Close');
                console.log('closed');
            };
        "#,
        )
        .replace("WSURL", &url),
        &[],
    );
    assert_eq!(out.stdout, b"open 1\nclosed\n");
    thread.join().unwrap();
}

#[test]
fn tls_handshake_against_plain_server_fails() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("wss://{}", listener.local_addr().unwrap());
    let thread = std::thread::spawn(move || {
        for _ in 0..1 {
            let mut stream = incoming(&listener);
            // Accept the TCP connection but never speak TLS; the client must fail.
            let mut buffer = [0; 64];
            let _ = stream.read(&mut buffer);
        }
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            const order = [];
            ws.onerror = () => order.push('error');
            ws.onclose = (event) => {
                if (order.join('|') !== 'error') throw Error('Error before close');
                if (event.wasClean || event.code !== 1006) throw Error('Failed close');
                console.log('tls failed');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"tls failed\n");
    thread.join().unwrap();
}

#[test]
fn close_during_connecting_fails_the_connection() {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let url = format!("ws://{}", listener.local_addr().unwrap());
    let out = common::run(
        &connect(
            &url,
            r#"
            const order = [];
            ws.close(1000, 'never opened');
            if (ws.readyState !== 2) throw Error('Ready state after close');
            ws.onopen = () => order.push('open');
            ws.onerror = () => order.push('error');
            ws.onclose = (event) => {
                order.push('close');
                if (order.join('|') !== 'error|close') throw Error('Order: ' + order.join('|'));
                if (event.wasClean || event.code !== 1006) throw Error('Failed close');
                if (ws.readyState !== 3) throw Error('Ready state');
                console.log('closed while connecting');
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"closed while connecting\n");
}

#[test]
fn sends_are_ordered() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        let mut received = Vec::new();
        loop {
            let frame = session.read()?;
            if frame.opcode == CLOSE {
                session.write(CLOSE, &frame.payload)?;
                break;
            }
            assert_eq!(frame.opcode, TEXT);
            session.write(TEXT, &frame.payload)?;
            received.push(String::from_utf8(frame.payload).unwrap());
        }
        assert_eq!(received.len(), 50);
        for (index, message) in received.iter().enumerate() {
            assert_eq!(message, &format!("message {index}"));
        }
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
            const replies = [];
            ws.onopen = () => {
                for (let i = 0; i < 50; i++) ws.send('message ' + i);
            };
            ws.onmessage = (event) => {
                replies.push(event.data);
                if (replies.length < 50) return;
                for (let i = 0; i < replies.length; i++) {
                    if (replies[i] !== 'message ' + i) throw Error('Out of order');
                }
                console.log('ordered', replies.length);
                ws.close(1000);
            };
        "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"ordered 50\n");
    thread.join().unwrap();
}

#[test]
fn handshake_response_validation_and_failure_events() {
    let (url, thread) = server(4, |mut session| {
        let key = accept(&session.key());
        let response = match session.path.as_str() {
            "/tokens" => format!("HTTP/1.1 101 Switching Protocols\r\nUpgrade: example, WebSocket\r\nConnection: keep-alive, Upgrade\r\nSec-WebSocket-Accept: {key}\r\n\r\n"),
            "/duplicate" => format!("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {key}\r\nSec-WebSocket-Accept: wrong\r\n\r\n"),
            "/extension" => format!("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: {key}\r\nSec-WebSocket-Extensions: permessage-deflate\r\n\r\n"),
            _ => "HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\nSec-WebSocket-Accept: wrong\r\n\r\n".to_owned(),
        };
        session.stream.write_all(response.as_bytes())?;
        if session.path == "/tokens" {
            let close = session.read()?;
            assert_eq!(close.opcode, CLOSE);
            session.write(CLOSE, &close.payload)?;
        } else {
            assert_eq!(
                session.read().err().unwrap().kind(),
                std::io::ErrorKind::UnexpectedEof
            );
        }
        Ok(())
    });
    let out = common::run(
        &format!(
            r#"
        for (const path of ['tokens', 'duplicate', 'extension', 'accept']) {{
            await new Promise(resolve => {{
                const ws = new WebSocket('{url}/' + path);
                let errors = 0, microtaskRan = false;
                ws.onopen = () => {{
                    if (path !== 'tokens') throw Error('Accepted bad handshake');
                    ws.close(1000);
                }};
                ws.onerror = () => {{
                    if (ws.readyState !== WebSocket.CLOSED) throw Error('Error readyState');
                    errors++;
                    queueMicrotask(() => microtaskRan = true);
                    ws.send('discarded');
                    if (ws.bufferedAmount !== 9) throw Error('Failed socket buffering');
                }};
                ws.onclose = event => {{
                    if (microtaskRan) throw Error('Error and close must share a turn');
                    if (path === 'tokens') {{
                        if (!event.wasClean || event.code !== 1000 || errors) throw Error('Valid tokens');
                    }} else if (event.wasClean || event.code !== 1006 || errors !== 1) {{
                        throw Error('Failed handshake');
                    }}
                    resolve();
                }};
            }});
        }}
        console.log('handshakes checked');
    "#
        ),
        &[],
    );
    assert_eq!(out.stdout, b"handshakes checked\n");
    thread.join().unwrap();
}

#[test]
fn close_reason_preserves_bom_and_registered_code() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        let payload = Frame::close(1014, "\u{feff}gateway").payload;
        session.write(CLOSE, &payload)?;
        let reply = session.read()?;
        assert_eq!(reply.opcode, CLOSE);
        assert_eq!(reply.payload, payload);
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
        ws.onclose = event => {
            if (!event.wasClean || event.code !== 1014 || event.reason !== '\ufeffgateway') {
                throw Error('Close payload changed');
            }
            console.log('close payload preserved');
        };
    "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"close payload preserved\n");
    thread.join().unwrap();
}

#[test]
fn invalid_control_frames_fail() {
    let (url, thread) = server(5, |mut session| {
        session.accept(None)?;
        let (opcode, payload, code) = match session.path.as_str() {
            "/pong" => (PONG, vec![0; 126], 1002),
            "/long-close" => (CLOSE, Frame::close(1000, &"x".repeat(124)).payload, 1002),
            "/short-close" => (CLOSE, vec![0], 1002),
            "/code" => (CLOSE, Frame::close(1005, "").payload, 1002),
            _ => (CLOSE, vec![3, 232, 255], 1007),
        };
        session.write(opcode, &payload)?;
        let close = session.read()?;
        assert_eq!(close.opcode, CLOSE);
        assert_eq!(close.code(), code);
        Ok(())
    });
    let out = common::run(
        &format!(
            r#"
        for (const path of ['pong', 'long-close', 'short-close', 'code', 'utf8']) {{
            await new Promise(resolve => {{
                const ws = new WebSocket('{url}/' + path);
                let errors = 0;
                ws.onerror = () => errors++;
                ws.onmessage = () => {{ throw Error('Control frame delivered'); }};
                ws.onclose = event => {{
                    if (event.wasClean || event.code !== 1006 || errors !== 1) throw Error('Invalid control frame');
                    resolve();
                }};
            }});
        }}
        console.log('invalid controls rejected');
    "#
        ),
        &[],
    );
    assert_eq!(out.stdout, b"invalid controls rejected\n");
    thread.join().unwrap();
}

#[test]
fn call_time_primordials_and_protocol_iterator() {
    let (url, thread) = server(1, |mut session| {
        assert_eq!(session.path, "/CaseSensitive");
        assert_eq!(session.protocol, "chat");
        session.accept(Some("chat"))?;
        let text = session.read()?;
        assert_eq!(text.opcode, TEXT);
        assert_eq!(text.payload, b"hello");
        session.write(TEXT, &text.payload)?;
        let close = session.read()?;
        assert_eq!(close.code(), 1000);
        session.write(CLOSE, &close.payload)?;
        Ok(())
    });
    let out = common::run(
        &format!(
            r#"
        const poison = () => {{ throw Error('Guest replacement invoked'); }};
        Math.floor = poison;
        Promise.prototype.then = poison;
        Object.defineProperty(Promise.prototype, 'constructor', {{ get: poison }});
        Object.setPrototypeOf(WebSocket, poison);
        Object.setPrototypeOf(CloseEvent, poison);
        for (const name of ['href', 'protocol', 'origin', 'hash']) {{
            Object.defineProperty(URL.prototype, name, {{ get: poison }});
        }}
        Object.defineProperty(Object.getPrototypeOf(Uint8Array.prototype), 'length', {{ get: poison }});
        let gets = 0;
        const protocols = {{
            get [Symbol.iterator]() {{
                if (++gets !== 1) throw Error('Iterator read twice');
                return function* () {{ yield 'chat'; }};
            }}
        }};
        const ws = new WebSocket('{url}/CaseSensitive', protocols);
        ws.onopen = () => {{
            if (ws.url !== '{url}/CaseSensitive' || gets !== 1) throw Error('Constructor');
            ws.send('hello');
            if (ws.bufferedAmount !== 5) throw Error('Buffered amount');
        }};
        ws.onmessage = event => {{
            if (event.data !== 'hello' || event.origin !== '{url}') throw Error('Message');
            ws.close(1000.4);
        }};
        ws.onclose = event => {{
            if (!event.wasClean || event.code !== 1000) throw Error('Close');
            console.log('primordials safe');
        }};
    "#
        ),
        &[],
    );
    assert_eq!(out.stdout, b"primordials safe\n");
    thread.join().unwrap();
}

#[test]
fn canceling_a_pending_handshake_drops_the_transport() {
    let (ready, handshake) = std::sync::mpsc::channel();
    let (closed, disconnected) = std::sync::mpsc::channel();
    let (url, thread) = server(1, move |mut session| {
        ready.send(()).unwrap();
        let mut byte = [0];
        assert_eq!(
            session.stream.read(&mut byte)?,
            0,
            "canceled handshake retained TCP socket"
        );
        closed.send(()).unwrap();
        Ok(())
    });
    let (_dir, mut child) = common::spawn(&connect(
        &url,
        r#"
        import { stdin } from "avril:std";
        const reader = stdin.getReader();
        ws.onopen = () => { throw Error('Pending handshake opened'); };
        ws.onclose = event => {
            if (event.wasClean || event.code !== 1006) throw Error('Failed close');
            console.log('canceled');
        };
        await reader.read();
        ws.close();
        await reader.read();
        reader.releaseLock();
    "#,
    ));
    handshake.recv_timeout(Duration::from_secs(5)).unwrap();
    let mut stdin = child.stdin.take().unwrap();
    stdin.write_all(b"a").unwrap();
    disconnected.recv_timeout(Duration::from_secs(5)).unwrap();
    stdin.write_all(b"b").unwrap();
    drop(stdin);
    let out = common::wait(child);
    assert!(
        out.status.success(),
        "{}",
        String::from_utf8_lossy(&out.stderr)
    );
    assert_eq!(out.stdout, b"canceled\n");
    thread.join().unwrap();
}

#[test]
fn malformed_wire_frames_fail_with_protocol_or_utf8_close() {
    let cases: &[(&str, &[u8], u16)] = &[
        ("masked-server-text", b"\x81\x82\0\0\0\0hi", 1002),
        (
            "overlapping-text",
            b"\x01\x03old\x01\x03new\x80\x03end",
            1002,
        ),
        ("interleaved-binary", b"\x02\x01a\x01\x01b\x80\x01c", 1002),
        ("interleaved-complete-message", b"\x01\x01a\x82\x01b", 1002),
        (
            "incomplete-fragmented-utf8",
            b"\x01\x03hi\xe2\x80\x00",
            1007,
        ),
        ("invalid-fragmented-utf8", b"\x01\x01\xc2\x80\x01x", 1007),
        ("nonminimal-16-bit-length", b"\x82\x7e\x00\x01x", 1002),
        (
            "nonminimal-64-bit-length",
            b"\x82\x7f\0\0\0\0\0\0\0\x01x",
            1002,
        ),
        (
            "negative-64-bit-length",
            b"\x82\x7f\x80\0\0\0\0\0\0\0",
            1002,
        ),
        ("unexpected-continuation", b"\x80\x00", 1002),
        ("reserved-bits", b"\xc1\x00", 1002),
        ("reserved-opcode", b"\x83\x00", 1002),
        ("fragmented-control", b"\x09\x00", 1002),
    ];
    for &(name, wire, code) in cases {
        let (url, thread) = server(1, move |mut session| {
            session.accept(None)?;
            session.stream.write_all(wire)?;
            let close = session.read()?;
            assert_eq!(close.opcode, CLOSE, "{name}");
            assert_eq!(close.code(), code, "{name}");
            Ok(())
        });
        let out = common::run(
            &connect(
                &url,
                r#"
                let errors = 0;
                ws.onmessage = () => { throw Error('Malformed message delivered'); };
                ws.onerror = () => errors++;
                ws.onclose = event => {
                    if (event.wasClean || event.code !== 1006 || errors !== 1) {
                        throw Error('Malformed frame did not fail the connection');
                    }
                    console.log('framing failed');
                };
            "#,
            ),
            &[],
        );
        assert_eq!(out.stdout, b"framing failed\n", "{name}");
        thread.join().unwrap();
    }
}

#[test]
fn fragmented_utf8_survives_control_frames_and_resets_between_messages() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        write_frame(&mut session.stream, false, TEXT, b"a\xf0")?;
        write_frame(&mut session.stream, false, CONTINUATION, b"\x9f")?;
        session.write(PING, b"between fragments")?;
        write_frame(&mut session.stream, true, CONTINUATION, b"\x92\xa9")?;
        session.write(TEXT, b"next")?;
        let mut pong = false;
        loop {
            let frame = session.read()?;
            match frame.opcode {
                PONG => {
                    assert_eq!(frame.payload, b"between fragments");
                    pong = true;
                }
                CLOSE => {
                    assert!(pong, "control frames must not interrupt message assembly");
                    session.write(CLOSE, &frame.payload)?;
                    break;
                }
                _ => panic!("unexpected client frame"),
            }
        }
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
        const messages = [];
        ws.onmessage = event => {
            messages.push(event.data);
            if (messages.length === 2) {
                if (messages[0] !== 'a\u{1f4a9}' || messages[1] !== 'next') throw Error('Fragmented UTF-8');
                ws.close(1000);
            }
        };
        ws.onclose = event => {
            if (!event.wasClean || event.code !== 1000) throw Error('Close');
            console.log('fragmented utf8 checked');
        };
    "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"fragmented utf8 checked\n");
    thread.join().unwrap();
}

#[test]
fn message_size_limit_applies_before_reading_excess_payload() {
    for fragmented in [false, true] {
        let (url, thread) = server(1, move |mut session| {
            session.accept(None)?;
            if fragmented {
                let fragment = vec![0; 8 * 1024 * 1024];
                write_frame(&mut session.stream, false, BINARY, &fragment)?;
                write_frame(&mut session.stream, false, CONTINUATION, &fragment)?;
                // No payload follows: the aggregate length alone must trigger failure.
                session.stream.write_all(b"\x80\x01")?;
            } else {
                session.stream.write_all(b"\x82\x7f")?;
                session
                    .stream
                    .write_all(&(16 * 1024 * 1024 + 1u64).to_be_bytes())?;
            }
            let close = session.read()?;
            assert_eq!(close.opcode, CLOSE);
            assert_eq!(close.code(), 1009);
            Ok(())
        });
        let out = common::run(
            &connect(
                &url,
                r#"
            let errors = 0;
            ws.onerror = () => errors++;
            ws.onmessage = () => { throw Error('Oversized message delivered'); };
            ws.onclose = event => {
                if (event.wasClean || event.code !== 1006 || errors !== 1) throw Error('Oversized message');
                console.log('size limit checked');
            };
        "#,
            ),
            &[],
        );
        assert_eq!(out.stdout, b"size limit checked\n");
        thread.join().unwrap();
    }
}

#[test]
fn fragmented_message_at_size_limit_is_accepted() {
    let (url, thread) = server(1, |mut session| {
        session.accept(None)?;
        let fragment = vec![0x5a; 8 * 1024 * 1024];
        write_frame(&mut session.stream, false, BINARY, &fragment)?;
        write_frame(&mut session.stream, true, CONTINUATION, &fragment)?;
        let close = session.read()?;
        assert_eq!(close.opcode, CLOSE);
        session.write(CLOSE, &close.payload)?;
        Ok(())
    });
    let out = common::run(
        &connect(
            &url,
            r#"
        ws.binaryType = 'arraybuffer';
        let messages = 0;
        ws.onmessage = event => {
            const bytes = new Uint8Array(event.data);
            if (++messages !== 1 || bytes.length !== 16 * 1024 * 1024 ||
                bytes[0] !== 0x5a || bytes[bytes.length - 1] !== 0x5a) throw Error('Size limit boundary');
            ws.close(1000);
        };
        ws.onclose = event => {
            if (!event.wasClean || event.code !== 1000 || messages !== 1) throw Error('Close');
            console.log('size boundary checked');
        };
    "#,
        ),
        &[],
    );
    assert_eq!(out.stdout, b"size boundary checked\n");
    thread.join().unwrap();
}