github/autogen
1
1
Fork 0
mirror of https://github.com/microsoft/autogen.git synced 2026-04-20 03:02:16 -04:00
Code Issues Packages Projects Releases Wiki Activity

Update tests for code_utils and use ThreadPoolExecutor for code execution in windows (#1472)

Browse Source 
* threadpoolexecutor

* update tests

* update tests

* fix tests on windows

* fix test

* update tests

* update tests

* update tests

* update tests

* update tests

* update tests

* fix build.yml

* build yaml

* test

* use skip-docker to explicitly skipping docker tests

* update tests

* contribution doc update

* Update website/docs/Contribute.md

Co-authored-by: Chi Wang <wang.chi@microsoft.com>

* update doc

---------

Co-authored-by: Chi Wang <wang.chi@microsoft.com>
This commit is contained in:
Eric Zhu
2024-02-04 07:08:14 -08:00
committed by GitHub
parent 392d381721
commit 1abbcf3c44
5 changed files with 140 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

166
test/test_code.py
View File

@@ -1,15 +1,12 @@
import importlib.metadata
import os
import sys
import tempfile
import unittest
import pytest
import autogen
from autogen.code_utils import (
PATH_SEPARATOR,
UNKNOWN,
WIN32,
content_str,
execute_code,
extract_code,
@@ -21,27 +18,13 @@ from autogen.code_utils import (
decide_use_docker,
check_can_use_docker_or_throw,
)
from conftest import skip_docker
KEY_LOC = "notebook"
OAI_CONFIG_LIST = "OAI_CONFIG_LIST"
here = os.path.abspath(os.path.dirname(__file__))
def is_package_installed(package_name):
"""Check if a package is installed. This is a preferred way to check if a
package is installed or not than doing a try-catch around an import
because it avoids name conflict with local modules and
code execution in the imported module."""
try:
importlib.metadata.version(package_name)
return True
except importlib.metadata.PackageNotFoundError:
return False
docker_package_installed = is_package_installed("docker")
# def test_find_code():
# try:
# import openai
@@ -313,71 +296,105 @@ def scrape(url):
assert len(codeblocks) == 1 and codeblocks[0] == ("", "source setup.sh")
# skip if os is windows
@pytest.mark.skipif(
sys.platform in ["win32"] or (not is_docker_running() and not in_docker_container()), reason="docker is not running"
skip_docker or not is_docker_running(),
reason="docker is not running or requested to skip docker tests",
)
def test_execute_code(use_docker=None):
try:
import docker
except ImportError as exc:
print(exc)
docker = None
if use_docker is None:
use_docker = docker is not None
exit_code, msg, image = execute_code("print('hello world')", filename="tmp/codetest.py", use_docker=use_docker)
assert exit_code == 0 and msg == "hello world\n", msg
# read a file
print(execute_code("with open('tmp/codetest.py', 'r') as f: a=f.read()", use_docker=use_docker))
# create a file
exit_code, msg, image = execute_code(
"with open('tmp/codetest.py', 'w') as f: f.write('b=1')",
work_dir=f"{here}/my_tmp",
filename="tmp2/codetest.py",
use_docker=use_docker,
)
assert exit_code and (
'File "tmp2/codetest.py"'.replace("/", PATH_SEPARATOR) in msg
or 'File ".\\tmp2/codetest.py' in msg # py3.8 + win32
), msg
print(
execute_code(
"with open('tmp/codetest.py', 'w') as f: f.write('b=1')", work_dir=f"{here}/my_tmp", use_docker=use_docker
def test_execute_code(use_docker=True):
# Test execute code and save the code to a file.
with tempfile.TemporaryDirectory() as tempdir:
filename = "temp_file_with_code.py"
# execute code and save the code to a file.
exit_code, msg, image = execute_code(
"print('hello world')",
filename=filename,
work_dir=tempdir,
use_docker=use_docker,
)
)
# execute code in a file
print(execute_code(filename="tmp/codetest.py", use_docker=use_docker))
print(execute_code("python tmp/codetest.py", lang="sh", use_docker=use_docker))
# execute code for assertion error
exit_code, msg, image = execute_code("assert 1==2", use_docker=use_docker)
assert exit_code, msg
assert 'File ""' in msg or 'File ".\\"' in msg # py3.8 + win32
# execute code which takes a long time
exit_code, error, image = execute_code("import time; time.sleep(2)", timeout=1, use_docker=use_docker)
assert exit_code and error == "Timeout" or WIN32
assert isinstance(image, str) or docker is None or os.path.exists("/.dockerenv") or use_docker is False
assert exit_code == 0 and msg == "hello world\n", msg
# read the file just saved
exit_code, msg, image = execute_code(
f"with open('{filename}', 'rt') as f: print(f.read())",
use_docker=use_docker,
work_dir=tempdir,
)
assert exit_code == 0 and "print('hello world')" in msg, msg
# execute code in a file
exit_code, msg, image = execute_code(
filename=filename,
use_docker=use_docker,
work_dir=tempdir,
)
assert exit_code == 0 and msg == "hello world\n", msg
# execute code in a file using shell command directly
exit_code, msg, image = execute_code(
f"python {filename}",
lang="sh",
use_docker=use_docker,
work_dir=tempdir,
)
assert exit_code == 0 and msg == "hello world\n", msg
with tempfile.TemporaryDirectory() as tempdir:
# execute code for assertion error
exit_code, msg, image = execute_code(
"assert 1==2",
use_docker=use_docker,
work_dir=tempdir,
)
assert exit_code, msg
assert "AssertionError" in msg
assert 'File "' in msg or 'File ".\\"' in msg # py3.8 + win32
with tempfile.TemporaryDirectory() as tempdir:
# execute code which takes a long time
exit_code, error, image = execute_code(
"import time; time.sleep(2)",
timeout=1,
use_docker=use_docker,
work_dir=tempdir,
)
assert exit_code and error == "Timeout"
if use_docker is True:
assert isinstance(image, str)
@pytest.mark.skipif(
sys.platform in ["win32"] or (not is_docker_running()) or in_docker_container(),
reason="docker is not running or in docker container already",
skip_docker or not is_docker_running(),
reason="docker is not running or requested to skip docker tests",
)
def test_execute_code_with_custom_filename_on_docker():
exit_code, msg, image = execute_code("print('hello world')", filename="tmp/codetest.py", use_docker=True)
assert exit_code == 0 and msg == "hello world\n", msg
assert image == "python:tmp_codetest.py"
with tempfile.TemporaryDirectory() as tempdir:
filename = "codetest.py"
exit_code, msg, image = execute_code(
"print('hello world')",
filename=filename,
use_docker=True,
work_dir=tempdir,
)
assert exit_code == 0 and msg == "hello world\n", msg
assert image == "python:codetest.py"
@pytest.mark.skipif(
sys.platform in ["win32"] or (not is_docker_running()) or in_docker_container(),
reason="docker is not running or in docker container already",
skip_docker or not is_docker_running(),
reason="docker is not running or requested to skip docker tests",
)
def test_execute_code_with_misformed_filename_on_docker():
exit_code, msg, image = execute_code(
"print('hello world')", filename="tmp/codetest.py (some extra information)", use_docker=True
)
assert exit_code == 0 and msg == "hello world\n", msg
assert image == "python:tmp_codetest.py__some_extra_information_"
with tempfile.TemporaryDirectory() as tempdir:
filename = "codetest.py (some extra information)"
exit_code, msg, image = execute_code(
"print('hello world')",
filename=filename,
use_docker=True,
work_dir=tempdir,
)
assert exit_code == 0 and msg == "hello world\n", msg
assert image == "python:codetest.py__some_extra_information_"
def test_execute_code_raises_when_code_and_filename_are_both_none():
@@ -385,14 +402,13 @@ def test_execute_code_raises_when_code_and_filename_are_both_none():
execute_code(code=None, filename=None)
def test_execute_code_nodocker():
def test_execute_code_no_docker():
test_execute_code(use_docker=False)
def test_execute_code_no_docker():
def test_execute_code_timeout_no_docker():
exit_code, error, image = execute_code("import time; time.sleep(2)", timeout=1, use_docker=False)
if sys.platform != "win32":
assert exit_code and error == "Timeout"
assert exit_code and error == "Timeout"
assert image is None