Clean up e2e test scripts ahead of open-sourcing

src/gst_webrtc/gpu_sink/core.py

@@ -41,14 +41,12 @@ def __init__(
         output_format: str = "RGBA",
         queue_size: int = 4,
         drop_when_full: bool = True,
-        copy_frame: bool = True,
     ) -> None:
         if output_format not in {"RGBA", "RGB", "BGR", "GRAY8"}:
             raise ValueError("output_format must be one of RGBA|RGB|BGR|GRAY8")
         self.name = name or f"appsink-{uuid.uuid4().hex[:8]}"
         self.output_format = output_format
         self.drop_when_full = drop_when_full
-        self.copy_frame = copy_frame
 
         self._q: "queue.Queue[Frame]" = queue.Queue(maxsize=max(1, queue_size))
         self._sink: Optional[Gst.Element] = None
@@ -203,11 +201,7 @@ def _on_new_sample(self, sink: Gst.Element):  # -> Gst.FlowReturn
                 return Gst.FlowReturn.OK
             try:
                 arr = self._map_to_numpy(map_info, width, height, str(fmt))
-                # Always copy to ensure safety across threads and buffer reuse.
-                if not self.copy_frame:
-                    # Minimal impl note: zero-copy is unsafe without keeping the buffer mapped.
-                    # Fallback to copy for correctness.
-                    pass
+                # Always copy: zero-copy is unsafe without keeping the buffer mapped.
                 arr = arr.copy()
             finally:
                 buffer.unmap(map_info)

src/gst_webrtc/sender/core.py

@@ -324,11 +324,10 @@ def _on_ice_state(self, _webrtc, _pspec) -> None:
         elif nick == "failed":
             self._restart_ice()
         else:
-            # If disconnected persists >3s, restart ICE
+            # If disconnected persists >1s, restart ICE
             if self._ice_disconnected_at is not None:
                 import time as _t
 
-                # Faster reaction: 1.0s
                 if _t.time() - self._ice_disconnected_at > 1.0:
                     self._restart_ice()
 

tests/e2e/.gitignore

@@ -2,3 +2,4 @@ logs/
 report.json
 artifacts/
 gpu_sink_report.json
+frames_msid/

tests/e2e/README.md

@@ -6,18 +6,18 @@ Quick start:
 
 ```
 pixi install
-./e2e/smoke.sh
+./tests/e2e/smoke.sh
 ```
 
 What it does:
 - Starts `signal-server` via `go run . serve --addr :18080`.
-- Runs `receiver.py` headlessly and `sender.py` via `pixi run`.
+- Runs `receiver.py` and `sender_sw.py` (software x264enc) headlessly via `pixi run`.
 - Watches logs for successful WebRTC link/answer.
-- Writes a JSON report to `e2e/report.json` and exits non-zero on failure.
+- Writes a JSON report to `tests/e2e/report.json` and exits non-zero on failure.
 
 Artifacts:
-- Logs: `e2e/logs/{signal,receiver,sender}.log`
-- Report: `e2e/report.json`
+- Logs: `tests/e2e/logs/{signal,receiver,sender}.log`
+- Report: `tests/e2e/report.json`
 
 Environment:
 - `TIMEOUT_SEC` (optional): overall wait timeout (default 25).
@@ -32,20 +32,20 @@ Run:
 
 ```
 pixi install
-./e2e/gpu_sink.sh
+./tests/e2e/gpu_sink.sh
 ```
 
 What it does:
 - Starts the Go signaling server on `:18080`.
 - Runs `gpu_sink_save.py` as the receiver to pull NumPy frames and save PNGs.
-- Runs `sender.py` to produce an H264 RTP stream.
+- Runs `sender_sw.py` to produce an H264 RTP stream.
 - Waits until the requested number of frames are saved, then verifies images using PIL.
-- Writes a JSON report to `e2e/gpu_sink_report.json`.
+- Writes a JSON report to `tests/e2e/gpu_sink_report.json`.
 
 Artifacts:
