proto-socket/dart/crosstest/dart_python.dart

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

import 'package:proto_socket/proto_socket.dart';

const _host = '127.0.0.1';
const _tcpPort = 29694;
const _wsPort = 29696;
const _tlsTcpPort = 29698;
const _wssPort = 29700;
const _heartbeatIntervalSeconds = 30;
const _heartbeatWaitSeconds = 10;
const _serverObservationWindow = Duration(milliseconds: 200);
const _processTimeout = Duration(seconds: 20);
const _streamTimeout = Duration(seconds: 5);
final _repoRoot = File.fromUri(Platform.script).parent.parent.parent.path;
final _pythonDir = Directory('$_repoRoot/python').path;
final _certFile = '$_repoRoot/dart/test/certs/server.crt';
final _keyFile = '$_repoRoot/dart/test/certs/server.key';

class _DartTcpClient extends ProtobufClient {
  _DartTcpClient(Socket socket)
      : super(socket, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, {
          TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
        });
}

class _DartWsClient extends WsProtobufClient {
  _DartWsClient(WebSocket ws)
      : super(ws, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, {
          TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer,
        });
}

class _DartTcpServer extends ProtobufServer {
  final void Function(ProtobufClient) onConnected;

  _DartTcpServer(int port, this.onConnected)
      : super(_host, port, (socket) => _DartTcpClient(socket));

  @override
  void onClientConnected(ProtobufClient client) => onConnected(client);
}

class _DartTlsTcpServer extends ProtobufServer {
  final void Function(ProtobufClient) onConnected;

  _DartTlsTcpServer(int port, SecurityContext ctx, this.onConnected)
      : super.secure(_host, port, ctx, (socket) => _DartTcpClient(socket));

  @override
  void onClientConnected(ProtobufClient client) => onConnected(client);
}

class _DartWsServer extends WsProtobufServer {
  final void Function(WsProtobufClient) onConnected;

  _DartWsServer(int port, this.onConnected)
      : super(_host, port, (ws) => _DartWsClient(ws));

  @override
  void onClientConnected(WsProtobufClient client) => onConnected(client);
}

class _DartWssServer extends WsProtobufServer {
  final void Function(WsProtobufClient) onConnected;

  _DartWssServer(int port, SecurityContext ctx, this.onConnected)
      : super.secure(_host, port, ctx, (ws) => _DartWsClient(ws));

  @override
  void onClientConnected(WsProtobufClient client) => onConnected(client);
}

Future<void> main() async {
  print(
      'INFO typeName dart=${TestData.getDefault().info_.qualifiedMessageName}');
  try {
    await _runTcp();
    await _runWs();
    await _runTls();
    print('PASS all dart-server/python-client crosstests passed');
  } catch (error) {
    stderr.writeln('FAIL crosstest error=$error');
    exitCode = 1;
  }
}

Future<void> _runTcp() async {
  await _runTcpSendPush();
  await Future<void>.delayed(const Duration(milliseconds: 150));
  await _runTcpRequests();
}

Future<void> _runWs() async {
  await _runWsSendPush();
  await Future<void>.delayed(const Duration(milliseconds: 150));
  await _runWsRequests();
}

Future<void> _runTcpSendPush() async {
  final received = Completer<bool>();
  final server = _DartTcpServer(_tcpPort, (client) {
    client.addListener<TestData>((data) {
      print('SERVER_RECEIVED index=${data.index} message=${data.message}');
      final valid =
          data.index == 101 && data.message == 'fire from python client';
      if (!received.isCompleted) {
        received.complete(valid);
      }
      if (valid) {
        unawaited(client.send(TestData()
          ..index = 200
          ..message = 'push from dart server'));
      }
    });
  });
  await _withServer(server.start, server.stop, () async {
    await _runPythonClient('tcp', _tcpPort, 'send-push', {'1', '2'});
    final ok = await received.future
        .timeout(_serverObservationWindow, onTimeout: () => false);
    if (!ok) {
      throw StateError('TCP send-push server did not receive expected data');
    }
  });
}

Future<void> _runTcpRequests() async {
  final server = _DartTcpServer(_tcpPort, (client) {
    client.addRequestListener<TestData, TestData>((req) async {
      return TestData()
        ..index = req.index * 2
        ..message = 'echo: ${req.message}';
    });
  });
  await _withServer(
    server.start,
    server.stop,
    () => _runPythonClient('tcp', _tcpPort, 'requests', {'3', '4'}),
  );
}

Future<void> _runWsSendPush() async {
  final received = Completer<bool>();
  final server = _DartWsServer(_wsPort, (client) {
    client.addListener<TestData>((data) {
      print('SERVER_RECEIVED index=${data.index} message=${data.message}');
      final valid =
          data.index == 101 && data.message == 'fire from python client';
      if (!received.isCompleted) {
        received.complete(valid);
      }
      if (valid) {
        unawaited(client.send(TestData()
          ..index = 200
          ..message = 'push from dart server'));
      }
    });
  });
  await _withServer(server.start, server.stop, () async {
    await _runPythonClient('ws', _wsPort, 'send-push', {'1', '2'});
    final ok = await received.future
        .timeout(_serverObservationWindow, onTimeout: () => false);
    if (!ok) {
      throw StateError('WS send-push server did not receive expected data');
    }
  });
}

Future<void> _runWsRequests() async {
  final server = _DartWsServer(_wsPort, (client) {
    client.addRequestListener<TestData, TestData>((req) async {
      return TestData()
        ..index = req.index * 2
        ..message = 'echo: ${req.message}';
    });
  });
  await _withServer(
    server.start,
    server.stop,
    () => _runPythonClient('ws', _wsPort, 'requests', {'3', '4'}),
  );
}

