Run functional tests against LibVNCServer examples

libvncserver.mk

@@ -13,7 +13,7 @@ LIBVNCSERVER_MAKEFILE=$(LIBVNCSERVER_DIR)/Makefile
 LIBVNCSERVER_MAKEFILE_SRCS=$(wildcard $(LIBVNCSERVER_DIR)/*.cmake)
 
 
-LIBVNCSERVER_EXAMPLES=vncev
+LIBVNCSERVER_EXAMPLES=vncev example
 LIBVNCSERVER_EXAMPLES:=$(addprefix $(LIBVNCSERVER_DIR)/examples/, $(LIBVNCSERVER_EXAMPLES))
 
 LIBVNCSERVER_EXAMPLES_SRCS=$(addsuffix .c, $(LIBVNCSERVER_EXAMPLES))

tests/__init__.py

@@ -0,0 +1 @@
+"""Test suite package for vncdotool."""

tests/functional/__init__.py

@@ -0,0 +1 @@
+"""Functional test helpers and fixtures."""

tests/functional/cli.py

@@ -0,0 +1,14 @@
+"""Helpers for invoking the vncdotool command line entry points."""
+
+from __future__ import annotations
+
+import sys
+
+import pexpect
+
+
+def spawn_command(command: str, *args: str, **kwargs) -> pexpect.spawn:
+    """Spawn a vncdotool CLI subprocess using ``python -m`` invocation."""
+    argv = ["-m", "tests.functional.cli_runner", command]
+    argv.extend(args)
+    return pexpect.spawn(sys.executable, argv, **kwargs)

tests/functional/cli_runner.py

@@ -0,0 +1,31 @@
+"""Invoke vncdotool CLI entry points without installing the package."""
+
+from __future__ import annotations
+
+import sys
+from typing import Callable
+
+from vncdotool import command as cli
+
+COMMANDS: dict[str, Callable[[], None]] = {
+    'vncdo': cli.vncdo,
+    'vnclog': cli.vnclog,
+}
+
+
+def main() -> None:
+    if len(sys.argv) < 2:
+        raise SystemExit('usage: cli_runner.py <command> [args...]')
+
+    cmd = sys.argv[1]
+    try:
+        entry = COMMANDS[cmd]
+    except KeyError as exc:
+        raise SystemExit(f'unknown vncdotool command: {cmd}') from exc
+
+    sys.argv = [cmd, *sys.argv[2:]]
+    entry()
+
+
+if __name__ == '__main__':
+    main()

tests/functional/libvncserver.py

@@ -0,0 +1,84 @@
+"""Utilities for building and running LibVNCServer example binaries."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from functools import lru_cache
+from pathlib import Path
+from typing import Dict, Optional, Tuple
+
+ROOT = Path(__file__).resolve().parents[2]
+MAKEFILE = ROOT / "libvncserver.mk"
+BUILD_ROOT = ROOT / ".vncdo"
+
+
+def _examples_dir() -> Optional[Path]:
+    """Return the directory containing compiled LibVNCServer examples."""
+    if not BUILD_ROOT.exists():
+        return None
+
+    # Prefer the most recent build directory if multiple versions exist.
+    candidates = sorted(BUILD_ROOT.glob("libvncserver-LibVNCServer-*/examples"))
+    for path in reversed(candidates):
+        if path.is_dir():
+            return path
+    return None
+
+
+def _build_examples() -> None:
+    """Invoke the makefile to build the required LibVNCServer examples."""
+    subprocess.run(
+        ["make", "-f", str(MAKEFILE), "libvnc-examples"],
+        check=True,
+        cwd=ROOT,
+    )
+
+
+@lru_cache(maxsize=None)
+def _ensure_examples_dir() -> Path:
+    """Ensure the LibVNCServer examples have been built and return their path."""
+    directory = _examples_dir()
+    if directory is None:
+        _build_examples()
+        directory = _examples_dir()
+    if directory is None:
+        raise RuntimeError(
+            "LibVNCServer examples were not produced by libvncserver.mk build"
+        )
+    return directory
+
+
+def ensure_example(name: str) -> Path:
+    """Return the full path to the named LibVNCServer example binary."""
+    examples = _ensure_examples_dir()
+    executable = examples / name
+    if not executable.exists():
+        # Attempt to rebuild and check once more in case the default target changes.
+        _build_examples()
+        examples = _ensure_examples_dir()
+        executable = examples / name
+    if not executable.exists():
+        raise FileNotFoundError(
+            f"Example binary {name!r} was not created by libvncserver.mk"
+        )
+    return executable
+
+
+@lru_cache(maxsize=None)
+def _runtime_env() -> Dict[str, str]:
+    """Environment variables required to run the LibVNCServer examples."""
+    lib_root = _ensure_examples_dir().parent
+    env = os.environ.copy()
+    existing = env.get("LD_LIBRARY_PATH", "")
+    if existing:
+        env["LD_LIBRARY_PATH"] = f"{lib_root}:{existing}"
+    else:
+        env["LD_LIBRARY_PATH"] = str(lib_root)
+    return env
+
+
+def example_command(name: str, *args: str) -> Tuple[str, Tuple[str, ...], Dict[str, str]]:
+    """Return command, arguments, and environment for launching an example binary."""
+    executable = ensure_example(name)
+    return str(executable), tuple(args), _runtime_env().copy()

tests/functional/test_proxy.py

@@ -1,21 +1,27 @@
+import shlex
 import sys
-from shutil import which
-from unittest import TestCase, skipUnless
+from unittest import TestCase
 
 import pexpect
 
 from vncdotool import rfb
 
+from .cli import spawn_command
+from .libvncserver import example_command
+
 
-@skipUnless(which("vncev"), reason="requires https://github.com/LibVNC/libvncserver")
 class TestLogEvents(TestCase):
     def setUp(self) -> None:
-        cmd = 'vncev -rfbport 5999 -rfbwait 1000'
-        self.server = pexpect.spawn(cmd, timeout=2)
+        server_cmd, server_args, server_env = example_command(
+            "vncev", "-rfbport", "5999", "-rfbwait", "1000"
+        )
+        self.server = pexpect.spawn(server_cmd, list(server_args), env=server_env, timeout=5)
         self.server.logfile_read = sys.stdout.buffer
+        self.server.expect('Listening for VNC connections on TCP port')
 
-        cmd = 'vnclog --listen 1842 -s :99 -'
-        self.recorder = pexpect.spawn(cmd, timeout=2)
+        self.recorder = spawn_command(
+            "vnclog", "--listen", "1842", "-s", ":99", "-", timeout=5
+        )
         self.recorder.logfile_read = sys.stdout.buffer
 
     def tearDown(self) -> None:
@@ -24,8 +30,8 @@ def tearDown(self) -> None:
             self.recorder.terminate(force=True)
 
     def run_vncdo(self, commands: str) -> None:
-        cmd = 'vncdo -s localhost::1842 ' + commands
-        vnc = pexpect.spawn(cmd, timeout=2)
+        args = shlex.split(commands)
+        vnc = spawn_command("vncdo", "-s", "localhost::1842", *args, timeout=5)
         vnc.logfile_read = sys.stdout.buffer
         retval = vnc.wait()
         assert retval == 0, (retval, str(vnc))

tests/functional/test_screen.py

@@ -1,9 +1,9 @@
 import os.path
+import shlex
 import sys
 import tempfile
-from shutil import which
 from typing import IO, List, Optional
-from unittest import TestCase, skipUnless
+from unittest import TestCase
 
 import pexpect
 
@@ -12,10 +12,12 @@
 EXAMPLE_PNG = os.path.join(DATADIR, 'example.png')
 EXAMPLE_NOCURSOR_PNG = os.path.join(DATADIR, 'example_nocursor.png')
 
+from .cli import spawn_command
+from .libvncserver import example_command
+
 SERVER = "example"
 
 
-@skipUnless(which(SERVER), reason=f"requires program {SERVER!r}")
 class TestVNCCapture(TestCase):
     server: Optional[pexpect.spawn] = None
 
@@ -36,14 +38,23 @@ def mktemp(self) -> str:
         return f.name
 
     def run_server(self, server: str) -> None:
-        cmd = f'{server} -rfbport 5910 -rfbwait 1000'
-        self.server = pexpect.spawn(cmd, timeout=2)
+        server_cmd, server_args, server_env = example_command(
+            server, '-rfbport', '5910', '-rfbwait', '1000'
+        )
+        self.server = pexpect.spawn(
+            server_cmd,
+            list(server_args),
+            env=server_env,
+            timeout=5,
+        )
         self.server.logfile_read = sys.stdout.buffer
+        self.server.expect('Listening for VNC connections on TCP port')
 
     def run_vncdo(self, commands: str, exitcode: int = 0) -> None:
-        cmd = f'vncdo -s :10 {commands}'
-        vnc = pexpect.spawn(cmd, logfile=sys.stdout.buffer, timeout=5)
-        vnc.logfile_read = sys.stdout.buffer
+        args = shlex.split(commands)
+        vnc = spawn_command(
+            'vncdo', '-s', ':10', *args, logfile=sys.stdout.buffer, timeout=5
+        )
         vnc.expect(pexpect.EOF)
         if vnc.isalive():
             vnc.wait()

tests/functional/test_send_events.py

@@ -1,7 +1,7 @@
 import os.path
+import shlex
 import sys
-from shutil import which
-from unittest import TestCase, skipUnless
+from unittest import TestCase
 
 import pexpect
 
@@ -10,12 +10,25 @@
 KEYB_VDO = os.path.join(DATADIR, 'sampleb.vdo')
 
 
-@skipUnless(which("vncev"), reason="requires https://github.com/LibVNC/libvncserver")
+from .cli import spawn_command
+from .libvncserver import example_command
+
+
 class TestSendEvents(TestCase):
 
     def setUp(self) -> None:
-        cmd = 'vncev -rfbport 5933 -rfbwait 1000'
-        self.server = pexpect.spawn(cmd, logfile=sys.stdout.buffer, timeout=2)
+        server_cmd, server_args, server_env = example_command(
+            'vncev', '-rfbport', '5933', '-rfbwait', '1000'
+        )
+        self.server = pexpect.spawn(
+            server_cmd,
+            list(server_args),
+            logfile=sys.stdout.buffer,
+            env=server_env,
+            timeout=5,
+        )
+        self.server.logfile_read = sys.stdout.buffer
+        self.server.expect('Listening for VNC connections on TCP port')
 
     def tearDown(self) -> None:
         self.server.terminate(force=True)
@@ -37,8 +50,10 @@ def assertDisconnect(self) -> None:
         self.server.expect(disco)
 
     def run_vncdo(self, commands: str) -> None:
-        cmd = 'vncdo -v -s :33 ' + commands
-        vnc = pexpect.spawn(cmd, logfile=sys.stdout.buffer, timeout=5)
+        args = shlex.split(commands)
+        vnc = spawn_command(
+            'vncdo', '-v', '-s', ':33', *args, logfile=sys.stdout.buffer, timeout=5
+        )
         retval = vnc.wait()
         assert retval == 0, retval