mirror of
https://git.quad4.io/RNS-Things/MeshChatX.git
synced 2026-05-11 18:14:51 +00:00
Ivan
139 lines
4.0 KiB
Python
139 lines
4.0 KiB
Python
# SPDX-License-Identifier: 0BSD
|
|
|
|
"""Property-based tests for Web Audio bridge PCM paths (robustness, no uncaught exceptions)."""
|
|
|
|
import asyncio
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import numpy as np
|
|
import pytest
|
|
from hypothesis import given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from meshchatx.src.backend.web_audio_bridge import (
|
|
WebAudioBridge,
|
|
WebAudioSink,
|
|
WebAudioSource,
|
|
)
|
|
|
|
|
|
class _DummySink:
|
|
def __init__(self):
|
|
self.frames = []
|
|
|
|
def can_receive(self, from_source=None):
|
|
return True
|
|
|
|
def handle_frame(self, frame, source):
|
|
self.frames.append(frame)
|
|
|
|
|
|
@settings(max_examples=200, deadline=None)
|
|
@given(pcm=st.binary(max_size=8192))
|
|
def test_web_audio_source_push_pcm_never_propagates(pcm):
|
|
sink = _DummySink()
|
|
src = WebAudioSource(target_frame_ms=60, sink=sink)
|
|
with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
|
|
src.push_pcm(pcm)
|
|
|
|
|
|
@settings(max_examples=120, deadline=None)
|
|
@given(pcm=st.binary(max_size=8192))
|
|
def test_web_audio_bridge_push_client_frame_never_propagates(pcm):
|
|
bridge = WebAudioBridge(MagicMock(), MagicMock())
|
|
sink = _DummySink()
|
|
bridge.tx_source = WebAudioSource(target_frame_ms=60, sink=sink)
|
|
with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
|
|
bridge.push_client_frame(pcm)
|
|
|
|
|
|
@settings(max_examples=120, deadline=None)
|
|
@given(pcm=st.binary(max_size=8192))
|
|
def test_web_audio_source_push_pcm_sink_can_receive_false_never_propagates(pcm):
|
|
sink = MagicMock()
|
|
sink.can_receive.return_value = False
|
|
src = WebAudioSource(target_frame_ms=60, sink=sink)
|
|
with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
|
|
src.push_pcm(pcm)
|
|
sink.handle_frame.assert_not_called()
|
|
|
|
|
|
@settings(max_examples=150, deadline=None)
|
|
@given(
|
|
values=st.lists(
|
|
st.floats(min_value=-4.0, max_value=4.0, allow_nan=False, allow_infinity=False),
|
|
max_size=2048,
|
|
),
|
|
shape=st.sampled_from(["col", "row", "empty"]),
|
|
)
|
|
def test_web_audio_sink_numpy_frame_never_propagates(values, shape):
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
sent = []
|
|
|
|
async def _send_bytes(data):
|
|
sent.append(data)
|
|
|
|
sink = WebAudioSink(loop, _send_bytes)
|
|
if shape == "empty":
|
|
arr = np.zeros((0, 1), dtype=np.float32)
|
|
elif shape == "row":
|
|
arr = (
|
|
np.array(values, dtype=np.float32).reshape(1, -1)
|
|
if values
|
|
else np.zeros((1, 0), dtype=np.float32)
|
|
)
|
|
else:
|
|
arr = (
|
|
np.array(values, dtype=np.float32).reshape(-1, 1)
|
|
if values
|
|
else np.zeros((0, 1), dtype=np.float32)
|
|
)
|
|
|
|
with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
|
|
sink.handle_frame(arr, None)
|
|
loop.run_until_complete(asyncio.sleep(0.05))
|
|
loop.close()
|
|
|
|
|
|
@settings(max_examples=100, deadline=None)
|
|
@given(raw=st.binary(max_size=1024))
|
|
def test_web_audio_sink_bytes_frame_never_propagates(raw):
|
|
loop = asyncio.new_event_loop()
|
|
asyncio.set_event_loop(loop)
|
|
sent = []
|
|
|
|
async def _send_bytes(data):
|
|
sent.append(data)
|
|
|
|
sink = WebAudioSink(loop, _send_bytes)
|
|
with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
|
|
sink.handle_frame(raw, None)
|
|
loop.run_until_complete(asyncio.sleep(0.05))
|
|
loop.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@settings(max_examples=80, deadline=None)
|
|
@given(
|
|
values=st.lists(
|
|
st.floats(min_value=-2.0, max_value=2.0, allow_nan=False, allow_infinity=False),
|
|
max_size=1024,
|
|
),
|
|
)
|
|
async def test_web_audio_sink_running_loop_numpy_never_propagates(values):
|
|
sent = []
|
|
|
|
async def _send_bytes(data):
|
|
sent.append(data)
|
|
|
|
sink = WebAudioSink(asyncio.get_running_loop(), _send_bytes)
|
|
arr = (
|
|
np.array(values, dtype=np.float32).reshape(-1, 1)
|
|
if values
|
|
else np.zeros((0, 1), dtype=np.float32)
|
|
)
|
|
with patch("meshchatx.src.backend.web_audio_bridge.RNS.log"):
|
|
sink.handle_frame(arr, None)
|
|
await asyncio.sleep(0.05)
|