Test send_file method
(self)
| 317 | |
| 318 | @pytest.mark.asyncio |
| 319 | async def test_send_file(self): |
| 320 | """Test send_file method""" |
| 321 | mock_transport = MagicMock(spec=WebSocketTransport) |
| 322 | mock_transport.send = AsyncMock() |
| 323 | mock_transport.send_binary = AsyncMock() |
| 324 | |
| 325 | protocol = AIPProtocol(mock_transport) |
| 326 | |
| 327 | # Create a temporary test file |
| 328 | with tempfile.NamedTemporaryFile(delete=False, suffix=".txt") as temp_file: |
| 329 | temp_file.write(b"Test file content for chunked transfer" * 1000) |
| 330 | temp_file_path = temp_file.name |
| 331 | |
| 332 | try: |
| 333 | await protocol.send_file(temp_file_path, chunk_size=1024) |
| 334 | |
| 335 | # Verify file_transfer_start was sent |
| 336 | assert mock_transport.send.called |
| 337 | start_msg = mock_transport.send.call_args_list[0][0][0] |
| 338 | assert b'"type": "file_transfer_start"' in start_msg |
| 339 | |
| 340 | # Verify chunks were sent |
| 341 | assert mock_transport.send_binary.called |
| 342 | |
| 343 | # Verify file_transfer_complete was sent |
| 344 | complete_msg = mock_transport.send.call_args_list[-1][0][0] |
| 345 | assert b'"type": "file_transfer_complete"' in complete_msg |
| 346 | |
| 347 | finally: |
| 348 | os.unlink(temp_file_path) |
| 349 | |
| 350 | @pytest.mark.asyncio |
| 351 | async def test_receive_file(self): |
nothing calls this directly
no test coverage detected