mirror of
https://github.com/tiennm99/serena.git
synced 2026-10-15 08:17:15 +00:00
Merge pull request #2035 from oraios/dev-v2
Add REPL (milestone for Serena v2)
This commit is contained in:
75 files changed
+7024
-1577
No files matched your search
@@ -4,12 +4,14 @@ import shutil
|
||||
import tempfile
|
||||
from copy import deepcopy
|
||||
from pathlib import Path
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.agent import SerenaAgent
|
||||
from serena.config.serena_config import (
|
||||
DEFAULT_PROJECT_SERENA_FOLDER_LOCATION,
|
||||
AgentInterface,
|
||||
LanguageBackend,
|
||||
ProjectConfig,
|
||||
RegisteredProject,
|
||||
@@ -218,6 +220,68 @@ class TestProjectConfigLanguageBackend:
|
||||
assert config.language_backend is None
|
||||
|
||||
|
||||
class TestAgentInterface:
|
||||
"""Tests for the agent_interface setting (global and per project)."""
|
||||
|
||||
@staticmethod
|
||||
def _project_config(agent_interface: AgentInterface | None) -> ProjectConfig:
|
||||
return ProjectConfig(project_name="test", language_servers=[LanguageServerId.PYTHON], agent_interface=agent_interface)
|
||||
|
||||
def test_agent_interface_roundtrips_through_project_yaml(self):
|
||||
assert self._project_config(AgentInterface.REPL)._to_yaml_dict()["agent_interface"] == "REPL"
|
||||
assert self._project_config(None)._to_yaml_dict()["agent_interface"] is None
|
||||
|
||||
def test_agent_interface_parsed_from_project_dict(self):
|
||||
data, _ = ProjectConfig._load_yaml_dict(PROJECT_TEMPLATE_FILE)
|
||||
data["project_name"] = "test"
|
||||
data["languages"] = ["python"]
|
||||
data["agent_interface"] = "repl" # case-insensitive
|
||||
assert ProjectConfig._from_dict(data, local_override_keys=[]).agent_interface == AgentInterface.REPL
|
||||
data.pop("agent_interface")
|
||||
assert ProjectConfig._from_dict(data, local_override_keys=[]).agent_interface is None
|
||||
|
||||
def test_determine_agent_interface_precedence(self):
|
||||
# default
|
||||
assert SerenaConfig().determine_agent_interface() == AgentInterface.TOOLS
|
||||
assert SerenaConfig().determine_agent_interface(self._project_config(None)) == AgentInterface.TOOLS
|
||||
# global configuration
|
||||
assert SerenaConfig(agent_interface=AgentInterface.REPL).determine_agent_interface() == AgentInterface.REPL
|
||||
# project configuration takes precedence
|
||||
config = SerenaConfig(agent_interface=AgentInterface.REPL)
|
||||
assert config.determine_agent_interface(self._project_config(AgentInterface.TOOLS)) == AgentInterface.TOOLS
|
||||
assert config.determine_agent_interface(self._project_config(None)) == AgentInterface.REPL
|
||||
|
||||
def test_repl_toolset_is_fixed_and_repl_follows_project_activation(self):
|
||||
"""
|
||||
In REPL mode, neither the exposed nor the active toolset is affected by tool inclusion/exclusion definitions
|
||||
(here: the project's exclusions and read-only setting), whereas the REPL's API scope follows the active project.
|
||||
"""
|
||||
config, name = _make_config_with_project("test_proj")
|
||||
config.agent_interface = AgentInterface.REPL
|
||||
project_config = config.projects[0].project_config
|
||||
project_config.excluded_tools = ["initial_instructions", "serena_repl"]
|
||||
project_config.excluded_apis = ["mem"]
|
||||
project_config.read_only = True
|
||||
|
||||
agent = SerenaAgent(project=None, serena_config=config)
|
||||
try:
|
||||
# before activation: the fixed toolset and the full set of facades
|
||||
fixed_toolset = {"serena_repl", "initial_instructions", "activate_project"}
|
||||
assert {t.get_name() for t in agent.get_exposed_tool_instances()} == fixed_toolset
|
||||
assert set(agent.get_active_tool_names()) == fixed_toolset
|
||||
overview = agent.get_repl().entrypoint.overview()
|
||||
assert "s.mem" in overview
|
||||
# the dashboard is disabled in the test configuration, so opening it is not offered
|
||||
assert "s.cfg" in overview and "open_dashboard" not in overview
|
||||
|
||||
# after activation: the toolset is unchanged, the REPL reflects the project's API exclusions
|
||||
agent.activate_project_from_path_or_name(name)
|
||||
assert set(agent.get_active_tool_names()) == fixed_toolset
|
||||
assert "s.mem" not in agent.get_repl().entrypoint.overview()
|
||||
finally:
|
||||
agent.on_shutdown(timeout=5)
|
||||
|
||||
|
||||
def _make_config_with_project(
|
||||
project_name: str,
|
||||
language_backend: LanguageBackend | None = None,
|
||||
@@ -300,6 +364,35 @@ class TestEffectiveLanguageBackend:
|
||||
finally:
|
||||
agent.on_shutdown(timeout=5)
|
||||
|
||||
def test_activate_project_switches_backend_with_repl_interface(self):
|
||||
"""With the REPL interface, post-init activation of a project with a different backend switches the backend."""
|
||||
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=LanguageBackend.LSP)
|
||||
config.agent_interface = AgentInterface.REPL
|
||||
jb_project = Project(
|
||||
project_root=str(Path(__file__).parent.parent / "resources" / "repos" / "java" / "test_repo"),
|
||||
project_config=ProjectConfig(
|
||||
project_name="jb_proj",
|
||||
language_servers=[LanguageServerId.JAVA],
|
||||
language_backend=LanguageBackend.JETBRAINS,
|
||||
),
|
||||
serena_config=config,
|
||||
)
|
||||
config.projects.append(RegisteredProject.from_project_instance(jb_project))
|
||||
|
||||
agent = SerenaAgent(project=name, serena_config=config)
|
||||
try:
|
||||
assert agent.get_language_backend() == LanguageBackend.LSP
|
||||
assert "s.lsp" in agent.get_repl().entrypoint.overview()
|
||||
|
||||
# the backend and everything depending on it follow the activated project
|
||||
agent.activate_project_from_path_or_name("jb_proj")
|
||||
assert agent.get_language_backend() == LanguageBackend.JETBRAINS
|
||||
overview = agent.get_repl().entrypoint.overview()
|
||||
assert "s.jb" in overview and "s.lsp" not in overview
|
||||
assert "jetbrains" in [m.name for m in agent.get_active_modes().get_modes(include_background_base_modes=True)]
|
||||
finally:
|
||||
agent.on_shutdown(timeout=5)
|
||||
|
||||
def test_activate_project_allows_matching_backend(self):
|
||||
"""Post-init activation of a project with matching backend succeeds."""
|
||||
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=LanguageBackend.LSP)
|
||||
@@ -543,6 +636,36 @@ class TestSerenaConfigLoadSave:
|
||||
config = SerenaConfig.from_config_file(generate_if_missing=False)
|
||||
assert config.projects == []
|
||||
|
||||
@pytest.mark.parametrize("setting", ["", "auth_secret: null\n", 'auth_secret: ""\n'])
|
||||
def test_unset_auth_secret_is_generated_and_persisted(self, setting: str) -> None:
|
||||
# load an existing configuration without a usable secret
|
||||
self.master_config_path.write_text("projects: []\n" + setting)
|
||||
config = SerenaConfig.from_config_file(generate_if_missing=False)
|
||||
|
||||
# subsequent loads retain the generated random UUID
|
||||
assert UUID(config.auth_secret).version == 4
|
||||
assert SerenaConfig.from_config_file(generate_if_missing=False).auth_secret == config.auth_secret
|
||||
|
||||
def test_configured_auth_secret_is_preserved(self) -> None:
|
||||
# retain a user-provided secret across loading and migration
|
||||
self.master_config_path.write_text("projects: []\nauth_secret: custom-secret\n")
|
||||
assert SerenaConfig.from_config_file(generate_if_missing=False).auth_secret == "custom-secret"
|
||||
assert SerenaConfig.from_config_file(generate_if_missing=False).auth_secret == "custom-secret"
|
||||
|
||||
def test_new_config_has_persistent_auth_secret(self) -> None:
|
||||
# generate the configuration from the template and retain its secret
|
||||
config = SerenaConfig.from_config_file()
|
||||
assert UUID(config.auth_secret).version == 4
|
||||
assert SerenaConfig.from_config_file().auth_secret == config.auth_secret
|
||||
|
||||
def test_direct_config_instances_have_distinct_auth_secrets(self) -> None:
|
||||
# directly constructed configurations receive independent secrets
|
||||
first = SerenaConfig()
|
||||
second = SerenaConfig()
|
||||
assert UUID(first.auth_secret).version == 4
|
||||
assert UUID(second.auth_secret).version == 4
|
||||
assert first.auth_secret != second.auth_secret
|
||||
|
||||
def test_malformed_project_is_skipped_with_warning(self, caplog):
|
||||
"""A malformed project.yml must not abort loading of the others."""
|
||||
good_project = self._make_project_dir(
|
||||
|
||||
@@ -0,0 +1,25 @@
|
||||
"""
|
||||
Tests for the configuration facade API.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from serena.repl.api.cfg_api import ConfigApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
def test_facade_exposes_config_operations() -> None:
|
||||
facade = Facade.from_api(ConfigApi(MagicMock()), ApiScope())
|
||||
assert facade.name == "cfg"
|
||||
assert set(facade.enabled_method_names) == {"get_current_config", "open_dashboard"}
|
||||
assert not any(facade.get_method(name).info.can_edit for name in facade.enabled_method_names)
|
||||
|
||||
|
||||
def test_operations_delegate_to_agent() -> None:
|
||||
agent = MagicMock()
|
||||
agent.get_current_config_overview.return_value = "overview"
|
||||
agent.open_dashboard.return_value = False
|
||||
agent.get_dashboard_url.return_value = "http://localhost:1"
|
||||
api = ConfigApi(agent)
|
||||
assert api.get_current_config() == "overview"
|
||||
assert "http://localhost:1" in api.open_dashboard()
|
||||
@@ -0,0 +1,68 @@
|
||||
"""
|
||||
Tests for the editing facade API (backend-independent parts, without a language server).
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.edit_api import EditApi, ReplacementPreview
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project(tmp_path: Path) -> Project:
|
||||
(tmp_path / "a.py").write_text("x = foo(1)\ny = foo(2)\n", encoding="utf-8")
|
||||
(tmp_path / "b.py").write_text("z = foo(3)\n", encoding="utf-8")
|
||||
return Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(project: Project) -> EditApi:
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return EditApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_editing_operations(api: EditApi) -> None:
|
||||
default_methods = {
|
||||
"replace_content",
|
||||
"replace_in_files",
|
||||
"replace_symbol_body",
|
||||
"insert_after_symbol",
|
||||
"insert_before_symbol",
|
||||
}
|
||||
optional_methods = {"delete_lines", "replace_lines", "insert_at_line"} # the line-level operations are optional (as are the tools)
|
||||
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "edit"
|
||||
assert set(facade.enabled_method_names) == default_methods
|
||||
for name in optional_methods:
|
||||
assert facade.get_method(name).info.optional
|
||||
assert all(facade.get_method(name).info.can_edit for name in default_methods | optional_methods)
|
||||
|
||||
|
||||
def test_replace_in_files_dry_run_returns_inspectable_preview(api: EditApi, project: Project) -> None:
|
||||
preview = api.replace_in_files("foo", "bar", mode="literal", dry_run=True)
|
||||
assert isinstance(preview, ReplacementPreview)
|
||||
|
||||
# the occurrences are accessible from code
|
||||
assert [o.relative_path for o in preview.occurrences] == ["a.py", "a.py", "b.py"]
|
||||
assert preview.affected_files == ["a.py", "b.py"]
|
||||
assert all(o.replacement == "bar" for o in preview.occurrences)
|
||||
|
||||
# the rendering lists the occurrence ids and diffs; nothing was modified
|
||||
rendered = preview.represent()
|
||||
assert "DRY RUN" in rendered
|
||||
assert all(o.occurrence_id in rendered for o in preview.occurrences)
|
||||
assert (Path(project.project_root) / "a.py").read_text(encoding="utf-8") == "x = foo(1)\ny = foo(2)\n"
|
||||
|
||||
|
||||
def test_replace_in_files_guard_failure_includes_preview(api: EditApi) -> None:
|
||||
with pytest.raises(ValueError, match="expected_count=1") as exc_info:
|
||||
api.replace_in_files("foo", "bar", mode="literal", expected_count=1)
|
||||
assert "b.py" in str(exc_info.value) # the listing of prospective changes is included
|
||||
@@ -0,0 +1,107 @@
|
||||
"""
|
||||
End-to-end test of querying an external project through the REPL: a project server executes the language server
|
||||
operations of the queried project, and the (pickled) results are usable in the querying agent's REPL.
|
||||
"""
|
||||
|
||||
import socket
|
||||
import threading
|
||||
from collections.abc import Iterator
|
||||
|
||||
import pytest
|
||||
from werkzeug.serving import make_server
|
||||
|
||||
from serena.agent import SerenaAgent
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project_server import ProjectServer, ProjectServerClient
|
||||
from serena.repl.api.lsp_api import LspSymbolCollection
|
||||
from serena.tools import SerenaReplTool
|
||||
from solidlsp.ls_config import LanguageServerId
|
||||
from test.conftest import language_server_tests_enabled
|
||||
from test.serena.test_serena_agent import serena_config # noqa: F401 (fixture)
|
||||
|
||||
|
||||
def _free_port() -> int:
|
||||
with socket.socket() as sock:
|
||||
sock.bind(("127.0.0.1", 0))
|
||||
return sock.getsockname()[1]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project_server(serena_config: SerenaConfig) -> Iterator[tuple[ProjectServer, int]]: # noqa: F811
|
||||
"""
|
||||
Runs a project server (backed by a real agent) on a free port for the duration of the test.
|
||||
"""
|
||||
config = serena_config
|
||||
port = _free_port()
|
||||
|
||||
# construct the server around an agent with the test configuration (the constructor would load the user's configuration)
|
||||
server = ProjectServer.__new__(ProjectServer)
|
||||
server._agent = SerenaAgent(serena_config=config)
|
||||
server._loaded_projects_by_root = {}
|
||||
server._project_load_locks_by_root = {}
|
||||
server._active_project_lock = threading.Lock()
|
||||
server._loaded_projects_lock = threading.Lock()
|
||||
server._port = port
|
||||
server._host = "127.0.0.1"
|
||||
from flask import Flask
|
||||
|
||||
server._app = Flask(__name__)
|
||||
server._setup_routes()
|
||||
|
||||
http_server = make_server("127.0.0.1", port, server._app, threaded=True)
|
||||
thread = threading.Thread(target=http_server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
yield server, port
|
||||
finally:
|
||||
http_server.shutdown()
|
||||
server._agent.on_shutdown(timeout=5)
|
||||
|
||||
|
||||
@pytest.mark.python
|
||||
@pytest.mark.skipif(not language_server_tests_enabled(LanguageServerId.PYTHON), reason="python tests are disabled in this environment")
|
||||
def test_facade_method_results_are_transferred_from_the_project_server(project_server: tuple[ProjectServer, int]) -> None:
|
||||
server, port = project_server
|
||||
client = ProjectServerClient(server.get_serena_config(), port=port)
|
||||
result = client.call_facade_method("test_repo_python", "lsp", "find_symbol", ["create_user"], {"include_body": True})
|
||||
|
||||
# the result is a self-contained object which can be processed and rendered locally
|
||||
assert isinstance(result, LspSymbolCollection)
|
||||
assert [s.name for s in result.symbols] == ["create_user"]
|
||||
assert result.symbols[0].body.startswith("def create_user")
|
||||
assert "create_user" in result.represent()
|
||||
|
||||
|
||||
@pytest.mark.python
|
||||
@pytest.mark.skipif(not language_server_tests_enabled(LanguageServerId.PYTHON), reason="python tests are disabled in this environment")
|
||||
def test_external_project_context_in_repl(
|
||||
project_server: tuple[ProjectServer, int],
|
||||
serena_config: SerenaConfig, # noqa: F811
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
) -> None:
|
||||
server, port = project_server
|
||||
monkeypatch.setattr(ProjectServer, "PORT", port) # let the REPL's external project context use the test server
|
||||
|
||||
# enable the optional "ext" facade
|
||||
serena_config.included_apis = ["ext"]
|
||||
|
||||
# the querying agent has another project active and queries the python test project
|
||||
serena_config.auth_secret = server.get_auth_secret()
|
||||
agent = SerenaAgent(project="test_repo_typescript", serena_config=serena_config)
|
||||
agent.execute_task(lambda: None)
|
||||
try:
|
||||
tool = agent.get_tool(SerenaReplTool)
|
||||
session_id = agent.create_session().session_id
|
||||
code = (
|
||||
'with s.ext.read_project_context("test_repo_python"):\n'
|
||||
' result = s.lsp.find_symbol("create_user")\n'
|
||||
"[s.name for s in result.symbols]"
|
||||
)
|
||||
assert tool.apply(session_id, code) == "create_user"
|
||||
assert server._loaded_projects_by_root, "the operation was not executed by the project server"
|
||||
|
||||
# the result persists and can be used after the context; the active project is restored
|
||||
assert "services.py" in tool.apply(session_id, "result.represent()")
|
||||
assert "test_repo_typescript" in agent.get_current_config_overview()
|
||||
finally:
|
||||
agent.on_shutdown(timeout=5)
|
||||
@@ -15,10 +15,8 @@ def read_file_tool(tmp_path: Path) -> ReadFileTool:
|
||||
project = Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
tool = ReadFileTool(agent)
|
||||
# bypass the length limit, which would otherwise depend on the agent configuration
|
||||
tool._limit_length = lambda result, max_answer_chars: result
|
||||
return tool
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return ReadFileTool(agent)
|
||||
|
||||
|
||||
def _deleted_by_delete_lines(content: str, line: int) -> str:
|
||||
|
||||
@@ -0,0 +1,81 @@
|
||||
"""
|
||||
Tests for the file system facade API.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.fs_api import FsApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project(tmp_path: Path) -> Project:
|
||||
(tmp_path / "src").mkdir()
|
||||
(tmp_path / "src" / "a.py").write_text("x = foo(1)\ny = foo(2)\nz = 3\n", encoding="utf-8")
|
||||
(tmp_path / "src" / "b.txt").write_text("foo in text\n", encoding="utf-8")
|
||||
(tmp_path / "README.md").write_text("# readme\n", encoding="utf-8")
|
||||
return Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(project: Project) -> FsApi:
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return FsApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_file_operations(api: FsApi) -> None:
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "fs"
|
||||
assert set(facade.enabled_method_names) == {"read_file", "create_text_file", "list_dir", "find_file", "search_for_pattern"}
|
||||
assert {name for name in facade.enabled_method_names if facade.get_method(name).info.can_edit} == {"create_text_file"}
|
||||
|
||||
|
||||
def test_read_file(api: FsApi) -> None:
|
||||
content = api.read_file("src/a.py")
|
||||
assert content.lines == ["x = foo(1)", "y = foo(2)", "z = 3", ""]
|
||||
assert content.represent() == content.text
|
||||
|
||||
assert api.read_file("src/a.py", start_line=1, end_line=1).text == "y = foo(2)"
|
||||
assert api.read_file("src/a.py", start_line=-2).lines == ["z = 3", ""]
|
||||
|
||||
|
||||
def test_create_text_file(api: FsApi, project: Project) -> None:
|
||||
result = api.create_text_file("sub/new.txt", "hello\n")
|
||||
assert "new.txt" in result
|
||||
assert (Path(project.project_root) / "sub" / "new.txt").read_text(encoding="utf-8") == "hello\n"
|
||||
|
||||
result = api.create_text_file("sub/new.txt", "changed\n")
|
||||
assert "Overwrote" in result
|
||||
|
||||
with pytest.raises(AssertionError):
|
||||
api.create_text_file("../outside.txt", "nope")
|
||||
|
||||
|
||||
def test_list_dir_and_find_file(api: FsApi) -> None:
|
||||
listing = api.list_dir(".", recursive=True)
|
||||
assert "src" in listing.dirs
|
||||
assert {"src/a.py", "src/b.txt", "README.md"} <= {f.replace("\\", "/") for f in listing.files}
|
||||
assert '"dirs"' in listing.represent() and '"files"' in listing.represent()
|
||||
|
||||
with pytest.raises(FileNotFoundError):
|
||||
api.list_dir("missing", recursive=False)
|
||||
|
||||
assert [f.replace("\\", "/") for f in api.find_file("*.py", ".")] == ["src/a.py"]
|
||||
|
||||
|
||||
def test_search_for_pattern(api: FsApi) -> None:
|
||||
matches = api.search_for_pattern("foo", relative_path="src")
|
||||
assert len(matches) == 3
|
||||
assert {m.source_file_path.replace("\\", "/") for m in matches.matches} == {"src/a.py", "src/b.txt"}
|
||||
|
||||
# restricting to code files excludes the text file; the rendering maps files to matched lines
|
||||
code_matches = api.search_for_pattern("foo", restrict_search_to_code_files=True)
|
||||
assert all(m.source_file_path.endswith("a.py") for m in code_matches.matches)
|
||||
assert "foo(1)" in code_matches.represent()
|
||||
@@ -0,0 +1,79 @@
|
||||
"""
|
||||
Tests for the JetBrains facade API, using a mocked plugin client (no IDE required).
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.repl.api.jb_api import JetBrainsApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def agent() -> MagicMock:
|
||||
agent = MagicMock()
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return agent
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> MagicMock:
|
||||
client = MagicMock()
|
||||
with patch("serena.repl.api.jb_api.JetBrainsPluginClient.from_project") as from_project:
|
||||
from_project.return_value.__enter__.return_value = client
|
||||
yield client
|
||||
|
||||
|
||||
def test_facade_exposes_all_jetbrains_operations(agent: MagicMock) -> None:
|
||||
facade = Facade.from_api(JetBrainsApi(agent), ApiScope())
|
||||
assert facade.name == "jb"
|
||||
assert set(facade.enabled_method_names) == {
|
||||
"find_symbol",
|
||||
"find_referencing_symbols",
|
||||
"get_symbols_overview",
|
||||
"get_type_hierarchy",
|
||||
"find_declaration",
|
||||
"find_implementations",
|
||||
"rename",
|
||||
"move",
|
||||
"safe_delete",
|
||||
"inline_symbol",
|
||||
"run_inspections",
|
||||
"list_inspections",
|
||||
"debug_eval",
|
||||
"debug_eval_info",
|
||||
}
|
||||
|
||||
|
||||
def test_find_symbol_renders_grouped_and_exposes_symbols(agent: MagicMock, client: MagicMock) -> None:
|
||||
symbols = [
|
||||
{"name_path": "Foo", "type": "class", "relative_path": "a.py", "quick_info": "class Foo"},
|
||||
{"name_path": "Bar/foo", "type": "method", "relative_path": "b.py", "quick_info": "def foo()"},
|
||||
]
|
||||
client.find_symbol.return_value = {"symbols": symbols}
|
||||
|
||||
result = JetBrainsApi(agent).find_symbol("foo")
|
||||
|
||||
# the underlying symbols are accessible from code
|
||||
assert [s["name_path"] for s in result.symbols] == ["Foo", "Bar/foo"]
|
||||
# and the rendering contains them, grouped by file
|
||||
rendered = result.represent()
|
||||
assert '"a.py"' in rendered and '"b.py"' in rendered
|
||||
assert "class Foo" in rendered
|
||||
|
||||
|
||||
def test_find_symbol_rejects_too_many_matches_with_identifiers(agent: MagicMock, client: MagicMock) -> None:
|
||||
client.find_symbol.return_value = {
|
||||
"symbols": [{"name_path": f"foo{i}", "type": "function", "relative_path": "a.py", "body": "..."} for i in range(3)]
|
||||
}
|
||||
with pytest.raises(ValueError, match="Matched 3>max_matches=1") as exc_info:
|
||||
JetBrainsApi(agent).find_symbol("foo*", max_matches=1)
|
||||
assert "foo2" in str(exc_info.value)
|
||||
assert '"body"' not in str(exc_info.value) # only identifiers, no content
|
||||
|
||||
|
||||
def test_find_symbol_rejects_wildcard_only_pattern(agent: MagicMock, client: MagicMock) -> None:
|
||||
with pytest.raises(ValueError, match="get_symbols_overview"):
|
||||
JetBrainsApi(agent).find_symbol("*")
|
||||
client.find_symbol.assert_not_called()
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Tests for the mcp.py module in serena."""
|
||||
|
||||
import pytest
|
||||
from mcp.server.fastmcp import Context
|
||||
from mcp.server.fastmcp.tools.base import Tool as MCPTool
|
||||
|
||||
from serena import __version__
|
||||
@@ -8,6 +9,8 @@ from serena.agent import Tool, ToolRegistry
|
||||
from serena.config.context_mode import SerenaAgentContext
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.mcp import SerenaMCPFactory
|
||||
from serena.repl.facade import ApiScope
|
||||
from serena.repl.repl import SerenaRepl
|
||||
|
||||
make_tool = SerenaMCPFactory.make_mcp_tool
|
||||
|
||||
@@ -22,6 +25,14 @@ class MockAgent:
|
||||
def get_context() -> SerenaAgentContext:
|
||||
return SerenaAgentContext.load_default()
|
||||
|
||||
@staticmethod
|
||||
def get_repl() -> SerenaRepl:
|
||||
return SerenaRepl([], ApiScope())
|
||||
|
||||
@staticmethod
|
||||
def is_single_project() -> bool:
|
||||
return False
|
||||
|
||||
|
||||
class BaseMockTool(Tool):
|
||||
"""A mock Tool class for testing."""
|
||||
@@ -46,6 +57,7 @@ class BasicTool(BaseMockTool):
|
||||
self,
|
||||
log_call: bool = True,
|
||||
catch_exceptions: bool = True,
|
||||
mcp_ctx: Context | None = None,
|
||||
**kwargs,
|
||||
) -> str:
|
||||
"""Mock implementation of apply_ex."""
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
"""
|
||||
Tests for the memory facade API.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.mem_api import MemoryApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(tmp_path: Path) -> MemoryApi:
|
||||
project = Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return MemoryApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_memory_operations(api: MemoryApi) -> None:
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "mem"
|
||||
assert set(facade.enabled_method_names) == {
|
||||
"list_memories",
|
||||
"read_memory",
|
||||
"write_memory",
|
||||
"edit_memory",
|
||||
"rename_memory",
|
||||
"delete_memory",
|
||||
"onboarding",
|
||||
}
|
||||
assert {name for name in facade.enabled_method_names if facade.get_method(name).info.can_edit} == {
|
||||
"write_memory",
|
||||
"edit_memory",
|
||||
"rename_memory",
|
||||
"delete_memory",
|
||||
}
|
||||
|
||||
|
||||
def test_memory_lifecycle(api: MemoryApi) -> None:
|
||||
api.write_memory("topic/first", "# First\nhello")
|
||||
api.write_memory("second", "see `mem:topic/first`")
|
||||
|
||||
# listing exposes the names to code and renders as JSON (global memories of the machine may be present, too)
|
||||
memory_list = api.list_memories()
|
||||
assert {"second", "topic/first"} <= set(memory_list.memories)
|
||||
assert api.list_memories("topic").memories == ["topic/first"]
|
||||
assert '"memories"' in memory_list.represent()
|
||||
|
||||
# reading, editing, renaming (with reference propagation) and deleting
|
||||
assert api.read_memory("topic/first") == "# First\nhello"
|
||||
api.edit_memory("topic/first", "hello", "world", mode="literal")
|
||||
assert api.read_memory("topic/first") == "# First\nworld"
|
||||
api.rename_memory("topic/first", "topic/renamed")
|
||||
assert "mem:topic/renamed" in api.read_memory("second")
|
||||
api.delete_memory("second")
|
||||
assert "second" not in api.list_memories().memories
|
||||
assert api.list_memories("topic").memories == ["topic/renamed"]
|
||||
|
||||
|
||||
def test_write_memory_rejects_overlong_content(api: MemoryApi) -> None:
|
||||
with pytest.raises(ValueError, match="too long"):
|
||||
api.write_memory("big", "x" * 100, max_chars=10)
|
||||
@@ -8,8 +8,11 @@ from typing import Any, cast
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
from flask import Flask
|
||||
from werkzeug.serving import make_server
|
||||
|
||||
from serena.project_server import ProjectServer, QueryProjectRequest
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project_server import ProjectServer, ProjectServerClient, QueryProjectRequest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
@@ -23,6 +26,53 @@ def project_server() -> ProjectServer:
|
||||
return server
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def authenticated_server(project_server: ProjectServer, monkeypatch: pytest.MonkeyPatch) -> ProjectServer:
|
||||
# expose the real HTTP routes with a query handler that needs no language servers
|
||||
project_server._agent.serena_config.auth_secret = "test-shared-secret"
|
||||
project_server._app = Flask(__name__)
|
||||
monkeypatch.setattr(project_server, "_query_project", lambda req: req.project_name)
|
||||
project_server._setup_routes()
|
||||
return project_server
|
||||
|
||||
|
||||
@pytest.mark.parametrize("authorization", [None, "Bearer wrong-secret", "test-shared-secret", "Bearer café"])
|
||||
@pytest.mark.parametrize("path", ["/heartbeat", "/query_project"])
|
||||
def test_project_server_rejects_invalid_credentials(authenticated_server: ProjectServer, authorization: str | None, path: str) -> None:
|
||||
# unauthorized requests are rejected even before query payload validation
|
||||
headers = {} if authorization is None else {"Authorization": authorization}
|
||||
with authenticated_server._app.test_client() as client:
|
||||
response = client.open(path, method="GET" if path == "/heartbeat" else "POST", headers=headers)
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.parametrize("use_wrong_password", [True, False])
|
||||
def test_project_server_client_authenticates_requests(authenticated_server: ProjectServer, use_wrong_password: bool) -> None:
|
||||
# run the authenticated endpoints on an ephemeral local port
|
||||
http_server = make_server("127.0.0.1", 0, authenticated_server._app)
|
||||
thread = threading.Thread(target=http_server.serve_forever, daemon=True)
|
||||
thread.start()
|
||||
try:
|
||||
|
||||
def check_client():
|
||||
serena_config = SerenaConfig()
|
||||
serena_config.auth_secret = "wrong-secret" if use_wrong_password else authenticated_server.get_auth_secret()
|
||||
client = ProjectServerClient(serena_config, port=http_server.server_port)
|
||||
assert client.query_project("other", "find_symbol", "{}") == "other"
|
||||
|
||||
# construction authenticates the heartbeat, raising a Connection error if using the wrong password
|
||||
if use_wrong_password:
|
||||
with pytest.raises(expected_exception=ConnectionError, match="401"):
|
||||
check_client()
|
||||
else:
|
||||
check_client()
|
||||
|
||||
finally:
|
||||
http_server.shutdown()
|
||||
thread.join(timeout=5)
|
||||
http_server.server_close()
|
||||
|
||||
|
||||
def test_cached_project_lookup_is_not_blocked_by_unrelated_cold_load(project_server: ProjectServer) -> None:
|
||||
cached_root = Path("/cached")
|
||||
cold_root = Path("/cold")
|
||||
|
||||
@@ -0,0 +1,380 @@
|
||||
"""
|
||||
Tests for the REPL tool, which executes Python code against the facade entrypoint `s`.
|
||||
"""
|
||||
|
||||
import os
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import ApiInclusionDefinition, LanguageBackend
|
||||
from serena.repl.api.edit_api import EditApi
|
||||
from serena.repl.api.lsp_api import LspApi
|
||||
from serena.repl.external_project import ExternalProjectExecution
|
||||
from serena.repl.facade import ApiScope, Facade, FacadeApi, FacadeMethodInfo, facade_method
|
||||
from serena.repl.repl import SerenaRepl
|
||||
from serena.session import SerenaSession
|
||||
from serena.tools import FindSymbolTool, SerenaReplTool
|
||||
from solidlsp.ls_config import LanguageServerId
|
||||
from test.conftest import agent_for_project_context
|
||||
|
||||
|
||||
class TestReplExecution:
|
||||
"""Tests the code execution mechanics of the REPL, which do not require a project."""
|
||||
|
||||
@pytest.fixture
|
||||
def repl(self) -> SerenaRepl:
|
||||
return SerenaRepl([Facade.from_api(LspApi(MagicMock()), ApiScope())], ApiScope())
|
||||
|
||||
def test_last_expression_defines_result(self, repl: SerenaRepl) -> None:
|
||||
assert repl.execute("x = 20\ny = 22\nx + y") == "42"
|
||||
assert repl.execute("x = 20\ny = 22") == "None" # no trailing expression
|
||||
assert repl.execute("") == "None"
|
||||
|
||||
def test_return_yields_a_hint(self, repl: SerenaRepl) -> None:
|
||||
result = repl.execute("x = 1\nreturn x")
|
||||
assert result.startswith("SyntaxError") and "line 2" in result
|
||||
assert "last expression is the result" in result
|
||||
|
||||
def test_single_expression_is_evaluated(self, repl: SerenaRepl) -> None:
|
||||
assert repl.execute("1 + 2") == "3"
|
||||
|
||||
def test_list_is_rendered_element_wise(self, repl: SerenaRepl) -> None:
|
||||
assert repl.execute('["a", "b"]') == "a\nb"
|
||||
|
||||
def test_error_reports_type_message_and_line(self, repl: SerenaRepl) -> None:
|
||||
result = repl.execute("x = 1\nraise ValueError('boom')")
|
||||
assert result.startswith("ValueError: boom")
|
||||
assert "line 2" in result
|
||||
|
||||
def test_syntax_error_reports_line(self, repl: SerenaRepl) -> None:
|
||||
result = repl.execute("x = 1\ny = (2")
|
||||
assert result.startswith("SyntaxError")
|
||||
assert "line 2" in result
|
||||
|
||||
def test_top_level_names_persist_within_session(self, repl: SerenaRepl) -> None:
|
||||
session = SerenaSession("a")
|
||||
repl.execute("x = 20\ndef double(v):\n return 2 * v\nfor i in range(3):\n pass\nimport os as os_module", session)
|
||||
assert repl.execute("double(x) + i", session) == "42"
|
||||
assert repl.execute("os_module.sep is not None", session) == "True"
|
||||
|
||||
# local variables of nested scopes do not persist; persisted items can be listed and cleared
|
||||
assert "v" not in repl.execute("s.vars()", session)
|
||||
listing = repl.execute("s.vars()", session)
|
||||
assert "x: int = 20" in listing and "double: function" in listing
|
||||
assert repl.execute("s.clear()", session) == "Removed 4 persisted item(s)."
|
||||
assert repl.execute("s.vars()", session) == "No persisted variables."
|
||||
assert "NameError" in repl.execute("x", session)
|
||||
|
||||
def test_sessions_have_separate_namespaces(self, repl: SerenaRepl) -> None:
|
||||
session_a = SerenaSession("a")
|
||||
repl.execute("x = 1", session_a)
|
||||
assert "NameError" in repl.execute("x", SerenaSession("b"))
|
||||
assert repl.execute("x", session_a) == "1"
|
||||
# ... and execution without a session persists nothing
|
||||
repl.execute("y = 1")
|
||||
assert "NameError" in repl.execute("y")
|
||||
|
||||
def test_persisted_functions_use_the_current_entrypoint(self) -> None:
|
||||
# a function defined against one REPL instance uses the entrypoint of the REPL that later calls it
|
||||
session = SerenaSession("a")
|
||||
SerenaRepl([Facade.from_api(LspApi(MagicMock()), ApiScope())], ApiScope()).execute("def facades():\n return s.info()", session)
|
||||
rebuilt_repl = SerenaRepl([Facade.from_api(EditApi(MagicMock()), ApiScope())], ApiScope())
|
||||
overview = rebuilt_repl.execute("facades()", session)
|
||||
assert "s.edit" in overview and "s.lsp" not in overview
|
||||
|
||||
@pytest.mark.parametrize("backend", [LanguageBackend.LSP, LanguageBackend.JETBRAINS])
|
||||
@pytest.mark.parametrize("read_only", [True, False])
|
||||
def test_external_project_dispatch(self, backend: LanguageBackend, read_only: bool) -> None:
|
||||
agent = MagicMock()
|
||||
agent.get_language_backend.return_value = backend
|
||||
|
||||
class FakeExternalProject(ExternalProjectExecution):
|
||||
def __init__(self) -> None:
|
||||
super().__init__("other", read_only=read_only, agent=agent)
|
||||
self.calls: list[tuple[str, str, tuple, dict]] = []
|
||||
|
||||
def call_remotely(self, facade_name: str, method_name: str, args: tuple, kwargs: dict) -> str:
|
||||
self.calls.append((facade_name, method_name, args, kwargs))
|
||||
return "remote result"
|
||||
|
||||
class LocalApi(FacadeApi):
|
||||
@facade_method(can_edit=True)
|
||||
def write(self, content: str) -> str:
|
||||
return f"local result: {content}"
|
||||
|
||||
# expose a server-backed read and a backend-dependent write
|
||||
facades = [Facade.from_api(LspApi(agent), ApiScope()), Facade.from_api(LocalApi(agent, "local", "local operations"), ApiScope())]
|
||||
repl = SerenaRepl(facades, ApiScope())
|
||||
external_project = FakeExternalProject()
|
||||
repl.entrypoint.set_external_project_(external_project)
|
||||
|
||||
# methods explicitly requiring the project server are executed remotely
|
||||
assert repl.execute('s.lsp.find_symbol("Foo", depth=1)') == "remote result"
|
||||
assert external_project.calls == [("lsp", "find_symbol", ("Foo",), {"depth": 1})]
|
||||
|
||||
# writes obey the context's access mode and use the selected backend
|
||||
result = repl.execute('s.local.write("content")')
|
||||
if read_only:
|
||||
assert "PermissionError" in result and "read-only" in result
|
||||
assert len(external_project.calls) == 1
|
||||
elif backend.is_lsp():
|
||||
assert result == "remote result"
|
||||
assert external_project.calls[-1] == ("local", "write", ("content",), {})
|
||||
else:
|
||||
assert result == "local result: content"
|
||||
assert len(external_project.calls) == 1
|
||||
|
||||
# leaving the external execution context restores local execution
|
||||
repl.entrypoint.set_external_project_(None)
|
||||
assert repl.execute('s.local.write("restored")') == "local result: restored"
|
||||
|
||||
def test_facade_discovery(self, repl: SerenaRepl) -> None:
|
||||
overview = repl.execute("s.info()")
|
||||
assert "s.lsp" in overview
|
||||
assert "find_symbol" in overview # method names are listed, but not signatures
|
||||
assert "name_path_pattern" not in overview
|
||||
facade_info = repl.execute('s.info("lsp")')
|
||||
assert "find_symbol(" in facade_info
|
||||
method_info = repl.execute('s.info("lsp.find_symbol")')
|
||||
assert "name_path_pattern" in method_info
|
||||
|
||||
def test_type_discovery(self, repl: SerenaRepl) -> None:
|
||||
# the overview names the result types of methods returning objects that can be processed in code
|
||||
assert "find_symbol -> LspSymbolCollection" in repl.execute("s.info()")
|
||||
|
||||
# signatures render type names without module paths, and point to the documentation of referenced return types
|
||||
method_info = repl.execute('s.info("lsp.find_symbol")')
|
||||
assert "-> LspSymbolCollection" in method_info and "lsp_api." not in method_info
|
||||
assert 's.info("LspSymbolCollection")' in method_info
|
||||
|
||||
# the facade description documents the operations only and lists the result types by name
|
||||
facade_info = repl.execute('s.info("lsp")')
|
||||
assert "type LspSymbolCollection" not in facade_info
|
||||
assert "Result types: " in facade_info and "LspSymbolCollection" in facade_info
|
||||
|
||||
# types can be requested via the facade or by bare name, and their curated members are documented
|
||||
type_info = repl.execute('s.info("lsp.LanguageServerSymbol")')
|
||||
assert type_info == repl.execute('s.info("LanguageServerSymbol")')
|
||||
assert "get_name_path() -> str" in type_info and "iter_children()" in type_info
|
||||
assert "to_dict" not in type_info # not among the curated members
|
||||
|
||||
# types reachable through annotations are documented without being declared: SymbolKind (a parameter type of
|
||||
# iter_ancestors) is documented as an enum with its members, both transitively and on request
|
||||
assert "enum SymbolKind" in type_info and "SymbolKind.Class = 5" in type_info
|
||||
assert "enum SymbolKind" in repl.execute('s.info("SymbolKind")')
|
||||
|
||||
# TypedDicts reachable through annotations are documented with their keys
|
||||
assert "keys:" in repl.execute('s.info("Diagnostic")') and "severity" in repl.execute('s.info("Diagnostic")')
|
||||
assert "represent()" not in repl.execute('s.info("LspSymbolCollection")') # the representation mechanism is not exposed
|
||||
|
||||
def test_contained_types_are_documented_once_per_session(self, repl: SerenaRepl) -> None:
|
||||
session = SerenaSession("test")
|
||||
|
||||
# a type's documentation includes the types it contains (transitively)
|
||||
first = repl.execute('s.info("LspReferenceCollection")', session)
|
||||
assert "type LspReferenceCollection" in first
|
||||
assert "type ReferenceInLanguageServerSymbol" in first and "type LanguageServerSymbol" in first
|
||||
|
||||
# a contained type documented earlier in the session is only pointed to; an explicit request yields it again
|
||||
second = repl.execute('s.info("LspSymbolCollection")', session)
|
||||
assert "type LspSymbolCollection" in second
|
||||
assert "type LanguageServerSymbol: documented earlier" in second and "get_name_path()" not in second
|
||||
assert "get_name_path()" in repl.execute('s.info("LanguageServerSymbol")', session)
|
||||
|
||||
# another session is unaffected
|
||||
assert "get_name_path()" in repl.execute('s.info("LspSymbolCollection")', SerenaSession("other"))
|
||||
|
||||
def test_info_documents_several_items(self, repl: SerenaRepl) -> None:
|
||||
info = repl.execute('s.info("lsp.find_symbol", "nope", "lsp.LspSymbolCollection")')
|
||||
assert "lsp.find_symbol(" in info and "type LspSymbolCollection" in info
|
||||
assert "Unknown item 'nope'" in info # an unknown item does not prevent the documentation of the others
|
||||
|
||||
|
||||
class TestFacade:
|
||||
"""Tests the indirection between facades and their implementations."""
|
||||
|
||||
class DummyApi(FacadeApi):
|
||||
def __init__(self, agent: MagicMock) -> None:
|
||||
super().__init__(agent, name="dummy", description="a dummy facade")
|
||||
|
||||
@facade_method()
|
||||
def add(self, a: int, b: int) -> int:
|
||||
"""Adds two numbers."""
|
||||
return a + b
|
||||
|
||||
@facade_method(can_edit=True)
|
||||
def secret(self) -> str:
|
||||
return "hidden"
|
||||
|
||||
@facade_method(optional=True, beta=True)
|
||||
def extra(self) -> str:
|
||||
return "extra"
|
||||
|
||||
@facade_method(niche=True)
|
||||
def rarely(self, x: int) -> str:
|
||||
"""Rarely needed operation.
|
||||
|
||||
:param x: some parameter
|
||||
"""
|
||||
return str(x)
|
||||
|
||||
def undecorated(self) -> str:
|
||||
"""Public within Serena, but not exposed, since it is not decorated."""
|
||||
return "internal"
|
||||
|
||||
def _internal(self) -> None:
|
||||
pass
|
||||
|
||||
@staticmethod
|
||||
def _scope(*definitions: ApiInclusionDefinition, **kwargs: list[str]) -> ApiScope:
|
||||
"""
|
||||
:param definitions: definitions to apply in order
|
||||
:param kwargs: an additional definition (`included_apis`/`excluded_apis`) to apply last
|
||||
"""
|
||||
scope = ApiScope()
|
||||
for definition in definitions:
|
||||
scope.process(definition)
|
||||
if kwargs:
|
||||
scope.process(ApiInclusionDefinition(**kwargs))
|
||||
return scope
|
||||
|
||||
def test_optional_facade_is_opt_in(self) -> None:
|
||||
# an optional facade is disabled unless it is included explicitly
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope(), is_optional=True)
|
||||
assert not facade.is_enabled()
|
||||
assert facade.enabled_method_names == []
|
||||
|
||||
# including the facade enables its non-optional methods
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), self._scope(included_apis=["dummy"]), is_optional=True)
|
||||
assert facade.is_enabled()
|
||||
assert set(facade.enabled_method_names) == {"add", "secret", "rarely"}
|
||||
|
||||
# including a single method enables the facade with just that method
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), self._scope(included_apis=["dummy.add"]), is_optional=True)
|
||||
assert facade.is_enabled()
|
||||
assert facade.enabled_method_names == ["add"]
|
||||
|
||||
def test_api_scope_facade_exclusion_and_method_inclusion(self) -> None:
|
||||
# excluding the facade disables everything
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), self._scope(excluded_apis=["dummy"]))
|
||||
assert facade.enabled_method_names == []
|
||||
|
||||
# an excluded facade is opt-in: a method inclusion enables exactly that method
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), self._scope(excluded_apis=["dummy"], included_apis=["dummy.add"]))
|
||||
assert facade.enabled_method_names == ["add"]
|
||||
|
||||
def test_entrypoint_omits_excluded_facades(self) -> None:
|
||||
def create_repl(scope: ApiScope) -> SerenaRepl:
|
||||
return SerenaRepl([Facade.from_api(self.DummyApi(MagicMock()), scope)], scope)
|
||||
|
||||
assert "s.dummy" in create_repl(ApiScope()).execute("s.info()")
|
||||
assert "s.dummy" not in create_repl(self._scope(excluded_apis=["dummy"])).execute("s.info()")
|
||||
# a method inclusion keeps the facade available (with just that method)
|
||||
overview = create_repl(self._scope(excluded_apis=["dummy"], included_apis=["dummy.add"])).execute("s.info()")
|
||||
assert "s.dummy" in overview and "methods: add" in overview
|
||||
|
||||
def test_api_scope_later_definitions_take_precedence(self) -> None:
|
||||
scope = self._scope(
|
||||
ApiInclusionDefinition(included_apis=["dummy.extra"]),
|
||||
ApiInclusionDefinition(excluded_apis=["dummy.extra", "dummy.add"]),
|
||||
ApiInclusionDefinition(included_apis=["dummy.add"]),
|
||||
)
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), scope)
|
||||
assert set(facade.enabled_method_names) == {"add", "secret", "rarely"}
|
||||
|
||||
def test_api_scope_read_only_excludes_editing_methods(self) -> None:
|
||||
scope = self._scope(included_apis=["dummy.secret"])
|
||||
scope.exclude_editing()
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), scope)
|
||||
assert set(facade.enabled_method_names) == {"add", "rarely"}
|
||||
|
||||
def test_enabled_methods_delegate_to_implementation(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
assert facade.add(1, 2) == 3
|
||||
assert "dummy.add(a: int, b: int) -> int" in facade.describe()
|
||||
assert "Adds two numbers." in facade.describe_member("add")
|
||||
|
||||
def test_disabled_methods_are_inaccessible_and_undocumented(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), self._scope(excluded_apis=["dummy.secret"]))
|
||||
assert facade.add(1, 2) == 3
|
||||
with pytest.raises(AttributeError):
|
||||
facade.secret()
|
||||
with pytest.raises(ValueError):
|
||||
facade.describe_member("secret")
|
||||
assert "secret" not in facade.describe()
|
||||
assert "_internal" not in facade.describe()
|
||||
|
||||
def test_undecorated_methods_are_not_exposed(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
assert self.DummyApi(MagicMock()).undecorated() == "internal" # usable from within Serena
|
||||
with pytest.raises(AttributeError):
|
||||
facade.undecorated()
|
||||
with pytest.raises(ValueError):
|
||||
facade.get_method("undecorated")
|
||||
assert "undecorated" not in facade.describe()
|
||||
assert "undecorated" not in facade.enabled_method_names
|
||||
|
||||
def test_niche_methods_are_summarised_in_facade_description(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
description = facade.describe()
|
||||
assert "dummy.rarely: Rarely needed operation." in description
|
||||
assert ":param x:" not in description # only the summary, no signature or full documentation
|
||||
assert ":param x:" in facade.describe_member("rarely") # full documentation on request
|
||||
|
||||
def test_optional_methods_are_disabled_by_default(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
assert "extra" not in facade.enabled_method_names
|
||||
with pytest.raises(AttributeError):
|
||||
facade.extra()
|
||||
facade.get_method("extra").enabled = True
|
||||
assert facade.extra() == "extra"
|
||||
|
||||
# an explicit inclusion enables it
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), self._scope(included_apis=["dummy.extra"]))
|
||||
assert "extra" in facade.enabled_method_names
|
||||
|
||||
def test_corresponding_tool(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
assert facade.get_method("add").info.get_corresponding_tool_name() is None
|
||||
|
||||
lsp_facade = Facade.from_api(LspApi(MagicMock()), ApiScope())
|
||||
info = lsp_facade.get_method("find_symbol").info
|
||||
assert info.corresponding_tool is FindSymbolTool
|
||||
assert info.get_corresponding_tool_name() == "find_symbol"
|
||||
|
||||
def test_method_info_mirrors_decorator(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
assert facade.get_method("add").info == FacadeMethodInfo(name="add")
|
||||
assert facade.get_method("secret").info.can_edit
|
||||
assert facade.get_method("extra").info == FacadeMethodInfo(name="extra", optional=True, beta=True)
|
||||
|
||||
def test_enablement_can_be_changed(self) -> None:
|
||||
facade = Facade.from_api(self.DummyApi(MagicMock()), ApiScope())
|
||||
facade.get_method("secret").enabled = False
|
||||
with pytest.raises(AttributeError):
|
||||
facade.secret()
|
||||
facade.get_method("secret").enabled = True
|
||||
assert facade.secret() == "hidden"
|
||||
|
||||
|
||||
@pytest.mark.python
|
||||
class TestLspFacade:
|
||||
_SERVICES_FILE = os.path.join("test_repo", "services.py")
|
||||
|
||||
def test_find_symbol_via_repl(self) -> None:
|
||||
with agent_for_project_context(LanguageServerId.PYTHON) as agent:
|
||||
tool = agent.get_tool(SerenaReplTool)
|
||||
session_id = agent.create_session().session_id
|
||||
|
||||
# a returned collection is rendered, identifying the symbol and its file
|
||||
rendered = tool.apply(session_id, 's.lsp.find_symbol("create_user")')
|
||||
assert "create_user" in rendered
|
||||
assert "services.py" in rendered
|
||||
|
||||
# the underlying symbols are accessible from code, e.g. to retrieve a body without rendering the collection
|
||||
body = tool.apply(
|
||||
session_id,
|
||||
f'result = s.lsp.find_symbol("create_user", relative_path={self._SERVICES_FILE!r})\nresult.symbols[0].body',
|
||||
)
|
||||
assert body.startswith("def create_user")
|
||||
@@ -14,12 +14,12 @@ from _pytest.mark import Mark, MarkDecorator, ParameterSet
|
||||
|
||||
from serena.agent import SerenaAgent
|
||||
from serena.config.context_mode import SerenaAgentContext
|
||||
from serena.config.serena_config import ProjectConfig, RegisteredProject, SerenaConfig
|
||||
from serena.config.serena_config import AgentInterface, ProjectConfig, RegisteredProject, SerenaConfig
|
||||
from serena.lsp.lsp_diagnostics import DiagnosticsContext
|
||||
from serena.project import Project
|
||||
from serena.tools import (
|
||||
SUCCESS_RESULT,
|
||||
ActivateProjectTool,
|
||||
EditingToolWithDiagnostics,
|
||||
FindDeclarationTool,
|
||||
FindImplementationsTool,
|
||||
FindReferencingSymbolsTool,
|
||||
@@ -30,6 +30,7 @@ from serena.tools import (
|
||||
ReplaceInFilesTool,
|
||||
ReplaceSymbolBodyTool,
|
||||
SafeDeleteSymbol,
|
||||
SerenaReplTool,
|
||||
Tool,
|
||||
)
|
||||
from solidlsp.ls_config import LanguageServerId
|
||||
@@ -824,9 +825,9 @@ def read_project_file(project: Project, relative_path: str) -> str:
|
||||
|
||||
def parse_edit_diagnostics_result(result: str) -> dict:
|
||||
"""Utility function to parse the diagnostic payload returned by edit tools."""
|
||||
assert EditingToolWithDiagnostics.DIAGNOSTICS_KEY in result
|
||||
assert DiagnosticsContext.DIAGNOSTICS_KEY in result
|
||||
d = json.loads(result)
|
||||
return d[EditingToolWithDiagnostics.DIAGNOSTICS_KEY]
|
||||
return d[DiagnosticsContext.DIAGNOSTICS_KEY]
|
||||
|
||||
|
||||
@contextmanager
|
||||
@@ -901,6 +902,43 @@ class TestSerenaAgent:
|
||||
finally:
|
||||
agent.on_shutdown(timeout=5)
|
||||
|
||||
@pytest.mark.python
|
||||
@pytest.mark.skipif(not language_server_tests_enabled(LanguageServerId.PYTHON), reason="python tests are disabled in this environment")
|
||||
@pytest.mark.parametrize("context_name", ["desktop-app", "grok"], ids=["multi_project", "single_project"])
|
||||
def test_repl_interface_exposes_fixed_toolset(self, serena_config, context_name: str):
|
||||
# the toolset is fixed regardless of tool inclusions/exclusions (e.g. the context's or the configuration's);
|
||||
# only the single-project property of the context matters (no project activation in that case)
|
||||
serena_config.agent_interface = AgentInterface.REPL
|
||||
serena_config.included_optional_tools = ["get_diagnostics_for_symbol"]
|
||||
context = SerenaAgentContext.from_name(context_name)
|
||||
agent = SerenaAgent(project="test_repo_python", serena_config=serena_config, context=context)
|
||||
agent.execute_task(lambda: None)
|
||||
try:
|
||||
exposed = {tool.get_name() for tool in agent.get_exposed_tool_instances()}
|
||||
expected = {"serena_repl", "initial_instructions"} | (set() if context.single_project else {"activate_project"})
|
||||
assert exposed == expected
|
||||
assert "s.lsp" in agent.get_tool(SerenaReplTool).apply(agent.create_session().session_id, "s.info()")
|
||||
|
||||
# the facade listing is part of the (fixed) tool description in single-project sessions,
|
||||
# and of the activation message otherwise (where the facades depend on the activated project)
|
||||
tool_description = agent.get_tool(SerenaReplTool).get_apply_docstring()
|
||||
activation_message = agent.get_project_activation_message("test_session")
|
||||
assert ("s.lsp:" in tool_description) == context.single_project
|
||||
assert ("s.lsp:" in activation_message) == (not context.single_project)
|
||||
|
||||
# prompts refer to operations by their qualified REPL names, e.g. `lsp.find_symbol` instead of the tool name
|
||||
system_prompt = agent.create_system_prompt()
|
||||
assert "`lsp.find_symbol`" in system_prompt
|
||||
assert "`find_symbol`" not in system_prompt
|
||||
|
||||
# the instructions establish a session, whose id can be used with session-aware tools
|
||||
session_id_match = re.search(r"session id is `(\w+)`", system_prompt)
|
||||
assert session_id_match is not None
|
||||
session_id = session_id_match.group(1)
|
||||
assert "s.lsp" in agent.get_tool(SerenaReplTool).apply(session_id, "s.info()")
|
||||
finally:
|
||||
agent.on_shutdown(timeout=5)
|
||||
|
||||
def _symbol_matches_expected_name(self, symbol: dict, expected_name: str) -> bool:
|
||||
return (
|
||||
symbol.get("name") == expected_name
|
||||
|
||||
@@ -0,0 +1,45 @@
|
||||
"""
|
||||
Tests for the shell facade API.
|
||||
"""
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
from serena.config.serena_config import SerenaConfig
|
||||
from serena.project import Project
|
||||
from serena.repl.api.shell_api import ShellApi
|
||||
from serena.repl.facade import ApiScope, Facade
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def api(tmp_path: Path) -> ShellApi:
|
||||
(tmp_path / "sub").mkdir()
|
||||
project = Project.load(str(tmp_path), serena_config=SerenaConfig(gui_log_window=False, web_dashboard=False))
|
||||
agent = MagicMock()
|
||||
agent.get_active_project_or_raise.return_value = project
|
||||
agent.serena_config.default_max_tool_answer_chars = 10000
|
||||
return ShellApi(agent)
|
||||
|
||||
|
||||
def test_facade_exposes_shell_command_as_editing_operation(api: ShellApi) -> None:
|
||||
facade = Facade.from_api(api, ApiScope())
|
||||
assert facade.name == "shell"
|
||||
assert facade.enabled_method_names == ["execute_shell_command"]
|
||||
assert facade.get_method("execute_shell_command").info.can_edit
|
||||
|
||||
|
||||
def test_execute_shell_command(api: ShellApi, tmp_path: Path) -> None:
|
||||
print_cwd = "cd" if sys.platform == "win32" else "pwd"
|
||||
|
||||
output = api.execute_shell_command(f"{print_cwd}")
|
||||
assert output.return_code == 0
|
||||
assert Path(output.stdout.strip()).resolve() == tmp_path.resolve()
|
||||
assert '"stdout"' in output.represent() and '"return_code"' in output.represent()
|
||||
|
||||
# a relative working directory is resolved against the project root and must exist
|
||||
assert Path(api.execute_shell_command(print_cwd, cwd="sub").stdout.strip()).resolve() == (tmp_path / "sub").resolve()
|
||||
with pytest.raises(FileNotFoundError):
|
||||
api.execute_shell_command(print_cwd, cwd="missing")
|
||||
@@ -9,7 +9,7 @@ import pytest
|
||||
|
||||
from solidlsp import SolidLanguageServer
|
||||
from solidlsp.ls_config import LanguageServerId
|
||||
from test.conftest import language_server_tests_enabled
|
||||
from test.conftest import is_ci, language_server_tests_enabled
|
||||
from test.solidlsp.conftest import format_symbol_for_assert, has_malformed_name, request_all_symbols
|
||||
|
||||
|
||||
@@ -41,6 +41,7 @@ class TestRLanguageServer:
|
||||
expected_functions = {"calculate_mean", "process_data", "create_data_frame"}
|
||||
assert expected_functions.issubset(function_names), f"Expected functions {expected_functions} but found {function_names}"
|
||||
|
||||
@pytest.mark.xfail(is_ci, reason="Test is flaky") # See #1040
|
||||
@pytest.mark.parametrize("language_server", [LanguageServerId.R], indirect=True)
|
||||
def test_find_definition_across_files(self, language_server: SolidLanguageServer):
|
||||
"""Test finding function definitions across files."""
|
||||
@@ -58,6 +59,7 @@ class TestRLanguageServer:
|
||||
# Definition should be around line 37 (0-indexed: 36) where create_data_frame is defined
|
||||
assert definition_location["range"]["start"]["line"] >= 35
|
||||
|
||||
@pytest.mark.xfail(is_ci, reason="Test is flaky") # See #1040
|
||||
@pytest.mark.parametrize("language_server", [LanguageServerId.R], indirect=True)
|
||||
def test_find_references_across_files(self, language_server: SolidLanguageServer):
|
||||
"""Test finding function references across files."""
|
||||
|
||||
Reference in new issue
Block a user