test(interop): load the local GStreamer plugin reliably

quest/m1/test-flakes-2/README.md

@@ -48,4 +48,3 @@ Public API: none. Wire: none.
 - [Echo report test on mocked time](/quest/m1/test-flakes-2/export-echo-time.md) - the moq-cli echo report test stops taking 35 s of wall time
 - [Stuck-window eviction test on macOS](/quest/m1/test-flakes-2/stuck-window-evicts.md) - the evicted writer reliably sees its stop on macOS, fixed at the race
 - [Swapped Rust to JS Opus](/quest/m1/test-flakes-2/swapped-js-opus.md) - the `rust-swapped -> js` interop cell stops timing out under load
-- [Interop loads moq-gst](/quest/m1/test-flakes-2/gst-plugin-load.md) - the interop gst cells always find the local plugin

rs/moq-gst/smoke.sh

@@ -31,7 +31,7 @@ if ! command -v gst-inspect-1.0 >/dev/null 2>&1; then
     exit 0
 fi
 
-echo "Smoke testing against $(gst-inspect-1.0 --version | head -1)..."
+echo "Smoke testing against $(gst-inspect-1.0 --version | sed -n '1p')..."
 
 # Isolate discovery to $plugin_dir:
 #   - GST_PLUGIN_PATH_1_0 alone is additive, so a system-installed moq
@@ -51,8 +51,10 @@ out="$(
         GST_REGISTRY_1_0="$tmp_registry" \
         gst-inspect-1.0 moq
 )"
-if ! echo "$out" | grep -qE '^[[:space:]]+moqsink:' ||
-    ! echo "$out" | grep -qE '^[[:space:]]+moqsrc:'; then
+# Read captured output without a producer pipe: grep -q can close it early,
+# making a successful inspection fail with SIGPIPE under pipefail.
+if ! grep -qE '^[[:space:]]+moqsink:' <<<"$out" ||
+    ! grep -qE '^[[:space:]]+moqsrc:' <<<"$out"; then
     echo "Error: gst-inspect-1.0 didn't find moqsink/moqsrc in the plugin." >&2
     echo "$out" >&2
     exit 1

rs/moq-gst/smoke.test.py

@@ -0,0 +1,65 @@
+"""Plugin discovery consumes inspection output and checks both factories."""
+
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+SMOKE = Path(__file__).with_name("smoke.sh")
+
+
+class SmokeTest(unittest.TestCase):
+    def inspect(self, mode):
+        with tempfile.TemporaryDirectory() as directory:
+            scratch = Path(directory)
+            inspect = scratch / "gst-inspect-1.0"
+            inspect.write_text('''#!/usr/bin/env python3
+import os
+import signal
+import sys
+
+signal.signal(signal.SIGPIPE, signal.SIG_DFL)
+if sys.argv[1] == "--version":
+    print("gst-inspect-1.0 test", flush=True)
+else:
+    assert sys.argv[1] == "moq"
+    assert os.environ["GST_PLUGIN_PATH_1_0"] == os.environ["PLUGIN_DIR"]
+    assert os.environ["GST_PLUGIN_SYSTEM_PATH_1_0"] == ""
+    assert os.environ["GST_REGISTRY_1_0"] != os.environ["INHERITED_REGISTRY"]
+    print("  moqsink: MoQ Sink", flush=True)
+    if os.environ["INSPECT_MODE"] != "missing":
+        print("  moqsrc: MoQ Src", flush=True)
+# A reader must allow the writer to finish after finding its first match.
+for _ in range(16):
+    sys.stdout.write("x" * 4096 + "\\n")
+    sys.stdout.flush()
+if sys.argv[1] == "moq" and os.environ["INSPECT_MODE"] == "error":
+    sys.exit(1)
+''')
+            inspect.chmod(0o755)
+            env = dict(os.environ, PATH=f"{scratch}:{os.environ['PATH']}", BASH_ENV=os.devnull,
+                       PLUGIN_DIR=str(scratch), INSPECT_MODE=mode,
+                       GST_PLUGIN_PATH_1_0="inherited", GST_PLUGIN_SYSTEM_PATH_1_0="inherited",
+                       GST_REGISTRY_1_0=str(scratch / "inherited-registry"),
+                       INHERITED_REGISTRY=str(scratch / "inherited-registry"))
+            return subprocess.run(["bash", str(SMOKE), str(scratch)], env=env,
+                                  text=True, capture_output=True)
+
+    def test_large_inspection_completes(self):
+        result = self.inspect("complete")
+        self.assertEqual(result.returncode, 0, result.stderr[:1000])
+        self.assertIn("Smoke test passed", result.stdout)
+
+    def test_missing_factory_is_refused(self):
+        result = self.inspect("missing")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("didn't find moqsink/moqsrc", result.stderr)
+
+    def test_inspection_failure_is_refused_even_with_both_factories(self):
+        self.assertNotEqual(self.inspect("error").returncode, 0)
+
+
+if __name__ == "__main__":
+    unittest.main()

test/interop/interop.sh

@@ -771,13 +771,10 @@ prepare_gst() {
         return
     fi
     GST_PLUGIN_DIR="$TARGET_BASE/$PROFILE"
-    # gst-inspect exits 0 even when the .so fails to load, so grep for the
-    # factory. Isolate discovery to our dir + a temp registry so a system-wide moq
-    # plugin can't shadow it (mirrors rs/moq-gst/smoke.sh).
-    if ! GST_PLUGIN_PATH_1_0="$GST_PLUGIN_DIR" GST_PLUGIN_SYSTEM_PATH_1_0="" \
-        GST_REGISTRY_1_0="$HARNESS_RUN/gst-registry.bin" \
-        gst-inspect-1.0 moq 2>/dev/null | grep -qE '^[[:space:]]+moqsrc:'; then
+    # Use the packaging smoke check's isolated discovery and factory assertions.
+    if ! "$WORKSPACE/rs/moq-gst/smoke.sh" "$GST_PLUGIN_DIR" >"$HARNESS_RUN/gst-inspect.log" 2>&1; then
         mark_broken gst "moqsrc not exposed (plugin failed to load against this GStreamer)"
+        sed 's/^/        /' "$HARNESS_RUN/gst-inspect.log" >&2 || true
         return
     fi
     # The audio check decodes through these, from the system's plugins.
@@ -1114,7 +1111,7 @@ run_native() {
     local out
     out=$( (cd "$NATIVE_DIR" && "$@") 2>&1) || true
     printf '%s\n' "$out" >&2
-    printf '%s\n' "$out" | grep -q '^received '
+    grep -q '^received ' <<<"$out"
 }
 
 # Write a track name and the group the compat client's live subscription saw on it to

test/justfile

@@ -42,6 +42,7 @@ default:
 harness:
     ./lib/harness.test.sh
     ./interop/relay-log.test.sh
+    python3 ../rs/moq-gst/smoke.test.py
     python3 interop/report.test.py
     python3 interop/tone.test.py
     cd .. && bun test {{ replace(harness_compat_tests, "\n", " ") }}