-- Saved frames: `e2e/artifacts/gpu_frames/*.png`
-- Logs: `e2e/logs/{signal_gpu,gpu_sink_receiver,gpu_sink_sender}.log`
-- Report: `e2e/gpu_sink_report.json`
+- Saved frames: `tests/e2e/artifacts/gpu_frames/*.png`
+- Logs: `tests/e2e/logs/{signal_gpu,gpu_sink_receiver,gpu_sink_sender}.log`
+- Report: `tests/e2e/gpu_sink_report.json`
 
 Environment:
 - `FRAMES` (optional): number of frames to save (default 5)

tests/e2e/appsink_msid_e2e.py

@@ -17,7 +17,6 @@
 import argparse
 import asyncio
 import pathlib
-import sys
 from typing import Dict, List
 
 import numpy as np
@@ -37,11 +36,6 @@
 gi.require_version("GstApp", "1.0")
 from gi.repository import Gst, GstApp  # type: ignore
 
-# Ensure project root on sys.path when running from tests/e2e/
-ROOT = pathlib.Path(__file__).resolve().parents[2]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
-
 from gst_webrtc import init_gst
 from gst_webrtc.receiver import WebRTCReceiver
 from gst_webrtc.sender import WebRTCSender

tests/e2e/gpu_sink.sh

@@ -5,7 +5,7 @@ set -Eeuo pipefail
 IFS=$'\n\t'
 
 ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")"/../.. && pwd)"
-E2E_DIR="$ROOT_DIR/e2e"
+E2E_DIR="$ROOT_DIR/tests/e2e"
 LOG_DIR="$E2E_DIR/logs"
 ART_DIR="$E2E_DIR/artifacts"
 OUT_DIR="$ART_DIR/gpu_frames"

tests/e2e/gpu_sink_save.py

@@ -6,7 +6,7 @@
   pixi run python tests/e2e/gpu_sink_save.py --frames 5 --out ./frames_gpu
 
 Prereqs:
