ppy4sjqvf
/
chatProgram
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

确保语音模块测试通过

Browse Source
main
杨博文 2 months ago
parent 0b2994e594
commit e4c02e8585
4 changed files with 1887 additions and 0 deletions
Show Stats Download Patch File Download Diff File

18
client/__init__.py
Unescape View File

@ -41,3 +41,21 @@ from client.media_player import (
PlaybackError,
MediaLoadError,
)
from client.voice_chat import (
VoiceChatModule,
CallState,
VoiceChatError,
AudioDeviceError,
CallError,
AudioConfig,
CallInfo,
NetworkStats,
JitterBuffer,
AudioCapture,
AudioPlayback,
AudioEncoder,
AudioDecoder,
CallStateCallback,
IncomingCallCallback,
)

1478
client/voice_chat.py
View File

File diff suppressed because it is too large Load Diff

1
requirements.txt
Unescape View File

@ -12,6 +12,7 @@ PyAudio>=0.2.13
opencv-python>=4.8.0
ffmpeg-python>=0.2.0
mutagen>=1.47.0
opuslib>=3.0.1
# Image Processing
Pillow>=10.0.0

390
tests/test_voice_chat.py
Unescape View File

@ -0,0 +1,390 @@
# P2P Network Communication - Voice Chat Module Tests
"""
语音聊天模块测试
测试音频采集、编码、通话控制和实时传输功能
"""
import asyncio
import pytest
import struct
import time
from unittest.mock import Mock, MagicMock, patch, AsyncMock
from datetime import datetime
from client.voice_chat import (
VoiceChatModule,
CallState,
VoiceChatError,
AudioDeviceError,
CallError,
AudioConfig,
CallInfo,
NetworkStats,
JitterBuffer,
AudioCapture,
AudioPlayback,
AudioEncoder,
AudioDecoder,
)
from shared.models import Message, MessageType, NetworkQuality
class TestAudioConfig:
"""音频配置测试"""
def test_default_config(self):
"""测试默认配置"""
config = AudioConfig()
assert config.sample_rate == 16000
assert config.channels == 1
assert config.chunk_duration == 0.02
assert config.bits_per_sample == 16
def test_chunk_size_calculation(self):
"""测试音频块大小计算"""
config = AudioConfig(sample_rate=16000, chunk_duration=0.02)
# 16000 * 0.02 = 320 samples
assert config.chunk_size == 320
def test_bytes_per_chunk_calculation(self):
"""测试每块字节数计算"""
config = AudioConfig(
sample_rate=16000,
channels=1,
chunk_duration=0.02,
bits_per_sample=16
)
# 320 samples * 1 channel * 2 bytes = 640 bytes
assert config.bytes_per_chunk == 640
class TestCallInfo:
"""通话信息测试"""
def test_call_info_creation(self):
"""测试通话信息创建"""
info = CallInfo(
peer_id="user123",
peer_name="Test User",
is_outgoing=True
)
assert info.peer_id == "user123"
assert info.peer_name == "Test User"
assert info.is_outgoing is True
assert info.start_time is None
def test_call_duration_not_started(self):
"""测试未开始通话的时长"""
info = CallInfo(peer_id="user123", peer_name="Test")
assert info.duration == 0.0
def test_call_duration_started(self):
"""测试已开始通话的时长"""
info = CallInfo(peer_id="user123", peer_name="Test")
info.start_time = datetime.now()
time.sleep(0.1)
assert info.duration >= 0.1
class TestNetworkStats:
"""网络统计测试"""
def test_default_stats(self):
"""测试默认统计值"""
stats = NetworkStats()
assert stats.packets_sent == 0
assert stats.packets_received == 0
assert stats.packets_lost == 0
assert stats.avg_latency == 0.0
def test_packet_loss_rate_no_packets(self):
"""测试无数据包时的丢包率"""
stats = NetworkStats()
assert stats.packet_loss_rate == 0.0
def test_packet_loss_rate_with_loss(self):
"""测试有丢包时的丢包率"""
stats = NetworkStats(
packets_sent=100,
packets_received=90,
packets_lost=10
)
# 10 / (100 + 90) ≈ 0.0526
assert 0.05 < stats.packet_loss_rate < 0.06
def test_network_quality_excellent(self):
"""测试优秀网络质量"""
stats = NetworkStats(avg_latency=30)
assert stats.get_quality() == NetworkQuality.EXCELLENT
def test_network_quality_good(self):
"""测试良好网络质量"""
stats = NetworkStats(avg_latency=75)
assert stats.get_quality() == NetworkQuality.GOOD
def test_network_quality_fair(self):
"""测试一般网络质量"""
stats = NetworkStats(avg_latency=150)
assert stats.get_quality() == NetworkQuality.FAIR
def test_network_quality_poor(self):
"""测试较差网络质量"""
stats = NetworkStats(avg_latency=250)
assert stats.get_quality() == NetworkQuality.POOR
def test_network_quality_bad(self):
"""测试很差网络质量"""
stats = NetworkStats(avg_latency=400)
assert stats.get_quality() == NetworkQuality.BAD
class TestJitterBuffer:
"""抖动缓冲区测试"""
def test_buffer_creation(self):
"""测试缓冲区创建"""
buffer = JitterBuffer()
assert buffer.size == 0
assert buffer.delay == 0.0
def test_push_and_pop(self):
"""测试数据包入队和出队"""
buffer = JitterBuffer(target_delay=0.0) # 禁用延迟等待
# 添加数据包
buffer.push(1, b"audio_data_1", time.time() - 0.1)
buffer.push(2, b"audio_data_2", time.time() - 0.05)
buffer.push(3, b"audio_data_3", time.time())
assert buffer.size == 3
# 取出数据包（按序列号顺序）
data = buffer.pop()
assert data == b"audio_data_1"
assert buffer.size == 2
def test_out_of_order_packets(self):
"""测试乱序数据包处理"""
buffer = JitterBuffer(target_delay=0.0)
# 乱序添加
buffer.push(3, b"data_3", time.time())
buffer.push(1, b"data_1", time.time())
buffer.push(2, b"data_2", time.time())
# 应该按序列号顺序取出
assert buffer.pop() == b"data_1"
assert buffer.pop() == b"data_2"
assert buffer.pop() == b"data_3"
def test_duplicate_packet_ignored(self):
"""测试重复数据包被忽略"""
buffer = JitterBuffer(target_delay=0.0)
buffer.push(1, b"data_1", time.time())
buffer.push(1, b"data_1_dup", time.time()) # 重复
assert buffer.size == 1
def test_old_packet_ignored(self):
"""测试过期数据包被忽略"""
buffer = JitterBuffer(target_delay=0.0)
buffer.push(5, b"data_5", time.time())
buffer.pop() # 取出序列号5
buffer.push(3, b"data_3", time.time()) # 旧包，应被忽略
assert buffer.size == 0
def test_clear_buffer(self):
"""测试清空缓冲区"""
buffer = JitterBuffer()
buffer.push(1, b"data", time.time())
buffer.push(2, b"data", time.time())
buffer.clear()
assert buffer.size == 0
class TestVoiceChatModule:
"""语音聊天模块测试"""
@pytest.fixture
def voice_chat(self):
"""创建语音聊天模块实例"""
module = VoiceChatModule()
module.set_user_info("test_user", "Test User")
return module
def test_initial_state(self, voice_chat):
"""测试初始状态"""
assert voice_chat.state == CallState.IDLE
assert voice_chat.is_in_call is False
assert voice_chat.is_muted is False
assert voice_chat.call_info is None
def test_mute_toggle(self, voice_chat):
"""测试静音切换"""
assert voice_chat.is_muted is False
voice_chat.mute(True)
assert voice_chat.is_muted is True
voice_chat.mute(False)
assert voice_chat.is_muted is False
def test_toggle_mute(self, voice_chat):
"""测试静音切换方法"""
assert voice_chat.toggle_mute() is True
assert voice_chat.toggle_mute() is False
def test_get_call_duration_no_call(self, voice_chat):
"""测试无通话时的时长"""
assert voice_chat.get_call_duration() == 0.0
def test_get_network_quality_default(self, voice_chat):
"""测试默认网络质量"""
# 默认延迟为0，应该是EXCELLENT
quality = voice_chat.get_network_quality()
assert quality == NetworkQuality.EXCELLENT
def test_state_callback(self, voice_chat):
"""测试状态回调"""
callback_called = []
def state_callback(state, reason):
callback_called.append((state, reason))
voice_chat.add_state_callback(state_callback)
voice_chat._set_state(CallState.CALLING, "test")
assert len(callback_called) == 1
assert callback_called[0][0] == CallState.CALLING
assert callback_called[0][1] == "test"
def test_remove_state_callback(self, voice_chat):
"""测试移除状态回调"""
callback_called = []
def state_callback(state, reason):
callback_called.append((state, reason))
voice_chat.add_state_callback(state_callback)
voice_chat.remove_state_callback(state_callback)
voice_chat._set_state(CallState.CALLING, "test")
assert len(callback_called) == 0
@pytest.mark.asyncio
async def test_start_call_no_callback(self, voice_chat):
"""测试无消息回调时发起通话"""
result = await voice_chat.start_call("peer123", "Peer User")
assert result is False
assert voice_chat.state == CallState.IDLE
@pytest.mark.asyncio
async def test_start_call_not_idle(self, voice_chat):
"""测试非空闲状态发起通话"""
voice_chat._state = CallState.CONNECTED
async def mock_send(peer_id, msg):
return True
voice_chat.set_send_message_callback(mock_send)
result = await voice_chat.start_call("peer123")
assert result is False
def test_reject_call_not_ringing(self, voice_chat):
"""测试非响铃状态拒绝通话"""
voice_chat.reject_call("peer123")
# 应该不会改变状态
assert voice_chat.state == CallState.IDLE
def test_end_call_idle(self, voice_chat):
"""测试空闲状态结束通话"""
voice_chat.end_call()
# 应该保持空闲状态
assert voice_chat.state == CallState.IDLE
@pytest.mark.asyncio
async def test_accept_call_not_ringing(self, voice_chat):
"""测试非响铃状态接听通话"""
result = await voice_chat.accept_call("peer123")
assert result is False
class TestAudioEncoderDecoder:
"""音频编解码器测试"""
def test_encoder_without_opus(self):
"""测试无Opus时的编码器"""
config = AudioConfig()
with patch.dict('sys.modules', {'opuslib': None}):
encoder = AudioEncoder(config)
# 应该回退到原始音频
assert encoder.is_opus_enabled is False or encoder._encoder is None
def test_decoder_without_opus(self):
"""测试无Opus时的解码器"""
config = AudioConfig()
with patch.dict('sys.modules', {'opuslib': None}):
decoder = AudioDecoder(config)
# 应该回退到原始音频
assert decoder.is_opus_enabled is False or decoder._decoder is None
def test_encoder_raw_passthrough(self):
"""测试编码器原始数据透传"""
config = AudioConfig()
encoder = AudioEncoder(config)
encoder._use_opus = False
test_data = b"test_audio_data"
result = encoder.encode(test_data)
assert result == test_data
def test_decoder_raw_passthrough(self):
"""测试解码器原始数据透传"""
config = AudioConfig()
decoder = AudioDecoder(config)
decoder._use_opus = False
test_data = b"test_audio_data"
result = decoder.decode(test_data)
assert result == test_data
def test_encoder_bitrate_property(self):
"""测试编码器比特率属性"""
config = AudioConfig()
encoder = AudioEncoder(config, bitrate=32000)
assert encoder.bitrate == 32000
class TestAudioPacketFormat:
"""音频数据包格式测试"""
def test_header_format(self):
"""测试数据包头格式"""
# 格式: 序列号(4字节) + 时间戳(8字节)
header_size = struct.calcsize("!Id")
assert header_size == VoiceChatModule.AUDIO_HEADER_SIZE
def test_pack_unpack_header(self):
"""测试数据包头打包和解包"""
sequence = 12345
timestamp = time.time()
# 打包
header = struct.pack("!Id", sequence, timestamp)
# 解包
unpacked_seq, unpacked_ts = struct.unpack("!Id", header)
assert unpacked_seq == sequence
assert abs(unpacked_ts - timestamp) < 0.001
if __name__ == "__main__":
pytest.main([__file__, "-v"])
Write Preview
Loading…
Cancel
Save