From 6eb7da6a1b08581aab70701e3e43e73b109ed3f6 Mon Sep 17 00:00:00 2001 From: James Hush Date: Thu, 5 Feb 2026 10:27:21 +0800 Subject: [PATCH] Align smallwebrtc test with repo conventions (unittest + optional dep guard) Migrate from pytest-style to unittest.IsolatedAsyncioTestCase to match the pattern used by other transport tests (e.g. test_livekit_transport.py). Guard the aiortc/av import with try/except and skipUnless so tests gracefully skip when webrtc dependencies aren't installed. Add pyright suppressions for false positives inherent to testing internals of optional-dependency classes. Co-Authored-By: Claude Opus 4.5 --- tests/test_smallwebrtc_transport.py | 74 ++++++++++++++++++----------- 1 file changed, 45 insertions(+), 29 deletions(-) diff --git a/tests/test_smallwebrtc_transport.py b/tests/test_smallwebrtc_transport.py index ad21e6ec1..8fff11ac3 100644 --- a/tests/test_smallwebrtc_transport.py +++ b/tests/test_smallwebrtc_transport.py @@ -4,12 +4,29 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import pytest +# pyright: reportConstantRedefinition=false +# pyright: reportPrivateUsage=false, reportUnknownMemberType=false +# pyright: reportUnknownArgumentType=false, reportUnknownVariableType=false +# pyright: reportOperatorIssue=false +# pyright: reportOptionalCall=false -from pipecat.transports.smallwebrtc.transport import RawAudioTrack +import unittest +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from pipecat.transports.smallwebrtc.transport import RawAudioTrack + +try: + from pipecat.transports.smallwebrtc.transport import RawAudioTrack + + WEBRTC_AVAILABLE = True +except (ImportError, Exception): + WEBRTC_AVAILABLE = False + RawAudioTrack = None # type: ignore[misc,assignment] -class TestRawAudioTrack: +@unittest.skipUnless(WEBRTC_AVAILABLE, "webrtc dependencies not installed") +class TestRawAudioTrack(unittest.IsolatedAsyncioTestCase): """Tests for the RawAudioTrack class.""" def test_default_chunk_size_is_10ms(self): @@ -19,8 +36,8 @@ class TestRawAudioTrack: # 10ms at 16kHz = 160 samples, 2 bytes per sample = 320 bytes expected_bytes = int(sample_rate * 10 / 1000) * 2 - assert track._bytes_per_chunk == expected_bytes - assert track._bytes_per_chunk == 320 + self.assertEqual(track._bytes_per_chunk, expected_bytes) + self.assertEqual(track._bytes_per_chunk, 320) def test_custom_chunk_size_40ms(self): """Test that num_10ms_chunks=4 produces 40ms chunks.""" @@ -29,8 +46,8 @@ class TestRawAudioTrack: # 40ms at 16kHz = 640 samples, 2 bytes per sample = 1280 bytes expected_bytes = int(sample_rate * 40 / 1000) * 2 - assert track._bytes_per_chunk == expected_bytes - assert track._bytes_per_chunk == 1280 + self.assertEqual(track._bytes_per_chunk, expected_bytes) + self.assertEqual(track._bytes_per_chunk, 1280) def test_custom_chunk_size_20ms(self): """Test that num_10ms_chunks=2 produces 20ms chunks.""" @@ -39,10 +56,9 @@ class TestRawAudioTrack: # 20ms at 16kHz = 320 samples, 2 bytes per sample = 640 bytes expected_bytes = int(sample_rate * 20 / 1000) * 2 - assert track._bytes_per_chunk == expected_bytes - assert track._bytes_per_chunk == 640 + self.assertEqual(track._bytes_per_chunk, expected_bytes) + self.assertEqual(track._bytes_per_chunk, 640) - @pytest.mark.asyncio async def test_add_audio_bytes_queues_correct_chunks(self): """Test that add_audio_bytes breaks audio into correct chunk sizes.""" sample_rate = 16000 @@ -54,15 +70,14 @@ class TestRawAudioTrack: track.add_audio_bytes(audio_bytes) # Should have exactly 2 chunks in the queue - assert len(track._chunk_queue) == 2 + self.assertEqual(len(track._chunk_queue), 2) # Each chunk should be the correct size chunk1, _ = track._chunk_queue[0] chunk2, _ = track._chunk_queue[1] - assert len(chunk1) == track._bytes_per_chunk - assert len(chunk2) == track._bytes_per_chunk + self.assertEqual(len(chunk1), track._bytes_per_chunk) + self.assertEqual(len(chunk2), track._bytes_per_chunk) - @pytest.mark.asyncio async def test_add_audio_bytes_rejects_invalid_size(self): """Test that add_audio_bytes rejects audio not a multiple of chunk size.""" sample_rate = 16000 @@ -71,12 +86,11 @@ class TestRawAudioTrack: # Create audio that's not a multiple of 40ms chunk size invalid_audio = bytes(track._bytes_per_chunk + 100) - with pytest.raises(ValueError) as exc_info: + with self.assertRaises(ValueError) as ctx: track.add_audio_bytes(invalid_audio) - assert "40ms" in str(exc_info.value) + self.assertIn("40ms", str(ctx.exception)) - @pytest.mark.asyncio async def test_recv_returns_correct_frame_size(self): """Test that recv() returns AudioFrames with correct sample count.""" sample_rate = 16000 @@ -92,9 +106,8 @@ class TestRawAudioTrack: # Frame should have correct number of samples (40ms worth) expected_samples = int(sample_rate * 40 / 1000) # 640 samples - assert frame.samples == expected_samples + self.assertEqual(frame.samples, expected_samples) - @pytest.mark.asyncio async def test_recv_silence_has_correct_size(self): """Test that silence frames have correct size when queue is empty.""" sample_rate = 16000 @@ -106,9 +119,8 @@ class TestRawAudioTrack: # Silence frame should have correct number of samples expected_samples = int(sample_rate * 40 / 1000) # 640 samples - assert frame.samples == expected_samples + self.assertEqual(frame.samples, expected_samples) - @pytest.mark.asyncio async def test_timestamp_advances_by_chunk_samples(self): """Test that timestamp advances correctly based on chunk size.""" sample_rate = 16000 @@ -117,14 +129,14 @@ class TestRawAudioTrack: # Receive first frame and check its timestamp frame1 = await track.recv() - first_pts = frame1.pts - # Receive second frame frame2 = await track.recv() # Timestamp should advance by samples_per_chunk between frames + self.assertIsNotNone(frame1.pts) + self.assertIsNotNone(frame2.pts) expected_samples = int(sample_rate * 40 / 1000) # 640 samples - assert frame2.pts - first_pts == expected_samples + self.assertEqual(frame2.pts - frame1.pts, expected_samples) def test_different_sample_rates(self): """Test chunk size calculation at different sample rates.""" @@ -137,18 +149,22 @@ class TestRawAudioTrack: for sample_rate, num_chunks, expected_bytes in test_cases: track = RawAudioTrack(sample_rate=sample_rate, num_10ms_chunks=num_chunks) - assert track._bytes_per_chunk == expected_bytes + self.assertEqual(track._bytes_per_chunk, expected_bytes) def test_invalid_num_10ms_chunks_zero(self): """Test that num_10ms_chunks=0 raises ValueError.""" - with pytest.raises(ValueError) as exc_info: + with self.assertRaises(ValueError) as ctx: RawAudioTrack(sample_rate=16000, num_10ms_chunks=0) - assert "positive integer" in str(exc_info.value) + self.assertIn("positive integer", str(ctx.exception)) def test_invalid_num_10ms_chunks_negative(self): """Test that negative num_10ms_chunks raises ValueError.""" - with pytest.raises(ValueError) as exc_info: + with self.assertRaises(ValueError) as ctx: RawAudioTrack(sample_rate=16000, num_10ms_chunks=-1) - assert "positive integer" in str(exc_info.value) + self.assertIn("positive integer", str(ctx.exception)) + + +if __name__ == "__main__": + unittest.main()