[#62830] server: test: Added postgres-based testing

Signed-off-by: Jan Bylicki <[email protected]>

server/pyproject.toml

@@ -41,6 +41,8 @@ pycodestyle = "^2.10.0"
 httpx = "^0.27.0"
 cryptography = "^42.0.5"
 pytest-asyncio = "^0.23.5.post1"
+pg-temp = "^0.9.1"
+pytest-lazy-fixture = "^0.6.3"
 
 [tool.poetry.group.types.dependencies]
 types-flask = "^1.1.6"

server/tests/conftest.py

@@ -2,6 +2,7 @@
 import os
 import subprocess
 from pathlib import Path
+import pg_temp
 from common import (
     DBPATH,
     SERVER_WAIT_TIMEOUT,
@@ -82,6 +83,38 @@ def process_gunicorn():
     process.kill()
     log_file.close()
 
+@pytest.fixture(scope="function")
+def process_postgres():
+    """Fixture to start the RDFM server with Werkzeug and Postgres"""
+
+    log_file = (Path(__file__).parent / "server.log").open("a")
+    dbname = DBPATH.split('.')[0].replace('-','_')
+    temp_db = pg_temp.TempDB(databases=[dbname])
+    db_url = f"postgresql:///{dbname}?host={temp_db.pg_socket_dir}"
+
+    print("Starting server..")
+    process = subprocess.Popen(
+        [
+            "python3",
+            "-m",
+            "rdfm_mgmt_server",
+            "--no-ssl",
+            "--no-api-auth",
+            "--test-mocks",
+            "--database",
+            db_url
+        ],
+        stdout=log_file,
+        stderr=log_file,
+    )
+    assert wait_for_api(SERVER_WAIT_TIMEOUT, SERVER), "server has started successfully"
+
+    yield process
+
+    print("Shutting down server..")
+    process.kill()
+    log_file.close()
+
 
 @pytest.fixture(scope="session")
 def install_manager_to_venv():

server/tests/test-server-ws.py

@@ -71,8 +71,17 @@ def spawn_shell_on_mock_device():
     except (simple_websocket.ConnectionError) as e:
         pytest.fail(f"connecting to the shell WebSocket failed: {e}")
 
-
-def test_ws_connect_device(process, connect_mock_device):
+@pytest.mark.parametrize('proc, conn', [
+    (
+        pytest.lazy_fixture('process'),
+        pytest.lazy_fixture('connect_mock_device')
+    ),
+    (
+        pytest.lazy_fixture('process_postgres'),
+        pytest.lazy_fixture('connect_mock_device')
+    )
+])
+def test_ws_connect_device(proc, conn):
     """ This tests if the device WebSocket connection works properly
 
     The device connects to this endpoint. This is a simple test to see
@@ -83,8 +92,17 @@ def test_ws_connect_device(process, connect_mock_device):
     pass
 
 
-def test_ws_device_capability_report(process,
-                                     connect_mock_device):
+@pytest.mark.parametrize('proc, conn', [
+    (
+        pytest.lazy_fixture('process'),
+        pytest.lazy_fixture('connect_mock_device')
+    ),
+    (
+        pytest.lazy_fixture('process_postgres'),
+        pytest.lazy_fixture('connect_mock_device')
+    )
+])
+def test_ws_device_capability_report(proc, conn):
     """ This tests if device capabilities are considered by the server
 
     We accomplish this by attempting to spawn a shell. The connection should
@@ -100,26 +118,46 @@ def test_ws_device_capability_report(process,
     assert "shell_attach" in client.close_message, "the close message should indicate the missing capability"
 
 
-def test_ws_device_receives_shell_attach_message(process,
-                                                 connect_mock_device_with_shell_capability,
-                                                 spawn_shell_on_mock_device):
+@pytest.mark.parametrize('proc, conn, spawn', [
+    (
+        pytest.lazy_fixture('process'),
+        pytest.lazy_fixture('connect_mock_device_with_shell_capability'),
+        pytest.lazy_fixture('spawn_shell_on_mock_device')
+    ),
+    (
+        pytest.lazy_fixture('process_postgres'),
+        pytest.lazy_fixture('connect_mock_device_with_shell_capability'),
+        pytest.lazy_fixture('spawn_shell_on_mock_device')
+    )
+])
+def test_ws_device_receives_shell_attach_message(proc, conn, spawn):
     """ This tests if the device receives a shell attach message
 
     When a manager spawns a shell, the device should receive a `shell_attach`
     message from the server indicating where it should connect to provide
     the shell contents.
     """
-    msg = receive_message(connect_mock_device_with_shell_capability, MESSAGE_WAIT_TIMEOUT)
+    msg = receive_message(conn, MESSAGE_WAIT_TIMEOUT)
     assert isinstance(msg, DeviceAttachToManager), "the device should have received a shell_attach message"
 
 
-def test_ws_device_attaching_to_shell_session(process,
-                                              connect_mock_device_with_shell_capability,
-                                              spawn_shell_on_mock_device):
+@pytest.mark.parametrize('proc, conn, spawn', [
+    (
+        pytest.lazy_fixture('process'),
+        pytest.lazy_fixture('connect_mock_device_with_shell_capability'),
+        pytest.lazy_fixture('spawn_shell_on_mock_device')
+    ),
+    (
+        pytest.lazy_fixture('process_postgres'),
+        pytest.lazy_fixture('connect_mock_device_with_shell_capability'),
+        pytest.lazy_fixture('spawn_shell_on_mock_device')
+    )
+])
+def test_ws_device_attaching_to_shell_session(proc, conn, spawn):
     """ This tests if the device can attach to the shell session
         indicated in the `shell_attach` message.
     """
-    msg: DeviceAttachToManager = receive_message(connect_mock_device_with_shell_capability, MESSAGE_WAIT_TIMEOUT)
+    msg: DeviceAttachToManager = receive_message(conn, MESSAGE_WAIT_TIMEOUT)
     try:
         device_client = simple_websocket.Client.connect(device_attach_shell_ws(FAKE_DEVICE_MAC, msg.uuid), headers={
             "Authorization": f"Bearer token={create_fake_device_token()}",
@@ -128,19 +166,29 @@ def test_ws_device_attaching_to_shell_session(process,
         pytest.fail(f"device failed to attach to the shell session: {e}")
 
 
-def test_ws_shell_bidirectional_communication(process,
-                                              connect_mock_device_with_shell_capability,
-                                              spawn_shell_on_mock_device):
+@pytest.mark.parametrize('proc, conn, spawn', [
+    (
+        pytest.lazy_fixture('process'),
+        pytest.lazy_fixture('connect_mock_device_with_shell_capability'),
+        pytest.lazy_fixture('spawn_shell_on_mock_device')
+    ),
+    (
+        pytest.lazy_fixture('process_postgres'),
+        pytest.lazy_fixture('connect_mock_device_with_shell_capability'),
+        pytest.lazy_fixture('spawn_shell_on_mock_device')
+    )
+])
+def test_ws_shell_bidirectional_communication(proc, conn, spawn):
     """ This tests if data is transferred between the manager and device
         in a established shell session.
     """
 
-    msg: DeviceAttachToManager = receive_message(connect_mock_device_with_shell_capability, MESSAGE_WAIT_TIMEOUT)
+    msg: DeviceAttachToManager = receive_message(conn, MESSAGE_WAIT_TIMEOUT)
     device_client = simple_websocket.Client.connect(device_attach_shell_ws(FAKE_DEVICE_MAC, msg.uuid), headers={
         "Authorization": f"Bearer token={create_fake_device_token()}",
     })
 
-    manager: simple_websocket.Client = spawn_shell_on_mock_device
+    manager: simple_websocket.Client = spawn
     TEST_MESSAGE = b'\x1b[1;31mRDFM TEST SHELL MESSAGE\033[m'
 
     device_client.send(TEST_MESSAGE)
@@ -152,7 +200,11 @@ def test_ws_shell_bidirectional_communication(process,
     assert received == TEST_MESSAGE, "manager to device data transfer should work"
 
 
-def test_ws_shell_spawn_on_nonexistent_device(process):
+@pytest.mark.parametrize('proc', [
+    (pytest.lazy_fixture('process')),
+    (pytest.lazy_fixture('process_postgres'))
+])
+def test_ws_shell_spawn_on_nonexistent_device(proc):
     """ This tests if trying to spawn a shell on a nonexistent device fails properly.
     """
     with pytest.raises(simple_websocket.ConnectionClosed):