toki
5df325bf96
- Dart Web 브라우저 E2E 테스트 프레임워크 추가 - browser_ws_dart_io_test, browser_ws_kotlin_test 등 브라우저 통합 테스트 추가 - Dart Web 크로스테스트 케이스 추가 (go, kotlin, python, typescript) - TypeScript 브라우저 웹소켓 클라이언트 개선 - Node.js 웹소켓 서버 클라이언트 업데이트 - 테스트 실행 매트릭스 스킬 및 스크립트 업데이트 - README.md 문서 업데이트
101 lines
3.2 KiB
Python
101 lines
3.2 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
from proto_socket.communicator import type_name_of
|
|
from proto_socket.packets.message_common_pb2 import TestData
|
|
from proto_socket.ws_server import WsServer
|
|
|
|
HOST = "127.0.0.1"
|
|
WS_PORT = 29698
|
|
WS_PATH = "/"
|
|
PROCESS_TIMEOUT = 60.0
|
|
SERVER_OBSERVATION_WINDOW = 0.5
|
|
|
|
|
|
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_web_ws()
|
|
except Exception as exc:
|
|
print(f"FAIL crosstest error={exc}", file=sys.stderr)
|
|
raise SystemExit(1) from exc
|
|
print("PASS scenario=send-push")
|
|
print("PASS scenario=request-response")
|
|
print("PASS all python-server/dart-web-client crosstests passed")
|
|
|
|
|
|
async def run_web_ws() -> None:
|
|
received = asyncio.get_running_loop().create_future()
|
|
server = WsServer(HOST, WS_PORT, WS_PATH, interval_sec=0, wait_sec=0, parser_map=parser_map())
|
|
|
|
def on_connected(client):
|
|
def on_request(req):
|
|
print(f"SERVER_RECEIVED index={req.index} message={req.message}")
|
|
if req.index == 101 and req.message == "fire from dart web client":
|
|
if not received.done():
|
|
received.set_result(True)
|
|
asyncio.create_task(
|
|
client.communicator.send(
|
|
TestData(index=200, message="push from python server")
|
|
)
|
|
)
|
|
return TestData(index=req.index * 2, message="echo: " + req.message)
|
|
|
|
client.communicator.add_request_listener(type_name_of(TestData), on_request)
|
|
|
|
server.on_client_connected = on_connected
|
|
await server.start()
|
|
try:
|
|
await run_dart_browser_test("test/browser_ws_python_test.dart")
|
|
ok = await asyncio.wait_for(received, SERVER_OBSERVATION_WINDOW)
|
|
if not ok:
|
|
raise RuntimeError("WS web send-push server received unexpected data")
|
|
finally:
|
|
await server.stop()
|
|
|
|
|
|
async def run_dart_browser_test(test_path: str) -> None:
|
|
dart_dir = dart_package_dir()
|
|
process = await asyncio.create_subprocess_exec(
|
|
"dart",
|
|
"test",
|
|
"-p",
|
|
"chrome",
|
|
test_path,
|
|
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("dart browser test timed out") from exc
|
|
for line in stdout.decode().splitlines():
|
|
print(line)
|
|
for line in stderr.decode().splitlines():
|
|
print(line, file=sys.stderr)
|
|
if process.returncode != 0:
|
|
raise RuntimeError(f"dart browser test failed exitCode={process.returncode}")
|
|
|
|
|
|
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())
|