| | from unittest.mock import MagicMock |
| |
|
| | import pytest |
| |
|
| | import src.schemas.sandbox_models as dm |
| | from src.tools import AsyncPythonSandBoxTool, PythonSandBoxToolResponse |
| |
|
| | |
| |
|
| |
|
| | |
| | @pytest.fixture |
| | def mock_client(): |
| | client = MagicMock() |
| | return client |
| |
|
| |
|
| | |
| | @pytest.fixture |
| | def tool(mock_client): |
| | tool = AsyncPythonSandBoxTool(name="Sandbox", description="Remote code sandbox") |
| | tool.client = mock_client |
| | return tool |
| |
|
| |
|
| | def test_upload_file(tool, mock_client): |
| | mock_client.upload_file.return_value = dm.UploadOutput(code=0, message="succeed", data="/mnt/file.txt") |
| | response = tool.upload_file("/path/to/file.txt") |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | assert response.output_text == "Successfully uploaded file to server's path: /mnt/file.txt" |
| |
|
| |
|
| | def test_download_file(tool, mock_client): |
| | mock_client.download_file.return_value = dm.DownloadSuccessOutput(file_name="file.txt", content="hello world") |
| | response = tool.download_file("file.txt") |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | assert response.output_text == "Successfully downloaded file: file.txt" |
| | assert response.raw_output.content == "hello world" |
| |
|
| | def test_extract_code(tool): |
| | test_str = '''```python |
| | import pandas as pd |
| | |
| | # Load the data |
| | df = pd.read_csv('/mnt/0020400390.csv') |
| | |
| | # Checking the first few records |
| | df.head() |
| | ```''' |
| |
|
| | test_str2 = '''``` |
| | import pandas as pd |
| | |
| | # Load the data |
| | df = pd.read_csv('/mnt/0020400390.csv') |
| | |
| | # Checking the first few records |
| | df.head() |
| | ```''' |
| |
|
| | extracted_code = tool.input_handler(test_str) |
| | assert extracted_code == '''import pandas as pd |
| | |
| | # Load the data |
| | df = pd.read_csv('/mnt/0020400390.csv') |
| | |
| | # Checking the first few records |
| | df.head()''' |
| |
|
| | extracted_code2 = tool.input_handler(test_str2) |
| | assert extracted_code2 == '''import pandas as pd |
| | |
| | # Load the data |
| | df = pd.read_csv('/mnt/0020400390.csv') |
| | |
| | # Checking the first few records |
| | df.head()''' |
| |
|
| |
|
| | def test_run_code_success(tool, mock_client): |
| | |
| | code = "```python\nabc\n```" |
| | mock_client.run_code.return_value = dm.RunCodeOutput( |
| | code=0, |
| | message="succeed", |
| | data=dm.CodeRunData(is_partial=False, |
| | result=dm.CodeRunResult( |
| | code_output_result=[dm.CodeOutput(type="stdout", content="hello\n")], |
| | deleted_files=[], |
| | new_generated_files=[]))) |
| | response = tool.run(code) |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | assert response.output_text == "Ran code\nSTDOUT:\nhello\n\n" |
| |
|
| | |
| | code = "```python\nabc\n```" |
| | mock_client.run_code.return_value = dm.RunCodeOutput(code=0, |
| | message="succeed", |
| | data=dm.CodeRunData( |
| | is_partial=False, |
| | result=dm.CodeRunResult(code_output_result=[ |
| | dm.CodeOutput(type="image", |
| | content="https://example.com/image.png") |
| | ], |
| | deleted_files=[], |
| | new_generated_files=[]))) |
| | response = tool.run(code) |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | assert response.output_text == "Ran code\nGenerated an image: https://example.com/image.png\n" |
| |
|
| | |
| | code = "```python\nabc\n```" |
| | mock_client.run_code.return_value = dm.RunCodeOutput( |
| | code=0, |
| | message="succeed", |
| | data=dm.CodeRunData(is_partial=False, |
| | result=dm.CodeRunResult(code_output_result=[], |
| | deleted_files=[], |
| | new_generated_files=["/mnt/qr.jpg", "/mnt/qr2.jpg"]))) |
| | mock_client.download_file.return_value = dm.DownloadSuccessOutput(file_name="qr.jpg", content="hello world") |
| | response = tool.run(code) |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | file_str = ",".join(response.raw_output.data.result.new_generated_files) |
| | assert response.output_text == f"Ran code\nGenerated files on server: {file_str}\n" |
| |
|
| | |
| | code = "```python\nabc\n```" |
| | mock_client.run_code.return_value = dm.RunCodeOutput( |
| | code=0, |
| | message="succeed", |
| | data=dm.CodeRunData(is_partial=False, |
| | result=dm.CodeRunResult(code_output_result=[], |
| | deleted_files=["/mnt/qr.jpg", "/mnt/qr2.jpg"], |
| | new_generated_files=[]))) |
| | response = tool.run(code) |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | assert response.output_text == "Ran code\nDeleted files from server: /mnt/qr.jpg,/mnt/qr2.jpg\n" |
| |
|
| | |
| | code = "```python\nabc\n```" |
| | mock_client.run_code.return_value = dm.RunCodeOutput( |
| | code=0, |
| | message="succeed", |
| | data=dm.CodeRunData(is_partial=False, |
| | result=dm.CodeRunResult(code_output_result=[ |
| | dm.CodeOutput(type="image", content="https://example.com/image.png"), |
| | dm.CodeOutput(type="stdout", content="Plotted and saved files"), |
| | dm.CodeOutput(type="stderr", content="Something is deprecated") |
| | ], |
| | deleted_files=["/mnt/qr.jpg", "/mnt/qr2.jpg"], |
| | new_generated_files=["/mnt/plot.png", "/mnt/plot2.png"]))) |
| | response = tool.run(code) |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | generated_file_str = ",".join(response.raw_output.data.result.new_generated_files) |
| | expected_str = "Ran code\n" + \ |
| | "Deleted files from server: /mnt/qr.jpg,/mnt/qr2.jpg\n" + \ |
| | f"Generated files on server: {generated_file_str}\n" + \ |
| | "Generated an image: https://example.com/image.png\n" + \ |
| | "STDOUT:\n" + \ |
| | "Plotted and saved files\n" + \ |
| | "STDERR:\n" + \ |
| | "Something is deprecated\n" |
| | assert response.output_text == expected_str |
| |
|
| |
|
| | def test_run_code_timeout(tool, mock_client): |
| | server_return = { |
| | "code": 0, |
| | "data": { |
| | "is_partial": True, |
| | "result": { |
| | "code_output_result": [], |
| | "deleted_files": [], |
| | "new_generated_files": [] |
| | } |
| | }, |
| | "message": "the code doesn't finish in timeout value 3" |
| | } |
| | mock_client.run_code.return_value = dm.RunCodeOutput(**server_return) |
| | response = tool.run("```python\nabc\n```") |
| | assert isinstance(response, PythonSandBoxToolResponse) |
| | expected_str = "Ran code but was not fully successful\n" + \ |
| | "What happened: the code doesn\'t finish in timeout value 3\n" |
| | assert response.output_text == expected_str |
| |
|