Commonwealth-Online-Server/server/tests/test_runtime_portability.py

from __future__ import annotations

import json
import socket
import threading
import time
from pathlib import Path

import pytest

from admin_server import send_admin_command
from lan_discovery import LanDiscoveryResponder
from server_core import FalloutTogetherServer, get_lan_addresses
from server_service import ServerConfig, ServerService


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind(("127.0.0.1", 0))
        return int(sock.getsockname()[1])


def test_get_lan_addresses_never_returns_wildcard() -> None:
    addresses = get_lan_addresses()
    assert "0.0.0.0" not in addresses
    for address in addresses:
        assert not address.startswith("127.")


def test_discovery_failure_does_not_stop_game_server(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    game_port = _free_port()
    admin_port = _free_port()

    def _fail_start(self: LanDiscoveryResponder) -> None:
        raise OSError("Could not bind LAN discovery to 0.0.0.0:7778. simulated failure")

    monkeypatch.setattr(LanDiscoveryResponder, "start", _fail_start)

    service = ServerService(
        ServerConfig(
            host="127.0.0.1",
            port=game_port,
            admin_port=admin_port,
            bans_path=str(tmp_path / "bans.json"),
        )
    )
    thread = threading.Thread(target=service.serve_forever, daemon=True)
    thread.start()
    deadline = time.time() + 5.0
    while time.time() < deadline and not service.is_running():
        time.sleep(0.05)

    assert service.is_running()
    with socket.create_connection(("127.0.0.1", game_port), timeout=2.0) as conn:
        data = conn.recv(4096)
        assert b"welcome" in data

    service.stop()
    thread.join(timeout=3.0)
    assert not service.is_running()


def test_stop_is_idempotent(tmp_path: Path) -> None:
    game_port = _free_port()
    admin_port = _free_port()
    service = ServerService(
        ServerConfig(
            host="127.0.0.1",
            port=game_port,
            admin_port=admin_port,
            bans_path=str(tmp_path / "bans.json"),
        )
    )
    service.start()
    deadline = time.time() + 5.0
    while time.time() < deadline and not service.is_running():
        time.sleep(0.05)
    assert service.is_running()

    service.stop()
    service.stop()
    assert not service.is_running()


def test_admin_and_game_bind_errors_include_address() -> None:
    occupied = _free_port()
    holder = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        holder.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        holder.bind(("127.0.0.1", occupied))
        holder.listen()

        server = FalloutTogetherServer(host="127.0.0.1", port=occupied)
        with pytest.raises(OSError, match=rf"127\.0\.0\.1:{occupied}"):
            server._prepare_server_socket()
    finally:
        holder.close()


def test_headless_admin_roundtrip(tmp_path: Path) -> None:
    game_port = _free_port()
    admin_port = _free_port()
    service = ServerService(
        ServerConfig(
            host="127.0.0.1",
            port=game_port,
            admin_port=admin_port,
            bans_path=str(tmp_path / "bans.json"),
        )
    )
    thread = threading.Thread(target=service.serve_forever, daemon=True)
    thread.start()
    deadline = time.time() + 5.0
    while time.time() < deadline and not service.is_running():
        time.sleep(0.05)
    assert service.is_running()

    response = send_admin_command({"cmd": "ping"}, port=admin_port)
    assert response.get("ok") is True
    assert response.get("data", {}).get("pong") is True

    service.stop()
    thread.join(timeout=3.0)


def test_lan_discovery_sets_broadcast_option() -> None:
    class DummyServer:
        def get_stats(self):
            return {
                "connectedClients": 0,
                "port": 7777,
                "serverName": "Test",
                "serverDescription": "",
                "maxPlayers": 16,
            }

        def _log(self, message: str, *, level: str = "info") -> None:
            return None

    # Bind an ephemeral discovery port to avoid colliding with a real server.
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind(("127.0.0.1", 0))
    port = int(sock.getsockname()[1])
    sock.close()

    responder = LanDiscoveryResponder(DummyServer(), discovery_port=port)
    responder.start()
    try:
        assert responder._socket is not None
        # SO_BROADCAST should be enabled; querying may return 0/1 depending on OS.
        value = responder._socket.getsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST)
        assert value in (0, 1)
        probe = {
            "type": "discover",
            "protocol": "commonwealth-online",
        }
        with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as client:
            client.settimeout(2.0)
            client.sendto(json.dumps(probe).encode("utf-8"), ("127.0.0.1", port))
            data, _addr = client.recvfrom(2048)
            packet = json.loads(data.decode("utf-8"))
            assert packet["type"] == "discoverResponse"
            assert packet["port"] == 7777
    finally:
        responder.stop()