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.
251 lines
7.8 KiB
251 lines
7.8 KiB
# P2P Network Communication - Data Models Tests
|
|
"""
|
|
测试数据模型的基本功能
|
|
"""
|
|
|
|
import pytest
|
|
from datetime import datetime
|
|
from shared.models import (
|
|
MessageType, UserStatus, TransferStatus, ConnectionMode, NetworkQuality,
|
|
Message, UserInfo, ChatMessage, FileChunk, TransferProgress, FileTransferRecord, PeerInfo
|
|
)
|
|
|
|
|
|
class TestMessageType:
|
|
"""测试消息类型枚举"""
|
|
|
|
def test_message_type_values(self):
|
|
"""测试消息类型枚举值"""
|
|
assert MessageType.TEXT.value == "text"
|
|
assert MessageType.FILE_REQUEST.value == "file_request"
|
|
assert MessageType.HEARTBEAT.value == "heartbeat"
|
|
assert MessageType.VOICE_DATA.value == "voice_data"
|
|
|
|
|
|
class TestUserStatus:
|
|
"""测试用户状态枚举"""
|
|
|
|
def test_user_status_values(self):
|
|
"""测试用户状态枚举值"""
|
|
assert UserStatus.ONLINE.value == "online"
|
|
assert UserStatus.OFFLINE.value == "offline"
|
|
assert UserStatus.BUSY.value == "busy"
|
|
assert UserStatus.AWAY.value == "away"
|
|
|
|
|
|
class TestTransferStatus:
|
|
"""测试传输状态枚举"""
|
|
|
|
def test_transfer_status_values(self):
|
|
"""测试传输状态枚举值"""
|
|
assert TransferStatus.PENDING.value == "pending"
|
|
assert TransferStatus.IN_PROGRESS.value == "in_progress"
|
|
assert TransferStatus.COMPLETED.value == "completed"
|
|
assert TransferStatus.FAILED.value == "failed"
|
|
assert TransferStatus.CANCELLED.value == "cancelled"
|
|
|
|
|
|
class TestMessage:
|
|
"""测试Message数据类"""
|
|
|
|
def test_message_creation(self):
|
|
"""测试消息创建"""
|
|
msg = Message(
|
|
msg_type=MessageType.TEXT,
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
timestamp=1234567890.0,
|
|
payload=b"Hello, World!"
|
|
)
|
|
|
|
assert msg.msg_type == MessageType.TEXT
|
|
assert msg.sender_id == "user1"
|
|
assert msg.receiver_id == "user2"
|
|
assert msg.timestamp == 1234567890.0
|
|
assert msg.payload == b"Hello, World!"
|
|
assert msg.checksum != ""
|
|
assert msg.message_id != ""
|
|
|
|
def test_message_checksum_verification(self):
|
|
"""测试消息校验和验证"""
|
|
msg = Message(
|
|
msg_type=MessageType.TEXT,
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
timestamp=1234567890.0,
|
|
payload=b"Test message"
|
|
)
|
|
|
|
assert msg.verify_checksum() is True
|
|
|
|
def test_message_to_dict_and_from_dict(self):
|
|
"""测试消息字典转换"""
|
|
original = Message(
|
|
msg_type=MessageType.TEXT,
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
timestamp=1234567890.0,
|
|
payload=b"Test"
|
|
)
|
|
|
|
msg_dict = original.to_dict()
|
|
restored = Message.from_dict(msg_dict)
|
|
|
|
assert restored.msg_type == original.msg_type
|
|
assert restored.sender_id == original.sender_id
|
|
assert restored.receiver_id == original.receiver_id
|
|
assert restored.timestamp == original.timestamp
|
|
assert restored.payload == original.payload
|
|
assert restored.checksum == original.checksum
|
|
assert restored.message_id == original.message_id
|
|
|
|
def test_message_serialize_deserialize(self):
|
|
"""测试消息序列化和反序列化"""
|
|
original = Message(
|
|
msg_type=MessageType.FILE_REQUEST,
|
|
sender_id="sender",
|
|
receiver_id="receiver",
|
|
timestamp=9876543210.0,
|
|
payload=b"file_data_here"
|
|
)
|
|
|
|
serialized = original.serialize()
|
|
restored = Message.deserialize(serialized)
|
|
|
|
assert restored.msg_type == original.msg_type
|
|
assert restored.sender_id == original.sender_id
|
|
assert restored.receiver_id == original.receiver_id
|
|
assert restored.payload == original.payload
|
|
|
|
|
|
class TestUserInfo:
|
|
"""测试UserInfo数据类"""
|
|
|
|
def test_user_info_creation(self):
|
|
"""测试用户信息创建"""
|
|
user = UserInfo(
|
|
user_id="uid123",
|
|
username="testuser",
|
|
display_name="Test User"
|
|
)
|
|
|
|
assert user.user_id == "uid123"
|
|
assert user.username == "testuser"
|
|
assert user.display_name == "Test User"
|
|
assert user.status == UserStatus.OFFLINE
|
|
|
|
def test_user_info_serialize_deserialize(self):
|
|
"""测试用户信息序列化和反序列化"""
|
|
original = UserInfo(
|
|
user_id="uid456",
|
|
username="user456",
|
|
display_name="User 456",
|
|
status=UserStatus.ONLINE,
|
|
ip_address="192.168.1.100",
|
|
port=8888
|
|
)
|
|
|
|
serialized = original.serialize()
|
|
restored = UserInfo.deserialize(serialized)
|
|
|
|
assert restored.user_id == original.user_id
|
|
assert restored.username == original.username
|
|
assert restored.status == original.status
|
|
assert restored.ip_address == original.ip_address
|
|
assert restored.port == original.port
|
|
|
|
|
|
class TestChatMessage:
|
|
"""测试ChatMessage数据类"""
|
|
|
|
def test_chat_message_creation(self):
|
|
"""测试聊天消息创建"""
|
|
chat_msg = ChatMessage(
|
|
message_id="msg001",
|
|
sender_id="user1",
|
|
receiver_id="user2",
|
|
content_type=MessageType.TEXT,
|
|
content="Hello!"
|
|
)
|
|
|
|
assert chat_msg.message_id == "msg001"
|
|
assert chat_msg.content == "Hello!"
|
|
assert chat_msg.is_read is False
|
|
assert chat_msg.is_sent is False
|
|
|
|
def test_chat_message_serialize_deserialize(self):
|
|
"""测试聊天消息序列化和反序列化"""
|
|
original = ChatMessage(
|
|
message_id="msg002",
|
|
sender_id="sender",
|
|
receiver_id="receiver",
|
|
content_type=MessageType.IMAGE,
|
|
content="/path/to/image.png",
|
|
is_read=True,
|
|
is_sent=True
|
|
)
|
|
|
|
serialized = original.serialize()
|
|
restored = ChatMessage.deserialize(serialized)
|
|
|
|
assert restored.message_id == original.message_id
|
|
assert restored.content_type == original.content_type
|
|
assert restored.content == original.content
|
|
assert restored.is_read == original.is_read
|
|
assert restored.is_sent == original.is_sent
|
|
|
|
|
|
class TestFileChunk:
|
|
"""测试FileChunk数据类"""
|
|
|
|
def test_file_chunk_creation(self):
|
|
"""测试文件块创建"""
|
|
chunk = FileChunk(
|
|
file_id="file001",
|
|
chunk_index=0,
|
|
total_chunks=10,
|
|
data=b"chunk_data_here"
|
|
)
|
|
|
|
assert chunk.file_id == "file001"
|
|
assert chunk.chunk_index == 0
|
|
assert chunk.total_chunks == 10
|
|
assert chunk.checksum != ""
|
|
|
|
def test_file_chunk_checksum_verification(self):
|
|
"""测试文件块校验和验证"""
|
|
chunk = FileChunk(
|
|
file_id="file002",
|
|
chunk_index=1,
|
|
total_chunks=5,
|
|
data=b"test_chunk_data"
|
|
)
|
|
|
|
assert chunk.verify_checksum() is True
|
|
|
|
|
|
class TestTransferProgress:
|
|
"""测试TransferProgress数据类"""
|
|
|
|
def test_transfer_progress_percent(self):
|
|
"""测试传输进度百分比计算"""
|
|
progress = TransferProgress(
|
|
file_id="file001",
|
|
file_name="test.txt",
|
|
total_size=1000,
|
|
transferred_size=500
|
|
)
|
|
|
|
assert progress.progress_percent == 50.0
|
|
|
|
def test_transfer_progress_zero_total(self):
|
|
"""测试总大小为0时的进度计算"""
|
|
progress = TransferProgress(
|
|
file_id="file002",
|
|
file_name="empty.txt",
|
|
total_size=0,
|
|
transferred_size=0
|
|
)
|
|
|
|
assert progress.progress_percent == 0.0
|