124 lines
3.9 KiB
Python
124 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
import time
|
|
import unittest
|
|
|
|
from client_session import ClientSession
|
|
from server_core import FalloutTogetherServer, MAX_NPCS_PER_PACKET
|
|
|
|
|
|
class NpcProtocolTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.host_server, self.host_peer = socket.socketpair()
|
|
self.client_server, self.client_peer = socket.socketpair()
|
|
self.server = FalloutTogetherServer()
|
|
self.host = ClientSession(self.host_server, ("local", 1), 1, time.time())
|
|
self.client = ClientSession(self.client_server, ("local", 2), 2, time.time())
|
|
self.host.mark_gameplay_active(1)
|
|
self.client.mark_gameplay_active(1)
|
|
self.server._init_rate_state(self.host, time.monotonic())
|
|
self.server._init_rate_state(self.client, time.monotonic())
|
|
self.server._clients = {
|
|
self.host_server: self.host,
|
|
self.client_server: self.client,
|
|
}
|
|
self.server._world_state_host_player_id = 1
|
|
|
|
def tearDown(self) -> None:
|
|
for connection in (
|
|
self.host_server,
|
|
self.host_peer,
|
|
self.client_server,
|
|
self.client_peer,
|
|
):
|
|
connection.close()
|
|
|
|
def _npc(self, index: int) -> dict:
|
|
return {
|
|
"npcId": index,
|
|
"sourceFormId": "0001A4D7",
|
|
"x": float(index),
|
|
"y": 2.0,
|
|
"z": 3.0,
|
|
"angleZ": 0.5,
|
|
"cellId": "0000003C",
|
|
}
|
|
|
|
def test_host_npc_state_is_authoritative_and_broadcast(self) -> None:
|
|
self.server._handle_line(
|
|
self.host,
|
|
json.dumps(
|
|
{
|
|
"type": "npcState",
|
|
"playerId": 999,
|
|
"npcs": [self._npc(1)],
|
|
}
|
|
),
|
|
)
|
|
|
|
relayed = json.loads(self.client_peer.recv(4096))
|
|
self.assertEqual(relayed["playerId"], 1)
|
|
self.assertTrue(relayed["fullReplace"])
|
|
self.assertEqual(relayed["npcs"][0]["npcId"], 1)
|
|
self.assertIsNotNone(self.server._last_npc_state)
|
|
|
|
def test_non_host_npc_state_is_dropped(self) -> None:
|
|
self.server._handle_line(
|
|
self.client,
|
|
'{"type":"npcState","npcs":[]}',
|
|
)
|
|
|
|
self.host_peer.setblocking(False)
|
|
with self.assertRaises(BlockingIOError):
|
|
self.host_peer.recv(1)
|
|
|
|
def test_late_joiner_receives_cached_npc_state(self) -> None:
|
|
self.server._last_npc_state = {
|
|
"type": "npcState",
|
|
"playerId": 1,
|
|
"npcs": [],
|
|
"fullReplace": True,
|
|
"serverTime": time.time(),
|
|
}
|
|
|
|
self.server._send_existing_npc_state_to_client(self.client)
|
|
relayed = json.loads(self.client_peer.recv(4096))
|
|
self.assertEqual(relayed["type"], "npcState")
|
|
self.assertEqual(relayed["playerId"], 1)
|
|
|
|
def test_host_reassignment_clears_cached_npc_state(self) -> None:
|
|
self.server._last_npc_state = {
|
|
"type": "npcState",
|
|
"playerId": 1,
|
|
"npcs": [{"npcId": 1}],
|
|
}
|
|
self.server._clients.pop(self.host_server)
|
|
|
|
self.server._reassign_world_state_host()
|
|
|
|
self.assertEqual(self.server._world_state_host_player_id, 2)
|
|
self.assertIsNone(self.server._last_npc_state)
|
|
reassignment = json.loads(self.client_peer.recv(4096))
|
|
self.assertEqual(reassignment["worldStateHostPlayerId"], 2)
|
|
|
|
def test_oversized_npc_batch_is_rejected(self) -> None:
|
|
self.server._handle_line(
|
|
self.host,
|
|
json.dumps(
|
|
{
|
|
"type": "npcState",
|
|
"npcs": [self._npc(index + 1) for index in range(MAX_NPCS_PER_PACKET + 1)],
|
|
}
|
|
),
|
|
)
|
|
|
|
self.client_peer.setblocking(False)
|
|
with self.assertRaises(BlockingIOError):
|
|
self.client_peer.recv(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|