-  - Start signaling server (see project docs) and a sender (`tests/e2e/sender.py`).
+  - Start signaling server (see project docs) and a sender (`tests/e2e/sender_sw.py`).
   - Pillow is recommended: `pixi run python -m pip install pillow` if missing.
 """
 
@@ -28,7 +28,8 @@
 gi.require_version("Gst", "1.0")
 gi.require_version("GstWebRTC", "1.0")
 gi.require_version("GstSdp", "1.0")
-from gi.repository import Gst  # type: ignore
+gi.require_version("GstApp", "1.0")
+from gi.repository import Gst, GstApp  # type: ignore
 
 from gst_webrtc import init_gst
 from gst_webrtc.gpu_sink import GpuFrameSink
@@ -64,10 +65,25 @@ async def main(nframes: int, out_dir: Path, out_fmt: str, timeout: float) -> Non
     # Start receiver loop
     recv_task = asyncio.create_task(receiver.run())
 
-    # Wait for appsink to be added by receiver, then bind
+    # Wait for the receiver to add its appsink, then bind. The receiver renames
+    # the appsink to the incoming stream's msid, so discover it by element type
+    # rather than by GpuFrameSink's original name.
+    def _find_appsink_name():
+        it = receiver.pipe.iterate_recurse()
+        while True:
+            res, elem = it.next()
+            if res == Gst.IteratorResult.OK:
+                if isinstance(elem, GstApp.AppSink):
+                    return elem.get_name()
+            elif res == Gst.IteratorResult.RESYNC:
+                it.resync()
+            else:
+                return None
+
     for _ in range(200):  # ~2s max
-        el = receiver.pipe.get_by_name(sink.name)
-        if el is not None:
+        found = _find_appsink_name()
+        if found is not None:
+            sink.name = found
             sink.bind(receiver.pipe)
             print(f"[gpu-sink-test] bound appsink: {sink.name}")
             break
@@ -81,7 +97,6 @@ async def main(nframes: int, out_dir: Path, out_fmt: str, timeout: float) -> Non
                 continue
 
             arr: np.ndarray = frame.array
-            h, w = arr.shape[:2]
             info = {
                 "shape": arr.shape,
                 "dtype": str(arr.dtype),

tests/e2e/inference_buffer_v2_ipc.py

@@ -73,7 +73,12 @@ def main() -> int:
     # Step 1: Child writes v1; parent verifies visibility (only for CUDA)
     v1 = 1.2345
     parent_conn.send(("write1", v1))
-    tag, mean1 = parent_conn.recv()
+    msg = parent_conn.recv()
+    if msg[0] == "child_error":
+        print(f"FAIL: child raised: {msg[1]}")
+        proc.terminate(); proc.join(5)
+        return 6
+    tag, mean1 = msg
     assert tag == "done1"
     if is_cuda:
         torch.cuda.synchronize()
@@ -92,7 +97,12 @@ def main() -> int:
     if is_cuda:
         torch.cuda.synchronize()
     parent_conn.send(("parent_written2", v2))
-    tag, mean2, ok2 = parent_conn.recv()
+    msg = parent_conn.recv()
+    if msg[0] == "child_error":
+        print(f"FAIL: child raised: {msg[1]}")
+        proc.terminate(); proc.join(5)
+        return 7
+    tag, mean2, ok2 = msg
     assert tag == "seen2"
     if is_cuda:
         print(f"parent->child CUDA propagation: child_mean={mean2:.6f} child_allclose={ok2}")

tests/e2e/ipc_two_procs/run.sh

@@ -11,7 +11,7 @@ cleanup() {
   [[ -p "$workdir/w_out" ]] && rm -f "$workdir/w_out"
   [[ -p "$workdir/r_in" ]] && rm -f "$workdir/r_in"
   [[ -p "$workdir/r_out" ]] && rm -f "$workdir/r_out"
-  [[ -f "$workdir" ]] && rm -rf "$workdir" || true
+  [[ -d "$workdir" ]] && rm -rf "$workdir" || true
 }
 trap cleanup EXIT
 

tests/e2e/receiver.py

@@ -1,37 +1,26 @@
 import asyncio
 import os
 
-import gi
-
-gi.require_version("Gst", "1.0")
-gi.require_version("GstWebRTC", "1.0")
-gi.require_version("GstSdp", "1.0")
-
-from gi.repository import Gst, GstSdp, GstWebRTC  # noqa: F401
-
 from gst_webrtc import init_gst
 from gst_webrtc.receiver import WebRTCReceiver
-from gst_webrtc.ws_signal.signal_client import SignalClient  # noqa: F401
 
 ROOM = os.environ.get("ROOM", "demo")
 SIGNAL_URL = os.environ.get("SIGNAL_URL", "ws://127.0.0.1:18080/ws")
 STUN = os.environ.get("STUN", "stun://stun.example.com")
 TURN = os.environ.get(
     "TURN", "turn://USERNAME:PASSWORD@TURN_HOST:3478?transport=udp"
-)  # not used
+)
 
 init_gst()
 
-wh = "width=640,height=480"
-fr = "framerate=30/1"
 queue = "queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream"
 
 def h264_decode_bin_sink() -> str:
     desc = f"""
 capsfilter caps="application/x-rtp" ! rtph264depay ! h264parse !
 {queue} !
