From bf0157d35d573adcd50303b1441547a797633a4f Mon Sep 17 00:00:00 2001 From: toki Date: Fri, 24 Apr 2026 03:42:25 +0900 Subject: [PATCH] feat: add Python-Dart cross-test files --- dart/crosstest/dart_python.dart | 229 ++++++++++++++++++++++ dart/crosstest/python_dart_client.dart | 193 +++++++++++++++++++ kotlin/crosstest/kotlin_python.kt | 234 +++++++++++++++++++++++ kotlin/crosstest/python_kotlin_client.kt | 196 +++++++++++++++++++ python/crosstest/dart_python_client.py | 139 ++++++++++++++ python/crosstest/kotlin_python_client.py | 139 ++++++++++++++ python/crosstest/python_dart.py | 178 +++++++++++++++++ python/crosstest/python_kotlin.py | 176 +++++++++++++++++ 8 files changed, 1484 insertions(+) create mode 100644 dart/crosstest/dart_python.dart create mode 100644 dart/crosstest/python_dart_client.dart create mode 100644 kotlin/crosstest/kotlin_python.kt create mode 100644 kotlin/crosstest/python_kotlin_client.kt create mode 100644 python/crosstest/dart_python_client.py create mode 100644 python/crosstest/kotlin_python_client.py create mode 100644 python/crosstest/python_dart.py create mode 100644 python/crosstest/python_kotlin.py diff --git a/dart/crosstest/dart_python.dart b/dart/crosstest/dart_python.dart new file mode 100644 index 0000000..49f127e --- /dev/null +++ b/dart/crosstest/dart_python.dart @@ -0,0 +1,229 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:io'; + +import 'package:toki_socket/toki_socket.dart'; + +const _host = '127.0.0.1'; +const _tcpPort = 29694; +const _wsPort = 29696; +const _heartbeatIntervalSeconds = 30; +const _heartbeatWaitSeconds = 10; +const _serverObservationWindow = Duration(milliseconds: 200); +const _processTimeout = Duration(seconds: 20); +final _repoRoot = File.fromUri(Platform.script).parent.parent.parent.path; +final _pythonDir = Directory('$_repoRoot/python').path; + +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 _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); +} + +Future main() async { + print( + 'INFO typeName dart=${TestData.getDefault().info_.qualifiedMessageName}'); + try { + await _runTcp(); + await _runWs(); + print('PASS all dart-server/python-client crosstests passed'); + } catch (error) { + stderr.writeln('FAIL crosstest error=$error'); + exitCode = 1; + } +} + +Future _runTcp() async { + await _runTcpSendPush(); + await Future.delayed(const Duration(milliseconds: 150)); + await _runTcpRequests(); +} + +Future _runWs() async { + await _runWsSendPush(); + await Future.delayed(const Duration(milliseconds: 150)); + await _runWsRequests(); +} + +Future _runTcpSendPush() async { + final received = Completer(); + final server = _DartTcpServer(_tcpPort, (client) { + client.addListener((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 _runTcpRequests() async { + final server = _DartTcpServer(_tcpPort, (client) { + client.addRequestListener((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 _runWsSendPush() async { + final received = Completer(); + final server = _DartWsServer(_wsPort, (client) { + client.addListener((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 _runWsRequests() async { + final server = _DartWsServer(_wsPort, (client) { + client.addRequestListener((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 _withServer(Future Function() start, + Future Function() stop, Future Function() body) async { + await start(); + try { + await body(); + } finally { + await stop(); + } +} + +Future _runPythonClient( + String mode, int port, String phase, Set expectedScenarios) async { + final process = await Process.start( + 'python3', + [ + 'crosstest/dart_python_client.py', + '--mode=$mode', + '--port=$port', + '--phase=$phase', + ], + workingDirectory: _pythonDir, + ); + + final resultLines = []; + 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(); + + final stderrDone = process.stderr + .transform(utf8.decoder) + .transform(const LineSplitter()) + .listen(stderr.writeln) + .asFuture(); + + final code = await process.exitCode.timeout(_processTimeout, onTimeout: () { + process.kill(ProcessSignal.sigkill); + return -1; + }); + await Future.wait([stdoutDone, stderrDone]); + + _validateResultLines( + 'python-client $mode/$phase', code, resultLines, expectedScenarios); +} + +void _validateResultLines( + String label, + int exitCode, + List lines, + Set expectedScenarios, +) { + final failed = lines.where((line) => line.startsWith('FAIL ')).toList(); + final passed = {}; + 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'); + } +} diff --git a/dart/crosstest/python_dart_client.dart b/dart/crosstest/python_dart_client.dart new file mode 100644 index 0000000..3fbb85a --- /dev/null +++ b/dart/crosstest/python_dart_client.dart @@ -0,0 +1,193 @@ +import 'dart:async'; +import 'dart:io'; + +import 'package:toki_socket/toki_socket.dart'; + +const _host = '127.0.0.1'; +const _wsPath = '/'; +const _heartbeatIntervalSeconds = 30; +const _heartbeatWaitSeconds = 10; +const _connectWindow = Duration(seconds: 3); +const _requestWindow = Duration(seconds: 2); + +class _TcpClient extends ProtobufClient { + _TcpClient(Socket socket) + : super(socket, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, { + TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer, + }); +} + +class _WsClient extends WsProtobufClient { + _WsClient(WebSocket ws) + : super(ws, _heartbeatIntervalSeconds, _heartbeatWaitSeconds, { + TestData.getDefault().info_.qualifiedMessageName: TestData.fromBuffer, + }); +} + +class _ClientHandle { + final Communicator client; + final Future Function() close; + + _ClientHandle(this.client, this.close); +} + +Future main(List args) async { + final mode = _argValue(args, 'mode') ?? 'tcp'; + final phase = _argValue(args, 'phase') ?? 'send-push'; + final port = int.tryParse(_argValue(args, 'port') ?? ''); + + print( + 'INFO typeName dart=${TestData.getDefault().info_.qualifiedMessageName}'); + + if (port == null) { + _fail('setup', 'port is required'); + exitCode = 1; + return; + } + + late _ClientHandle handle; + try { + handle = await _connectWithRetry(mode, port); + } catch (error) { + _fail('setup', error.toString()); + exitCode = 1; + return; + } + + var ok = false; + try { + switch (phase) { + case 'send-push': + ok = await _runSendPush(handle.client); + break; + case 'requests': + ok = await _runRequests(handle.client); + break; + default: + _fail('setup', 'unknown phase "$phase"'); + } + } finally { + await handle.close(); + } + + if (!ok) { + exitCode = 1; + } + exit(exitCode); +} + +Future<_ClientHandle> _connectWithRetry(String mode, int port) async { + final deadline = DateTime.now().add(_connectWindow); + Object? lastError; + + while (DateTime.now().isBefore(deadline)) { + try { + switch (mode) { + case 'tcp': + final socket = await ProtobufClient.connect(_host, port) + .timeout(const Duration(milliseconds: 300)); + final client = _TcpClient(socket); + return _ClientHandle(client, client.close); + case 'ws': + final ws = await WsProtobufClient.connect(_host, port, path: _wsPath) + .timeout(const Duration(milliseconds: 300)); + final client = _WsClient(ws); + return _ClientHandle(client, client.close); + default: + throw ArgumentError('unknown mode "$mode"'); + } + } catch (error) { + lastError = error; + await Future.delayed(const Duration(milliseconds: 100)); + } + } + + throw TimeoutException( + 'connect $mode:$port timed out: $lastError', _connectWindow); +} + +Future _runSendPush(Communicator client) async { + final pushCompleter = Completer(); + client.addListener((data) { + if (!pushCompleter.isCompleted) { + pushCompleter.complete(data); + } + }); + + await client.send(TestData() + ..index = 101 + ..message = 'fire from dart client'); + _pass('1', 'fire-and-forget sent'); + + try { + final push = await pushCompleter.future.timeout(_requestWindow); + if (push.index != 200 || push.message != 'push from python server') { + _fail( + '2', 'unexpected push index=${push.index} message="${push.message}"'); + return false; + } + _pass('2', 'received push from python server'); + return true; + } catch (error) { + _fail('2', error.toString()); + return false; + } +} + +Future _runRequests(Communicator client) async { + try { + final single = await client + .sendRequest(TestData() + ..index = 21 + ..message = 'single request from dart') + .timeout(_requestWindow); + if (single.index != 42 || + single.message != 'echo: single request from dart') { + _fail('3', + 'unexpected response index=${single.index} message="${single.message}"'); + return false; + } + _pass('3', 'single request response matched'); + + final futures = >[]; + for (var i = 0; i < 5; i++) { + futures.add(() async { + final index = 30 + i; + final message = 'multi request $i from dart'; + final res = await client + .sendRequest(TestData() + ..index = index + ..message = message) + .timeout(_requestWindow); + if (res.index != index * 2 || res.message != 'echo: $message') { + throw StateError( + 'request $i got index=${res.index} message="${res.message}"'); + } + }()); + } + await Future.wait(futures); + _pass('4', 'concurrent request responses matched'); + return true; + } catch (error) { + _fail('4', error.toString()); + return false; + } +} + +String? _argValue(List args, String name) { + final prefix = '--$name='; + for (final arg in args) { + if (arg.startsWith(prefix)) { + return arg.substring(prefix.length); + } + } + return null; +} + +void _pass(String scenario, String detail) { + print('PASS scenario=$scenario detail=$detail'); +} + +void _fail(String scenario, String error) { + print('FAIL scenario=$scenario error=$error'); +} diff --git a/kotlin/crosstest/kotlin_python.kt b/kotlin/crosstest/kotlin_python.kt new file mode 100644 index 0000000..5987e79 --- /dev/null +++ b/kotlin/crosstest/kotlin_python.kt @@ -0,0 +1,234 @@ +@file:JvmName("KotlinPythonKt") + +package com.tokilabs.toki_socket.crosstest + +import com.tokilabs.toki_socket.TcpClient +import com.tokilabs.toki_socket.TcpServer +import com.tokilabs.toki_socket.ParserMap +import com.tokilabs.toki_socket.WsClient +import com.tokilabs.toki_socket.WsServer +import com.tokilabs.toki_socket.addListenerTyped +import com.tokilabs.toki_socket.addRequestListenerTyped +import com.tokilabs.toki_socket.typeNameOf +import com.tokilabs.toki_socket.packets.TestData +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withContext +import kotlinx.coroutines.withTimeoutOrNull +import java.io.File +import java.util.concurrent.TimeUnit +import kotlin.system.exitProcess + +private const val HOST = "127.0.0.1" +private const val TCP_PORT = 29394 +private const val WS_PORT = 29396 +private const val WS_PATH = "/" +private const val PROCESS_TIMEOUT_MS = 20_000L +private const val SERVER_OBSERVATION_MS = 200L + +fun main() = runBlocking { + println("INFO typeName kotlin=${typeNameOf()}") + try { + runTcpSendPush() + delay(150) + runTcpRequests() + delay(150) + runWsSendPush() + delay(150) + runWsRequests() + println("PASS all kotlin-server/python-client crosstests passed") + } catch (error: Throwable) { + System.err.println("FAIL crosstest error=${error.message ?: error}") + exitProcess(1) + } +} + +private suspend fun runTcpSendPush() = coroutineScope { + val received = CompletableDeferred() + val server = TcpServer(HOST, TCP_PORT) { socket -> + TcpClient.fromSocket(socket, 0, 0, parserMap()) + } + server.onClientConnected = { client -> + addListenerTyped(client.communicator) { data -> + println("SERVER_RECEIVED index=${data.index} message=${data.message}") + val valid = data.index == 101 && data.message == "fire from python client" + received.complete(valid) + if (valid) { + launch { + client.send( + TestData.newBuilder() + .setIndex(200) + .setMessage("push from kotlin server") + .build(), + ) + } + } + } + } + withServer(server::start, server::stop) { + runPythonClient("tcp", TCP_PORT, "send-push", setOf("1", "2")) + val ok = withTimeoutOrNull(SERVER_OBSERVATION_MS) { received.await() } ?: false + check(ok) { "TCP send-push server did not receive expected data" } + } +} + +private suspend fun runTcpRequests() { + val server = TcpServer(HOST, TCP_PORT) { socket -> + TcpClient.fromSocket(socket, 0, 0, parserMap()) + } + server.onClientConnected = { client -> + addRequestListenerTyped(client.communicator) { req -> + TestData.newBuilder() + .setIndex(req.index * 2) + .setMessage("echo: ${req.message}") + .build() + } + } + withServer(server::start, server::stop) { + runPythonClient("tcp", TCP_PORT, "requests", setOf("3", "4")) + } +} + +private suspend fun runWsSendPush() = coroutineScope { + val received = CompletableDeferred() + val server = WsServer(HOST, WS_PORT, WS_PATH) { conn -> + WsClient.forServer(conn, 0, 0, parserMap()) + } + server.onClientConnected = { client -> + addListenerTyped(client.communicator) { data -> + println("SERVER_RECEIVED index=${data.index} message=${data.message}") + val valid = data.index == 101 && data.message == "fire from python client" + received.complete(valid) + if (valid) { + launch { + client.send( + TestData.newBuilder() + .setIndex(200) + .setMessage("push from kotlin server") + .build(), + ) + } + } + } + } + withServer(server::start, server::stop) { + runPythonClient("ws", WS_PORT, "send-push", setOf("1", "2")) + val ok = withTimeoutOrNull(SERVER_OBSERVATION_MS) { received.await() } ?: false + check(ok) { "WS send-push server did not receive expected data" } + } +} + +private suspend fun runWsRequests() { + val server = WsServer(HOST, WS_PORT, WS_PATH) { conn -> + WsClient.forServer(conn, 0, 0, parserMap()) + } + server.onClientConnected = { client -> + addRequestListenerTyped(client.communicator) { req -> + TestData.newBuilder() + .setIndex(req.index * 2) + .setMessage("echo: ${req.message}") + .build() + } + } + withServer(server::start, server::stop) { + runPythonClient("ws", WS_PORT, "requests", setOf("3", "4")) + } +} + +private suspend fun withServer( + start: () -> Unit, + stop: () -> Unit, + body: suspend () -> Unit, +) { + start() + try { + body() + } finally { + stop() + } +} + +private suspend fun runPythonClient( + mode: String, + port: Int, + phase: String, + expectedScenarios: Set, +) = withContext(Dispatchers.IO) { + val process = ProcessBuilder( + "python3", + "crosstest/kotlin_python_client.py", + "--mode=$mode", + "--port=$port", + "--phase=$phase", + ) + .directory(pythonDir()) + .redirectErrorStream(false) + .start() + + val stdoutLines = mutableListOf() + val stdoutThread = Thread { + process.inputStream.bufferedReader().forEachLine { line -> + println(line) + if (line.startsWith("PASS ") || line.startsWith("FAIL ")) { + stdoutLines.add(line) + } + } + } + val stderrThread = Thread { + process.errorStream.bufferedReader().forEachLine { line -> + System.err.println(line) + } + } + stdoutThread.start() + stderrThread.start() + + val finished = process.waitFor(PROCESS_TIMEOUT_MS, TimeUnit.MILLISECONDS) + if (!finished) { + process.destroyForcibly() + } + stdoutThread.join() + stderrThread.join() + + validateResultLines( + "python-client $mode/$phase", + if (finished) process.exitValue() else -1, + stdoutLines, + expectedScenarios, + ) +} + +private fun validateResultLines( + label: String, + exitCode: Int, + lines: List, + expectedScenarios: Set, +) { + val failed = lines.filter { it.startsWith("FAIL ") } + val passed = lines + .filter { it.startsWith("PASS ") } + .mapNotNull { Regex("""scenario=([^ ]+)""").find(it)?.groupValues?.get(1) } + .toSet() + val missing = expectedScenarios - passed + check(exitCode == 0 && failed.isEmpty() && missing.isEmpty()) { + "$label failed exitCode=$exitCode failed=$failed missing=$missing" + } +} + +private fun parserMap(): ParserMap = + mapOf(typeNameOf() to { TestData.parseFrom(it) }) + +private fun pythonDir(): File { + var dir = File(System.getProperty("user.dir")).absoluteFile + while (dir.parentFile != null) { + val candidate = File(dir, "../python").canonicalFile + if (File(candidate, "pyproject.toml").isFile) return candidate + val direct = File(dir, "python").canonicalFile + if (File(direct, "pyproject.toml").isFile) return direct + dir = dir.parentFile + } + error("cannot resolve python directory") +} diff --git a/kotlin/crosstest/python_kotlin_client.kt b/kotlin/crosstest/python_kotlin_client.kt new file mode 100644 index 0000000..8df5398 --- /dev/null +++ b/kotlin/crosstest/python_kotlin_client.kt @@ -0,0 +1,196 @@ +@file:JvmName("PythonKotlinClientKt") + +package com.tokilabs.toki_socket.crosstest + +import com.tokilabs.toki_socket.DialTcp +import com.tokilabs.toki_socket.DialWs +import com.tokilabs.toki_socket.ParserMap +import com.tokilabs.toki_socket.WsClient +import com.tokilabs.toki_socket.TcpClient +import com.tokilabs.toki_socket.addListenerTyped +import com.tokilabs.toki_socket.sendRequestTyped +import com.tokilabs.toki_socket.typeNameOf +import com.tokilabs.toki_socket.packets.TestData +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.async +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.delay +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import kotlin.system.exitProcess + +private const val HOST = "127.0.0.1" +private const val WS_PATH = "/" +private const val CONNECT_WINDOW_MS = 3_000L +private const val REQUEST_WINDOW_MS = 2_000L + +private interface ClientHandle { + suspend fun send(data: TestData) + suspend fun close() + val communicator: com.tokilabs.toki_socket.Communicator +} + +private class TcpHandle( + private val client: TcpClient, +) : ClientHandle { + override val communicator = client.communicator + override suspend fun send(data: TestData) { client.send(data) } + override suspend fun close() { client.close() } +} + +private class WsHandle( + private val client: WsClient, +) : ClientHandle { + override val communicator = client.communicator + override suspend fun send(data: TestData) { client.send(data) } + override suspend fun close() { client.close() } +} + +fun main(args: Array) = runBlocking { + val mode = argValue(args, "mode") ?: "tcp" + val phase = argValue(args, "phase") ?: "send-push" + val port = argValue(args, "port")?.toIntOrNull() + + println("INFO typeName kotlin=${typeNameOf()}") + + if (port == null) { + fail("setup", "port is required") + exitProcess(1) + } + + val client = try { + connectWithRetry(mode, port) + } catch (error: Throwable) { + fail("setup", error.message ?: error.toString()) + exitProcess(1) + } + + val ok: Boolean = try { + when (phase) { + "send-push" -> runSendPush(client) + "requests" -> runRequests(client) + else -> { + fail("setup", "unknown phase $phase") + false + } + } + } finally { + client.close() + } + + if (!ok) exitProcess(1) +} + +private suspend fun connectWithRetry(mode: String, port: Int): ClientHandle { + val deadline = System.nanoTime() + CONNECT_WINDOW_MS * 1_000_000L + var lastError: Throwable? = null + while (System.nanoTime() < deadline) { + try { + return when (mode) { + "tcp" -> TcpHandle(DialTcp(HOST, port, 0, 0, parserMap())) + "ws" -> WsHandle(DialWs(HOST, port, WS_PATH, 0, 0, parserMap())) + else -> error("unknown mode $mode") + } + } catch (error: Throwable) { + lastError = error + delay(100) + } + } + error("connect $mode:$port timed out: $lastError") +} + +private suspend fun runSendPush(client: ClientHandle): Boolean { + val push = CompletableDeferred() + addListenerTyped(client.communicator) { + push.complete(it) + } + + return try { + client.send( + TestData.newBuilder() + .setIndex(101) + .setMessage("fire from kotlin client") + .build(), + ) + pass("1", "fire-and-forget sent") + + val msg = withTimeout(REQUEST_WINDOW_MS) { push.await() } + if (msg.index != 200 || msg.message != "push from python server") { + fail("2", "unexpected push index=${msg.index} message=${msg.message}") + false + } else { + pass("2", "received push from python server") + true + } + } catch (error: Throwable) { + fail("2", error.message ?: error.toString()) + false + } +} + +private suspend fun runRequests(client: ClientHandle): Boolean = + try { + val single = sendRequestTyped( + client.communicator, + TestData.newBuilder() + .setIndex(21) + .setMessage("single request from kotlin") + .build(), + REQUEST_WINDOW_MS, + ) + if (single.index != 42 || single.message != "echo: single request from kotlin") { + fail("3", "unexpected response index=${single.index} message=${single.message}") + false + } else { + pass("3", "single request response matched") + runConcurrentRequests(client) + } + } catch (error: Throwable) { + fail("3", error.message ?: error.toString()) + false + } + +private suspend fun runConcurrentRequests(client: ClientHandle): Boolean = + coroutineScope { + val jobs = (0 until 5).map { i -> + async { + val index = 30 + i + val message = "multi request $i from kotlin" + val res = sendRequestTyped( + client.communicator, + TestData.newBuilder() + .setIndex(index) + .setMessage(message) + .build(), + REQUEST_WINDOW_MS, + ) + check(res.index == index * 2 && res.message == "echo: $message") { + "request $i got index=${res.index} message=${res.message}" + } + } + } + try { + jobs.forEach { it.await() } + pass("4", "concurrent request responses matched") + true + } catch (error: Throwable) { + fail("4", error.message ?: error.toString()) + false + } + } + +private fun parserMap(): ParserMap = + mapOf(typeNameOf() to { TestData.parseFrom(it) }) + +private fun argValue(args: Array, name: String): String? { + val prefix = "--$name=" + return args.firstOrNull { it.startsWith(prefix) }?.substring(prefix.length) +} + +private fun pass(scenario: String, detail: String) { + println("PASS scenario=$scenario detail=$detail") +} + +private fun fail(scenario: String, error: String) { + println("FAIL scenario=$scenario error=$error") +} diff --git a/python/crosstest/dart_python_client.py b/python/crosstest/dart_python_client.py new file mode 100644 index 0000000..9b9d96c --- /dev/null +++ b/python/crosstest/dart_python_client.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import argparse +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from toki_socket.communicator import type_name_of +from toki_socket.packets.message_common_pb2 import TestData +from toki_socket.tcp_client import connect_tcp +from toki_socket.ws_client import connect_ws + +HOST = "127.0.0.1" +WS_PATH = "/" +CONNECT_WINDOW = 3.0 +REQUEST_WINDOW = 2.0 + + +def parser_map(): + return {type_name_of(TestData): TestData.FromString} + + +async def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--mode", choices=["tcp", "ws"], default="tcp") + parser.add_argument("--port", type=int, required=True) + parser.add_argument("--phase", choices=["send-push", "requests"], default="send-push") + args = parser.parse_args() + + print(f"INFO typeName python={type_name_of(TestData)}") + + try: + client = await dial_with_retry(args.mode, args.port) + except Exception as exc: + fail("setup", str(exc)) + raise SystemExit(1) from exc + + try: + if args.phase == "send-push": + ok = await run_send_push(client) + else: + ok = await run_requests(client) + finally: + await client.close() + + if not ok: + raise SystemExit(1) + + +async def dial_with_retry(mode: str, port: int): + deadline = asyncio.get_running_loop().time() + CONNECT_WINDOW + last_error: Exception | None = None + while asyncio.get_running_loop().time() < deadline: + try: + return await asyncio.wait_for(dial(mode, port), 0.3) + except Exception as exc: + last_error = exc + await asyncio.sleep(0.1) + raise TimeoutError(f"connect {mode}:{port} timed out: {last_error}") + + +async def dial(mode: str, port: int): + if mode == "tcp": + return await connect_tcp(HOST, port, 0, 0, parser_map()) + if mode == "ws": + return await connect_ws(HOST, port, WS_PATH, 0, 0, parser_map()) + raise ValueError(f"unknown mode {mode!r}") + + +async def run_send_push(client) -> bool: + loop = asyncio.get_running_loop() + push = loop.create_future() + client.communicator.add_listener( + type_name_of(TestData), + lambda message: push.set_result(message) if not push.done() else None, + ) + + try: + await client.communicator.send(TestData(index=101, message="fire from python client")) + passed("1", "fire-and-forget sent") + + message = await asyncio.wait_for(push, REQUEST_WINDOW) + if message.index != 200 or message.message != "push from dart server": + fail("2", f"unexpected push index={message.index} message={message.message!r}") + return False + passed("2", "received push from dart server") + return True + except Exception as exc: + fail("2", str(exc)) + return False + + +async def run_requests(client) -> bool: + try: + single = await client.communicator.send_request( + TestData(index=21, message="single request from python"), + TestData, + timeout=REQUEST_WINDOW, + ) + if single.index != 42 or single.message != "echo: single request from python": + fail("3", f"unexpected response index={single.index} message={single.message!r}") + return False + passed("3", "single request response matched") + except Exception as exc: + fail("3", str(exc)) + return False + + async def request_one(i: int) -> None: + index = 30 + i + message = f"multi request {i} from python" + res = await client.communicator.send_request( + TestData(index=index, message=message), + TestData, + timeout=REQUEST_WINDOW, + ) + if res.index != index * 2 or res.message != "echo: " + message: + raise AssertionError(f"request {i} got index={res.index} message={res.message!r}") + + try: + await asyncio.gather(*(request_one(i) for i in range(5))) + passed("4", "concurrent request responses matched") + return True + except Exception as exc: + fail("4", str(exc)) + return False + + +def passed(scenario: str, detail: str) -> None: + print(f"PASS scenario={scenario} detail={detail}") + + +def fail(scenario: str, error: str) -> None: + print(f"FAIL scenario={scenario} error={error}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/crosstest/kotlin_python_client.py b/python/crosstest/kotlin_python_client.py new file mode 100644 index 0000000..21ec99c --- /dev/null +++ b/python/crosstest/kotlin_python_client.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import argparse +import asyncio +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from toki_socket.communicator import type_name_of +from toki_socket.packets.message_common_pb2 import TestData +from toki_socket.tcp_client import connect_tcp +from toki_socket.ws_client import connect_ws + +HOST = "127.0.0.1" +WS_PATH = "/" +CONNECT_WINDOW = 3.0 +REQUEST_WINDOW = 2.0 + + +def parser_map(): + return {type_name_of(TestData): TestData.FromString} + + +async def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--mode", choices=["tcp", "ws"], default="tcp") + parser.add_argument("--port", type=int, required=True) + parser.add_argument("--phase", choices=["send-push", "requests"], default="send-push") + args = parser.parse_args() + + print(f"INFO typeName python={type_name_of(TestData)}") + + try: + client = await dial_with_retry(args.mode, args.port) + except Exception as exc: + fail("setup", str(exc)) + raise SystemExit(1) from exc + + try: + if args.phase == "send-push": + ok = await run_send_push(client) + else: + ok = await run_requests(client) + finally: + await client.close() + + if not ok: + raise SystemExit(1) + + +async def dial_with_retry(mode: str, port: int): + deadline = asyncio.get_running_loop().time() + CONNECT_WINDOW + last_error: Exception | None = None + while asyncio.get_running_loop().time() < deadline: + try: + return await asyncio.wait_for(dial(mode, port), 0.3) + except Exception as exc: + last_error = exc + await asyncio.sleep(0.1) + raise TimeoutError(f"connect {mode}:{port} timed out: {last_error}") + + +async def dial(mode: str, port: int): + if mode == "tcp": + return await connect_tcp(HOST, port, 0, 0, parser_map()) + if mode == "ws": + return await connect_ws(HOST, port, WS_PATH, 0, 0, parser_map()) + raise ValueError(f"unknown mode {mode!r}") + + +async def run_send_push(client) -> bool: + loop = asyncio.get_running_loop() + push = loop.create_future() + client.communicator.add_listener( + type_name_of(TestData), + lambda message: push.set_result(message) if not push.done() else None, + ) + + try: + await client.communicator.send(TestData(index=101, message="fire from python client")) + passed("1", "fire-and-forget sent") + + message = await asyncio.wait_for(push, REQUEST_WINDOW) + if message.index != 200 or message.message != "push from kotlin server": + fail("2", f"unexpected push index={message.index} message={message.message!r}") + return False + passed("2", "received push from kotlin server") + return True + except Exception as exc: + fail("2", str(exc)) + return False + + +async def run_requests(client) -> bool: + try: + single = await client.communicator.send_request( + TestData(index=21, message="single request from python"), + TestData, + timeout=REQUEST_WINDOW, + ) + if single.index != 42 or single.message != "echo: single request from python": + fail("3", f"unexpected response index={single.index} message={single.message!r}") + return False + passed("3", "single request response matched") + except Exception as exc: + fail("3", str(exc)) + return False + + async def request_one(i: int) -> None: + index = 30 + i + message = f"multi request {i} from python" + res = await client.communicator.send_request( + TestData(index=index, message=message), + TestData, + timeout=REQUEST_WINDOW, + ) + if res.index != index * 2 or res.message != "echo: " + message: + raise AssertionError(f"request {i} got index={res.index} message={res.message!r}") + + try: + await asyncio.gather(*(request_one(i) for i in range(5))) + passed("4", "concurrent request responses matched") + return True + except Exception as exc: + fail("4", str(exc)) + return False + + +def passed(scenario: str, detail: str) -> None: + print(f"PASS scenario={scenario} detail={detail}") + + +def fail(scenario: str, error: str) -> None: + print(f"FAIL scenario={scenario} error={error}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/crosstest/python_dart.py b/python/crosstest/python_dart.py new file mode 100644 index 0000000..be74a96 --- /dev/null +++ b/python/crosstest/python_dart.py @@ -0,0 +1,178 @@ +from __future__ import annotations + +import asyncio +import re +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from toki_socket.communicator import type_name_of +from toki_socket.packets.message_common_pb2 import TestData +from toki_socket.tcp_server import TcpServer +from toki_socket.ws_server import WsServer + +HOST = "127.0.0.1" +PYTHON_DART_TCP_PORT = 29494 +PYTHON_DART_WS_PORT = 29496 +WS_PATH = "/" +PROCESS_TIMEOUT = 20.0 +SERVER_OBSERVATION_WINDOW = 0.2 + + +def parser_map(): + return {type_name_of(TestData): TestData.FromString} + + +async def main() -> None: + print(f"INFO typeName python={type_name_of(TestData)}") + try: + await run_tcp_send_push() + await asyncio.sleep(0.15) + await run_tcp_requests() + await asyncio.sleep(0.15) + await run_ws_send_push() + await asyncio.sleep(0.15) + await run_ws_requests() + except Exception as exc: + print(f"FAIL crosstest error={exc}", file=sys.stderr) + raise SystemExit(1) from exc + print("PASS all python-server/dart-client crosstests passed") + + +async def run_tcp_send_push() -> None: + received = asyncio.get_running_loop().create_future() + server = TcpServer(HOST, PYTHON_DART_TCP_PORT, interval_sec=0, wait_sec=0, parser_map=parser_map()) + + def on_connected(client): + def on_message(data): + print(f"SERVER_RECEIVED index={data.index} message={data.message}") + valid = data.index == 101 and data.message == "fire from dart client" + if not received.done(): + received.set_result(valid) + if valid: + asyncio.create_task( + client.communicator.send(TestData(index=200, message="push from python server")) + ) + + client.communicator.add_listener(type_name_of(TestData), on_message) + + server.on_client_connected = on_connected + await server.start() + try: + await run_dart_client("tcp", PYTHON_DART_TCP_PORT, "send-push", {"1", "2"}) + ok = await asyncio.wait_for(received, SERVER_OBSERVATION_WINDOW) + if not ok: + raise RuntimeError("TCP send-push server received unexpected data") + finally: + await server.stop() + + +async def run_tcp_requests() -> None: + server = TcpServer(HOST, PYTHON_DART_TCP_PORT, interval_sec=0, wait_sec=0, parser_map=parser_map()) + server.on_client_connected = lambda client: client.communicator.add_request_listener( + type_name_of(TestData), + lambda req: TestData(index=req.index * 2, message="echo: " + req.message), + ) + await server.start() + try: + await run_dart_client("tcp", PYTHON_DART_TCP_PORT, "requests", {"3", "4"}) + finally: + await server.stop() + + +async def run_ws_send_push() -> None: + received = asyncio.get_running_loop().create_future() + server = WsServer(HOST, PYTHON_DART_WS_PORT, WS_PATH, interval_sec=0, wait_sec=0, parser_map=parser_map()) + + def on_connected(client): + def on_message(data): + print(f"SERVER_RECEIVED index={data.index} message={data.message}") + valid = data.index == 101 and data.message == "fire from dart client" + if not received.done(): + received.set_result(valid) + if valid: + asyncio.create_task( + client.communicator.send(TestData(index=200, message="push from python server")) + ) + + client.communicator.add_listener(type_name_of(TestData), on_message) + + server.on_client_connected = on_connected + await server.start() + try: + await run_dart_client("ws", PYTHON_DART_WS_PORT, "send-push", {"1", "2"}) + ok = await asyncio.wait_for(received, SERVER_OBSERVATION_WINDOW) + if not ok: + raise RuntimeError("WS send-push server received unexpected data") + finally: + await server.stop() + + +async def run_ws_requests() -> None: + server = WsServer(HOST, PYTHON_DART_WS_PORT, WS_PATH, interval_sec=0, wait_sec=0, parser_map=parser_map()) + server.on_client_connected = lambda client: client.communicator.add_request_listener( + type_name_of(TestData), + lambda req: TestData(index=req.index * 2, message="echo: " + req.message), + ) + await server.start() + try: + await run_dart_client("ws", PYTHON_DART_WS_PORT, "requests", {"3", "4"}) + finally: + await server.stop() + + +async def run_dart_client(mode: str, port: int, phase: str, expected: set[str]) -> None: + dart_dir = dart_package_dir() + process = await asyncio.create_subprocess_exec( + "dart", + "run", + "crosstest/python_dart_client.dart", + f"--mode={mode}", + f"--port={port}", + f"--phase={phase}", + cwd=dart_dir, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), PROCESS_TIMEOUT) + except TimeoutError as exc: + process.kill() + await process.wait() + raise TimeoutError(f"dart client {mode}/{phase} timed out") from exc + + result_lines: list[str] = [] + for line in stdout.decode().splitlines(): + print(line) + if line.startswith(("PASS ", "FAIL ")): + result_lines.append(line) + for line in stderr.decode().splitlines(): + print(line, file=sys.stderr) + + validate_result_lines(f"dart-client {mode}/{phase}", process.returncode, result_lines, expected) + + +def validate_result_lines(label: str, return_code: int | None, lines: list[str], expected: set[str]) -> None: + failed = [line for line in lines if line.startswith("FAIL ")] + passed: set[str] = set() + for line in lines: + if line.startswith("PASS "): + match = re.search(r"scenario=([^ ]+)", line) + if match: + passed.add(match.group(1)) + missing = expected - passed + if return_code or failed or missing: + raise RuntimeError(f"{label} failed returnCode={return_code} failed={failed} missing={sorted(missing)}") + + +def dart_package_dir() -> Path: + repo_root = Path(__file__).resolve().parents[2] + candidate = repo_root / "dart" + if (candidate / "pubspec.yaml").is_file(): + return candidate + raise RuntimeError(f"cannot resolve dart package directory from {candidate}") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/python/crosstest/python_kotlin.py b/python/crosstest/python_kotlin.py new file mode 100644 index 0000000..20801d5 --- /dev/null +++ b/python/crosstest/python_kotlin.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +import asyncio +import re +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from toki_socket.communicator import type_name_of +from toki_socket.packets.message_common_pb2 import TestData +from toki_socket.tcp_server import TcpServer +from toki_socket.ws_server import WsServer + +HOST = "127.0.0.1" +PYTHON_KOTLIN_TCP_PORT = 29498 +PYTHON_KOTLIN_WS_PORT = 29500 +WS_PATH = "/" +PROCESS_TIMEOUT = 20.0 +SERVER_OBSERVATION_WINDOW = 0.2 + + +def parser_map(): + return {type_name_of(TestData): TestData.FromString} + + +async def main() -> None: + print(f"INFO typeName python={type_name_of(TestData)}") + try: + await run_tcp_send_push() + await asyncio.sleep(0.15) + await run_tcp_requests() + await asyncio.sleep(0.15) + await run_ws_send_push() + await asyncio.sleep(0.15) + await run_ws_requests() + except Exception as exc: + print(f"FAIL crosstest error={exc}", file=sys.stderr) + raise SystemExit(1) from exc + print("PASS all python-server/kotlin-client crosstests passed") + + +async def run_tcp_send_push() -> None: + received = asyncio.get_running_loop().create_future() + server = TcpServer(HOST, PYTHON_KOTLIN_TCP_PORT, interval_sec=0, wait_sec=0, parser_map=parser_map()) + + def on_connected(client): + def on_message(data): + print(f"SERVER_RECEIVED index={data.index} message={data.message}") + valid = data.index == 101 and data.message == "fire from kotlin client" + if not received.done(): + received.set_result(valid) + if valid: + asyncio.create_task( + client.communicator.send(TestData(index=200, message="push from python server")) + ) + + client.communicator.add_listener(type_name_of(TestData), on_message) + + server.on_client_connected = on_connected + await server.start() + try: + await run_kotlin_client("tcp", PYTHON_KOTLIN_TCP_PORT, "send-push", {"1", "2"}) + ok = await asyncio.wait_for(received, SERVER_OBSERVATION_WINDOW) + if not ok: + raise RuntimeError("TCP send-push server received unexpected data") + finally: + await server.stop() + + +async def run_tcp_requests() -> None: + server = TcpServer(HOST, PYTHON_KOTLIN_TCP_PORT, interval_sec=0, wait_sec=0, parser_map=parser_map()) + server.on_client_connected = lambda client: client.communicator.add_request_listener( + type_name_of(TestData), + lambda req: TestData(index=req.index * 2, message="echo: " + req.message), + ) + await server.start() + try: + await run_kotlin_client("tcp", PYTHON_KOTLIN_TCP_PORT, "requests", {"3", "4"}) + finally: + await server.stop() + + +async def run_ws_send_push() -> None: + received = asyncio.get_running_loop().create_future() + server = WsServer(HOST, PYTHON_KOTLIN_WS_PORT, WS_PATH, interval_sec=0, wait_sec=0, parser_map=parser_map()) + + def on_connected(client): + def on_message(data): + print(f"SERVER_RECEIVED index={data.index} message={data.message}") + valid = data.index == 101 and data.message == "fire from kotlin client" + if not received.done(): + received.set_result(valid) + if valid: + asyncio.create_task( + client.communicator.send(TestData(index=200, message="push from python server")) + ) + + client.communicator.add_listener(type_name_of(TestData), on_message) + + server.on_client_connected = on_connected + await server.start() + try: + await run_kotlin_client("ws", PYTHON_KOTLIN_WS_PORT, "send-push", {"1", "2"}) + ok = await asyncio.wait_for(received, SERVER_OBSERVATION_WINDOW) + if not ok: + raise RuntimeError("WS send-push server received unexpected data") + finally: + await server.stop() + + +async def run_ws_requests() -> None: + server = WsServer(HOST, PYTHON_KOTLIN_WS_PORT, WS_PATH, interval_sec=0, wait_sec=0, parser_map=parser_map()) + server.on_client_connected = lambda client: client.communicator.add_request_listener( + type_name_of(TestData), + lambda req: TestData(index=req.index * 2, message="echo: " + req.message), + ) + await server.start() + try: + await run_kotlin_client("ws", PYTHON_KOTLIN_WS_PORT, "requests", {"3", "4"}) + finally: + await server.stop() + + +async def run_kotlin_client(mode: str, port: int, phase: str, expected: set[str]) -> None: + kotlin_dir = kotlin_package_dir() + process = await asyncio.create_subprocess_exec( + "./gradlew", + "run", + f"-PmainClass=com.tokilabs.toki_socket.crosstest.PythonKotlinClientKt", + f"--args=--mode={mode} --port={port} --phase={phase}", + cwd=kotlin_dir, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + try: + stdout, stderr = await asyncio.wait_for(process.communicate(), PROCESS_TIMEOUT) + except TimeoutError as exc: + process.kill() + await process.wait() + raise TimeoutError(f"kotlin client {mode}/{phase} timed out") from exc + + result_lines: list[str] = [] + for line in stdout.decode().splitlines(): + print(line) + if line.startswith(("PASS ", "FAIL ")): + result_lines.append(line) + for line in stderr.decode().splitlines(): + print(line, file=sys.stderr) + + validate_result_lines(f"kotlin-client {mode}/{phase}", process.returncode, result_lines, expected) + + +def validate_result_lines(label: str, return_code: int | None, lines: list[str], expected: set[str]) -> None: + failed = [line for line in lines if line.startswith("FAIL ")] + passed: set[str] = set() + for line in lines: + if line.startswith("PASS "): + match = re.search(r"scenario=([^ ]+)", line) + if match: + passed.add(match.group(1)) + missing = expected - passed + if return_code or failed or missing: + raise RuntimeError(f"{label} failed returnCode={return_code} failed={failed} missing={sorted(missing)}") + + +def kotlin_package_dir() -> Path: + repo_root = Path(__file__).resolve().parents[2] + candidate = repo_root / "kotlin" + if (candidate / "build.gradle.kts").is_file(): + return candidate + raise RuntimeError(f"cannot resolve kotlin package directory from {candidate}") + + +if __name__ == "__main__": + asyncio.run(main())