Change name from LocalCommandlineCodeExecutor to LocalCommandLineCodeExecutor (#1873)

* Change name from LocalCommandlineCodeExecutor to LocalCommandLineCodeExecutor

* formatting

* name

* name

* CommandLineCodeResult rename too

* formatting
This commit is contained in:
Jack Gerrits 2024-03-06 10:41:19 -05:00 committed by GitHub
parent 09a4918921
commit d9078210e7
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 91 additions and 24 deletions

View File

@ -34,8 +34,8 @@ class CodeExecutorFactory:
return EmbeddedIPythonCodeExecutor(**code_execution_config.get("ipython-embedded", {}))
elif executor == "commandline-local":
from .local_commandline_code_executor import LocalCommandlineCodeExecutor
from .local_commandline_code_executor import LocalCommandLineCodeExecutor
return LocalCommandlineCodeExecutor(**code_execution_config.get("commandline-local", {}))
return LocalCommandLineCodeExecutor(**code_execution_config.get("commandline-local", {}))
else:
raise ValueError(f"Unknown code executor {executor}")

View File

@ -11,12 +11,12 @@ from .base import CodeBlock, CodeExtractor, CodeResult
from .markdown_code_extractor import MarkdownCodeExtractor
__all__ = (
"LocalCommandlineCodeExecutor",
"CommandlineCodeResult",
"LocalCommandLineCodeExecutor",
"CommandLineCodeResult",
)
class CommandlineCodeResult(CodeResult):
class CommandLineCodeResult(CodeResult):
"""(Experimental) A code result class for command line code executor."""
code_file: Optional[str] = Field(
@ -25,7 +25,7 @@ class CommandlineCodeResult(CodeResult):
)
class LocalCommandlineCodeExecutor(BaseModel):
class LocalCommandLineCodeExecutor(BaseModel):
"""(Experimental) A code executor class that executes code through a local command line
environment.
@ -49,7 +49,7 @@ class LocalCommandlineCodeExecutor(BaseModel):
directory is the current directory ".".
system_message_update (str): The system message update for agent that
produces code to run on this executor.
Default is `LocalCommandlineCodeExecutor.DEFAULT_SYSTEM_MESSAGE_UPDATE`.
Default is `LocalCommandLineCodeExecutor.DEFAULT_SYSTEM_MESSAGE_UPDATE`.
"""
DEFAULT_SYSTEM_MESSAGE_UPDATE: ClassVar[
@ -92,10 +92,10 @@ If you want the user to save the code in a file before executing it, put # filen
raise ValueError(f"Working directory {v} does not exist.")
@property
def user_capability(self) -> "LocalCommandlineCodeExecutor.UserCapability":
def user_capability(self) -> "LocalCommandLineCodeExecutor.UserCapability":
"""Export a user capability for this executor that can be added to
an agent that produces code to be executed by this executor."""
return LocalCommandlineCodeExecutor.UserCapability(self.system_message_update)
return LocalCommandLineCodeExecutor.UserCapability(self.system_message_update)
@property
def code_extractor(self) -> CodeExtractor:
@ -124,19 +124,19 @@ If you want the user to save the code in a file before executing it, put # filen
if re.search(pattern, code):
raise ValueError(f"Potentially dangerous command detected: {message}")
def execute_code_blocks(self, code_blocks: List[CodeBlock]) -> CommandlineCodeResult:
def execute_code_blocks(self, code_blocks: List[CodeBlock]) -> CommandLineCodeResult:
"""(Experimental) Execute the code blocks and return the result.
Args:
code_blocks (List[CodeBlock]): The code blocks to execute.
Returns:
CommandlineCodeResult: The result of the code execution."""
CommandLineCodeResult: The result of the code execution."""
logs_all = ""
for i, code_block in enumerate(code_blocks):
lang, code = code_block.language, code_block.code
LocalCommandlineCodeExecutor.sanitize_command(lang, code)
LocalCommandLineCodeExecutor.sanitize_command(lang, code)
filename_uuid = uuid.uuid4().hex
filename = None
if lang in ["bash", "shell", "sh", "pwsh", "powershell", "ps1"]:
@ -166,8 +166,75 @@ If you want the user to save the code in a file before executing it, put # filen
if exitcode != 0:
break
code_filename = os.path.join(self.work_dir, filename) if filename is not None else None
return CommandlineCodeResult(exit_code=exitcode, output=logs_all, code_file=code_filename)
return CommandLineCodeResult(exit_code=exitcode, output=logs_all, code_file=code_filename)
def restart(self) -> None:
"""(Experimental) Restart the code executor."""
warnings.warn("Restarting local command line code executor is not supported. No action is taken.")
# From stack overflow: https://stackoverflow.com/a/52087847/2214524
class _DeprecatedClassMeta(type):
def __new__(cls, name, bases, classdict, *args, **kwargs):
alias = classdict.get("_DeprecatedClassMeta__alias")
if alias is not None:
def new(cls, *args, **kwargs):
alias = getattr(cls, "_DeprecatedClassMeta__alias")
if alias is not None:
warnings.warn(
"{} has been renamed to {}, the alias will be "
"removed in the future".format(cls.__name__, alias.__name__),
DeprecationWarning,
stacklevel=2,
)
return alias(*args, **kwargs)
classdict["__new__"] = new
classdict["_DeprecatedClassMeta__alias"] = alias
fixed_bases = []
for b in bases:
alias = getattr(b, "_DeprecatedClassMeta__alias", None)
if alias is not None:
warnings.warn(
"{} has been renamed to {}, the alias will be "
"removed in the future".format(b.__name__, alias.__name__),
DeprecationWarning,
stacklevel=2,
)
# Avoid duplicate base classes.
b = alias or b
if b not in fixed_bases:
fixed_bases.append(b)
fixed_bases = tuple(fixed_bases)
return super().__new__(cls, name, fixed_bases, classdict, *args, **kwargs)
def __instancecheck__(cls, instance):
return any(cls.__subclasscheck__(c) for c in {type(instance), instance.__class__})
def __subclasscheck__(cls, subclass):
if subclass is cls:
return True
else:
return issubclass(subclass, getattr(cls, "_DeprecatedClassMeta__alias"))
class LocalCommandlineCodeExecutor(metaclass=_DeprecatedClassMeta):
"""LocalCommandlineCodeExecutor renamed to LocalCommandLineCodeExecutor"""
_DeprecatedClassMeta__alias = LocalCommandLineCodeExecutor
class CommandlineCodeResult(metaclass=_DeprecatedClassMeta):
"""CommandlineCodeResult renamed to CommandLineCodeResult"""
_DeprecatedClassMeta__alias = CommandLineCodeResult

View File

@ -4,7 +4,7 @@ import pytest
from autogen.agentchat.conversable_agent import ConversableAgent
from autogen.coding.base import CodeBlock, CodeExecutor
from autogen.coding.factory import CodeExecutorFactory
from autogen.coding.local_commandline_code_executor import LocalCommandlineCodeExecutor
from autogen.coding.local_commandline_code_executor import LocalCommandLineCodeExecutor
from autogen.oai.openai_utils import config_list_from_json
from conftest import MOCK_OPEN_AI_API_KEY, skip_openai
@ -13,25 +13,25 @@ from conftest import MOCK_OPEN_AI_API_KEY, skip_openai
def test_create() -> None:
config = {"executor": "commandline-local"}
executor = CodeExecutorFactory.create(config)
assert isinstance(executor, LocalCommandlineCodeExecutor)
assert isinstance(executor, LocalCommandLineCodeExecutor)
config = {"executor": LocalCommandlineCodeExecutor()}
config = {"executor": LocalCommandLineCodeExecutor()}
executor = CodeExecutorFactory.create(config)
assert executor is config["executor"]
def test_local_commandline_executor_init() -> None:
executor = LocalCommandlineCodeExecutor(timeout=10, work_dir=".")
executor = LocalCommandLineCodeExecutor(timeout=10, work_dir=".")
assert executor.timeout == 10 and executor.work_dir == "."
# Try invalid working directory.
with pytest.raises(ValueError, match="Working directory .* does not exist."):
executor = LocalCommandlineCodeExecutor(timeout=111, work_dir="/invalid/directory")
executor = LocalCommandLineCodeExecutor(timeout=111, work_dir="/invalid/directory")
def test_local_commandline_executor_execute_code() -> None:
with tempfile.TemporaryDirectory() as temp_dir:
executor = LocalCommandlineCodeExecutor(work_dir=temp_dir)
executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
_test_execute_code(executor=executor)
@ -81,7 +81,7 @@ def _test_execute_code(executor: CodeExecutor) -> None:
@pytest.mark.skipif(sys.platform in ["win32"], reason="do not run on windows")
def test_local_commandline_code_executor_timeout() -> None:
with tempfile.TemporaryDirectory() as temp_dir:
executor = LocalCommandlineCodeExecutor(timeout=1, work_dir=temp_dir)
executor = LocalCommandLineCodeExecutor(timeout=1, work_dir=temp_dir)
_test_timeout(executor)
@ -92,7 +92,7 @@ def _test_timeout(executor: CodeExecutor) -> None:
def test_local_commandline_code_executor_restart() -> None:
executor = LocalCommandlineCodeExecutor()
executor = LocalCommandLineCodeExecutor()
_test_restart(executor)
@ -105,7 +105,7 @@ def _test_restart(executor: CodeExecutor) -> None:
@pytest.mark.skipif(skip_openai, reason="requested to skip openai tests")
def test_local_commandline_executor_conversable_agent_capability() -> None:
with tempfile.TemporaryDirectory() as temp_dir:
executor = LocalCommandlineCodeExecutor(work_dir=temp_dir)
executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
_test_conversable_agent_capability(executor=executor)
@ -150,7 +150,7 @@ def _test_conversable_agent_capability(executor: CodeExecutor) -> None:
def test_local_commandline_executor_conversable_agent_code_execution() -> None:
with tempfile.TemporaryDirectory() as temp_dir:
executor = LocalCommandlineCodeExecutor(work_dir=temp_dir)
executor = LocalCommandLineCodeExecutor(work_dir=temp_dir)
with pytest.MonkeyPatch.context() as mp:
mp.setenv("OPENAI_API_KEY", MOCK_OPEN_AI_API_KEY)
_test_conversable_agent_code_execution(executor)
@ -192,7 +192,7 @@ def _test_conversable_agent_code_execution(executor: CodeExecutor) -> None:
)
def test_dangerous_commands(lang, code, expected_message):
with pytest.raises(ValueError) as exc_info:
LocalCommandlineCodeExecutor.sanitize_command(lang, code)
LocalCommandLineCodeExecutor.sanitize_command(lang, code)
assert expected_message in str(
exc_info.value
), f"Expected message '{expected_message}' not found in '{str(exc_info.value)}'"