outputStream.addStream() reports success when the transport is destroyed mid-write #2

import 'dart:async';
import 'dart:io';

import 'package:socks_socket/socks_socket.dart';
import 'package:test/test.dart';

void main() {
  test('write cut short by destroy() reports success', () async {
    // Minimal SOCKS5 proxy: accepts CONNECT, then never reads tunnel data.
    final proxy = await ServerSocket.bind(InternetAddress.loopbackIPv4, 0);
    proxy.listen((peer) {
      final buffer = <int>[];
      var tunnel = false;
      late StreamSubscription<List<int>> input;
      input = peer.listen((bytes) {
        if (tunnel) return;
        buffer.addAll(bytes);
        if (buffer.length == 3) {
          peer.add([5, 0]);
        } else if (buffer.length > 8 && buffer.length == 10 + buffer[7]) {
          tunnel = true;
          peer.add([5, 0, 0, 1, 0, 0, 0, 0, 0, 0]);
          input.pause();
        }
      }, onError: (Object _) {});
    });

    final socket = await SOCKSSocket.create(
        proxyHost: InternetAddress.loopbackIPv4.address, proxyPort: proxy.port);
    await socket.connect();
    await socket.connectTo('example.com', 80);

    // 8 MiB can't fit in the socket buffers, so this write stays in flight.
    final write = socket.outputStream
        .addStream(Stream.value(List.filled(8 * 1024 * 1024, 1)));
    await Future<void>.delayed(const Duration(milliseconds: 200));
    socket.socket.destroy();

    // Expected: the write fails. Actual: it completes normally.
    await expectLater(write, throwsA(anything));
    await socket.close().catchError((Object _) {});
    await proxy.close();
  });
}