wip2

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

@@ -57,16 +57,23 @@ jobs:
         shell: bash
         run: |
           sudo apt-get install ffmpeg xdotool xvfb
+      - name: Install python dependencies
+        shell: bash
+        run: |
+          cd tests
+          python -m venv .env
+          . .env/bin/activate
+          pip install -r dev-requirements.txt
       - name: Run tests
         shell: bash
         run: |
           chmod +x zig-out/bin/klawa
-          cd tests && ./test.sh
+          cd tests && . .env/bin/activate && python -m pytest
       - uses: actions/upload-artifact@v4
         if: always()
         with:
-          name: output.webm
-          path: tests/output.webm
+          name: report
+          path: tests/report/
           retention-days: 2
           if-no-files-found: error
 
@@ -79,7 +86,7 @@ jobs:
       - uses: actions/checkout@v4
       - uses: actions/download-artifact@v4
         with:
-          name: output.webm
+          name: report
           path: tests
       - name: Install Vercel CLI
         run: npm install --global vercel@latest

.gitignore

@@ -1,5 +1,8 @@
 .cache/
+.env/
 .zig-cache/
 zig-out/
-tools/.env
 .vercel
+tests/.env
+tests/report
+tests/__pycache__

.justfile

@@ -38,5 +38,5 @@ test-kle:
 test-queue:
   zig test src/spsc_queue.zig
 
-functional-tests:
-  cd tests && ./test.sh
+pytest:
+  . .env/bin/activate && cd tests && python -m pytest

src/main.zig

