Merge remote-tracking branch 'origin/main' into perf/search-text-line-lookup-pr

Conflicts:
  src/serena/util/text_utils.py
This commit is contained in:
Dominik Jain authored and Dominik Jain committed 2026-09-23 14:07:36 +02:00
commit fba55b6844
133 files changed
+8889 -2040

No files matched your search

+143 -18
View File
@@ -4,19 +4,21 @@ 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,
LanguageBackend,
AgentInterface,
ProjectConfig,
RegisteredProject,
SerenaConfig,
SerenaConfigError,
)
from serena.constants import PROJECT_TEMPLATE_FILE, SERENA_MANAGED_DIR_NAME
from serena.language_backend import BuiltinLanguageBackend
from serena.project import MemoryManager, Project
from solidlsp.ls_config import LanguageServerId
from test.conftest import create_default_serena_config
@@ -176,15 +178,16 @@ class TestProjectConfigLanguageBackend:
config = ProjectConfig(
project_name="test",
language_servers=[LanguageServerId.PYTHON],
language_backend=LanguageBackend.JETBRAINS,
language_backend=BuiltinLanguageBackend.JETBRAINS.get_instance(),
)
assert config.language_backend == LanguageBackend.JETBRAINS
assert config.language_backend is not None
assert config.language_backend.is_jetbrains()
def test_language_backend_roundtrips_through_yaml(self):
config = ProjectConfig(
project_name="test",
language_servers=[LanguageServerId.PYTHON],
language_backend=LanguageBackend.JETBRAINS,
language_backend=BuiltinLanguageBackend.JETBRAINS.get_instance(),
)
d = config._to_yaml_dict()
assert d["language_backend"] == "JetBrains"
@@ -205,7 +208,8 @@ class TestProjectConfigLanguageBackend:
data["languages"] = ["python"]
data["language_backend"] = "JetBrains"
config = ProjectConfig._from_dict(data, local_override_keys=[])
assert config.language_backend == LanguageBackend.JETBRAINS
assert config.language_backend is not None
assert config.language_backend.is_jetbrains()
def test_language_backend_none_when_missing_from_dict(self):
"""Test that _from_dict handles missing language_backend gracefully."""
@@ -218,22 +222,84 @@ 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,
global_backend: LanguageBackend = LanguageBackend.LSP,
language_backend: BuiltinLanguageBackend | None = None,
global_backend: BuiltinLanguageBackend = BuiltinLanguageBackend.LSP,
) -> tuple[SerenaConfig, str]:
"""Create a SerenaConfig with a single registered project and return (config, project_name)."""
config = SerenaConfig(
log_level=logging.ERROR,
language_backend=global_backend,
language_backend=global_backend.get_instance(),
).with_headless_mode_overrides()
project = Project(
project_root=str(Path(__file__).parent.parent / "resources" / "repos" / "python" / "test_repo"),
project_config=ProjectConfig(
project_name=project_name,
language_servers=[LanguageServerId.PYTHON],
language_backend=language_backend,
language_backend=language_backend.get_instance() if language_backend is not None else None,
),
serena_config=config,
)
@@ -246,7 +312,7 @@ class TestEffectiveLanguageBackend:
def test_default_backend_is_global(self):
"""When no project override, effective backend matches global config."""
config, name = _make_config_with_project("test_proj", language_backend=None, global_backend=LanguageBackend.LSP)
config, name = _make_config_with_project("test_proj", language_backend=None, global_backend=BuiltinLanguageBackend.LSP)
agent = SerenaAgent(project=name, serena_config=config)
try:
assert agent.get_language_backend().is_lsp()
@@ -256,7 +322,7 @@ class TestEffectiveLanguageBackend:
def test_project_overrides_global_backend(self):
"""When startup project has language_backend set, it overrides the global."""
config, name = _make_config_with_project(
"test_jetbrains", language_backend=LanguageBackend.JETBRAINS, global_backend=LanguageBackend.LSP
"test_jetbrains", language_backend=BuiltinLanguageBackend.JETBRAINS, global_backend=BuiltinLanguageBackend.LSP
)
agent = SerenaAgent(project=name, serena_config=config)
try:
@@ -268,18 +334,18 @@ class TestEffectiveLanguageBackend:
"""When no startup project is provided, effective backend is the global one."""
config = SerenaConfig(
log_level=logging.ERROR,
language_backend=LanguageBackend.LSP,
language_backend=BuiltinLanguageBackend.LSP.get_instance(),
).with_headless_mode_overrides()
agent = SerenaAgent(project=None, serena_config=config)
try:
assert agent.get_language_backend() == LanguageBackend.LSP
assert agent.get_language_backend().is_lsp()
finally:
agent.on_shutdown(timeout=5)
def test_activate_project_rejects_backend_mismatch(self):
"""Post-init activation of a project with mismatched backend raises ValueError."""
# Start with LSP backend
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=LanguageBackend.LSP)
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=BuiltinLanguageBackend.LSP)
# Add a second project that requires JetBrains
jb_project = Project(
@@ -287,7 +353,7 @@ class TestEffectiveLanguageBackend:
project_config=ProjectConfig(
project_name="jb_proj",
language_servers=[LanguageServerId.JAVA],
language_backend=LanguageBackend.JETBRAINS,
language_backend=BuiltinLanguageBackend.JETBRAINS.get_instance(),
),
serena_config=config,
)
@@ -300,9 +366,38 @@ 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=BuiltinLanguageBackend.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=BuiltinLanguageBackend.JETBRAINS.get_instance(),
),
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().is_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().is_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)
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=BuiltinLanguageBackend.LSP)
# Add a second project that also uses LSP
lsp_project2 = Project(
@@ -310,7 +405,7 @@ class TestEffectiveLanguageBackend:
project_config=ProjectConfig(
project_name="lsp_proj2",
language_servers=[LanguageServerId.PYTHON],
language_backend=LanguageBackend.LSP,
language_backend=BuiltinLanguageBackend.LSP.get_instance(),
),
serena_config=config,
)
@@ -325,7 +420,7 @@ class TestEffectiveLanguageBackend:
def test_activate_project_allows_none_backend(self):
"""Post-init activation of a project with no backend override succeeds."""
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=LanguageBackend.LSP)
config, name = _make_config_with_project("lsp_proj", language_backend=None, global_backend=BuiltinLanguageBackend.LSP)
# Add a second project with no backend override
proj2 = Project(
@@ -543,6 +638,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(
+29
View File
@@ -0,0 +1,29 @@
"""The anthropic package must only be loaded when the Anthropic token counter is used (#2012)."""
import subprocess
import sys
from unittest.mock import MagicMock
import pytest
from serena.analytics import AnthropicTokenCount
@pytest.mark.parametrize("module", ["serena.cli", "serena.agent", "serena.analytics"])
def test_importing_serena_does_not_load_anthropic(module: str) -> None:
code = f"import sys, {module}; print('anthropic' in sys.modules)"
result = subprocess.run([sys.executable, "-c", code], capture_output=True, text=True, check=True, timeout=120)
assert result.stdout.strip() == "False", f"importing {module} loaded the anthropic package"
def test_anthropic_token_count_sends_a_plain_user_message() -> None:
estimator = AnthropicTokenCount.__new__(AnthropicTokenCount)
estimator._model_name = "claude-sonnet-4-20250514"
estimator._anthropic_client = MagicMock()
estimator._anthropic_client.messages.count_tokens.return_value = MagicMock(input_tokens=7)
assert estimator.estimate_token_count("hello") == 7
estimator._anthropic_client.messages.count_tokens.assert_called_once_with(
model="claude-sonnet-4-20250514",
messages=[{"role": "user", "content": "hello"}],
)
+25
View File
@@ -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()
+123
View File
@@ -0,0 +1,123 @@
"""Tests for the CLI's ``project remove`` command."""
import shutil
import tempfile
from pathlib import Path
import pytest
from click.testing import CliRunner
from serena.cli import ProjectCommands
from serena.config.serena_config import SerenaConfig
from serena.constants import SERENA_MANAGED_DIR_NAME
class TestProjectRemove:
"""
Drives ``serena project remove`` against a temporary Serena configuration file, so the
user's real project registry is never touched.
"""
@pytest.fixture(autouse=True)
def setup(self, monkeypatch):
self.test_dir = Path(tempfile.mkdtemp())
self.master_config_path = self.test_dir / "serena_config.yml"
monkeypatch.setattr(
SerenaConfig,
"_determine_config_file_path",
classmethod(lambda cls: str(self.master_config_path)),
)
self.runner = CliRunner()
def teardown_method(self):
shutil.rmtree(self.test_dir, ignore_errors=True)
def _make_project_dir(self, dir_name: str, project_name: str | None = None) -> Path:
project_dir = self.test_dir / dir_name
(project_dir / SERENA_MANAGED_DIR_NAME).mkdir(parents=True)
(project_dir / SERENA_MANAGED_DIR_NAME / "project.yml").write_text(
f'project_name: "{project_name or dir_name}"\nlanguages: ["python"]\n'
)
return project_dir
def _write_master_config(self, project_paths: list[Path]) -> None:
self.master_config_path.write_text("projects:\n" + "".join(f" - {p}\n" for p in project_paths))
def _registered_roots(self) -> set[Path]:
config = SerenaConfig.from_config_file(generate_if_missing=False)
return {Path(project.project_root) for project in config.projects}
def test_remove_by_name_unregisters_only_that_project(self):
keep = self._make_project_dir("keep")
drop = self._make_project_dir("drop")
self._write_master_config([keep, drop])
result = self.runner.invoke(ProjectCommands.remove, ["drop"])
assert result.exit_code == 0, f"Command failed: {result.output}"
assert self._registered_roots() == {keep.resolve()}
def test_remove_by_path_unregisters_only_that_project(self):
keep = self._make_project_dir("keep")
drop = self._make_project_dir("drop")
self._write_master_config([keep, drop])
result = self.runner.invoke(ProjectCommands.remove, [str(drop)])
assert result.exit_code == 0, f"Command failed: {result.output}"
assert self._registered_roots() == {keep.resolve()}
def test_remove_names_the_project_it_removed(self):
drop = self._make_project_dir("drop")
self._write_master_config([drop])
result = self.runner.invoke(ProjectCommands.remove, [str(drop)])
assert result.exit_code == 0, f"Command failed: {result.output}"
assert "drop" in result.output
assert str(drop.resolve()) in result.output
def test_remove_keeps_the_project_configuration_file_on_disk(self):
"""Unregistering must not delete the project's own files; re-registering it must remain possible."""
drop = self._make_project_dir("drop")
self._write_master_config([drop])
result = self.runner.invoke(ProjectCommands.remove, [str(drop)])
assert result.exit_code == 0, f"Command failed: {result.output}"
assert (drop / SERENA_MANAGED_DIR_NAME / "project.yml").is_file()
def test_remove_unknown_project_fails_without_changing_the_registry(self):
keep = self._make_project_dir("keep")
self._write_master_config([keep])
result = self.runner.invoke(ProjectCommands.remove, ["no_such_project"])
assert result.exit_code != 0
assert "no_such_project" in result.output
assert self._registered_roots() == {keep.resolve()}
def test_remove_by_path_picks_the_entry_at_that_path_when_names_collide(self):
"""Two directories may carry the same ``project_name``; a path must remove the entry at that path."""
first = self._make_project_dir("first_dir", project_name="twin")
second = self._make_project_dir("second_dir", project_name="twin")
self._write_master_config([first, second])
result = self.runner.invoke(ProjectCommands.remove, [str(second)])
assert result.exit_code == 0, f"Command failed: {result.output}"
assert self._registered_roots() == {first.resolve()}
def test_remove_by_ambiguous_name_fails_with_a_message_rather_than_a_traceback(self):
first = self._make_project_dir("first_dir", project_name="twin")
second = self._make_project_dir("second_dir", project_name="twin")
self._write_master_config([first, second])
result = self.runner.invoke(ProjectCommands.remove, ["twin"])
assert result.exit_code != 0
assert result.exception is None or isinstance(result.exception, SystemExit), (
f"Expected a handled CLI error, got: {result.exception!r}"
)
assert "twin" in result.output
assert self._registered_roots() == {first.resolve(), second.resolve()}
@@ -0,0 +1,199 @@
"""Tests that saving an edited source file cannot destroy the previous content (issue #1958).
``CodeEditor._save_edited_file`` is the third of the three direct ``open(path, "w")`` writes the
issue lists; the two in ``MemoryManager`` were addressed in #1969, which is also where
``write_file_atomic`` comes from.
The editor is exercised through a stub subclass so that the inherited save path can be driven
without a language server.
"""
import os
import stat
import sys
from collections.abc import Iterator
from contextlib import contextmanager
from typing import Any
import pytest
from serena.code_editor import CodeEditor
from serena.language_backend import BuiltinLanguageBackend
from serena.util import file_system
class _InMemoryEditedFile(CodeEditor.EditedFile):
"""An ``EditedFile`` that holds its contents in memory."""
def __init__(self, relative_path: str, contents: str) -> None:
super().__init__(relative_path)
self._contents = contents
def get_contents(self) -> str:
return self._contents
def set_contents(self, contents: str) -> None:
self._contents = contents
def delete_text_between_positions(self, start_pos: Any, end_pos: Any) -> None:
raise NotImplementedError
def insert_text_at_position(self, pos: Any, text: str) -> None:
raise NotImplementedError
class _StubCodeEditor(CodeEditor[Any]):
"""A ``CodeEditor`` whose only inherited behaviour under test is the file-saving path."""
class DummyProject:
"""A dummy project object with only the attributes needed to construct a ``CodeEditor``."""
def __init__(self) -> None:
self.language_backend = BuiltinLanguageBackend.LSP.get_instance()
def __init__(self, project_root: str, encoding: str = "utf-8", newline: str | None = None) -> None:
self.project = self.DummyProject()
self.project_root = project_root
self.encoding = encoding
self.newline = newline
@contextmanager
def _open_file_context(self, relative_path: str) -> Iterator[CodeEditor.EditedFile]:
abs_path = os.path.join(self.project_root, relative_path)
with open(abs_path, encoding=self.encoding) as f:
contents = f.read()
yield _InMemoryEditedFile(relative_path, contents)
def _find_unique_symbol(self, name_path: str, relative_file_path: str) -> Any:
raise NotImplementedError
def rename_symbol(self, name_path: str, relative_path: str, new_name: str) -> str:
raise NotImplementedError
class TestSourceFileSaveIsAtomic:
def _editor(self, tmp_path: Any, **kwargs: Any) -> _StubCodeEditor:
return _StubCodeEditor(str(tmp_path), **kwargs)
def test_interrupted_save_keeps_the_previous_file_content(self, tmp_path, monkeypatch):
"""A crash partway through the write must leave the file holding its old content.
The crash is injected into the temp-file write that ``write_file_atomic`` performs, so on
a non-atomic implementation nothing raises at all and this fails with ``DID NOT RAISE``,
which is exactly the state the issue describes: the real file is truncated first and there
is no intermediate copy to fall back to.
"""
source = tmp_path / "module.py"
original = "def original():\n return 1\n" * 40
source.write_text(original, encoding="utf-8")
real_fdopen = os.fdopen
def crashing_fdopen(fd: int, *args: Any, **kwargs: Any) -> Any:
f = real_fdopen(fd, *args, **kwargs)
real_write = f.write
def crashing_write(data: str) -> int:
real_write(data[: len(data) // 4])
f.flush()
raise RuntimeError("simulated crash mid-write")
f.write = crashing_write
return f
monkeypatch.setattr(file_system.os, "fdopen", crashing_fdopen)
editor = self._editor(tmp_path)
with pytest.raises(RuntimeError, match="simulated crash mid-write"):
with editor.edited_file_context("module.py") as edited:
edited.set_contents("def replacement():\n return 2\n" * 40)
assert source.read_text(encoding="utf-8") == original
assert list(tmp_path.iterdir()) == [source], "the partial temp file must not be left behind"
def test_successful_save_writes_the_new_content(self, tmp_path):
"""Control: the ordinary path still writes what was asked for."""
source = tmp_path / "module.py"
source.write_text("old\n", encoding="utf-8")
editor = self._editor(tmp_path)
with editor.edited_file_context("module.py") as edited:
edited.set_contents("new\n")
assert source.read_text(encoding="utf-8") == "new\n"
assert list(tmp_path.iterdir()) == [source]
def test_save_into_a_subdirectory(self, tmp_path):
"""Every other test writes at the project root; the relative path is joined and resolved,
so a nested file has to work the same way.
"""
package = tmp_path / "pkg" / "sub"
package.mkdir(parents=True)
source = package / "module.py"
source.write_text("old\n", encoding="utf-8")
editor = self._editor(tmp_path)
with editor.edited_file_context("pkg/sub/module.py") as edited:
edited.set_contents("new\n")
assert source.read_text(encoding="utf-8") == "new\n"
assert list(package.iterdir()) == [source], "no temp file may be left beside the source"
def test_save_writes_through_a_symlinked_source_file(self, tmp_path):
"""A symlinked source file must keep being written through to its target, as
``open(path, "w")`` did; the link itself must not be replaced by a regular file.
"""
target_dir = tmp_path / "shared"
target_dir.mkdir()
target = target_dir / "shared.py"
target.write_text("old\n", encoding="utf-8")
project = tmp_path / "project"
project.mkdir()
link = project / "module.py"
try:
link.symlink_to(target)
except OSError as e:
pytest.skip(f"cannot create symlinks on this platform/permissions: {e}")
editor = self._editor(project)
with editor.edited_file_context("module.py") as edited:
edited.set_contents("new\n")
assert link.is_symlink(), "the source file's symlink must survive the edit"
assert target.read_text(encoding="utf-8") == "new\n", "the edit must reach the link's target"
@pytest.mark.skipif(
sys.platform == "win32", reason="Windows does not model POSIX permission bits; chmod only toggles the read-only flag"
)
def test_save_preserves_the_executable_bit(self, tmp_path):
script = tmp_path / "run.sh"
script.write_text("#!/bin/sh\necho old\n", encoding="utf-8")
os.chmod(script, 0o755)
editor = self._editor(tmp_path)
with editor.edited_file_context("run.sh") as edited:
edited.set_contents("#!/bin/sh\necho new\n")
assert stat.S_IMODE(os.stat(script).st_mode) == 0o755
def test_save_respects_the_configured_newline(self, tmp_path):
source = tmp_path / "module.py"
source.write_bytes(b"old\n")
editor = self._editor(tmp_path, newline="\r\n")
with editor.edited_file_context("module.py") as edited:
edited.set_contents("a\nb\n")
assert source.read_bytes() == b"a\r\nb\r\n"
def test_save_respects_the_configured_encoding(self, tmp_path):
source = tmp_path / "module.py"
source.write_text("alt\n", encoding="latin-1")
# newline is pinned so this test is about the encoding alone: LineEnding.NATIVE yields
# newline=None, under which Python translates "\n" to os.linesep on write
editor = self._editor(tmp_path, encoding="latin-1", newline="\n")
with editor.edited_file_context("module.py") as edited:
edited.set_contents("café\n")
assert source.read_bytes() == "café\n".encode("latin-1")
+68
View File
@@ -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
+107
View File
@@ -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)
+2 -4
View File
@@ -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:
+81
View File
@@ -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"} # type: ignore
# 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) # type: ignore
assert "foo(1)" in code_matches.represent()
+79
View File
@@ -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()
+5 -2
View File
@@ -62,7 +62,8 @@ class FileSystemSyncTestCase:
symbol_names = [ref["name_path"].split("/")[-1] for ref in ref_symbols]
return symbol_names
else:
ls = next(iter(agent.get_active_project_or_raise().language_server_manager.iter_language_servers()))
ls_manager = agent.get_active_project_or_raise().get_language_server_manager_or_raise()
ls = next(iter(ls_manager.iter_language_servers()))
document_symbols = ls.request_document_symbols(self._TARGET_FILE).get_all_symbols_and_roots()
target = next((s for s in document_symbols[0] if s.get("name") == self._TARGET_SYMBOL), None)
assert target is not None and "selectionRange" in target, f"{self._TARGET_SYMBOL} not found in {self._TARGET_FILE}"
@@ -162,7 +163,9 @@ class SymbolPositionStaleAfterExternalEditTestCase:
with agent_for_project_context(LanguageServerId.PYTHON, str(repo_root)) as agent:
project = agent.get_active_project_or_raise()
ls = next(iter(project.language_server_manager.iter_language_servers()))
ls_manager = project.language_server_manager
assert ls_manager is not None
ls = next(iter(ls_manager.iter_language_servers()))
tool = agent.get_tool(FindSymbolTool)
# Hold the file's buffer open across the external edit, mirroring the state left
+13 -21
View File
@@ -1,13 +1,14 @@
"""Tests for the mcp.py module in serena."""
import pytest
from mcp.server.fastmcp.tools.base import Tool as MCPTool
from mcp.server.mcpserver import Context
from mcp.server.mcpserver.tools.base import Tool as MCPTool
from serena import __version__
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 +23,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,30 +55,13 @@ class BasicTool(BaseMockTool):
self,
log_call: bool = True,
catch_exceptions: bool = True,
mcp_ctx: Context | None = None,
**kwargs,
) -> str:
"""Mock implementation of apply_ex."""
return self.apply(**kwargs)
def test_create_mcp_server_reports_serena_version(monkeypatch: pytest.MonkeyPatch) -> None:
"""MCP initialize must report Serena's version, not the installed mcp SDK version."""
class MinimalAgent:
def create_connection_prompt(self) -> str:
return ""
monkeypatch.setattr(SerenaConfig, "from_config_file", classmethod(lambda cls: SerenaConfig()))
factory = SerenaMCPFactory(transport="stdio")
monkeypatch.setattr(factory, "_create_serena_agent", lambda *args, **kwargs: MinimalAgent())
mcp_server = factory.create_mcp_server()
initialization_options = mcp_server._mcp_server.create_initialization_options()
assert initialization_options.server_name == "Serena"
assert initialization_options.server_version == __version__
def test_make_tool_basic() -> None:
"""Test that make_tool correctly creates an MCP tool from a Tool object."""
mock_tool = BasicTool()
+68
View File
@@ -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)
+38
View File
@@ -844,3 +844,41 @@ class TestAutoPrefixBareReferences:
# idempotent: the second run should not touch anything
assert second.total_replacements == 0
assert fs_manager.load_memory("docs") == "the mem:auth/login process"
class TestRenameMemorySparesReadOnlyMemories:
"""Regression: a tool-context rename enumerated read-only memories, so propagating the
reference into one raised ``PermissionError`` after the rename itself had already been applied.
"""
@staticmethod
def _manager(tmp_path, monkeypatch) -> MemoryManager:
manager = MemoryManager(serena_data_folder=tmp_path, read_only_memory_patterns=[r"frozen/.*"])
# the global memories of the machine would otherwise join the enumeration as well
global_dir = tmp_path / "global"
global_dir.mkdir()
monkeypatch.setattr(manager, "_global_memory_dir", global_dir)
_write(manager, "auth/login", "# login notes")
_write(manager, "frozen/notes", "see `mem:auth/login`")
_write(manager, "docs", "first `mem:auth/login`, then `mem:auth/login`")
return manager
def test_tool_context_rename_completes_and_leaves_read_only_reference_alone(self, tmp_path, monkeypatch) -> None:
manager = self._manager(tmp_path, monkeypatch)
message, n_updated = manager.rename_memory_and_propagate_references("auth/login", "auth/signin", is_tool_context=True)
assert "auth/signin" in message
assert manager.load_memory("auth/signin") == "# login notes"
assert manager.load_memory("docs") == "first `mem:auth/signin`, then `mem:auth/signin`"
assert manager.load_memory("frozen/notes") == "see `mem:auth/login`"
assert n_updated == 2
def test_cli_context_rename_still_propagates_into_read_only_memories(self, tmp_path, monkeypatch) -> None:
manager = self._manager(tmp_path, monkeypatch)
_, n_updated = manager.rename_memory_and_propagate_references("auth/login", "auth/signin", is_tool_context=False)
assert manager.load_memory("frozen/notes") == "see `mem:auth/signin`"
assert manager.load_memory("docs") == "first `mem:auth/signin`, then `mem:auth/signin`"
assert n_updated == 3
+51 -1
View File
@@ -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")
+382
View File
@@ -0,0 +1,382 @@
"""
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
from serena.language_backend import BuiltinLanguageBackend
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("builtin_backend", [BuiltinLanguageBackend.LSP, BuiltinLanguageBackend.JETBRAINS])
@pytest.mark.parametrize("read_only", [True, False])
def test_external_project_dispatch(self, builtin_backend: BuiltinLanguageBackend, read_only: bool) -> None:
agent = MagicMock()
backend = builtin_backend.get_instance()
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")
+54 -10
View File
@@ -14,12 +14,13 @@ 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.session import SessionRegistry
from serena.tools import (
SUCCESS_RESULT,
ActivateProjectTool,
EditingToolWithDiagnostics,
FindDeclarationTool,
FindImplementationsTool,
FindReferencingSymbolsTool,
@@ -30,6 +31,7 @@ from serena.tools import (
ReplaceInFilesTool,
ReplaceSymbolBodyTool,
SafeDeleteSymbol,
SerenaReplTool,
Tool,
)
from solidlsp.ls_config import LanguageServerId
@@ -824,15 +826,15 @@ 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
def project_file_modification_context(serena_agent: SerenaAgent, relative_path: str) -> Iterator[None]:
"""Context manager to modify a project file and revert the changes after use."""
project = serena_agent.get_active_project()
project = serena_agent.get_active_project_or_raise()
file_path = os.path.join(project.project_root, relative_path)
# Read the original content
@@ -901,6 +903,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
@@ -1362,13 +1401,17 @@ class TestSerenaAgent:
class TestPromptProvision:
class MockContext:
def __init__(self, session_id: str):
self.session = session_id
@classmethod
def _call_tool(cls, agent: SerenaAgent, tool_class: type[Tool], session_id: str = "global", **kwargs) -> str:
result = agent.get_tool(tool_class).apply_ex(mcp_ctx=cls.MockContext(session_id), catch_exceptions=False, **kwargs)
old_method = SessionRegistry._next_session_id
if tool_class == InitialInstructionsTool:
SessionRegistry._next_session_id = lambda x: session_id # type: ignore
else:
kwargs["session_id"] = session_id
try:
result = agent.get_tool(tool_class).apply_ex(catch_exceptions=False, **kwargs)
finally:
SessionRegistry._next_session_id = old_method
return result
@staticmethod
@@ -1417,6 +1460,7 @@ class TestPromptProvision:
# now activate another project which dynamically enables a new mode (no-onboarding)
reg_project = serena_agent.serena_config.get_registered_project(project_name2)
assert reg_project is not None
reg_project.project_config.default_modes = ["no-onboarding"]
expected_new_mode_message = "The onboarding process is not applied."
result2 = self._call_tool(serena_agent, ActivateProjectTool, project=project_name2, session_id=session1)
+45
View File
@@ -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")
+3 -3
View File
@@ -251,10 +251,10 @@ class TestSymbolDictTypes:
:param key_type: the corresponding key type (Literal[...]) that the dict should have for keys
"""
dict_type_keys = dict_type.__annotations__.keys()
assert len(dict_type_keys) == len(key_type.__args__), (
f"Expected {len(key_type.__args__)} keys in {dict_type}, but got {len(dict_type_keys)}"
assert len(dict_type_keys) == len(key_type.__args__), ( # type: ignore
f"Expected {len(key_type.__args__)} keys in {dict_type}, but got {len(dict_type_keys)}" # type: ignore
)
for expected_key in key_type.__args__:
for expected_key in key_type.__args__: # type: ignore
assert expected_key in dict_type_keys, f"Expected key '{expected_key}' not found in {dict_type}"
def test_ls_symbol_dict_type(self):
+56 -1
View File
@@ -3,7 +3,14 @@ from collections.abc import Callable
import pytest
from serena.util.file_proxy import FileCollection, FileProxy
from serena.util.text_utils import GlobMatcher, LineType, MultiFileContentReplacer, search_files, search_text
from serena.util.text_utils import (
ContentReplacer,
GlobMatcher,
LineType,
MultiFileContentReplacer,
search_files,
search_text,
)
class TestSearchText:
@@ -691,3 +698,51 @@ class TestMultiFileContentReplacer:
occ = replacer.find_occurrences([(path, content)], "old_pkg", "new_pkg")[0]
with pytest.raises(AssertionError):
replacer.apply_to_content("completely different content", [occ])
class TestBackreferenceExpansion:
"""$!N backreferences in regex-mode replacements refer to matched groups. A group that
exists but did not participate in the match (e.g. inside an optional construct that was
skipped) must expand to the empty string; a reference to a group that the search
expression does not define at all must fail with an error naming the problem instead of
a raw IndexError. Literal mode has no backreference expansion at all: the replacement is
used verbatim (observed in practice when an agent tried to document the $!N convention
itself and the literal-mode replacement crashed instead of writing the text).
"""
def test_unmatched_group_expands_to_empty_string(self):
replacer = ContentReplacer(mode="regex", allow_multiple_occurrences=False)
needle = r"EA_INPUT(?:\((\w*)\))?"
# the group participated and captured an empty string (empty parentheses)
assert replacer.replace("EA_INPUT()\n", needle, r"EA_INPUT$!1(...)") == "EA_INPUT(...)\n"
# the group did not participate at all (no parentheses)
assert replacer.replace("EA_INPUT\n", needle, r"EA_INPUT$!1(...)") == "EA_INPUT(...)\n"
def test_matched_group_expands_to_its_value(self):
replacer = ContentReplacer(mode="regex", allow_multiple_occurrences=False)
assert replacer.replace("id=alpha", r"id=(\w+)", r"[$!1]") == "[alpha]"
def test_nonexistent_group_reference_raises_clear_error(self):
replacer = ContentReplacer(mode="regex", allow_multiple_occurrences=False)
with pytest.raises(ValueError, match="does not exist"):
replacer.replace("id=alpha", r"id=(\w+)", r"[$!2]")
def test_literal_mode_repl_is_verbatim(self):
"""Literal mode has no groups at all and no backreference expansion: a replacement
containing $!N sequences is written as-is instead of failing with a backreference error.
"""
replacer = ContentReplacer(mode="literal", allow_multiple_occurrences=False)
assert replacer.replace("literal needle", "literal needle", "$!1 stuff $!2") == "$!1 stuff $!2"
def test_multi_file_replacer_expands_unmatched_group_to_empty_string(self):
replacer = MultiFileContentReplacer(mode="regex")
files = [("f.txt", "EA_INPUT\n")]
occurrences = replacer.find_occurrences(files, r"EA_INPUT(?:\((\w*)\))?", r"EA_INPUT$!1(...)")
assert [o.replacement for o in occurrences] == ["EA_INPUT(...)"]
def test_multi_file_replacer_nonexistent_group_reference_raises_clear_error(self):
replacer = MultiFileContentReplacer(mode="regex")
files = [("f.txt", "id=alpha\n")]
with pytest.raises(ValueError, match="does not exist"):
replacer.find_occurrences(files, r"id=(\w+)", r"[$!2]")
+98
View File
@@ -920,3 +920,101 @@ class TestGitignoreParserPermissionError:
finally:
# Restore permissions so teardown can clean up
os.chmod(unreadable, old_mode)
class TestWriteFileAtomicSymlinks:
"""``write_file_atomic`` replaces ``open(path, "w")`` at its call sites, so it has to agree
with it about symlinks: a plain write follows the link and updates its target, whereas a bare
``os.replace`` onto the link path would swap the link itself out for a regular file and leave
the target holding stale content (issue #1958 asks for symlink behaviour to be preserved
before source files use this).
"""
@staticmethod
def _symlink_or_skip(link: Path, target: Path) -> None:
"""Windows needs developer mode or admin rights to create a symlink; skip there rather
than fail, matching how ``test_memories_manager.py`` handles the same limitation.
"""
try:
link.symlink_to(target)
except OSError as e:
pytest.skip(f"cannot create symlinks on this platform/permissions: {e}")
def test_writes_through_a_symlink_instead_of_replacing_it(self, tmp_path):
target = tmp_path / "real.txt"
target.write_text("old", encoding="utf-8")
link = tmp_path / "link.txt"
self._symlink_or_skip(link, target)
write_file_atomic(str(link), "new", encoding="utf-8")
assert link.is_symlink(), "the symlink must survive the write, not be replaced by a regular file"
assert target.read_text(encoding="utf-8") == "new", "the content must reach the link's target"
def test_writes_through_a_symlink_pointing_outside_its_directory(self, tmp_path):
outside = tmp_path / "outside"
outside.mkdir()
target = outside / "real.txt"
target.write_text("old", encoding="utf-8")
inside = tmp_path / "inside"
inside.mkdir()
link = inside / "link.txt"
self._symlink_or_skip(link, target)
write_file_atomic(str(link), "new", encoding="utf-8")
assert link.is_symlink()
assert target.read_text(encoding="utf-8") == "new"
assert list(inside.iterdir()) == [link], "no temp file may be left beside the link"
def test_broken_symlink_creates_its_target(self, tmp_path):
"""``open(path, "w")`` on a dangling link creates the target; this must do the same."""
target = tmp_path / "missing.txt"
link = tmp_path / "link.txt"
self._symlink_or_skip(link, target)
write_file_atomic(str(link), "new", encoding="utf-8")
assert link.is_symlink()
assert target.read_text(encoding="utf-8") == "new"
def test_writes_through_a_symlinked_parent_directory(self, tmp_path):
"""The path is resolved in full, so a symlinked *directory* on the way to the file is
followed too, and the temporary file is created in the destination's real directory (it has
to be on the same filesystem as the destination for the rename to be atomic).
"""
real_dir = tmp_path / "real_dir"
real_dir.mkdir()
target = real_dir / "file.txt"
target.write_text("old", encoding="utf-8")
link_dir = tmp_path / "link_dir"
self._symlink_or_skip(link_dir, real_dir)
write_file_atomic(str(link_dir / "file.txt"), "new", encoding="utf-8")
assert link_dir.is_symlink(), "the directory symlink must survive"
assert target.read_text(encoding="utf-8") == "new"
assert list(real_dir.iterdir()) == [target], "no temp file may be left in the real directory"
def test_non_ascii_filename_round_trips(self, tmp_path):
target = tmp_path / "測試檔案.txt"
try:
target.write_text("old", encoding="utf-8")
except (OSError, UnicodeError) as e:
pytest.skip(f"cannot create non-ASCII filenames on this filesystem: {e}")
write_file_atomic(str(target), "new", encoding="utf-8")
assert target.read_text(encoding="utf-8") == "new"
assert list(tmp_path.iterdir()) == [target]
def test_regular_file_is_written_in_place(self, tmp_path):
"""Control: the symlink handling must not change the ordinary case."""
target = tmp_path / "plain.txt"
target.write_text("old", encoding="utf-8")
write_file_atomic(str(target), "new", encoding="utf-8")
assert not target.is_symlink()
assert target.read_text(encoding="utf-8") == "new"
assert list(tmp_path.iterdir()) == [target]
+8 -8
View File
@@ -264,7 +264,7 @@ class TestALHoverInjection:
char = start.get("character", 0)
hover = language_server.request_hover(file_path, line, char)
if hover and "contents" in hover:
return hover, hover["contents"].get("value", "")
return hover, hover["contents"].get("value", "") # type: ignore
return hover, None
return None, None
@@ -286,7 +286,7 @@ class TestALHoverInjection:
char = start.get("character", 0)
hover = language_server.request_hover(file_path, line, char)
if hover and "contents" in hover:
return hover, hover["contents"].get("value", "")
return hover, hover["contents"].get("value", "") # type: ignore
return hover, None
return None, None
@@ -373,7 +373,7 @@ class TestALHoverInjection:
hover = language_server.request_hover(file_path, line, char)
assert hover is not None, "Hover should return a result for field"
value = hover.get("contents", {}).get("value", "")
value = hover.get("contents", {}).get("value", "") # type: ignore
# Field hover should NOT start with ** (no injection)
assert not value.startswith("**"), f"Field hover should not have injected name. Got: {value[:200]}"
return
@@ -445,7 +445,7 @@ class TestALPathNormalization:
hover = language_server.request_hover(file_path, line, char)
assert hover is not None, "Hover should return a result"
value = hover.get("contents", {}).get("value", "")
value = hover.get("contents", {}).get("value", "") # type: ignore
assert '**Table 50000 "TEST Customer"**' in value, f"Hover should have injection. Got: {value[:200]}"
return
@@ -466,7 +466,7 @@ class TestALPathNormalization:
hover = language_server.request_hover(file_path, line, char)
assert hover is not None, "Hover should return a result"
value = hover.get("contents", {}).get("value", "")
value = hover.get("contents", {}).get("value", "") # type: ignore
assert '**Table 50000 "TEST Customer"**' in value, f"Hover should have injection. Got: {value[:200]}"
return
@@ -491,7 +491,7 @@ class TestALPathNormalization:
# Request hover with forward slash path (different format)
hover = language_server.request_hover(file_path_forward, line, char)
assert hover is not None, "Hover should return a result"
value = hover.get("contents", {}).get("value", "")
value = hover.get("contents", {}).get("value", "") # type: ignore
assert '**Table 50000 "TEST Customer"**' in value, (
f"Hover injection should work with mixed path formats. Got: {value[:200]}"
)
@@ -518,7 +518,7 @@ class TestALPathNormalization:
# Request hover with backslash path (different format)
hover = language_server.request_hover(file_path_backslash, line, char)
assert hover is not None, "Hover should return a result"
value = hover.get("contents", {}).get("value", "")
value = hover.get("contents", {}).get("value", "") # type: ignore
assert '**Table 50000 "TEST Customer"**' in value, (
f"Hover injection should work with mixed path formats. Got: {value[:200]}"
)
@@ -553,7 +553,7 @@ class TestALPathNormalization:
# Request hover with different path format
hover = language_server.request_hover(hover_path, line, char)
assert hover is not None, f"Hover should return a result for {symbol_name}"
value = hover.get("contents", {}).get("value", "")
value = hover.get("contents", {}).get("value", "") # type: ignore
assert f"**{expected_injection}**" in value, (
f"Hover for {symbol_name} should have injection with mixed paths. Got: {value[:200]}"
)
@@ -125,6 +125,7 @@ class TestAngularLanguageServerBasics:
# +1 puts the cursor inside the identifier rather than on its leading boundary.
refs = language_server.request_references(src_path, coords.line, coords.col + 1)
ref_paths = {r.get("relativePath", "") for r in refs}
ref_paths = {p for p in ref_paths if p} # filter out any empty relativePath entries
assert any(p.endswith("app.component.html") for p in ref_paths), (
f"Expected references for setName to include its template callsite in app.component.html, got: {ref_paths}"
)
@@ -21,6 +21,7 @@ import time
import pytest
from solidlsp import SolidLanguageServer
from solidlsp.language_servers.angular_language_server import AngularLanguageServer
from solidlsp.ls_config import LanguageServerId
from solidlsp.ls_exceptions import SolidLSPException
from test.conftest import _create_ls
@@ -344,6 +345,7 @@ class TestAngularStartupCleanup:
with pytest.raises(RuntimeError, match="simulated ngserver init"):
ls.start()
assert isinstance(ls, AngularLanguageServer)
assert ls._ts_server is None, "TS companion was not cleared after startup failure"
assert ls._html_server is None, "HTML companion was not cleared after startup failure"
@@ -54,7 +54,7 @@ class TestClojureProjectIndexing:
# extra.clj contains two real call sites (in double-product and triple-product);
# they must be returned regardless of whether the file was opened beforehand
extra_refs = [r for r in refs if r.get("relativePath", "").endswith("extra.clj")]
extra_refs = [r for r in refs if r.get("relativePath", "").endswith("extra.clj")] # type: ignore
assert extra_refs, (
"Expected references to 'multiply' to include call sites from extra.clj, "
f"but got files: {sorted(ref_paths)}. "
@@ -82,7 +82,9 @@ class TestClojureProjectIndexing:
ref_paths = {r.get("relativePath", "") for r in refs}
consumer_refs = [
r for r in refs if r.get("relativePath", "").replace("\\", "/").endswith("sub_module/src/sub_module_app/consumer.clj")
r
for r in refs
if r.get("relativePath", "").replace("\\", "/").endswith("sub_module/src/sub_module_app/consumer.clj") # type: ignore
]
assert consumer_refs, (
"Expected references to 'multiply' to include call sites from the sibling module "
+3 -1
View File
@@ -14,6 +14,7 @@ import os
import pytest
from solidlsp import SolidLanguageServer
from solidlsp.language_servers.crystal_language_server import CrystalLanguageServer
from solidlsp.ls_config import LanguageServerId
from test.conftest import language_server_tests_enabled
from test.solidlsp.conftest import format_symbol_for_assert, has_malformed_name, request_all_symbols
@@ -94,7 +95,8 @@ class TestCrystalDefinition:
file_path = os.path.join("src", "main.cr")
# wait for Crystalline to compile the project
language_server.language_server._wait_for_compilation()
assert isinstance(language_server, CrystalLanguageServer)
language_server._wait_for_compilation()
# Calculator.new on line 35 (0-indexed: 34), col 13 -> Calculator class on line 3 (0-indexed: 2)
definitions = language_server.request_definition(file_path, 34, 13)
+23
View File
@@ -7,6 +7,7 @@ import pytest
from solidlsp import SolidLanguageServer
from solidlsp.language_servers.csharp_language_server import (
CSharpLanguageServer,
breadth_first_file_scan,
find_solution_or_project_file,
)
@@ -200,6 +201,28 @@ class TestCSharpLanguageServer:
), f"Expected ConsoleGreeter.FormatGreeting symbol, got: {implementing_symbols}"
class TestCSharpExtractBaseNameAndType:
"""Regression tests for _extract_base_name_and_type, no running language server needed."""
@pytest.mark.parametrize(
("roslyn_name", "expected"),
[
# Property whose type is a tuple: the literal '(' in the type must not be
# mistaken for a method's parameter list.
("Position : (int X, string Y)", ("Position", ": (int X, string Y)")),
("Name : string", ("Name", ": string")),
("Add(int, int) : int", ("Add", "(int, int) : int")),
("ToString()", ("ToString", "()")),
("SimpleMethod", ("SimpleMethod", "")),
# Both still have a '(' before the first " : ", so they keep the method branch.
("GetPair() : (int, int)", ("GetPair", "() : (int, int)")),
("Merge((int, int) a, (int, int) b) : void", ("Merge", "((int, int) a, (int, int) b) : void")),
],
)
def test_extract_base_name_and_type(self, roslyn_name: str, expected: tuple[str, str]) -> None:
assert CSharpLanguageServer._extract_base_name_and_type(roslyn_name) == expected
@pytest.mark.csharp
class TestCSharpSolutionProjectOpening:
"""Test C# language server solution and project opening functionality."""
+1
View File
@@ -249,6 +249,7 @@ class TestDartLanguageServer:
# Find coordinates of 'final result = a + b;' - test position on 'result'
with language_server.open_file(file_path, open_in_ls=False) as f:
pos = find_text_coordinates(f.contents, r"final (result) = a \+ b;")
assert pos is not None
defining_symbol = language_server.request_defining_symbol(file_path, pos.line, pos.col)
@@ -4,6 +4,7 @@ from pathlib import Path
import pytest
from solidlsp import SolidLanguageServer
from solidlsp.language_servers.erlang_language_server import ErlangLanguageServer
from solidlsp.ls_config import LanguageServerId
from test.conftest import language_server_tests_enabled, start_ls_context
@@ -146,6 +147,8 @@ def test_symbol_tree_excludes_build_dirs(language_server: SolidLanguageServer):
@pytest.mark.parametrize("language_server", [LanguageServerId.ERLANG], indirect=True)
def test_ignore_compiled_files(language_server: SolidLanguageServer):
"""Test that compiled Erlang files are ignored."""
assert isinstance(language_server, ErlangLanguageServer)
# Test that beam files are ignored
assert language_server.is_ignored_filename("module.beam"), "BEAM files should be ignored"
assert language_server.is_ignored_filename("app.beam"), "BEAM files should be ignored"
@@ -164,6 +167,7 @@ def test_rebar_directories_ignored(language_server: SolidLanguageServer):
assert language_server.is_ignored_dirname(".rebar3"), "rebar3 cache should be ignored"
# Test that rebar.lock and rebar.config are not ignored (they are configuration files)
assert isinstance(language_server, ErlangLanguageServer)
assert not language_server.is_ignored_filename("rebar.config"), "rebar.config should not be ignored"
assert not language_server.is_ignored_filename("rebar.lock"), "rebar.lock should not be ignored"
+1 -1
View File
@@ -25,7 +25,7 @@ class TestFatouLanguageServer:
def test_cross_file_references(self, language_server: SolidLanguageServer) -> None:
references = language_server.request_references("src/fatou_a.jl", line=0, column=2)
locations = {(reference["relativePath"].replace("\\", "/"), reference["range"]["start"]["line"]) for reference in references}
locations = {(reference["relativePath"].replace("\\", "/"), reference["range"]["start"]["line"]) for reference in references} # type: ignore
assert locations >= {("src/fatou_a.jl", 1), ("src/fatou_b.jl", 0)}
def test_file_matching(self) -> None:
@@ -38,42 +38,42 @@ class TestKotlinDependencyProvider:
".win.zip",
"zip",
("bin", "intellij-server.exe"),
"f2daaa476f26d99301b406f76de6d87c437d04dc72f06845154619d8f991c51f",
"a9b471b16025b1bfb3b0a097862580abb40e3c35406c44242c18b1d70f5d0e44",
),
(
PlatformId.WIN_arm64,
"-aarch64.win.zip",
"zip",
("bin", "intellij-server.exe"),
"73a552a6a420158622e5ad8d96b53da8aa8ced3f88a24fded01575927a2fd8e7",
"3bf008d8c94fa70eb13fc998eaa42f29b9d13f368984d4cec46277808f94e1de",
),
(
PlatformId.LINUX_x64,
".tar.gz",
"gztar",
(f"kotlin-server-{DEFAULT_KOTLIN_LSP_VERSION}", "bin", "intellij-server"),
"2d99d8e198fbe4aa8f4481e37799724ce94803b4ea12a60b416040e3fcd7cc5e",
"1e11d2e5fefbf9ea215ad8dd6be95f2222897cd086e8cb7a661a52084a590405",
),
(
PlatformId.LINUX_arm64,
"-aarch64.tar.gz",
"gztar",
(f"kotlin-server-{DEFAULT_KOTLIN_LSP_VERSION}", "bin", "intellij-server"),
"2317831c6e5607d05b7ebc1da655330125ce0e3d66fbf24517dfce442debc14e",
"ec7cb254a6662a07fff9f10e4365226afab6c40008f8a974c10ac5e785d6510f",
),
(
PlatformId.OSX_x64,
".sit",
"zip",
(f"kotlin-server-{DEFAULT_KOTLIN_LSP_VERSION}", "bin", "intellij-server"),
"17369fda97c85418ac24ab38a9df56b21522a3468dfe193832fe455c13920745",
"62ab735947b1c855b505f64f5db8fbd7ff0b52a35ab1897938c6dbfc7b24c8a3",
),
(
PlatformId.OSX_arm64,
"-aarch64.sit",
"zip",
(f"kotlin-server-{DEFAULT_KOTLIN_LSP_VERSION}", "bin", "intellij-server"),
"6ba6021a706b21e64cef33f7e2b79f187c0910320722bb2d3ed05ad1115ec43f",
"95da3fc6d3b9092c7616345044a05edb85e5408dc648d081e4e433595c892bec",
),
],
)
@@ -176,12 +176,12 @@ class TestKotlinDependencyProvider:
"kotlin-lsp-261.13587.0-linux-aarch64.zip",
"kotlin-lsp-261.13587.0-mac-x64.zip",
"kotlin-lsp-261.13587.0-mac-aarch64.zip",
"kotlin-server-262.9593.0.win.zip",
"kotlin-server-262.9593.0-aarch64.win.zip",
"kotlin-server-262.9593.0.tar.gz",
"kotlin-server-262.9593.0-aarch64.tar.gz",
"kotlin-server-262.9593.0.sit",
"kotlin-server-262.9593.0-aarch64.sit",
"kotlin-server-263.4702.0.win.zip",
"kotlin-server-263.4702.0-aarch64.win.zip",
"kotlin-server-263.4702.0.tar.gz",
"kotlin-server-263.4702.0-aarch64.tar.gz",
"kotlin-server-263.4702.0.sit",
"kotlin-server-263.4702.0-aarch64.sit",
}
@pytest.mark.parametrize(
+1 -1
View File
@@ -192,7 +192,7 @@ class TestPascalLanguageServerBasics:
contents = hover.get("contents", {})
value = contents.get("value", "") if isinstance(contents, dict) else str(contents)
else:
value = hover.contents.value if hasattr(hover.contents, "value") else str(hover.contents)
value = hover.contents.value if hasattr(hover.contents, "value") else str(hover.contents) # type: ignore
# Should contain the function signature
assert "CalculateSum" in value, f"Hover should show function name. Got: {value[:500]}"
@@ -46,6 +46,7 @@ class TestLanguageServerSymbols:
with language_server.open_file(file_path, open_in_ls=False) as f:
file_content = f.contents
coords = find_text_coordinates(file_content, r"(status): str")
assert coords is not None
ref_symbols = [ref.symbol for ref in language_server.request_referencing_symbols(file_path, coords.line, coords.col)]
assert len(ref_symbols) > 0
+3 -1
View File
@@ -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."""
@@ -590,7 +590,9 @@ class TestRubyLanguageServerSymbols:
pos = find_text_coordinates(fb.contents, r"user = @service\.(create_user)")
# Verify that we can find the method definition
assert pos is not None
defining_symbol = language_server.request_defining_symbol(file_path, pos.line, pos.col)
assert defining_symbol is not None
assert "name" in defining_symbol
assert "kind" in defining_symbol
assert defining_symbol.get("name") == "create_user"
+2 -2
View File
@@ -68,7 +68,7 @@ class TestRustLanguageServer:
implementations = language_server.request_implementation(os.path.join("src", "lib.rs"), *pos)
assert implementations, "Expected at least one implementation of Greeter.format_greeting"
assert any("src/lib.rs" in implementation.get("relativePath", "").replace("\\", "/") for implementation in implementations), (
assert any("src/lib.rs" in implementation.get("relativePath", "").replace("\\", "/") for implementation in implementations), ( # type: ignore
f"Expected ConsoleGreeter.format_greeting in implementations, got: {implementations}"
)
@@ -81,7 +81,7 @@ class TestRustLanguageServer:
implementing_symbols = language_server.request_implementing_symbols(os.path.join("src", "lib.rs"), *pos)
assert implementing_symbols, "Expected implementing symbols for Greeter.format_greeting"
assert any(
symbol.get("name") == "format_greeting" and "src/lib.rs" in symbol["location"].get("relativePath", "").replace("\\", "/")
symbol.get("name") == "format_greeting" and "src/lib.rs" in symbol["location"].get("relativePath", "").replace("\\", "/") # type: ignore
for symbol in implementing_symbols
), f"Expected ConsoleGreeter.format_greeting symbol, got: {implementing_symbols}"
+2
View File
@@ -130,6 +130,7 @@ class TestScssReferences:
line, col = coords.line, coords.col
refs = language_server.request_references(path, line, col + 2)
ref_paths = {r.get("relativePath", "") for r in refs}
ref_paths = {r for r in ref_paths if r} # filter out empty strings
assert any(p.endswith("buttons.scss") for p in ref_paths), (
f"Expected card-surface references to include buttons.scss, got: {ref_paths}"
)
@@ -148,6 +149,7 @@ class TestScssReferences:
line, col = coords.line, coords.col
refs = language_server.request_references(path, line, col + 2)
ref_paths = {r.get("relativePath", "") for r in refs}
ref_paths = {r for r in ref_paths if r} # filter out empty strings
assert any(p.endswith("buttons.scss") for p in ref_paths), (
f"Expected $color-primary references to include buttons.scss, got: {ref_paths}"
)
+4 -3
View File
@@ -20,7 +20,7 @@ class TestSvelteLanguageServer:
def test_svelte_language_server_root_matches_repo_path(self, language_server: SolidLanguageServer, repo_path: Path) -> None:
assert language_server.is_running()
assert repo_path.resolve() == svelte_test_conftest.repo_path.resolve()
assert Path(language_server.language_server.repo_path).resolve() == repo_path.resolve()
assert Path(language_server.repository_root_path).resolve() == repo_path.resolve()
@pytest.mark.parametrize("language_server", [LanguageServerId.SVELTE], indirect=True)
def test_svelte_and_typescript_files_in_symbol_tree(self, language_server: SolidLanguageServer) -> None:
@@ -74,12 +74,13 @@ class TestSvelteLanguageServer:
def test_definition_from_component_import_to_svelte_file(self, language_server: SolidLanguageServer) -> None:
file_path = os.path.join("src", "lib", "components", "Header.svelte")
coords = find_text_coordinates(read_repo_file(language_server, file_path), r"(count)")
assert coords is not None
definitions = language_server.request_definition(file_path, coords.line, coords.col)
definition_paths = sorted(definition["relativePath"].replace("\\", "/") for definition in definitions)
definition_paths = sorted(definition["relativePath"].replace("\\", "/") for definition in definitions) # type: ignore
assert len(definitions) == 1, definition_paths
assert definitions[0]["relativePath"].replace("\\", "/") == "src/lib/components/Counter.svelte", definition_paths
assert definitions[0]["relativePath"].replace("\\", "/") == "src/lib/components/Counter.svelte", definition_paths # type: ignore
@pytest.mark.parametrize("language_server", [LanguageServerId.SVELTE], indirect=True)
def test_diagnostics_in_typescript_file(self, language_server: SolidLanguageServer) -> None:
@@ -12,7 +12,7 @@ class TestSvelteReferences:
@pytest.mark.parametrize("language_server", [LanguageServerId.SVELTE], indirect=True)
def test_references_across_svelte_and_typescript(self, language_server: SolidLanguageServer) -> None:
refs = language_server.request_references(os.path.join("src", "lib", "components", "Words.svelte"), 1, 17)
ref_paths = {ref["relativePath"].replace("\\", "/") for ref in refs}
ref_paths = {ref["relativePath"].replace("\\", "/") for ref in refs} # type: ignore
assert "src/routes/(sverdle)/words.server.ts" in ref_paths, sorted(ref_paths)
assert "src/lib/game.ts" in ref_paths, sorted(ref_paths)
@@ -21,6 +21,6 @@ class TestSvelteReferences:
@pytest.mark.parametrize("language_server", [LanguageServerId.SVELTE], indirect=True)
def test_references_from_typescript_file(self, language_server: SolidLanguageServer) -> None:
refs = language_server.request_references(os.path.join("src", "lib", "game.ts"), 3, 13)
ref_paths = {ref["relativePath"].replace("\\", "/") for ref in refs}
ref_paths = {ref["relativePath"].replace("\\", "/") for ref in refs} # type: ignore
assert "src/routes/(sverdle)/+page.server.ts" in ref_paths, sorted(ref_paths)
@@ -56,6 +56,7 @@ class TestSvelteRename:
def test_rename_svelte_export_updates_svelte_importers(self, language_server: SolidLanguageServer) -> None:
file_path = os.path.join("src", "lib", "components", "Counter.svelte")
coords = find_text_coordinates(read_repo_file(language_server, file_path), r"(count)")
assert coords is not None
workspace_edit = language_server.request_rename_symbol_edit(file_path, coords.line, coords.col, "score")
@@ -69,6 +70,7 @@ class TestSvelteRename:
def test_rename_svelte_export_updates_ts_and_svelte_files(self, language_server: SolidLanguageServer) -> None:
file_path = os.path.join("src", "lib", "components", "Words.svelte")
coords = find_text_coordinates(read_repo_file(language_server, file_path), r"(words)")
assert coords is not None
workspace_edit = language_server.request_rename_symbol_edit(file_path, coords.line, coords.col, "vocabulary")
+38
View File
@@ -0,0 +1,38 @@
# SPDX-License-Identifier: MIT
import tempfile
from pathlib import Path
from solidlsp.language_servers.dart_language_server import DartLanguageServer
def _make_dart_ls() -> DartLanguageServer:
ls = object.__new__(DartLanguageServer)
ls._custom_settings = {}
# Windows: as_uri() rejects drive-less paths like "/tmp/..."
project_dir = Path(tempfile.mkdtemp(prefix="fake-dart-project-")) / "project"
project_dir.mkdir(parents=True, exist_ok=True)
ls.repository_root_path = str(project_dir)
class _Cfg:
@staticmethod
def get_absolute_workspace_folders(root):
return [root]
@staticmethod
def get_absolute_additional_workspace_folders(root):
return []
ls.config = _Cfg()
# custom_settings property reads from _custom_settings on SolidLanguageServer
return ls
def test_dart_omits_root_uri():
builder = _make_dart_ls()._create_initialize_params_builder()
params = builder.build()
# rootUri must be present (some servers reject an undefined key) but null, so that only
# workspaceFolders determine the analysis roots (oraios/serena#2045).
assert params["rootUri"] is None
assert params["rootPath"] is None
assert params["workspaceFolders"]
@@ -0,0 +1,39 @@
# SPDX-License-Identifier: MIT
import tempfile
from pathlib import Path
from solidlsp.initialize_params import DefaultInitializeParamsBuilder
class _FakeLS:
# Windows: as_uri() rejects drive-less paths like "/tmp/..."
repository_root_path = str(Path(tempfile.mkdtemp(prefix="fake-project-")) / "root")
class config:
@staticmethod
def get_absolute_workspace_folders(root):
return [root]
@staticmethod
def get_absolute_additional_workspace_folders(root):
return []
custom_settings: dict = {}
def test_default_builder_sets_root_uri():
builder = DefaultInitializeParamsBuilder(_FakeLS())
params = builder.build()
assert "rootUri" in params
assert "rootPath" in params
def test_builder_sends_null_root_uri_when_disabled():
builder = DefaultInitializeParamsBuilder(_FakeLS(), set_root_uri=False)
params = builder.build()
# keys must be present (some servers reject an undefined rootUri); values are null
assert params["rootUri"] is None
assert params["rootPath"] is None
assert params["processId"] is not None
assert params["clientInfo"] == {"name": "Serena"}
+2 -2
View File
@@ -38,7 +38,7 @@ def test_start_stops_process_when_start_server_raises_after_spawning():
with pytest.raises(RuntimeError, match="capability assertion"):
server.start()
server.server.stop.assert_called_once()
server.server.stop.assert_called_once() # type: ignore
assert server.server_started is False
@@ -51,5 +51,5 @@ def test_start_does_not_call_stop_when_start_server_raises_before_spawning():
with pytest.raises(RuntimeError, match="capability assertion"):
server.start()
server.server.stop.assert_not_called()
server.server.stop.assert_not_called() # type: ignore
assert server.server_started is False
+6 -4
View File
@@ -115,8 +115,8 @@ def test_language_server_process_survives_a_short_lived_calling_thread() -> None
text=True,
)
try:
ready_line = driver.stdout.readline()
assert ready_line.strip() == "READY", f"driver failed to start the language server: {driver.stderr.read()}"
ready_line = driver.stdout.readline() # type: ignore
assert ready_line.strip() == "READY", f"driver failed to start the language server: {driver.stderr.read()}" # type: ignore
time.sleep(2)
# The driver's own argv also contains `marker` (it's passed as sys.argv[1]), so exclude
@@ -143,8 +143,10 @@ def test_language_server_process_dies_with_a_sigkilled_serena() -> None:
text=True,
)
try:
ready_line = driver.stdout.readline()
assert ready_line.strip() == "READY", f"driver failed to start the language server: {driver.stderr.read()}"
ready_line = driver.stdout.readline() # type: ignore
stderr = driver.stderr
assert stderr is not None, "stderr should be captured"
assert ready_line.strip() == "READY", f"driver failed to start the language server: {stderr.read()}"
assert _find_marked_processes(marker), "language server process never started"
driver.kill() # SIGKILL: simulates Serena being killed without a chance to clean up
+7 -3
View File
@@ -80,7 +80,9 @@ def _spawn_ready(src: str) -> subprocess.Popen:
test_pdeathsig.py's driver pattern (deterministic sync instead of a blind sleep).
"""
proc = subprocess.Popen([sys.executable, "-c", src], start_new_session=True, stdout=subprocess.PIPE, text=True)
ready_line = proc.stdout.readline()
stdout = proc.stdout
assert stdout is not None
ready_line = stdout.readline()
assert ready_line.strip() == "READY", f"helper process failed to start: {ready_line!r}"
return proc
@@ -341,8 +343,10 @@ class TestPsutilDenialConsequences:
"""
)
proc = subprocess.Popen([sys.executable, "-c", src], start_new_session=True, stdout=subprocess.PIPE, text=True)
child_pid = int(proc.stdout.readline().strip())
ready_line = proc.stdout.readline()
stdout = proc.stdout
assert stdout is not None
child_pid = int(stdout.readline().strip())
ready_line = stdout.readline()
assert ready_line.strip() == "READY", f"helper process failed to start: {ready_line!r}"
return proc, child_pid
@@ -0,0 +1,80 @@
"""Unit tests for GodotLanguageServer's document-symbol range correction.
See oraios/serena#1974: Godot's GDScript parser can report a symbol's end column one
column past the line-end convention every other language server follows, which silently
rolls over into the following line's content (or, at the last line of a body, into the
separating blank line) when the position is later turned into a text index.
"""
from __future__ import annotations
from solidlsp.language_servers.godot_language_server import GodotLanguageServer
_fix_range_end = GodotLanguageServer._fix_range_end
_fix_symbol_ranges = GodotLanguageServer._fix_symbol_ranges
def _range(end_line: int, end_char: int, start_line: int = 0, start_char: int = 0) -> dict:
return {"start": {"line": start_line, "character": start_char}, "end": {"line": end_line, "character": end_char}}
# The exact fixture from the reported issue: a 13-char line whose reported end column is 14.
_ISSUE_1952_LINES = ["extends Node", "", "func first():", '\tprint("old")', "", "func second():", '\tprint("keep")']
def test_fix_range_end_corrects_the_measured_off_by_one() -> None:
rng = _range(end_line=3, end_char=14)
_fix_range_end(rng, _ISSUE_1952_LINES)
assert rng["end"]["character"] == 13
def test_fix_range_end_leaves_a_correct_one_past_end_column_alone() -> None:
rng = _range(end_line=3, end_char=13)
_fix_range_end(rng, _ISSUE_1952_LINES)
assert rng["end"]["character"] == 13
def test_fix_range_end_does_not_guess_at_a_larger_overshoot() -> None:
# Two past the line's end is not the measured Godot mechanism; leave it as reported
# rather than assume the same +1 correction applies.
rng = _range(end_line=3, end_char=15)
_fix_range_end(rng, _ISSUE_1952_LINES)
assert rng["end"]["character"] == 15
def test_fix_range_end_ignores_an_out_of_range_line() -> None:
rng = _range(end_line=99, end_char=5)
_fix_range_end(rng, _ISSUE_1952_LINES)
assert rng["end"]["character"] == 5
def test_fix_range_end_tolerates_a_missing_end() -> None:
_fix_range_end({"start": {"line": 0, "character": 0}}, _ISSUE_1952_LINES) # must not raise
def test_fix_symbol_ranges_corrects_range_selection_range_and_location_recursively() -> None:
# line 2, "func first():", is 13 chars: end_char=14 is the off-by-one, end_char=5 is not.
child = {
"name": "inner",
"range": _range(end_line=3, end_char=14),
"selectionRange": _range(end_line=2, end_char=14, start_line=2),
}
root = {
"name": "first",
"range": _range(end_line=3, end_char=14),
"selectionRange": _range(end_line=2, end_char=5, start_line=2),
"children": [child],
}
_fix_symbol_ranges(root, _ISSUE_1952_LINES)
assert root["range"]["end"]["character"] == 13
assert root["selectionRange"]["end"]["character"] == 5 # unaffected, no overshoot
assert child["range"]["end"]["character"] == 13
assert child["selectionRange"]["end"]["character"] == 13 # line 2 is 13 chars, 14 was the overshoot
def test_fix_symbol_ranges_corrects_symbol_information_style_location() -> None:
# SymbolInformation (the flat, non-hierarchical shape) nests its range under "location".
symbol = {"name": "first", "location": {"uri": "file:///script.gd", "range": _range(end_line=3, end_char=14)}}
_fix_symbol_ranges(symbol, _ISSUE_1952_LINES)
assert symbol["location"]["range"]["end"]["character"] == 13