rsx-modloader/testserver/selftest.py

#!/usr/bin/env python3
"""Self-test for server.py: runs the stand-in server on a spare port and talks
to it with a hand-rolled 1.12.2 client, checking handshake, login and the
play-state packets the game needs to reach a world."""

import socket
import struct
import sys
import threading
import time
import uuid

sys.path.insert(0, r"G:\rsx-mod\testserver")

import server as srv  # noqa: E402


def read_varint(sock):
    value = 0
    for shift in range(0, 35, 7):
        chunk = sock.recv(1)
        if not chunk:
            raise AssertionError("closed")
        byte = chunk[0]
        value |= (byte & 0x7F) << shift
        if not byte & 0x80:
            return value
    raise AssertionError("varint too long")


def read_packet(sock):
    length = read_varint(sock)
    data = b""
    while len(data) < length:
        chunk = sock.recv(length - len(data))
        if not chunk:
            raise AssertionError("closed mid packet")
        data += chunk
    pid_val = 0
    shift = 0
    pos = 0
    while True:
        byte = data[pos]
        pos += 1
        pid_val |= (byte & 0x7F) << shift
        if not byte & 0x80:
            break
        shift += 7
    return pid_val, data[pos:]


def send_packet(sock, pid, payload=b""):
    body = srv.varint(pid) + payload
    sock.sendall(srv.varint(len(body)) + body)


def drain(sock, wanted, budget=8.0):
    seen = {}
    deadline = time.time() + budget
    while time.time() < deadline and not wanted.issubset(seen):
        try:
            pid, payload = read_packet(sock)
        except (socket.timeout, TimeoutError, AssertionError, OSError):
            break
        seen.setdefault(pid, payload)
    return seen


def main():
    port = 25599
    listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    listener.bind(("127.0.0.1", port))
    listener.listen(4)

    def accept_loop():
        try:
            sock, peer = listener.accept()
            srv.handle(srv.Conn(sock, "%s:%s" % peer))
        except Exception:
            pass

    threading.Thread(target=accept_loop, daemon=True).start()

    sock = socket.create_connection(("127.0.0.1", port), timeout=10)
    sock.settimeout(10)
    results = []

    handshake = (srv.varint(srv.PROTOCOL) + srv.mc_string("127.0.0.1")
                 + struct.pack(">H", port) + srv.varint(2))
    send_packet(sock, 0x00, handshake)
    results.append(("handshake sent", True))

    send_packet(sock, 0x00,
                srv.mc_string("TestPilot")
                + srv.mc_string("00000000-0000-0000-0000-000000000000"))
    pid, payload = read_packet(sock)
    cursor = srv.Cursor(payload)
    uuid_txt = cursor.string()
    nick = cursor.string()
    results.append(("login success", pid == 0x02 and nick == "TestPilot"))
    uuid_ok = len(uuid_txt) in (32, 36)
    try:
        uuid.UUID(uuid_txt)
    except (ValueError, AttributeError):
        uuid_ok = False
    results.append(("uuid parseable (%d chars)" % len(uuid_txt), uuid_ok))

    send_packet(sock, 0x03, b"")
    seen = drain(sock, {0x01, 0x08, 0x23})
    results.append(("join game received", 0x01 in seen))
    results.append(("spawn position received", 0x08 in seen))
    results.append(("brand received", 0x23 in seen))

    if 0x01 in seen:
        cur = srv.Cursor(seen[0x01])
        entity = struct.unpack(">i", cur.take(4))[0]
        gamemode = cur.ubyte()
        dimension = struct.unpack(">i", cur.take(4))[0]
        cur.ubyte()
        cur.ubyte()
        level = cur.string()
        results.append(("join game parses (entity=%d gm=%d dim=%d %s)"
                        % (entity, gamemode, dimension, level),
                        level == srv.WORLD_NAME))

    if 0x08 in seen:
        cur = srv.Cursor(seen[0x08])
        x = struct.unpack(">d", cur.take(8))[0]
        y = struct.unpack(">d", cur.take(8))[0]
        z = struct.unpack(">d", cur.take(8))[0]
        cur.take(8)
        cur.ubyte()
        tid = cur.varint()
        results.append(("spawn parses (%.1f/%.1f/%.1f tid=%d)"
                        % (x, y, z, tid), y == 64.0 and tid == 0))

    send_packet(sock, 0x00, srv.varint(0x1234))
    send_packet(sock, 0x03, srv.mc_string("ru_ru") + bytes([8])
                + srv.varint(0) + b"\x01" + bytes([2]) + bytes([127])
                + srv.varint(1))
    send_packet(sock, 0x07,
                struct.pack(">dddff?", 0.5, 64.0, 0.5, 0.0, 0.0, True))
    alive = drain(sock, {0x00}, budget=18.0)
    results.append(("keep alive flow", 0x00 in alive))

    sock.close()
    listener.close()

    print("== self-test ==")
    failed = 0
    for label, ok in results:
        print("   %-40s %s" % (label, "OK" if ok else "FAIL"))
        failed += 0 if ok else 1
    print("   %d check(s) failed" % failed)
    return 1 if failed else 0


if __name__ == "__main__":
    sys.exit(main())