Future<void> _withServer(Future<void> Function() start,
    Future<void> Function() stop, Future<void> Function() body) async {
  await start();
  try {
    await body();
  } finally {
    await stop();
  }
}

Future<void> _runTls() async {
  final ctx = SecurityContext()
    ..useCertificateChain(_certFile)
    ..usePrivateKey(_keyFile);
  await _runTlsTcpSendPush(ctx);
  await Future<void>.delayed(const Duration(milliseconds: 150));
  await _runTlsTcpRequests(ctx);
  await Future<void>.delayed(const Duration(milliseconds: 150));
  await _runWssSendPush(ctx);
  await Future<void>.delayed(const Duration(milliseconds: 150));
  await _runWssRequests(ctx);
}

Future<void> _runTlsTcpSendPush(SecurityContext ctx) async {
  final received = Completer<bool>();
  final server = _DartTlsTcpServer(_tlsTcpPort, ctx, (client) {
    client.addListener<TestData>((data) {
      final valid =
          data.index == 101 && data.message == 'fire from python client';
      if (!received.isCompleted) {
        received.complete(valid);
      }
      if (valid) {
        unawaited(client.send(TestData()
          ..index = 200
          ..message = 'push from dart server'));
      }
    });
  });
  await _withServer(server.start, server.stop, () async {
    await _runPythonClient('tls', _tlsTcpPort, 'send-push', {'1', '2'},
        certFile: _certFile);
    final ok = await received.future
        .timeout(_serverObservationWindow, onTimeout: () => false);
    if (!ok) {
      throw StateError(
          'TLS TCP send-push server did not receive expected data');
    }
  });
}

Future<void> _runTlsTcpRequests(SecurityContext ctx) async {
  final server = _DartTlsTcpServer(_tlsTcpPort, ctx, (client) {
    client.addRequestListener<TestData, TestData>((req) async {
      return TestData()
        ..index = req.index * 2
        ..message = 'echo: ${req.message}';
    });
  });
  await _withServer(
    server.start,
    server.stop,
    () => _runPythonClient('tls', _tlsTcpPort, 'requests', {'3', '4'},
        certFile: _certFile),
  );
}

Future<void> _runWssSendPush(SecurityContext ctx) async {
  final received = Completer<bool>();
  final server = _DartWssServer(_wssPort, ctx, (client) {
    client.addListener<TestData>((data) {
      final valid =
          data.index == 101 && data.message == 'fire from python client';
      if (!received.isCompleted) {
        received.complete(valid);
      }
      if (valid) {
        unawaited(client.send(TestData()
          ..index = 200
          ..message = 'push from dart server'));
      }
    });
  });
  await _withServer(server.start, server.stop, () async {
    await _runPythonClient('wss', _wssPort, 'send-push', {'1', '2'},
        certFile: _certFile);
    final ok = await received.future
        .timeout(_serverObservationWindow, onTimeout: () => false);
    if (!ok) {
      throw StateError('WSS send-push server did not receive expected data');
    }
  });
}

Future<void> _runWssRequests(SecurityContext ctx) async {
  final server = _DartWssServer(_wssPort, ctx, (client) {
    client.addRequestListener<TestData, TestData>((req) async {
      return TestData()
        ..index = req.index * 2
        ..message = 'echo: ${req.message}';
    });
  });
  await _withServer(
    server.start,
    server.stop,
    () => _runPythonClient('wss', _wssPort, 'requests', {'3', '4'},
        certFile: _certFile),
  );
}

Future<void> _runPythonClient(
    String mode, int port, String phase, Set<String> expectedScenarios,
    {String? certFile}) async {
  final process = await Process.start(
    'python3',
    [
      'crosstest/dart_python_client.py',
      '--mode=$mode',
      '--port=$port',
      '--phase=$phase',
      if (certFile != null) '--cert=$certFile',
    ],
    workingDirectory: _pythonDir,
  );

  final resultLines = <String>[];
  final stdoutDone = process.stdout
      .transform(utf8.decoder)
      .transform(const LineSplitter())
      .listen((line) {
    print(line);
    if (line.startsWith('PASS ') || line.startsWith('FAIL ')) {
      resultLines.add(line);
    }
  }).asFuture<void>();

  final stderrDone = process.stderr
      .transform(utf8.decoder)
      .transform(const LineSplitter())
      .listen(stderr.writeln)
      .asFuture<void>();

  final code = await process.exitCode.timeout(_processTimeout, onTimeout: () {
    process.kill(ProcessSignal.sigkill);
    return -1;
  });
  await Future.wait([stdoutDone, stderrDone])
      .timeout(_streamTimeout, onTimeout: () {
    throw StateError('python-client $mode/$phase streams did not finish');
  });

  _validateResultLines(
      'python-client $mode/$phase', code, resultLines, expectedScenarios);
}

void _validateResultLines(
  String label,
  int exitCode,
  List<String> lines,
  Set<String> expectedScenarios,
) {
  final failed = lines.where((line) => line.startsWith('FAIL ')).toList();
  final passed = <String>{};
  for (final line in lines.where((line) => line.startsWith('PASS '))) {
    final match = RegExp(r'scenario=([^ ]+)').firstMatch(line);
    if (match != null) {
      passed.add(match.group(1)!);
    }
  }
  final missing = expectedScenarios.difference(passed);
  if (exitCode != 0 || failed.isNotEmpty || missing.isNotEmpty) {
    throw StateError(
        '$label failed exitCode=$exitCode failed=$failed missing=$missing');
  }
}