Method test_receive_file

tests/aip/test_binary_transfer.py:351–414

Test receive_file method

(self)

    @pytest.mark.asyncio
    async def test_receive_file(self):
        """Test receive_file method"""
        import json

        mock_transport = MagicMock(spec=WebSocketTransport)

        # Prepare file transfer messages
        start_msg = {
            "type": "file_transfer_start",
            "filename": "test.bin",
            "size": 2048,
            "chunk_size": 1024,
            "total_chunks": 2,
        }

        chunk1_meta = {"type": "binary_data", "chunk_num": 0, "size": 1024}
        chunk2_meta = {"type": "binary_data", "chunk_num": 1, "size": 1024}

        complete_msg = {
            "type": "file_transfer_complete",
            "filename": "test.bin",
            "total_chunks": 2,
            "checksum": "abc123",
        }

        # Mock transport to return messages in sequence
        mock_transport.receive = AsyncMock(
            side_effect=[
                json.dumps(start_msg).encode("utf-8"),
                json.dumps(chunk1_meta).encode("utf-8"),
                json.dumps(chunk2_meta).encode("utf-8"),
                json.dumps(complete_msg).encode("utf-8"),
            ]
        )

        mock_transport.receive_binary = AsyncMock(
            side_effect=[
                b"A" * 1024,  # Chunk 1
                b"B" * 1024,  # Chunk 2
            ]
        )

        protocol = AIPProtocol(mock_transport)

        # Receive file
        with tempfile.NamedTemporaryFile(delete=False, suffix=".bin") as temp_file:
            output_path = temp_file.name

        try:
            metadata = await protocol.receive_file(output_path, validate_checksum=False)

            assert metadata["filename"] == "test.bin"
            assert metadata["size"] == 2048

            # Verify file was written
            with open(output_path, "rb") as f:
                content = f.read()
                assert len(content) == 2048