Nexus Agent
57 lines
1.5 KiB
Python
57 lines
1.5 KiB
Python
"""Guacamole protocol encode/decode and internal opcode detection."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from server.infrastructure.guacamole.protocol import (
|
|
GuacdStream,
|
|
encode_instruction,
|
|
is_internal_instruction,
|
|
parse_instruction,
|
|
tunnel_open_instruction,
|
|
)
|
|
|
|
|
|
def test_encode_instruction():
|
|
assert encode_instruction("select", "rdp") == "6.select,3.rdp;"
|
|
|
|
|
|
def test_parse_instruction_roundtrip():
|
|
raw = encode_instruction("connect", "host", "3389").rstrip(";")
|
|
opcode, args = parse_instruction(raw)
|
|
assert opcode == "connect"
|
|
assert args == ["host", "3389"]
|
|
|
|
|
|
def test_tunnel_open_instruction():
|
|
msg = tunnel_open_instruction("abc-uuid")
|
|
assert is_internal_instruction(msg)
|
|
opcode, args = parse_instruction(msg.rstrip(";"))
|
|
assert opcode == ""
|
|
assert args == ["abc-uuid"]
|
|
|
|
|
|
def test_is_internal_ping():
|
|
ping = encode_instruction("", "ping", "12345")
|
|
assert is_internal_instruction(ping)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_guacd_stream_splits_tcp_chunks_on_semicolon():
|
|
class FakeReader:
|
|
def __init__(self, chunks: list[bytes]):
|
|
self._chunks = list(chunks)
|
|
|
|
async def read(self, n: int) -> bytes:
|
|
if not self._chunks:
|
|
return b""
|
|
return self._chunks.pop(0)
|
|
|
|
ins = encode_instruction("sync", "123").encode("utf-8")
|
|
reader = FakeReader([ins[:10], ins[10:]])
|
|
stream = GuacdStream(reader) # type: ignore[arg-type]
|
|
raw = await stream.read_raw_instruction()
|
|
assert raw == ins
|
|
assert raw.endswith(b";")
|