Improve logic and error messages for file reading and writing with Python code (#4567)
* Fix issues with file reading and writing with Python code * Change error message, use Workspace.get_path --------- Co-authored-by: Reinier van der Leer <github@pwuts.nl>pull/2821/head^2
parent
20a4922b40
commit
fdc6e12945
|
@ -9,6 +9,8 @@ from docker.errors import ImageNotFound
|
|||
from autogpt.commands.command import command
|
||||
from autogpt.config import Config
|
||||
from autogpt.logs import logger
|
||||
from autogpt.setup import CFG
|
||||
from autogpt.workspace.workspace import Workspace
|
||||
|
||||
|
||||
@command("execute_python_file", "Execute Python File", '"filename": "<filename>"')
|
||||
|
@ -21,17 +23,28 @@ def execute_python_file(filename: str, config: Config) -> str:
|
|||
Returns:
|
||||
str: The output of the file
|
||||
"""
|
||||
logger.info(f"Executing file '{filename}'")
|
||||
logger.info(
|
||||
f"Executing python file '{filename}' in working directory '{CFG.workspace_path}'"
|
||||
)
|
||||
|
||||
if not filename.endswith(".py"):
|
||||
return "Error: Invalid file type. Only .py files are allowed."
|
||||
|
||||
if not os.path.isfile(filename):
|
||||
return f"Error: File '{filename}' does not exist."
|
||||
workspace = Workspace(config.workspace_path, config.restrict_to_workspace)
|
||||
|
||||
path = workspace.get_path(filename)
|
||||
if not path.is_file():
|
||||
# Mimic the response that you get from the command line so that it's easier to identify
|
||||
return (
|
||||
f"python: can't open file '{filename}': [Errno 2] No such file or directory"
|
||||
)
|
||||
|
||||
if we_are_running_in_a_docker_container():
|
||||
result = subprocess.run(
|
||||
["python", filename], capture_output=True, encoding="utf8"
|
||||
["python", str(path)],
|
||||
capture_output=True,
|
||||
encoding="utf8",
|
||||
cwd=CFG.workspace_path,
|
||||
)
|
||||
if result.returncode == 0:
|
||||
return result.stdout
|
||||
|
@ -63,7 +76,7 @@ def execute_python_file(filename: str, config: Config) -> str:
|
|||
logger.info(status)
|
||||
container = client.containers.run(
|
||||
image_name,
|
||||
["python", str(Path(filename).relative_to(config.workspace_path))],
|
||||
["python", str(path.relative_to(workspace.root))],
|
||||
volumes={
|
||||
config.workspace_path: {
|
||||
"bind": "/workspace",
|
||||
|
|
|
@ -146,7 +146,9 @@ def is_file_binary_fn(file_path: str):
|
|||
|
||||
def read_textual_file(file_path: str, logger: logs.Logger) -> str:
|
||||
if not os.path.isfile(file_path):
|
||||
raise FileNotFoundError(f"{file_path} not found!")
|
||||
raise FileNotFoundError(
|
||||
f"read_file {file_path} failed: no such file or directory"
|
||||
)
|
||||
is_binary = is_file_binary_fn(file_path)
|
||||
file_extension = os.path.splitext(file_path)[1].lower()
|
||||
parser = extension_to_parser.get(file_extension)
|
||||
|
|
|
@ -1,6 +1,7 @@
|
|||
import random
|
||||
import string
|
||||
import tempfile
|
||||
from typing import Callable
|
||||
|
||||
import pytest
|
||||
from pytest_mock import MockerFixture
|
||||
|
@ -10,12 +11,12 @@ from autogpt.config import Config
|
|||
|
||||
|
||||
@pytest.fixture
|
||||
def config_allow_execute(config: Config, mocker: MockerFixture):
|
||||
def config_allow_execute(config: Config, mocker: MockerFixture) -> Callable:
|
||||
yield mocker.patch.object(config, "execute_local_commands", True)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def python_test_file(config: Config, random_string):
|
||||
def python_test_file(config: Config, random_string) -> Callable:
|
||||
temp_file = tempfile.NamedTemporaryFile(dir=config.workspace_path, suffix=".py")
|
||||
temp_file.write(str.encode(f"print('Hello {random_string}!')"))
|
||||
temp_file.flush()
|
||||
|
@ -34,18 +35,24 @@ def test_execute_python_file(python_test_file: str, random_string: str, config):
|
|||
assert result.replace("\r", "") == f"Hello {random_string}!\n"
|
||||
|
||||
|
||||
def test_execute_python_file_invalid(config):
|
||||
def test_execute_python_file_invalid(config: Config):
|
||||
assert all(
|
||||
s in sut.execute_python_file("not_python", config).lower()
|
||||
for s in ["error:", "invalid", ".py"]
|
||||
)
|
||||
|
||||
|
||||
def test_execute_python_file_not_found(config: Config):
|
||||
assert all(
|
||||
s in sut.execute_python_file("notexist.py", config).lower()
|
||||
for s in ["error:", "does not exist"]
|
||||
for s in [
|
||||
"python: can't open file 'notexist.py'",
|
||||
"[errno 2] no such file or directory",
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def test_execute_shell(config_allow_execute, random_string, config):
|
||||
def test_execute_shell(config_allow_execute: bool, random_string: str, config: Config):
|
||||
result = sut.execute_shell(f"echo 'Hello {random_string}!'", config)
|
||||
assert f"Hello {random_string}!" in result
|
||||
|
||||
|
|
|
@ -229,6 +229,12 @@ def test_read_file(
|
|||
assert content.replace("\r", "") == file_content
|
||||
|
||||
|
||||
def test_read_file_not_found(config: Config):
|
||||
filename = "does_not_exist.txt"
|
||||
content = file_ops.read_file(filename, config)
|
||||
assert "Error:" in content and filename in content and "no such file" in content
|
||||
|
||||
|
||||
def test_write_to_file(test_file_path: Path, config):
|
||||
new_content = "This is new content.\n"
|
||||
file_ops.write_to_file(str(test_file_path), new_content, config)
|
||||
|
|
Loading…
Reference in New Issue