-nvh264dec ! {queue} !
-videoconvert ! autovideosink
+avdec_h264 ! {queue} !
+videoconvert ! fakesink sync=false
 """
     return desc
 

tests/e2e/receiver_msid_save.py

@@ -37,9 +37,9 @@
 
 
 def _rtp_h264_to_appsink_desc(appsink_name: str = "recv_app", to_format: str = "RGBA") -> str:
-    # Keep CPU decode for stability across environments
+    # CPU decode for stability across environments (no GPU context needed)
     q = "queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream"
-    dec = f"nvh264dec ! {q}"
+    dec = f"avdec_h264 ! {q}"
     return (
         f"capsfilter caps=\"application/x-rtp\" ! rtph264depay ! h264parse ! "
         f"{dec} ! videoconvert ! video/x-raw,format={to_format} ! {q} ! "
@@ -58,7 +58,6 @@ def _frame_from_sample(sample: Gst.Sample, expect_fmt: str = "RGBA") -> np.ndarr
     s = caps.get_structure(0) if caps else None
     width = int(s.get_value("width")) if s and s.has_field("width") else 0
     height = int(s.get_value("height")) if s and s.has_field("height") else 0
-    fmt = str(s.get_value("format")) if s and s.has_field("format") else expect_fmt
 
     ok, map_info = buf.map(Gst.MapFlags.READ)
     if not ok:
@@ -72,14 +71,9 @@ def _frame_from_sample(sample: Gst.Sample, expect_fmt: str = "RGBA") -> np.ndarr
         bpr = nbytes // height
         row_pixels = bpr // ch if ch > 0 else 0
         arr = np.frombuffer(mv, dtype=np.uint8)
-        if ch == 4:
-            arr = arr.reshape(height, row_pixels, 4)
-            if width > 0 and row_pixels != width:
-                arr = arr[:, :width, :]
-        else:
-            arr = arr.reshape(height, row_pixels, 3)
-            if width > 0 and row_pixels != width:
-                arr = arr[:, :width, :]
+        arr = arr.reshape(height, row_pixels, ch)
+        if width > 0 and row_pixels != width:
+            arr = arr[:, :width, :]
         return arr.copy()
     finally:
         buf.unmap(map_info)

tests/e2e/sender.py

@@ -6,49 +6,24 @@
 import gi
 
 gi.require_version("Gst", "1.0")
-gi.require_version("GstWebRTC", "1.0")
-gi.require_version("GstSdp", "1.0")
 
-from gi.repository import Gst, GstSdp, GstWebRTC  # noqa: F401
+from gi.repository import Gst
 
 from gst_webrtc import init_gst
 from gst_webrtc.sender import WebRTCSender
-from gst_webrtc.ws_signal.signal_client import SignalClient  # noqa: F401
 
 ROOM = os.environ.get("ROOM", "demo")
 SIGNAL_URL = os.environ.get("SIGNAL_URL", "ws://127.0.0.1:18080/ws")
 STUN = os.environ.get("STUN", "stun://stun.example.com")
 TURN = os.environ.get(
     "TURN", "turn://USERNAME:PASSWORD@TURN_HOST:3478?transport=udp"
-)  # not used
+)
 
 init_gst()
 
-wh = "width=640,height=480"
-fr = "framerate=30/1"
 queue = "queue max-size-buffers=1 max-size-time=0 max-size-bytes=0 leaky=downstream"
 
 
-def create_test_video_source(name: str) -> str:
-    """Build a low-latency H264 RTP source with an identifiable name.
-
-    NOTE: We insert an `identity name=<name>` so the multi-source case is easy to
-    debug on the sender side. This does not affect WebRTC semantics directly,
-    but helps correlate logs and pads per source.
-    """
-    src = f"""\
-videotestsrc is-live=true pattern=ball ! video/x-raw,{wh},{fr} ! \
-identity name={name} silent=true ! \
-{queue} ! \
-nvh264enc ! \
-h264parse config-interval=-1 ! video/x-h264,alignment=au ! \
-{queue} ! \
-rtph264pay aggregate-mode=zero-latency pt=96 ! \
-capsfilter caps=\"application/x-rtp,media=video,encoding-name=H264,payload=96,clock-rate=90000\"
-"""
-    return src
-
-
 # RealSense D405 color stream parameters (pushed into GStreamer via appsrc).
 RS_W, RS_H, RS_FPS = (
     int(os.environ.get("RS_W", 640)),

tests/e2e/sender_sw.py

@@ -4,24 +4,10 @@
 Use x264enc (CPU) to avoid GPU dependency during CI/E2E runs.
 
 Run:
-  ROOM=demo SIGNAL_URL=ws://127.0.0.1:18080/ws pixi run python e2e/sender_sw.py
+  ROOM=demo SIGNAL_URL=ws://127.0.0.1:18080/ws pixi run python tests/e2e/sender_sw.py
 """
 
 import asyncio
-import pathlib
-import sys
-
-import gi
-
-gi.require_version("Gst", "1.0")
-gi.require_version("GstWebRTC", "1.0")
-gi.require_version("GstSdp", "1.0")
-from gi.repository import Gst  # type: ignore  # noqa: F401
-
-# Ensure project root on sys.path when running from e2e/
-ROOT = pathlib.Path(__file__).resolve().parents[1]
-if str(ROOT) not in sys.path:
-    sys.path.insert(0, str(ROOT))
 
 from gst_webrtc import init_gst
 from gst_webrtc.sender import WebRTCSender