generated from mononoki/pyproj
51 lines
2.4 KiB
Python
51 lines
2.4 KiB
Python
"""Проверка настоящего main.py и TCP.run() в отдельном процессе."""
|
||
|
||
import asyncio
|
||
from pathlib import Path
|
||
import subprocess
|
||
import sys
|
||
|
||
|
||
def test_main_process_handshake_and_heartbeat(client_module, tmp_path):
|
||
"""Запускает main.py на порту ОС из другого cwd, проверяет протокол и завершает процесс."""
|
||
async def scenario():
|
||
main = Path(__file__).parents[1] / "src" / "main.py"
|
||
runner = Path(__file__).with_name("entrypoint_runner.py")
|
||
process = await asyncio.create_subprocess_exec(
|
||
sys.executable, "-B", "-u", str(runner), str(main),
|
||
cwd=tmp_path, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE,
|
||
stdin=asyncio.subprocess.PIPE,
|
||
creationflags=subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0,
|
||
)
|
||
client = None
|
||
try:
|
||
line = await asyncio.wait_for(process.stdout.readline(), 4)
|
||
if not line:
|
||
diagnostic = await asyncio.wait_for(process.stderr.read(), 1)
|
||
raise AssertionError("main.py не запустил TCP:\n" + diagnostic.decode(errors="replace"))
|
||
assert line.startswith(b"TCP listening "), line.decode(errors="replace")
|
||
port = int(line.decode().strip().rsplit(":", 1)[1])
|
||
assert 0 < port < 65536
|
||
client = client_module.Client("process-rover", port=port, timeout=2)
|
||
await client.connect()
|
||
await client.telemetry()
|
||
assert await client.heartbeat() == {"command": "heartbeat_answer"}
|
||
finally:
|
||
if client is not None:
|
||
await client.close()
|
||
if process.returncode is None:
|
||
try:
|
||
process.stdin.write(b"stop\n")
|
||
await process.stdin.drain()
|
||
process.stdin.close()
|
||
except (BrokenPipeError, ConnectionResetError):
|
||
pass
|
||
try:
|
||
stdout, stderr = await asyncio.wait_for(process.communicate(), 4)
|
||
except TimeoutError:
|
||
process.kill()
|
||
stdout, stderr = await process.communicate()
|
||
assert process.returncode is not None
|
||
assert process.returncode == 0, stderr.decode(errors="replace")
|
||
asyncio.run(asyncio.wait_for(scenario(), 12))
|