Adjust end to end tests to support websocket infrastructure (#987)

* Rename integration_tests to e2e_tests

* Add websocket support for e2e tests

* Fix flaky test

DEVELOPMENT.md

@@ -124,10 +124,10 @@ For further information on available command line arguments run `pytest --help`
 or see the official
 [pytest documentation](https://docs.pytest.org/en/latest/usage.html).
 
-There are also some integration tests which simulate real traffic to the test
+There are also some end to end tests which simulate real traffic to the test
 server.
 ```
-$ pipenv run integration
+$ pipenv run e2e
 ```
 
 Some of them may fail depending on the configuration deployed on the test

Pipfile

@@ -1,7 +1,7 @@
 [scripts]
 devserver = "python main.py --configuration-file dev-config.yml"
 tests = "py.test --doctest-modules --doctest-continue-on-failure --cov-report=term-missing --cov-branch --cov=server --mysql_database=faf -o testpaths=tests -m 'not rabbitmq'"
-integration = "py.test -o testpaths=integration_tests"
+e2e = "py.test -o testpaths=e2e_tests"
 vulture = "vulture main.py server/ --sort-by-size"
 doc = "pdoc3 --html --force server"
 
@@ -36,6 +36,7 @@ pytest-asyncio = "*"
 pytest-cov = "*"
 pytest-mock = "*"
 vulture = "*"
+websockets = "*"
 
 [requires]
 python_version = "3.10"

integration_tests/conftest.py → e2e_tests/conftest.py

@@ -16,9 +16,9 @@ class Manager():
         def __init__(self):
             self.clients = []
 
-        async def add_client(self, host="test.faforever.com", port=8001):
+        async def add_client(self, uri="wss://ws.faforever.xyz"):
             client = FAFClient()
-            await client.connect(host, port)
+            await client.ws_connect(uri)
             self.clients.append(client)
             return client
 

integration_tests/fafclient.py → e2e_tests/fafclient.py

@@ -2,12 +2,16 @@
 import subprocess
 from hashlib import sha256
 
-from server.protocol import QDataStreamProtocol
+from websockets.client import connect as ws_connect
+
+from server.protocol import QDataStreamProtocol, SimpleJsonProtocol
 from tests.integration_tests.conftest import read_until, read_until_command
 
+from .websocket_protocol import WebsocketProtocol
+
 
-class FAFClient(object):
-    """docstring for FAFClient."""
+class FAFClient:
+    """Simulates a FAF client."""
 
     def __init__(self, user_agent="faf-client", version="1.0.0-dev"):
         self.proto = None
@@ -28,11 +32,15 @@ async def close(self):
 
         await self.proto.close()
 
-    async def connect(self, host, port):
-        self.proto = QDataStreamProtocol(
+    async def connect(self, host, port, protocol_class=QDataStreamProtocol):
+        self.proto = protocol_class(
             *(await asyncio.open_connection(host, port))
         )
 
+    async def ws_connect(self, uri, protocol_class=SimpleJsonProtocol):
+        websocket = await ws_connect(uri, open_timeout=60)
+        self.proto = WebsocketProtocol(websocket, protocol_class)
+
     async def send_message(self, message):
         """Send a message to the server"""
         if not self.is_connected():
@@ -50,8 +58,13 @@ async def read_until(self, predicate, timeout=5):
             timeout=timeout
         )
 
-    async def read_until_command(self, command, timeout=5):
-        return await read_until_command(self.proto, command, timeout=timeout)
+    async def read_until_command(self, command, timeout=5, **kwargs):
+        return await read_until_command(
+            self.proto,
+            command,
+            timeout=timeout,
+            **kwargs,
+        )
 
     async def read_until_game_launch(self, uid):
         return await self.read_until(
@@ -116,12 +129,14 @@ async def host_game(self, **kwargs):
         game_id = int(msg["uid"])
 
         await self.open_fa()
+        await self.read_until_command("HostGame", target="game")
         return game_id
 
     async def join_game(self, game_id, **kwargs):
         await self.send_message({
             "command": "game_join",
-            "uid": game_id
+            "uid": game_id,
+            **kwargs
         })
         await self.read_until_command("game_launch")
 

integration_tests/test_game.py → e2e_tests/test_game.py

@@ -103,8 +103,10 @@ async def test_custom_game_1v1_game_stats(client_factory, json_stats_1v1):
     for client in (client1, client2):
         await client.send_gpg_command("GameState", "Ended")
 
-    await client1.read_until_command("updated_achievements", timeout=10)
-    await client2.read_until_command("updated_achievements", timeout=2)
+    # The client no longer gets a `updated_achievements` notification, but we
+    # keep the test in case it could generate an exception on the server side.
+    await client1.get_player_ratings("test", "test2", timeout=10)
+    await client2.get_player_ratings("test", "test2", timeout=2)
 
 
 async def test_custom_game_1v1_extra_gameresults(client_factory):

integration_tests/test_matchmaking.py → e2e_tests/test_matchmaking.py

@@ -3,6 +3,9 @@
 
 from .test_game import simulate_result_reports
 
+# NOTE: Tests will cause matchmaker violations, so after running them a few
+# times, they may start to fail.
+
 
 async def test_ladder_1v1_match(client_factory):
     """More or less the same as the regression test version"""

e2e_tests/websocket_protocol.py

@@ -0,0 +1,87 @@
+import asyncio
+import contextlib
+from unittest import mock
+
+import websockets
+
+from server.protocol import DisconnectedError, Protocol
+
+
+class WebsocketProtocol:
+    def __init__(
+        self,
+        websocket: websockets.client.WebSocketClientProtocol,
+        protocol_class: type[Protocol],
+    ):
+        self.websocket = websocket
+        reader = asyncio.StreamReader()
+        reader.set_transport(asyncio.ReadTransport())
+        self.proto = protocol_class(
+            reader,
+            mock.create_autospec(asyncio.StreamWriter)
+        )
+
+    def is_connected(self) -> bool:
+        """
+        Return whether or not the connection is still alive
+        """
+        return self.websocket.open
+
+    async def read_message(self) -> dict:
+        if self.proto.reader._buffer:
+            # If buffer contains partial data, this await call could hang.
+            return await self.proto.read_message()
+
+        msg = await self.websocket.recv()
+        self.proto.reader.feed_data(msg)
+        # msg should always contain at least 1 complete message.
+        # If it contains partial data, this await call could hang.
+        return await self.proto.read_message()
+
+    async def send_message(self, message: dict) -> None:
+        """
+        Send a single message in the form of a dictionary
+
+        # Errors
+        May raise `DisconnectedError`.
+        """
+        await self.send_raw(self.proto.encode_message(message))
+
+    async def send_messages(self, messages: list[dict]) -> None:
+        """
+        Send multiple messages in the form of a list of dictionaries.
+
+        # Errors
+        May raise `DisconnectedError`.
+        """
+        for message in messages:
+            await self.send_message(message)
+
+    async def send_raw(self, data: bytes) -> None:
+        """
+        Send raw bytes. Should generally not be used.
+
+        # Errors
+        May raise `DisconnectedError`.
+        """
+        try:
+            await self.websocket.send(data)
+        except websockets.exceptions.ConnectionClosedOK:
+            raise DisconnectedError("The websocket connection was closed")
+        except websockets.exceptions.ConnectionClosed as e:
+            raise DisconnectedError("Websocket connection lost!") from e
+
+    def abort(self) -> None:
+        # SelectorTransport only
+        self.websocket.transport.abort()
+
+    async def close(self) -> None:
+        """
+        Close the websocket connection.
+
+        # Errors
+        Never raises. Any exceptions that occur while waiting to close are
+        ignored.
+        """
+        with contextlib.suppress(Exception):
+            await self.websocket.close()

server/protocol/protocol.py

@@ -29,6 +29,14 @@ def encode_message(message: dict) -> bytes:
         """
         pass  # pragma: no cover
 
+    @staticmethod
+    @abstractmethod
+    def decode_message(data: bytes) -> dict:
+        """
+        Decode a message from raw bytes.
+        """
+        pass  # pragma: no cover
+
     def is_connected(self) -> bool:
         """
         Return whether or not the connection is still alive

server/protocol/qdatastream.py

@@ -80,29 +80,15 @@ def encode_message(message: dict) -> bytes:
 
         return QDataStreamProtocol.pack_message(json_encoder.encode(message))
 
-    async def read_message(self):
-        """
-        Read a message from the stream
-
-        # Errors
-        Raises `IncompleteReadError` on malformed stream.
-        """
-        try:
-            length, *_ = struct.unpack("!I", await self.reader.readexactly(4))
-            block = await self.reader.readexactly(length)
-        except IncompleteReadError as e:
-            if self.reader.at_eof() and not e.partial:
-                raise DisconnectedError()
-            # Otherwise reraise
-            raise
-
-        pos, action = self.read_qstring(block)
+    @staticmethod
+    def decode_message(data: bytes) -> dict:
+        _, action = QDataStreamProtocol.read_qstring(data)
         if action in ("PING", "PONG"):
             return {"command": action.lower()}
 
         message = json.loads(action)
         try:
-            for part in self.read_block(block):
+            for part in QDataStreamProtocol.read_block(data):
                 try:
                     message_part = json.loads(part)
                     if part != action:
@@ -115,6 +101,24 @@ async def read_message(self):
             pass
         return message
 
+    async def read_message(self):
+        """
+        Read a message from the stream
+
+        # Errors
+        Raises `IncompleteReadError` on malformed stream.
+        """
+        try:
+            length, *_ = struct.unpack("!I", await self.reader.readexactly(4))
+            block = await self.reader.readexactly(length)
+        except IncompleteReadError as e:
+            if self.reader.at_eof() and not e.partial:
+                raise DisconnectedError()
+            # Otherwise reraise
+            raise
+
+        return QDataStreamProtocol.decode_message(block)
+
 
 PING_MSG = QDataStreamProtocol.pack_message("PING")
 PONG_MSG = QDataStreamProtocol.pack_message("PONG")

server/protocol/simple_json.py

@@ -8,8 +8,12 @@ class SimpleJsonProtocol(Protocol):
     def encode_message(message: dict) -> bytes:
         return (json_encoder.encode(message) + "\n").encode()
 
+    @staticmethod
+    def decode_message(data: bytes) -> dict:
+        return json.loads(data.strip())
+
     async def read_message(self) -> dict:
         line = await self.reader.readline()
         if not line:
             raise DisconnectedError()
-        return json.loads(line.strip())
+        return SimpleJsonProtocol.decode_message(line)

tests/integration_tests/conftest.py

@@ -554,6 +554,10 @@ async def consume(self):
     def encode_message(message: dict) -> bytes:
         raise NotImplementedError("AioQueueProtocol is read-only")
 
+    @staticmethod
+    def decode_message(data: bytes) -> dict:
+        raise NotImplementedError("AioQueueProtocol doesn't user bytes")
+
     async def read_message(self) -> dict:
         return await self.aio_queue.get()