wip3

.github/workflows/build-and-test.yml

@@ -56,7 +56,7 @@ jobs:
       - name: Install dependencies
         shell: bash
         run: |
-          sudo apt-get install ffmpeg xdotool xvfb
+          sudo apt-get install ffmpeg xdotool x11-apps xvfb
       - name: Install python dependencies
         shell: bash
         run: |

tests/test_rendering.py

@@ -50,8 +50,7 @@ def is_xvfb_avaiable() -> bool:
     return False
 
 
-@pytest.fixture
-def screen_manager():
+def get_screen_manager():
     if sys.platform == "linux":
         if is_xvfb_avaiable():
             return LinuxVirtualScreenManager()
@@ -61,92 +60,129 @@ def screen_manager():
         pytest.skip(f"Platform '{sys.platform}' is not supported")
 
 
+def set_keyboard_layout(lang):
+    # Run "setxkbmap <lang> -print | xkbcomp - $DISPLAY"
+    setxkbmap_process = subprocess.Popen(
+        ["setxkbmap", lang, "-print"],
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True
+    )
+    xkbcomp_process = subprocess.Popen(
+        ["xkbcomp", "-", os.environ["DISPLAY"]],
+        stdin=setxkbmap_process.stdout,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True
+    )
+    # Close the stdout of the first process to allow it to receive a SIGPIPE if `xkbcomp` exits
+    if setxkbmap_process.stdout:
+        setxkbmap_process.stdout.close()
+
+    output, errors = xkbcomp_process.communicate()
+    if xkbcomp_process.returncode != 0:
+        print(f"Error: {errors}")
+    else:
+        print(f"Success: {output}")
+
+
+@pytest.fixture(scope="session", autouse=True)
+def screen_manager():
+    with get_screen_manager() as _:
+
+        # this is a trick to keep one keyboard layout for full display lifetime,
+        # otherwise server would regenerate layout on last client disconection
+        # https://stackoverflow.com/questions/75919741/how-to-add-keyboard-layouts-to-xvfb
+        # 1. run some app in the background for full duration on test
+        # 2. configure keyboard layout
+        # 3. test
+        dummy = subprocess.Popen(["xlogo"])
+        time.sleep(0.5)
+
+        set_keyboard_layout("pl")
+
+        yield
+
+        dummy.kill()
+
+
 @pytest.fixture
 def app_isolation(tmpdir, app_path):
     @contextmanager
     def _isolation():
         new_path = shutil.copy(app_path, tmpdir)
+        logger.info(f"New app path: {new_path}")
+
+        app_dir = Path(new_path).parent
+        os.mkdir(app_dir / "render-out")
+
         yield new_path
 
-        # some checks should be here
+        # will be a part of klawa some day:
+        subprocess.run(
+            [
+                "ffmpeg", "-y",
+                "-framerate", "60",
+                "-s", "960x320",
+                "-pix_fmt", "rgba",
+                "-i", "frame%05d.raw",
+                "-c:v", "libvpx-vp9",
+                "../output.webm",
+            ],
+            cwd=app_dir / "render-out",
+        )
 
     yield _isolation
 
 
-def run_process_wait(command, cwd, name, process_holder) -> None:
-    env = os.environ.copy()
-    logger.info(f"env: {env}")
+def run_process_capture_logs(command, cwd, name="", process_holder=None) -> None:
     process = subprocess.Popen(
         command,
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
         text=True,
-        env=env,
         cwd=cwd,
     )
     assert process, "Process creation failed"
+    assert process.stdout, "Could not get stdout"
 
-    process_holder[name] = process
+    if process_holder != None:
+        process_holder[name] = process
 
     for line in process.stdout:
         logger.info(line.strip())
 
     process.wait()
-    if process.returncode != 0:
-        logger.error("Command failed with return code %d", process.returncode)
 
 
 @pytest.mark.parametrize(
     "text",
     [
         "The quick brown fox jumps over the lazy dog",
-        # fails inside xfvb, something to do with keyboard layout
         "Dość błazeństw, żrą mój pęk luźnych fig",
     ],
 )
-def test_record_and_render(app_isolation, screen_manager, text: str) -> None:
+def test_record_and_render(app_isolation, text: str) -> None:
     with app_isolation() as app:
-        with screen_manager as _:
-            app_dir = Path(app).parent
-            logger.info(f"New app path: {app}")
-            os.mkdir(app_dir / "render-out")
-
-            processes = {}
-            thread = threading.Thread(target=run_process_wait, args=([app, "--record", "events.bin"], app_dir, "klawa", processes,))
-            thread.start()
-            time.sleep(0.5)
-
-            run_process_wait(
-                ["xdotool", "type", "--delay", "500", text],
-                app_dir,
-                "xdotool",
-                processes
-            )
-
-            process = processes.get("klawa")
-            if process and process.poll() is None:
-                os.kill(process.pid, signal.SIGTERM)
-
-            thread.join()
-
-            run_process_wait(
-                [app, "--replay", "events.bin", "--render", "render-out"],
-                app_dir,
-                "klawa",
-                processes
-            )
-
-            run_process_wait(
-                [
-                    "ffmpeg", "-y",
-                    "-framerate", "60",
-                    "-s", "960x320",
-                    "-pix_fmt", "rgba",
-                    "-i", "frame%05d.raw",
-                    "-c:v", "libvpx-vp9",
-                    "../output.webm",
-                ],
-                app_dir / "render-out",
-                "ffmpeg",
-                processes
-            )
+        app_dir = Path(app).parent
+        processes = {}
+
+        thread = threading.Thread(
+            target=run_process_capture_logs,
+            args=([app, "--record", "events.bin"], app_dir, "klawa", processes,)
+        )
+        thread.start()
+        time.sleep(0.5)
+
+        subprocess.run(["xdotool", "type", "--delay", "200", text])
+
+        process = processes.get("klawa")
+        if process and process.poll() is None:
+            os.kill(process.pid, signal.SIGTERM)
+
+        thread.join()
+
+        run_process_capture_logs(
+            [app, "--replay", "events.bin", "--render", "render-out"],
+            app_dir,
+        )