You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
644 lines
18 KiB
644 lines
18 KiB
# P2P Network Communication - Integration Tests
|
|
"""
|
|
客户端-服务器集成测试
|
|
测试端到端消息传递、文件传输和语音通话流程
|
|
|
|
需求: 全部
|
|
"""
|
|
|
|
import asyncio
|
|
import pytest
|
|
import time
|
|
import uuid
|
|
import os
|
|
import tempfile
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
from datetime import datetime
|
|
|
|
from client.app import P2PClientApp, create_client, quick_connect
|
|
from client.connection_manager import ConnectionManager, ConnectionState, Connection
|
|
from server.relay_server import RelayServer, ClientConnection
|
|
from shared.models import (
|
|
Message, MessageType, UserInfo, UserStatus,
|
|
ChatMessage, ConnectionMode, TransferProgress
|
|
)
|
|
from shared.message_handler import MessageHandler
|
|
from config import ClientConfig, ServerConfig
|
|
|
|
|
|
class TestP2PClientAppInit:
|
|
"""测试客户端应用程序初始化"""
|
|
|
|
def test_init_with_default_config(self):
|
|
"""测试使用默认配置初始化"""
|
|
app = P2PClientApp()
|
|
|
|
assert app.config is not None
|
|
assert app.user_info is None
|
|
assert app.is_connected is False
|
|
assert app.connection_state == ConnectionState.DISCONNECTED
|
|
|
|
def test_init_with_custom_config(self):
|
|
"""测试使用自定义配置初始化"""
|
|
config = ClientConfig(
|
|
server_host="192.168.1.100",
|
|
server_port=9999
|
|
)
|
|
app = P2PClientApp(config)
|
|
|
|
assert app.config.server_host == "192.168.1.100"
|
|
assert app.config.server_port == 9999
|
|
|
|
def test_components_initialized(self):
|
|
"""测试组件初始化"""
|
|
app = P2PClientApp()
|
|
|
|
assert app.connection_manager is not None
|
|
assert app.file_transfer is not None
|
|
assert app.image_processor is not None
|
|
assert app.media_player is not None
|
|
# voice_chat 在 start() 后初始化
|
|
assert app.voice_chat is None
|
|
|
|
|
|
class TestCallbackRegistration:
|
|
"""测试回调注册"""
|
|
|
|
def test_message_callback_registration(self):
|
|
"""测试消息回调注册"""
|
|
app = P2PClientApp()
|
|
callbacks_called = []
|
|
|
|
def callback(msg):
|
|
callbacks_called.append(msg)
|
|
|
|
app.add_message_callback(callback)
|
|
assert callback in app._message_callbacks
|
|
|
|
app.remove_message_callback(callback)
|
|
assert callback not in app._message_callbacks
|
|
|
|
def test_state_callback_registration(self):
|
|
"""测试状态回调注册"""
|
|
app = P2PClientApp()
|
|
callbacks_called = []
|
|
|
|
def callback(state, reason):
|
|
callbacks_called.append((state, reason))
|
|
|
|
app.add_state_callback(callback)
|
|
assert callback in app._state_callbacks
|
|
|
|
app.remove_state_callback(callback)
|
|
assert callback not in app._state_callbacks
|
|
|
|
def test_user_list_callback_registration(self):
|
|
"""测试用户列表回调注册"""
|
|
app = P2PClientApp()
|
|
callbacks_called = []
|
|
|
|
def callback(users):
|
|
callbacks_called.append(users)
|
|
|
|
app.add_user_list_callback(callback)
|
|
assert callback in app._user_list_callbacks
|
|
|
|
app.remove_user_list_callback(callback)
|
|
assert callback not in app._user_list_callbacks
|
|
|
|
|
|
class TestChatHistory:
|
|
"""测试聊天历史管理"""
|
|
|
|
def test_add_to_history(self):
|
|
"""测试添加消息到历史"""
|
|
app = P2PClientApp()
|
|
|
|
msg = ChatMessage(
|
|
message_id="msg1",
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
content_type=MessageType.TEXT,
|
|
content="Hello",
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
app._add_to_history("user1", msg)
|
|
|
|
history = app.get_chat_history("user1")
|
|
assert len(history) == 1
|
|
assert history[0].content == "Hello"
|
|
|
|
def test_history_sorted_by_time(self):
|
|
"""测试历史按时间排序"""
|
|
app = P2PClientApp()
|
|
|
|
msg1 = ChatMessage(
|
|
message_id="msg1",
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
content_type=MessageType.TEXT,
|
|
content="First",
|
|
timestamp=datetime(2024, 1, 1, 10, 0, 0)
|
|
)
|
|
|
|
msg2 = ChatMessage(
|
|
message_id="msg2",
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
content_type=MessageType.TEXT,
|
|
content="Second",
|
|
timestamp=datetime(2024, 1, 1, 9, 0, 0) # 更早的时间
|
|
)
|
|
|
|
app._add_to_history("user1", msg1)
|
|
app._add_to_history("user1", msg2)
|
|
|
|
history = app.get_chat_history("user1")
|
|
assert len(history) == 2
|
|
assert history[0].content == "Second" # 更早的消息在前
|
|
assert history[1].content == "First"
|
|
|
|
def test_clear_history(self):
|
|
"""测试清除历史"""
|
|
app = P2PClientApp()
|
|
|
|
msg = ChatMessage(
|
|
message_id="msg1",
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
content_type=MessageType.TEXT,
|
|
content="Hello",
|
|
timestamp=datetime.now()
|
|
)
|
|
|
|
app._add_to_history("user1", msg)
|
|
assert len(app.get_chat_history("user1")) == 1
|
|
|
|
app.clear_chat_history("user1")
|
|
assert len(app.get_chat_history("user1")) == 0
|
|
|
|
def test_get_empty_history(self):
|
|
"""测试获取空历史"""
|
|
app = P2PClientApp()
|
|
|
|
history = app.get_chat_history("nonexistent")
|
|
assert history == []
|
|
|
|
|
|
class TestMessageHandling:
|
|
"""测试消息处理"""
|
|
|
|
def test_handle_text_message(self):
|
|
"""测试处理文本消息"""
|
|
app = P2PClientApp()
|
|
app._user_info = UserInfo(
|
|
user_id="user2",
|
|
username="user2",
|
|
display_name="User 2"
|
|
)
|
|
|
|
message = Message(
|
|
msg_type=MessageType.TEXT,
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
timestamp=time.time(),
|
|
payload=b"Hello, World!"
|
|
)
|
|
|
|
app._handle_text_message(message)
|
|
|
|
history = app.get_chat_history("user1")
|
|
assert len(history) == 1
|
|
assert history[0].content == "Hello, World!"
|
|
assert history[0].sender_id == "user1"
|
|
|
|
def test_handle_user_list_response(self):
|
|
"""测试处理用户列表响应"""
|
|
import json
|
|
|
|
app = P2PClientApp()
|
|
|
|
users_data = [
|
|
{
|
|
"user_id": "user1",
|
|
"username": "user1",
|
|
"display_name": "User 1",
|
|
"status": "online"
|
|
},
|
|
{
|
|
"user_id": "user2",
|
|
"username": "user2",
|
|
"display_name": "User 2",
|
|
"status": "online"
|
|
}
|
|
]
|
|
|
|
message = Message(
|
|
msg_type=MessageType.USER_LIST_RESPONSE,
|
|
sender_id="server",
|
|
receiver_id="me",
|
|
timestamp=time.time(),
|
|
payload=json.dumps(users_data).encode('utf-8')
|
|
)
|
|
|
|
app._handle_user_list_response(message)
|
|
|
|
assert len(app._online_users) == 2
|
|
assert "user1" in app._online_users
|
|
assert "user2" in app._online_users
|
|
|
|
def test_handle_ack_message(self):
|
|
"""测试处理确认消息"""
|
|
app = P2PClientApp()
|
|
|
|
message = Message(
|
|
msg_type=MessageType.ACK,
|
|
sender_id="server",
|
|
receiver_id="me",
|
|
timestamp=time.time(),
|
|
payload=b"Message delivered"
|
|
)
|
|
|
|
# 不应该抛出异常
|
|
app._handle_ack_message(message)
|
|
|
|
def test_handle_error_message(self):
|
|
"""测试处理错误消息"""
|
|
app = P2PClientApp()
|
|
|
|
message = Message(
|
|
msg_type=MessageType.ERROR,
|
|
sender_id="server",
|
|
receiver_id="me",
|
|
timestamp=time.time(),
|
|
payload=b"User not found"
|
|
)
|
|
|
|
# 不应该抛出异常
|
|
app._handle_error_message(message)
|
|
|
|
|
|
class TestOnlineUsers:
|
|
"""测试在线用户管理"""
|
|
|
|
def test_get_online_users_empty(self):
|
|
"""测试获取空的在线用户列表"""
|
|
app = P2PClientApp()
|
|
|
|
users = app.get_online_users()
|
|
assert users == []
|
|
|
|
def test_get_user_info(self):
|
|
"""测试获取用户信息"""
|
|
app = P2PClientApp()
|
|
|
|
user = UserInfo(
|
|
user_id="user1",
|
|
username="user1",
|
|
display_name="User 1"
|
|
)
|
|
app._online_users["user1"] = user
|
|
|
|
info = app.get_user_info("user1")
|
|
assert info is not None
|
|
assert info.username == "user1"
|
|
|
|
info = app.get_user_info("nonexistent")
|
|
assert info is None
|
|
|
|
|
|
class TestConnectionMode:
|
|
"""测试连接模式"""
|
|
|
|
def test_get_connection_mode(self):
|
|
"""测试获取连接模式"""
|
|
app = P2PClientApp()
|
|
|
|
# 默认应该是中转模式
|
|
mode = app.get_connection_mode("unknown_peer")
|
|
assert mode == ConnectionMode.RELAY
|
|
|
|
|
|
class TestStats:
|
|
"""测试统计信息"""
|
|
|
|
def test_get_stats_initial(self):
|
|
"""测试获取初始统计信息"""
|
|
app = P2PClientApp()
|
|
|
|
stats = app.get_stats()
|
|
|
|
assert stats["user"] is None
|
|
assert stats["online_users"] == 0
|
|
assert stats["chat_sessions"] == 0
|
|
assert stats["total_messages"] == 0
|
|
assert stats["voice_call_active"] is False
|
|
|
|
def test_get_stats_with_data(self):
|
|
"""测试获取有数据的统计信息"""
|
|
app = P2PClientApp()
|
|
app._user_info = UserInfo(
|
|
user_id="user1",
|
|
username="testuser",
|
|
display_name="Test User"
|
|
)
|
|
|
|
# 添加一些聊天历史
|
|
msg = ChatMessage(
|
|
message_id="msg1",
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
content_type=MessageType.TEXT,
|
|
content="Hello",
|
|
timestamp=datetime.now()
|
|
)
|
|
app._add_to_history("user2", msg)
|
|
|
|
stats = app.get_stats()
|
|
|
|
assert stats["user"] == "testuser"
|
|
assert stats["chat_sessions"] == 1
|
|
assert stats["total_messages"] == 1
|
|
|
|
|
|
class TestCreateClient:
|
|
"""测试便捷函数"""
|
|
|
|
def test_create_client_default(self):
|
|
"""测试使用默认参数创建客户端"""
|
|
client = create_client()
|
|
|
|
assert client is not None
|
|
assert client.config.server_host == "127.0.0.1"
|
|
assert client.config.server_port == 8888
|
|
|
|
def test_create_client_custom(self):
|
|
"""测试使用自定义参数创建客户端"""
|
|
client = create_client(
|
|
server_host="192.168.1.100",
|
|
server_port=9999
|
|
)
|
|
|
|
assert client.config.server_host == "192.168.1.100"
|
|
assert client.config.server_port == 9999
|
|
|
|
|
|
class TestMessageFlow:
|
|
"""测试完整消息流程"""
|
|
|
|
@pytest.fixture
|
|
def mock_connection_manager(self):
|
|
"""创建模拟的连接管理器"""
|
|
cm = MagicMock(spec=ConnectionManager)
|
|
cm.is_connected = True
|
|
cm.state = ConnectionState.CONNECTED
|
|
cm.send_message = AsyncMock(return_value=True)
|
|
cm.get_connection_mode = MagicMock(return_value=ConnectionMode.RELAY)
|
|
return cm
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_text_message_flow(self, mock_connection_manager):
|
|
"""测试发送文本消息流程"""
|
|
app = P2PClientApp()
|
|
app._connection_manager = mock_connection_manager
|
|
app._user_info = UserInfo(
|
|
user_id="sender",
|
|
username="sender",
|
|
display_name="Sender"
|
|
)
|
|
|
|
success = await app.send_text_message("receiver", "Hello!")
|
|
|
|
assert success
|
|
mock_connection_manager.send_message.assert_called_once()
|
|
|
|
# 检查消息被添加到历史
|
|
history = app.get_chat_history("receiver")
|
|
assert len(history) == 1
|
|
assert history[0].content == "Hello!"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_send_text_message_not_logged_in(self):
|
|
"""测试未登录时发送消息"""
|
|
app = P2PClientApp()
|
|
app._user_info = None
|
|
|
|
success = await app.send_text_message("receiver", "Hello!")
|
|
|
|
assert not success
|
|
|
|
def test_receive_text_message_flow(self):
|
|
"""测试接收文本消息流程"""
|
|
app = P2PClientApp()
|
|
app._user_info = UserInfo(
|
|
user_id="receiver",
|
|
username="receiver",
|
|
display_name="Receiver"
|
|
)
|
|
|
|
# 模拟接收消息
|
|
message = Message(
|
|
msg_type=MessageType.TEXT,
|
|
sender_id="sender",
|
|
receiver_id="receiver",
|
|
timestamp=time.time(),
|
|
payload=b"Hello from sender!"
|
|
)
|
|
|
|
# 触发消息处理
|
|
app._on_message_received(message)
|
|
|
|
# 检查消息被添加到历史
|
|
history = app.get_chat_history("sender")
|
|
assert len(history) == 1
|
|
assert history[0].content == "Hello from sender!"
|
|
|
|
def test_message_callback_triggered(self):
|
|
"""测试消息回调被触发"""
|
|
app = P2PClientApp()
|
|
received_messages = []
|
|
|
|
def callback(msg):
|
|
received_messages.append(msg)
|
|
|
|
app.add_message_callback(callback)
|
|
|
|
message = Message(
|
|
msg_type=MessageType.TEXT,
|
|
sender_id="sender",
|
|
receiver_id="receiver",
|
|
timestamp=time.time(),
|
|
payload=b"Test"
|
|
)
|
|
|
|
app._on_message_received(message)
|
|
|
|
assert len(received_messages) == 1
|
|
assert received_messages[0].msg_type == MessageType.TEXT
|
|
|
|
|
|
class TestStateManagement:
|
|
"""测试状态管理"""
|
|
|
|
def test_state_callback_triggered(self):
|
|
"""测试状态回调被触发"""
|
|
app = P2PClientApp()
|
|
state_changes = []
|
|
|
|
def callback(state, reason):
|
|
state_changes.append((state, reason))
|
|
|
|
app.add_state_callback(callback)
|
|
|
|
# 模拟状态变化
|
|
app._on_connection_state_changed(ConnectionState.CONNECTING, "Test")
|
|
|
|
assert len(state_changes) == 1
|
|
assert state_changes[0][0] == ConnectionState.CONNECTING
|
|
assert state_changes[0][1] == "Test"
|
|
|
|
|
|
class TestMediaPlayback:
|
|
"""测试媒体播放"""
|
|
|
|
def test_pause_media(self):
|
|
"""测试暂停媒体"""
|
|
app = P2PClientApp()
|
|
|
|
# 不应该抛出异常
|
|
app.pause_media()
|
|
|
|
def test_stop_media(self):
|
|
"""测试停止媒体"""
|
|
app = P2PClientApp()
|
|
|
|
# 不应该抛出异常
|
|
app.stop_media()
|
|
|
|
def test_set_volume(self):
|
|
"""测试设置音量"""
|
|
app = P2PClientApp()
|
|
|
|
# 不应该抛出异常
|
|
app.set_volume(0.5)
|
|
|
|
|
|
class TestVoiceCall:
|
|
"""测试语音通话"""
|
|
|
|
def test_voice_call_not_initialized(self):
|
|
"""测试语音模块未初始化时的操作"""
|
|
app = P2PClientApp()
|
|
|
|
# voice_chat 在 start() 后才初始化
|
|
assert app.voice_chat is None
|
|
|
|
# 这些操作不应该抛出异常
|
|
app.reject_voice_call("peer")
|
|
app.end_voice_call()
|
|
app.mute_voice_call(True)
|
|
|
|
|
|
class TestEndToEndIntegration:
|
|
"""端到端集成测试"""
|
|
|
|
@pytest.fixture
|
|
def server_config(self):
|
|
"""服务器配置"""
|
|
return ServerConfig(
|
|
host="127.0.0.1",
|
|
port=18888, # 使用不同端口避免冲突
|
|
max_connections=10
|
|
)
|
|
|
|
@pytest.fixture
|
|
def client_config(self, server_config):
|
|
"""客户端配置"""
|
|
return ClientConfig(
|
|
server_host=server_config.host,
|
|
server_port=server_config.port
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_client_server_message_relay(self, server_config, client_config):
|
|
"""测试客户端-服务器消息中转"""
|
|
# 这是一个集成测试框架，实际测试需要启动真实服务器
|
|
# 这里只验证组件可以正确初始化
|
|
|
|
server = RelayServer(server_config)
|
|
client1 = P2PClientApp(client_config)
|
|
client2 = P2PClientApp(client_config)
|
|
|
|
assert server is not None
|
|
assert client1 is not None
|
|
assert client2 is not None
|
|
|
|
# 验证配置正确
|
|
assert server.port == client_config.server_port
|
|
assert client1.config.server_host == server_config.host
|
|
|
|
|
|
class TestComponentIntegration:
|
|
"""组件集成测试"""
|
|
|
|
def test_file_transfer_integration(self):
|
|
"""测试文件传输模块集成"""
|
|
app = P2PClientApp()
|
|
|
|
# 验证文件传输模块已集成
|
|
assert app.file_transfer is not None
|
|
assert app.file_transfer._send_message is not None
|
|
|
|
def test_image_processor_integration(self):
|
|
"""测试图片处理器集成"""
|
|
app = P2PClientApp()
|
|
|
|
# 验证图片处理器已集成
|
|
assert app.image_processor is not None
|
|
|
|
def test_media_player_integration(self):
|
|
"""测试媒体播放器集成"""
|
|
app = P2PClientApp()
|
|
|
|
# 验证媒体播放器已集成
|
|
assert app.media_player is not None
|
|
|
|
|
|
class TestUserListCallback:
|
|
"""测试用户列表回调"""
|
|
|
|
def test_user_list_callback_triggered(self):
|
|
"""测试用户列表回调被触发"""
|
|
import json
|
|
|
|
app = P2PClientApp()
|
|
received_users = []
|
|
|
|
def callback(users):
|
|
received_users.extend(users)
|
|
|
|
app.add_user_list_callback(callback)
|
|
|
|
users_data = [
|
|
{
|
|
"user_id": "user1",
|
|
"username": "user1",
|
|
"display_name": "User 1",
|
|
"status": "online"
|
|
}
|
|
]
|
|
|
|
message = Message(
|
|
msg_type=MessageType.USER_LIST_RESPONSE,
|
|
sender_id="server",
|
|
receiver_id="me",
|
|
timestamp=time.time(),
|
|
payload=json.dumps(users_data).encode('utf-8')
|
|
)
|
|
|
|
app._handle_user_list_response(message)
|
|
|
|
assert len(received_users) == 1
|
|
assert received_users[0].user_id == "user1"
|