# 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