reshaping code and fixing tests
Evgeny Vakhteev
parent
9fd80a8660
commit
894026cdd4
|
|
@ -11,7 +11,7 @@ from autogpt.logs import logger
|
|||
from autogpt.memory import get_memory
|
||||
|
||||
from autogpt.prompts.prompt import construct_main_ai_config
|
||||
from autogpt.plugins import load_plugins
|
||||
from autogpt.plugins import scan_plugins
|
||||
|
||||
|
||||
# Load environment variables from .env file
|
||||
|
|
@ -24,7 +24,7 @@ def main() -> None:
|
|||
check_openai_api_key()
|
||||
parse_arguments()
|
||||
logger.set_level(logging.DEBUG if cfg.debug_mode else logging.INFO)
|
||||
cfg.set_plugins(load_plugins(cfg, cfg.debug_mode))
|
||||
cfg.set_plugins(scan_plugins(cfg, cfg.debug_mode))
|
||||
# Create a CommandRegistry instance and scan default folder
|
||||
command_registry = CommandRegistry()
|
||||
command_registry.import_commands("scripts.ai_functions")
|
||||
|
|
|
|||
|
|
@ -2,6 +2,8 @@
|
|||
from typing import Any, Dict, List, Optional, Tuple, TypedDict
|
||||
from typing import TypeVar
|
||||
|
||||
from auto_gpt_plugin_template import AutoGPTPluginTemplate
|
||||
|
||||
PromptGenerator = TypeVar("PromptGenerator")
|
||||
|
||||
|
||||
|
|
@ -10,9 +12,9 @@ class Message(TypedDict):
|
|||
content: str
|
||||
|
||||
|
||||
class BaseOpenAIPlugin:
|
||||
class BaseOpenAIPlugin(AutoGPTPluginTemplate):
|
||||
"""
|
||||
This is a template for Auto-GPT plugins.
|
||||
This is a BaseOpenAIPlugin class for generating Auto-GPT plugins.
|
||||
"""
|
||||
|
||||
def __init__(self, manifests_specs_clients: dict):
|
||||
|
|
@ -20,9 +22,9 @@ class BaseOpenAIPlugin:
|
|||
self._name = manifests_specs_clients["manifest"]["name_for_model"]
|
||||
self._version = manifests_specs_clients["manifest"]["schema_version"]
|
||||
self._description = manifests_specs_clients["manifest"]["description_for_model"]
|
||||
self.client = manifests_specs_clients["client"]
|
||||
self.manifest = manifests_specs_clients["manifest"]
|
||||
self.openapi_spec = manifests_specs_clients["openapi_spec"]
|
||||
self._client = manifests_specs_clients["client"]
|
||||
self._manifest = manifests_specs_clients["manifest"]
|
||||
self._openapi_spec = manifests_specs_clients["openapi_spec"]
|
||||
|
||||
def can_handle_on_response(self) -> bool:
|
||||
"""This method is called to check that the plugin can
|
||||
|
|
@ -196,4 +198,3 @@ class BaseOpenAIPlugin:
|
|||
str: The resulting response.
|
||||
"""
|
||||
pass
|
||||
|
||||
|
|
|
|||
|
|
@ -176,7 +176,7 @@ def instantiate_openai_plugin_clients(manifests_specs_clients: dict, cfg: Config
|
|||
|
||||
|
||||
def scan_plugins(cfg: Config, debug: bool = False) -> List[Tuple[str, Path]]:
|
||||
"""Scan the plugins directory for plugins.
|
||||
"""Scan the plugins directory for plugins and loads them.
|
||||
|
||||
Args:
|
||||
cfg (Config): Config instance including plugins config
|
||||
|
|
@ -185,46 +185,37 @@ def scan_plugins(cfg: Config, debug: bool = False) -> List[Tuple[str, Path]]:
|
|||
Returns:
|
||||
List[Tuple[str, Path]]: List of plugins.
|
||||
"""
|
||||
plugins = []
|
||||
loaded_plugins = []
|
||||
# Generic plugins
|
||||
plugins_path_path = Path(cfg.plugins_dir)
|
||||
for plugin in plugins_path_path.glob("*.zip"):
|
||||
if module := inspect_zip_for_module(str(plugin), debug):
|
||||
plugins.append((module, plugin))
|
||||
plugin = Path(plugin)
|
||||
module = Path(module)
|
||||
if debug:
|
||||
print(f"Plugin: {plugin} Module: {module}")
|
||||
zipped_package = zipimporter(plugin)
|
||||
zipped_module = zipped_package.load_module(str(module.parent))
|
||||
for key in dir(zipped_module):
|
||||
if key.startswith("__"):
|
||||
continue
|
||||
a_module = getattr(zipped_module, key)
|
||||
a_keys = dir(a_module)
|
||||
if (
|
||||
"_abc_impl" in a_keys
|
||||
and a_module.__name__ != "AutoGPTPluginTemplate"
|
||||
and blacklist_whitelist_check(a_module.__name__, cfg)
|
||||
):
|
||||
loaded_plugins.append(a_module())
|
||||
# OpenAI plugins
|
||||
if cfg.plugins_openai:
|
||||
manifests_specs = fetch_openai_plugins_manifest_and_spec(cfg)
|
||||
if manifests_specs.keys():
|
||||
manifests_specs_clients = initialize_openai_plugins(manifests_specs, cfg, debug)
|
||||
for url, openai_plugin_meta in manifests_specs_clients.items():
|
||||
plugin = BaseOpenAIPlugin(openai_plugin_meta)
|
||||
plugins.append((plugin, url))
|
||||
return plugins
|
||||
|
||||
|
||||
def blacklist_whitelist_check(plugins: List[AbstractSingleton], cfg: Config):
|
||||
"""Check if the plugin is in the whitelist or blacklist.
|
||||
|
||||
Args:
|
||||
plugins (List[Tuple[str, Path]]): List of plugins.
|
||||
cfg (Config): Config object.
|
||||
|
||||
Returns:
|
||||
List[Tuple[str, Path]]: List of plugins.
|
||||
"""
|
||||
loaded_plugins = []
|
||||
for plugin in plugins:
|
||||
if plugin.__name__ in cfg.plugins_blacklist:
|
||||
continue
|
||||
if plugin.__name__ in cfg.plugins_whitelist:
|
||||
loaded_plugins.append(plugin())
|
||||
else:
|
||||
ack = input(
|
||||
f"WARNNG Plugin {plugin.__name__} found. But not in the"
|
||||
" whitelist... Load? (y/n): "
|
||||
)
|
||||
if ack.lower() == "y":
|
||||
loaded_plugins.append(plugin())
|
||||
if blacklist_whitelist_check(url, cfg):
|
||||
plugin = BaseOpenAIPlugin(openai_plugin_meta)
|
||||
loaded_plugins.append(plugin)
|
||||
|
||||
if loaded_plugins:
|
||||
print(f"\nPlugins found: {len(loaded_plugins)}\n" "--------------------")
|
||||
|
|
@ -232,30 +223,22 @@ def blacklist_whitelist_check(plugins: List[AbstractSingleton], cfg: Config):
|
|||
print(f"{plugin._name}: {plugin._version} - {plugin._description}")
|
||||
return loaded_plugins
|
||||
|
||||
|
||||
def load_plugins(cfg: Config = Config(), debug: bool = False) -> List[object]:
|
||||
"""Load plugins from the plugins directory.
|
||||
def blacklist_whitelist_check(plugin_name: str, cfg: Config) -> bool:
|
||||
"""Check if the plugin is in the whitelist or blacklist.
|
||||
|
||||
Args:
|
||||
cfg (Config): Config instance including plugins config
|
||||
debug (bool, optional): Enable debug logging. Defaults to False.
|
||||
plugin_name (str): Name of the plugin.
|
||||
cfg (Config): Config object.
|
||||
|
||||
Returns:
|
||||
List[AbstractSingleton]: List of plugins initialized.
|
||||
True or False
|
||||
"""
|
||||
plugins = scan_plugins(cfg)
|
||||
plugin_modules = []
|
||||
for module, plugin in plugins:
|
||||
plugin = Path(plugin)
|
||||
module = Path(module)
|
||||
if debug:
|
||||
print(f"Plugin: {plugin} Module: {module}")
|
||||
zipped_package = zipimporter(plugin)
|
||||
zipped_module = zipped_package.load_module(str(module.parent))
|
||||
for key in dir(zipped_module):
|
||||
if key.startswith("__"):
|
||||
continue
|
||||
a_module = getattr(zipped_module, key)
|
||||
a_keys = dir(a_module)
|
||||
if "_abc_impl" in a_keys and a_module.__name__ != "AutoGPTPluginTemplate":
|
||||
plugin_modules.append(a_module)
|
||||
return blacklist_whitelist_check(plugin_modules, cfg)
|
||||
if plugin_name in cfg.plugins_blacklist:
|
||||
return False
|
||||
if plugin_name in cfg.plugins_whitelist:
|
||||
return True
|
||||
ack = input(
|
||||
f"WARNNG Plugin {plugin_name} found. But not in the"
|
||||
" whitelist... Load? (y/n): "
|
||||
)
|
||||
return ack.lower() == "y"
|
||||
|
|
|
|||
|
|
@ -30,7 +30,7 @@ pytest-mock
|
|||
tweepy
|
||||
|
||||
|
||||
# OpenAI plugins import
|
||||
# OpenAI and Generic plugins import
|
||||
openapi-python-client==0.13.4
|
||||
abstract-singleton
|
||||
auto-vicuna
|
||||
auto-gpt-plugin-template
|
||||
|
|
|
|||
Binary file not shown.
|
|
@ -1,10 +1,51 @@
|
|||
import pytest
|
||||
from autogpt.plugins import inspect_zip_for_module, scan_plugins, load_plugins
|
||||
from autogpt.plugins import inspect_zip_for_module, scan_plugins, blacklist_whitelist_check
|
||||
from autogpt.config import Config
|
||||
|
||||
PLUGINS_TEST_DIR = "tests/unit/data/test_plugins"
|
||||
PLUGIN_TEST_ZIP_FILE = "Auto-GPT-Plugin-Test-master.zip"
|
||||
PLUGIN_TEST_INIT_PY = "Auto-GPT-Plugin-Test-master/src/auto_gpt_plugin_template/__init__.py"
|
||||
PLUGIN_TEST_INIT_PY = "Auto-GPT-Plugin-Test-master/src/auto_gpt_vicuna/__init__.py"
|
||||
PLUGIN_TEST_OPENAI = 'https://weathergpt.vercel.app/'
|
||||
|
||||
|
||||
def test_inspect_zip_for_module():
|
||||
result = inspect_zip_for_module(str(f'{PLUGINS_TEST_DIR}/{PLUGIN_TEST_ZIP_FILE}'))
|
||||
assert result == PLUGIN_TEST_INIT_PY
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_config_blacklist_whitelist_check():
|
||||
class MockConfig:
|
||||
plugins_blacklist = ["BadPlugin"]
|
||||
plugins_whitelist = ["GoodPlugin"]
|
||||
|
||||
return MockConfig()
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_blacklist(mock_config_blacklist_whitelist_check,
|
||||
monkeypatch):
|
||||
monkeypatch.setattr("builtins.input", lambda _: "y")
|
||||
assert not blacklist_whitelist_check("BadPlugin", mock_config_blacklist_whitelist_check)
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_whitelist(mock_config_blacklist_whitelist_check, monkeypatch):
|
||||
monkeypatch.setattr("builtins.input", lambda _: "y")
|
||||
assert blacklist_whitelist_check("GoodPlugin", mock_config_blacklist_whitelist_check)
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_user_input_yes(mock_config_blacklist_whitelist_check, monkeypatch):
|
||||
monkeypatch.setattr("builtins.input", lambda _: "y")
|
||||
assert blacklist_whitelist_check("UnknownPlugin", mock_config_blacklist_whitelist_check)
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_user_input_no(mock_config_blacklist_whitelist_check, monkeypatch):
|
||||
monkeypatch.setattr("builtins.input", lambda _: "n")
|
||||
assert not blacklist_whitelist_check("UnknownPlugin", mock_config_blacklist_whitelist_check)
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_user_input_invalid(mock_config_blacklist_whitelist_check, monkeypatch):
|
||||
monkeypatch.setattr("builtins.input", lambda _: "invalid")
|
||||
assert not blacklist_whitelist_check("UnknownPlugin", mock_config_blacklist_whitelist_check)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
|
|
@ -15,17 +56,33 @@ def config_with_plugins():
|
|||
return cfg
|
||||
|
||||
|
||||
def test_inspect_zip_for_module():
|
||||
result = inspect_zip_for_module(str(f'{PLUGINS_TEST_DIR}/{PLUGIN_TEST_ZIP_FILE}'))
|
||||
assert result == PLUGIN_TEST_INIT_PY
|
||||
@pytest.fixture
|
||||
def mock_config_openai_plugin():
|
||||
class MockConfig:
|
||||
plugins_dir = PLUGINS_TEST_DIR
|
||||
plugins_openai = [PLUGIN_TEST_OPENAI]
|
||||
plugins_blacklist = ["AutoGPTPVicuna"]
|
||||
plugins_whitelist = [PLUGIN_TEST_OPENAI]
|
||||
|
||||
def test_scan_plugins(config_with_plugins):
|
||||
result = scan_plugins(config_with_plugins, debug=True)
|
||||
return MockConfig()
|
||||
|
||||
|
||||
def test_scan_plugins_openai(mock_config_openai_plugin):
|
||||
result = scan_plugins(mock_config_openai_plugin, debug=True)
|
||||
assert len(result) == 1
|
||||
assert result[0][0] == PLUGIN_TEST_INIT_PY
|
||||
|
||||
|
||||
def test_load_plugins_blacklisted(config_with_plugins):
|
||||
config_with_plugins.plugins_blacklist = ['AbstractSingleton']
|
||||
result = load_plugins(cfg=config_with_plugins)
|
||||
assert len(result) == 0
|
||||
@pytest.fixture
|
||||
def mock_config_generic_plugin():
|
||||
class MockConfig:
|
||||
plugins_dir = PLUGINS_TEST_DIR
|
||||
plugins_openai = []
|
||||
plugins_blacklist = []
|
||||
plugins_whitelist = ["AutoGPTPVicuna"]
|
||||
|
||||
return MockConfig()
|
||||
|
||||
|
||||
def test_scan_plugins_generic(mock_config_generic_plugin):
|
||||
result = scan_plugins(mock_config_generic_plugin, debug=True)
|
||||
assert len(result) == 1
|
||||
|
|
|
|||
|
|
@ -1,222 +0,0 @@
|
|||
import json
|
||||
import os
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
from urllib.parse import urlparse
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from autogpt.plugins import create_directory_if_not_exists, inspect_zip_for_module, load_plugins, \
|
||||
blacklist_whitelist_check, instantiate_openai_plugin_clients, scan_plugins, fetch_openai_plugins_manifest_and_spec, \
|
||||
initialize_openai_plugins
|
||||
|
||||
|
||||
PLUGINS_TEST_DIR = "tests/unit/data/test_plugins"
|
||||
PLUGIN_TEST_ZIP_FILE = "Auto-GPT-Plugin-Test-master.zip"
|
||||
PLUGIN_TEST_INIT_PY = "Auto-GPT-Plugin-Test-master/src/auto_gpt_plugin_template/__init__.py"
|
||||
|
||||
|
||||
def test_inspect_zip_for_module():
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
zip_path = os.path.join(temp_dir, "sample.zip")
|
||||
with zipfile.ZipFile(zip_path, "w") as zf:
|
||||
zf.writestr("test_module/__init__.py", "")
|
||||
|
||||
result = plugins.inspect_zip_for_module(zip_path)
|
||||
assert result == "test_module/__init__.py"
|
||||
|
||||
result = plugins.inspect_zip_for_module(zip_path, debug=True)
|
||||
assert result == "test_module/__init__.py"
|
||||
|
||||
with zipfile.ZipFile(zip_path, "w") as zf:
|
||||
zf.writestr("not_a_module.py", "")
|
||||
|
||||
result = plugins.inspect_zip_for_module(zip_path)
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_write_dict_to_json_file():
|
||||
with tempfile.NamedTemporaryFile(mode="w+", delete=False) as file:
|
||||
test_data = {"test_key": "test_value"}
|
||||
plugins.write_dict_to_json_file(test_data, file.name)
|
||||
|
||||
file.seek(0)
|
||||
loaded_data = json.load(file)
|
||||
assert loaded_data == test_data
|
||||
|
||||
|
||||
def test_create_directory_if_not_exists():
|
||||
with tempfile.TemporaryDirectory() as temp_dir:
|
||||
new_dir = os.path.join(temp_dir, "test_dir")
|
||||
assert not os.path.exists(new_dir)
|
||||
|
||||
result = create_directory_if_not_exists(new_dir)
|
||||
assert result is True
|
||||
assert os.path.exists(new_dir)
|
||||
|
||||
result = create_directory_if_not_exists(new_dir)
|
||||
assert result is True
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config_mock():
|
||||
config = MagicMock()
|
||||
config.plugins_dir = "/plugins"
|
||||
config.plugins_openai = []
|
||||
return config
|
||||
|
||||
|
||||
@patch("autogpt.plugins.write_dict_to_json_file")
|
||||
@patch("requests.get")
|
||||
@patch("autogpt.plugins.create_directory_if_not_exists")
|
||||
def test_fetch_openai_plugins_manifest_and_spec(create_directory_mock, write_dict_mock, requests_get_mock, config_mock):
|
||||
requests_get_mock.side_effect = [
|
||||
MagicMock(status_code=200, json=lambda: {
|
||||
"schema_version": "v1",
|
||||
"api": {"type": "openapi", "url": "http://example.com/openapi.json"}
|
||||
}),
|
||||
MagicMock(status_code=404),
|
||||
]
|
||||
|
||||
config_mock.plugins_openai = ["http://example.com"]
|
||||
|
||||
result = fetch_openai_plugins_manifest_and_spec(config_mock)
|
||||
assert len(result) == 1
|
||||
assert "http://example.com" in result
|
||||
assert "manifest" in result["http://example.com"]
|
||||
assert "openapi_spec" in result["http://example.com"]
|
||||
|
||||
create_directory_mock.assert_called_once()
|
||||
write_dict_mock.assert_called_once()
|
||||
requests_get_mock.assert_has_calls([
|
||||
patch("requests.get", args=("http://example.com/.well-known/ai-plugin.json",)),
|
||||
patch("requests.get", args=("http://example.com/openapi.json",)),
|
||||
])
|
||||
|
||||
|
||||
# @patch("BaseOpenAIPlugin")
|
||||
# def test_instantiate_openai_plugin_clients(base_openai_plugin_client_mock, config_mock):
|
||||
# manifests_specs_clients = {
|
||||
# "http://example.com": {
|
||||
# "manifest": {},
|
||||
# "openapi_spec": {},
|
||||
# "client": MagicMock(),
|
||||
# }
|
||||
# }
|
||||
#
|
||||
# result = instantiate_openai_plugin_clients(manifests_specs_clients, config_mock)
|
||||
# assert len(result) == 1
|
||||
|
||||
|
||||
def test_scan_plugins(config_mock):
|
||||
with patch("inspect_zip_for_module", return_value="test_module/__init__.py"):
|
||||
plugins = scan_plugins(config_mock)
|
||||
assert len(plugins) == 0
|
||||
|
||||
|
||||
# @patch("BaseOpenAIPlugin")
|
||||
# def test_initialize_openai_plugins(base_openai_plugin_client_mock, config_mock):
|
||||
# manifests_specs = {
|
||||
# "http://example.com": {
|
||||
# "manifest": {},
|
||||
# "openapi_spec": {},
|
||||
# }
|
||||
# }
|
||||
#
|
||||
# with patch("Path.cwd") as cwd_mock:
|
||||
# cwd_mock.return_value = Path("/fake_cwd")
|
||||
# result = initialize_openai_plugins(manifests_specs, config_mock)
|
||||
# assert len(result) == 1
|
||||
# assert "http://example.com" in result
|
||||
# assert "client" in result["http://example.com"]
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check(config_mock):
|
||||
class Plugin1(MagicMock):
|
||||
__name__ = "Plugin1"
|
||||
|
||||
class Plugin2(MagicMock):
|
||||
__name__ = "Plugin2"
|
||||
|
||||
config_mock.plugins_blacklist = ["Plugin1"]
|
||||
config_mock.plugins_whitelist = ["Plugin2"]
|
||||
|
||||
plugins = [Plugin1, Plugin2]
|
||||
result = blacklist_whitelist_check(plugins, config_mock)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], Plugin2)
|
||||
|
||||
config_mock.plugins_blacklist = []
|
||||
config_mock.plugins_whitelist = []
|
||||
|
||||
with patch("builtins.input", side_effect=["y", "n"]):
|
||||
result = blacklist_whitelist_check(plugins, config_mock)
|
||||
assert len(result) == 1
|
||||
assert isinstance(result[0], Plugin1)
|
||||
|
||||
|
||||
@patch("autogpt.plugins.scan_plugins")
|
||||
@patch("autogpt.plugins.blacklist_whitelist_check")
|
||||
def test_load_plugins(blacklist_whitelist_check_mock, scan_plugins_mock, config_mock):
|
||||
load_plugins(cfg=config_mock, debug=True)
|
||||
|
||||
scan_plugins_mock.assert_called_once_with(config_mock)
|
||||
blacklist_whitelist_check_mock.assert_called_once_with(scan_plugins_mock.return_value, config_mock)
|
||||
|
||||
def test_inspect_zip_for_module_no_init_py():
|
||||
with patch("zipfile.ZipFile") as zip_mock:
|
||||
zip_mock.return_value.__enter__.return_value.namelist.return_value = ["test_module/file1.py"]
|
||||
|
||||
result = inspect_zip_for_module("test_module.zip")
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_create_directory_if_not_exists_error():
|
||||
with patch("os.makedirs") as makedirs_mock:
|
||||
makedirs_mock.side_effect = OSError("Error creating directory")
|
||||
|
||||
result = create_directory_if_not_exists("non_existent_dir")
|
||||
assert result is False
|
||||
|
||||
|
||||
def test_fetch_openai_plugins_manifest_and_spec_invalid_manifest():
|
||||
with patch("requests.get") as get_mock:
|
||||
get_mock.return_value.status_code = 200
|
||||
get_mock.return_value.json.return_value = {
|
||||
"schema_version": "v2",
|
||||
"api": {"type": "openapi"},
|
||||
}
|
||||
|
||||
config = MagicMock()
|
||||
config.plugins_openai = ["http://example.com"]
|
||||
|
||||
result = fetch_openai_plugins_manifest_and_spec(config)
|
||||
assert result == {}
|
||||
|
||||
|
||||
# @patch("BaseOpenAIPlugin")
|
||||
# def test_instantiate_openai_plugin_clients_invalid_input(base_openai_plugin_client_mock):
|
||||
# with pytest.raises(TypeError):
|
||||
# instantiate_openai_plugin_clients("invalid_input", MagicMock())
|
||||
|
||||
|
||||
def test_scan_plugins_invalid_config():
|
||||
with pytest.raises(AttributeError):
|
||||
scan_plugins("invalid_config")
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_invalid_plugins_input():
|
||||
with pytest.raises(TypeError):
|
||||
blacklist_whitelist_check("invalid_plugins_input", MagicMock())
|
||||
|
||||
|
||||
def test_blacklist_whitelist_check_invalid_config_input():
|
||||
with pytest.raises(TypeError):
|
||||
blacklist_whitelist_check([], "invalid_config_input")
|
||||
|
||||
|
||||
def test_load_plugins_invalid_config_input():
|
||||
with pytest.raises(TypeError):
|
||||
load_plugins("invalid_config_input")
|
||||
Loading…
Reference in New Issue