@@ -564,6 +564,7 @@ pub fn main() !void {
         }
 
         if (keys.pop()) |k| {
+            if (k.symbol == null) continue;
             std.debug.print("Consumed: '{s}'\n", .{k.symbol});
 
             var text_: [*:0]const u8 = undefined;

tests/conftest.py

@@ -1,4 +1,6 @@
+import glob
 import os
+import shutil
 from pathlib import Path
 
 import pytest
@@ -19,3 +21,22 @@ def app_path(request) -> Path:
     assert app_path, "App path is required"
     return Path(os.path.realpath(app_path))
 
+
+@pytest.hookimpl(hookwrapper=True)
+def pytest_runtest_makereport(item, call):
+    pytest_html = item.config.pluginmanager.getplugin("html")
+    outcome = yield
+    report = outcome.get_result()
+    extras = getattr(report, "extras", [])
+
+    if report.when == "call" and not report.skipped:
+        artifact_dir = Path(item.config.option.htmlpath).parent
+        webm_count = len(glob.glob(f"{artifact_dir}/*webm"))
+        if tmpdir := item.funcargs.get("tmpdir"):
+            videos = glob.glob(f"{tmpdir}/*webm")
+            for f in videos:
+                dest = shutil.copy(f, artifact_dir / f"output{webm_count}.webm")
+                dest = Path(dest).relative_to(artifact_dir)
+                html = f'<video controls><source src="{dest}" type="video/webm"></video>'
+                extras.append(pytest_html.extras.html(html))
+        report.extras = extras

tests/deploy.sh

@@ -5,9 +5,9 @@ set -u
 
 SCRIPT_DIR=$( cd -- "$( dirname -- "${BASH_SOURCE[0]}" )" &> /dev/null && pwd )
 DEPLOY_DIR=${SCRIPT_DIR}/../.vercel/output
+REPORT_DIR=${SCRIPT_DIR}/report
 
 rm -rf $DEPLOY_DIR
 mkdir $DEPLOY_DIR
 
-cp ${SCRIPT_DIR}/index.html $DEPLOY_DIR
-cp ${SCRIPT_DIR}/*.webm $DEPLOY_DIR
+cp $REPORT_DIR/* $DEPLOY_DIR

pytest.ini → tests/pytest.ini

@@ -1,8 +1,9 @@
 [pytest]
 addopts =
-  --html=report.html
-  --app-path=zig-out/bin/klawa
+  --html=report/report.html
+  --app-path=../zig-out/bin/klawa
 generate_report_on_test = True
+render_collapsed =
 log_cli = True
 log_cli_level = DEBUG
 log_cli_format = %(asctime)s [%(levelname)8s] %(message)s

tests/test_rendering.py

@@ -1,8 +1,10 @@
 import logging
 import os
+import signal
 import shutil
 import subprocess
 import sys
+import threading
 import time
 from contextlib import contextmanager
 from pathlib import Path
@@ -12,13 +14,10 @@
 
 logger = logging.getLogger(__name__)
 
-PREFFERED_X11_BACKEND="xvfb"
-#PREFFERED_X11_BACKEND="xephyr"
-
 
 class LinuxVirtualScreenManager:
     def __enter__(self):
-        self.display = SmartDisplay(backend=PREFFERED_X11_BACKEND, size=(960, 320))
+        self.display = SmartDisplay(backend="xvfb", size=(960, 320))
         self.display.start()
         return self
 
@@ -35,20 +34,6 @@ def __exit__(self, *exc):
         return False
 
 
-def run_process(args, cwd):
-    logger.info(f"cwd: {cwd} args: {args}")
-    env = os.environ.copy()
-    return subprocess.Popen(
-        args,
-        stdout=subprocess.PIPE,
-        stderr=subprocess.PIPE,
-        stdin=subprocess.PIPE,
-        #text=True,
-        cwd=cwd,
-        env=env,
-    )
-
-
 def is_xvfb_avaiable() -> bool:
     try:
         p = subprocess.Popen(
@@ -82,84 +67,86 @@ def app_isolation(tmpdir, app_path):
     def _isolation():
         new_path = shutil.copy(app_path, tmpdir)
         yield new_path
+
         # some checks should be here
 
     yield _isolation
 
 
-def test_record_and_render(screen_manager, app_isolation) -> None:
-    with app_isolation() as app:
-        app_dir = Path(app).parent
-        logger.info(f"New app path: {app}")
+def run_process_wait(command, cwd, name, process_holder) -> None:
+    env = os.environ.copy()
+    logger.info(f"env: {env}")
+    process = subprocess.Popen(
+        command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        env=env,
+        cwd=cwd,
+    )
+    assert process, "Process creation failed"
 
-        render_dir = "render-out"
-        os.mkdir(f"{app_dir}/{render_dir}")
+    process_holder[name] = process
 
-        with screen_manager as _:
-            app_process = run_process(
-                [
-                    app,
-                    "--record",
-                    "test_events.bin",
-                ],
-                app_dir
-            )
-            time.sleep(1)
+    for line in process.stdout:
+        logger.info(line.strip())
 
-            setxkbmap_process = run_process(["setxkbmap", "-layout", "pl"], app_dir)
-            setxkbmap_process.wait()
+    process.wait()
+    if process.returncode != 0:
+        logger.error("Command failed with return code %d", process.returncode)
 
-            xdotool_process = run_process(
-                [
-                    "xdotool",
-                    "type",
-                    "--delay",
-                    "500",
-                    "Dość błazeństw, żrą mój pęk luźnych fig",
-                ],
-                app_dir
+
+@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:
+    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
             )
-            xdotool_process.wait()
 
-            app_process.kill()
-            outs, errs = app_process.communicate()
+            process = processes.get("klawa")
+            if process and process.poll() is None:
+                os.kill(process.pid, signal.SIGTERM)
 
-            logger.info(f"Process stdout: {outs}")
-            logger.info(f"Process stderr: {errs}")
+            thread.join()
 
-            app_process = run_process(
-                [
-                    app,
-                    "--replay",
-                    "test_events.bin",
-                    "--render",
-                    render_dir,
-                ],
-                app_dir
+            run_process_wait(
+                [app, "--replay", "events.bin", "--render", "render-out"],
+                app_dir,
+                "klawa",
+                processes
             )
-            app_process.wait()
-            outs, errs = app_process.communicate()
-
-            logger.info(f"Process stdout: {outs}")
-            logger.info(f"Process stderr: {errs}")
 
-            ffmpeg_process = run_process(
+            run_process_wait(
                 [
-                    "ffmpeg",
-                    "-y",
-                    "-framerate",
-                    "60",
-                    "-s",
-                    "960x320",
-                    "-pix_fmt",
-                    "rgba",
-                    "-i",
-                    "frame%05d.raw",
-                    "-c:v",
-                    "libvpx-vp9",
+                    "ffmpeg", "-y",
+                    "-framerate", "60",
+                    "-s", "960x320",
+                    "-pix_fmt", "rgba",
+                    "-i", "frame%05d.raw",
+                    "-c:v", "libvpx-vp9",
                     "../output.webm",
                 ],
-                f"{app_dir}/{render_dir}",
+                app_dir / "render-out",
+                "ffmpeg",
+                processes
             )
-            ffmpeg_process.wait()
-