test(playout): liquidsoap wait for version

playout/libretime_playout/liquidsoap/client/__init__.py

@@ -1,2 +1,2 @@
-from ._client import LiquidsoapClient
+from ._client import LiquidsoapClient, LiquidsoapClientError
 from ._connection import LiquidsoapConnection

playout/tests/liquidsoap/client/client_test.py

@@ -1,4 +1,6 @@
-from libretime_playout.liquidsoap.client import LiquidsoapClient
+import pytest
+
+from libretime_playout.liquidsoap.client import LiquidsoapClient, LiquidsoapClientError
 
 
 def test_liq_client():
@@ -7,3 +9,16 @@ def test_liq_client():
         port=1234,
         timeout=15,
     )
+
+
+def test_liq_client_wait_for_version(liq_client: LiquidsoapClient):
+    assert liq_client.wait_for_version()
+
+
+def test_liq_client_wait_for_version_invalid_host():
+    liq_client = LiquidsoapClient(
+        host="invalid",
+        port=1234,
+    )
+    with pytest.raises(LiquidsoapClientError):
+        liq_client.wait_for_version(timeout=1)

playout/tests/liquidsoap/client/conftest.py

@@ -1,13 +1,15 @@
 import logging
+from contextlib import contextmanager
 from pathlib import Path
 from random import randint
 from subprocess import PIPE, STDOUT, Popen
 from time import sleep
+from typing import Generator, Protocol
 
 import pytest
 from libretime_shared.logging import setup_logger
 
-from libretime_playout.liquidsoap.client import LiquidsoapConnection
+from libretime_playout.liquidsoap.client import LiquidsoapClient, LiquidsoapConnection
 
 logger = logging.getLogger(__name__)
 
@@ -34,24 +36,74 @@ set("server.socket.path", "{socket_path}")
 """
 
 
-@pytest.fixture(
+class LiquidsoapManager(Protocol):
-    name="liq_conn",
+    def generate_entrypoint(self) -> str:
-    scope="session",
+        pass
-    params=["telnet", "socket"],
-)
-def liq_conn_fixture(request, tmp_path_factory):
-    tmp_path: Path = tmp_path_factory.mktemp(__name__)
 
+    def wait_start(self, process: Popen) -> None:
+        pass
+
+    def make_connection(self) -> LiquidsoapConnection:
+        pass
+
+    def make_client(self) -> LiquidsoapClient:
+        pass
+
+
+class LiquidsoapManagerTelnet:
+    def __init__(self) -> None:
+        self.telnet_port = randint(32768, 65535)
+
+    def generate_entrypoint(self) -> str:
+        liq_settings = LIQ_TELNET_SETTINGS.format(telnet_port=self.telnet_port)
+        liq_script = LIQ_SCRIPT.format(settings=liq_settings.strip())
+        return liq_script
+
+    # pylint: disable=unused-argument
+    def wait_start(self, process: Popen) -> None:
+        sleep(2)
+
+    def make_connection(self) -> LiquidsoapConnection:
+        return LiquidsoapConnection(host="localhost", port=self.telnet_port)
+
+    def make_client(self) -> LiquidsoapClient:
+        return LiquidsoapClient(host="localhost", port=self.telnet_port)
+
+
+class LiquidsoapManagerSocket:
+    def __init__(self, tmp_path: Path) -> None:
+        self.socket_path = tmp_path / "main.sock"
+
+    def generate_entrypoint(self) -> str:
+        liq_settings = LIQ_SOCKET_SETTINGS.format(socket_path=self.socket_path)
+        liq_script = LIQ_SCRIPT.format(settings=liq_settings.strip())
+        return liq_script
+
+    def wait_start(self, process: Popen):
+        while process.poll() is None and not self.socket_path.is_socket():
+            sleep(0.1)
+
+    def make_connection(self) -> LiquidsoapConnection:
+        return LiquidsoapConnection(path=self.socket_path)
+
+    def make_client(self) -> LiquidsoapClient:
+        return LiquidsoapClient(path=self.socket_path)
+
+
+@contextmanager
+def run_liq_server(
+    kind: str,
+    tmp_path: Path,
+) -> Generator[LiquidsoapManager, None, None]:
     entrypoint = tmp_path / "main.liq"
 
-    if request.param == "telnet":
+    manager: LiquidsoapManager
-        telnet_port = randint(32768, 65535)
+    if kind == "telnet":
-        liq_settings = LIQ_TELNET_SETTINGS.format(telnet_port=telnet_port)
+        manager = LiquidsoapManagerTelnet()
-    elif request.param == "socket":
+    elif kind == "socket":
-        socket_path = entrypoint.with_name("main.sock")
+        manager = LiquidsoapManagerSocket(tmp_path)
-        liq_settings = LIQ_SOCKET_SETTINGS.format(socket_path=socket_path)
 
-    liq_script = LIQ_SCRIPT.format(settings=liq_settings.strip())
+    liq_script = manager.generate_entrypoint()
     logger.debug(liq_script)
     entrypoint.write_text(liq_script)
 
@@ -62,20 +114,37 @@ def liq_conn_fixture(request, tmp_path_factory):
         stderr=STDOUT,
         text=True,
     ) as process:
-        if request.param == "telnet":
+        manager.wait_start(process)
-            sleep(2)
-        elif request.param == "socket":
-            while process.poll() is None and not socket_path.is_socket():
-                sleep(0.1)
 
         if process.poll() is not None:
             pytest.fail(process.stdout.read())
 
-        if request.param == "telnet":
+        yield manager
-            conn = LiquidsoapConnection(host="localhost", port=telnet_port)
-        elif request.param == "socket":
-            conn = LiquidsoapConnection(path=socket_path)
 
+        process.terminate()
+
+
+@pytest.fixture(
+    name="liq_conn",
+    scope="session",
+    params=["telnet", "socket"],
+)
+def liq_conn_fixture(request, tmp_path_factory):
+    tmp_path: Path = tmp_path_factory.mktemp(__name__)
+
+    with run_liq_server(request.param, tmp_path) as manager:
+        conn = manager.make_connection()
         with conn:
             yield conn
-        process.terminate()
+
+
+@pytest.fixture(
+    name="liq_client",
+    scope="session",
+    params=["telnet", "socket"],
+)
+def liq_client_fixture(request, tmp_path_factory):
+    tmp_path: Path = tmp_path_factory.mktemp(__name__)
+
+    with run_liq_server(request.param, tmp_path) as manager:
+        yield manager.make_client()