Added base class for SerenaConfig, use in-memory config for tests

Also:
1. now the normal config will always save to disk (no longer handled by a flag in the methods)
2. the commented yaml is always loaded in the beginning (instead of loading every time on save). Has the upside of simpler code and less i/o, inisignificant "downside" of comments permanently staying in RAM
3. small improvements in typing
This commit is contained in:
Michael Panchenko committed 2025-05-28 15:11:33 +02:00
1 parent d242217c4b
commit 9d493d8aab
4 files changed
+160 -130

No files matched your search

+124 -114
View File
@@ -21,6 +21,7 @@ from pathlib import Path
from typing import TYPE_CHECKING, Any, Self, TypeVar, Union
import yaml
from overrides import override
from ruamel.yaml.comments import CommentedMap
from sensai.util import logging
from sensai.util.logging import FallbackHandler
@@ -197,19 +198,16 @@ class Project:
return os.path.join(self.project_root, self.project_config.rel_path_to_project_yml())
@dataclass
class SerenaConfig:
@dataclass(kw_only=True)
class SerenaConfigBase(ABC):
"""
Handles user-defined Serena configuration based on the configuration file
Abstract base class for Serena configuration handling
"""
projects: list[Project] = field(default_factory=list)
gui_log_window_enabled: bool = False
gui_log_window_level: int = logging.INFO
web_dashboard: bool = True
loaded_original_yaml: dict | CommentedMap = field(default_factory=dict)
CONFIG_FILE = "serena_config.yml"
@cached_property
def project_paths(self) -> list[str]:
@@ -219,83 +217,7 @@ class SerenaConfig:
def project_names(self) -> list[str]:
return sorted(project.project_config.project_name for project in self.projects)
@classmethod
def get_config_file_path(cls) -> str:
return os.path.join(serena_root_path(), cls.CONFIG_FILE)
@classmethod
def from_config_file(cls, preserve_comments: bool = False) -> "SerenaConfig":
"""
Static constructor to create SerenaConfig from the configuration file
"""
config_file = cls.get_config_file_path()
if not os.path.exists(config_file):
raise FileNotFoundError(f"Serena configuration file not found: {config_file}")
log.info(f"Loading Serena configuration from {config_file}")
try:
loaded_original_yaml = load_yaml(config_file, preserve_comments)
except Exception as e:
raise ValueError(f"Error loading Serena configuration from {config_file}: {e}") from e
# Create instance
instance = cls()
# read projects
if "projects" not in loaded_original_yaml:
raise SerenaConfigError("`projects` key not found in Serena configuration. Please update your `serena_config.yml` file.")
# load list of known projects
instance.projects = []
num_project_migrations = 0
for path in loaded_original_yaml["projects"]:
path = Path(path).resolve()
if not path.exists():
log.warning(f"Project path {path} does not exist, skipping.")
continue
if path.is_file():
path = cls._migrate_out_of_project_config_file(path)
if path is None:
continue
num_project_migrations += 1
project = Project.load(path)
instance.projects.append(project)
instance.gui_log_window_enabled = loaded_original_yaml.get("gui_log_window", False)
instance.gui_log_window_level = loaded_original_yaml.get("gui_log_level", logging.INFO)
instance.web_dashboard = loaded_original_yaml.get("web_dashboard", True)
instance.loaded_original_yaml = loaded_original_yaml
# re-save the configuration file if any migrations were performed
if num_project_migrations > 0:
log.info(
f"Migrated {num_project_migrations} project configurations from legacy format to in-project configuration; re-saving configuration"
)
instance.save()
return instance
@classmethod
def _migrate_out_of_project_config_file(self, path: Path) -> Path | None:
"""
Migrates a legacy project configuration file (which is a YAML file containing the project root) to the
in-project configuration file (project.yml) inside the project root directory.
:param path: the path to the legacy project configuration file
:return: the project root path if the migration was successful, None otherwise.
"""
log.info(f"Found legacy project configuration file {path}, migrating to in-project configuration.")
try:
with open(path, encoding="utf-8") as f:
project_config_data = yaml.safe_load(f)
project_root = project_config_data["project_root"]
shutil.move(str(path), str(Path(project_root) / ProjectConfig.rel_path_to_project_yml()))
return Path(project_root).resolve()
except Exception as e:
log.error(f"Error migrating configuration file: {e}")
return None
def find_project(self, project_root_or_name: str) -> Project | None:
def get_project(self, project_root_or_name: str) -> Project | None:
for project in self.projects:
if project.project_config.project_name == project_root_or_name:
return project
@@ -306,22 +228,14 @@ class SerenaConfig:
return project
return None
def save(self, preserve_comments: bool = True) -> None:
loaded_original_yaml = deepcopy(self.loaded_original_yaml)
loaded_original_yaml["projects"] = [project.project_root for project in self.projects]
save_yaml(self.get_config_file_path(), loaded_original_yaml, preserve_comments)
def add_project_from_path(
self, project_root: Path | str, project_name: str | None = None, update_config_file: bool = True
) -> tuple[Project, bool]:
def add_project_from_path(self, project_root: Path | str, project_name: str | None = None) -> tuple[Project, bool]:
"""
Add a project to the Serena configuration from a given path. Will raise a FileExistsError if the
name or path is already registered.
:param path: the path to the project to add
:param project_root: the path to the project to add
:param project_name: the name of the project to add; if None, the name of the project will be the name of the directory
containing the project
:param update_config_file: whether to update the Serena configuration file on disk
:return: the project that was added and a boolean indicating whether a new project configuration was generated and
saved to disk. It may be that no new project configuration was generated if the project configuration already
exists on disk but the project itself was not added yet to the Serena configuration.
@@ -352,16 +266,18 @@ class SerenaConfig:
new_project_config_generated = True
new_project = Project(project_root=str(project_root), project_config=project_config)
self.projects.append(new_project)
self._add_new_project(new_project)
# Update the config file on disk and preserve comments
if update_config_file:
config_with_comments = SerenaConfig.from_config_file(preserve_comments=True)
config_with_comments.projects.append(new_project)
config_with_comments.save(preserve_comments=True)
return new_project, new_project_config_generated
def remove_project(self, project_name: str, update_config_file: bool = True) -> None:
def _add_new_project(self, project: Project) -> None:
"""
Adds a new project to the Serena configuration. No checks are performed here,
this method is intended to be overridden by subclasses.
"""
self.projects.append(project)
def remove_project(self, project_name: str) -> None:
# find the index of the project with the desired name and remove it
for i, project in enumerate(self.projects):
if project.project_name == project_name:
@@ -370,14 +286,110 @@ class SerenaConfig:
else:
raise ValueError(f"Project '{project_name}' not found in Serena configuration; valid project names: {self.project_names}")
# Update the config file on disk and preserve comments
if update_config_file:
config_with_comments = SerenaConfig.from_config_file(preserve_comments=True)
for i, project in enumerate(config_with_comments.projects):
if project.project_name == project_name:
del config_with_comments.projects[i]
break
config_with_comments.save(preserve_comments=True)
@dataclass(kw_only=True)
class SerenaConfig(SerenaConfigBase):
"""
Handles user-defined Serena configuration based on the (fixed) Serena configuration file.
Updates to the instance will be automatically saved to the configuration file.
Usually, there should be only one instance of this class in the application.
"""
loaded_commented_yaml: CommentedMap
CONFIG_FILE = "serena_config.yml"
@classmethod
def get_config_file_path(cls) -> str:
return os.path.join(serena_root_path(), cls.CONFIG_FILE)
@classmethod
def from_config_file(cls) -> "SerenaConfig":
"""
Static constructor to create SerenaConfig from the configuration file
"""
config_file = cls.get_config_file_path()
if not os.path.exists(config_file):
raise FileNotFoundError(f"Serena configuration file not found: {config_file}")
log.info(f"Loading Serena configuration from {config_file}")
try:
loaded_commented_yaml = load_yaml(config_file, preserve_comments=True)
except Exception as e:
raise ValueError(f"Error loading Serena configuration from {config_file}: {e}") from e
# Create instance
instance = cls(loaded_commented_yaml=loaded_commented_yaml)
# read projects
if "projects" not in loaded_commented_yaml:
raise SerenaConfigError("`projects` key not found in Serena configuration. Please update your `serena_config.yml` file.")
# load list of known projects
instance.projects = []
num_project_migrations = 0
for path in loaded_commented_yaml["projects"]:
path = Path(path).resolve()
if not path.exists():
log.warning(f"Project path {path} does not exist, skipping.")
continue
if path.is_file():
path = cls._migrate_out_of_project_config_file(path)
if path is None:
continue
num_project_migrations += 1
project = Project.load(path)
instance.projects.append(project)
instance.gui_log_window_enabled = loaded_commented_yaml.get("gui_log_window", False)
instance.gui_log_window_level = loaded_commented_yaml.get("gui_log_level", logging.INFO)
instance.web_dashboard = loaded_commented_yaml.get("web_dashboard", True)
# re-save the configuration file if any migrations were performed
if num_project_migrations > 0:
log.info(
f"Migrated {num_project_migrations} project configurations from legacy format to in-project configuration; re-saving configuration"
)
instance.save()
return instance
@classmethod
def _migrate_out_of_project_config_file(cls, path: Path) -> Path | None:
"""
Migrates a legacy project configuration file (which is a YAML file containing the project root) to the
in-project configuration file (project.yml) inside the project root directory.
:param path: the path to the legacy project configuration file
:return: the project root path if the migration was successful, None otherwise.
"""
log.info(f"Found legacy project configuration file {path}, migrating to in-project configuration.")
try:
with open(path, encoding="utf-8") as f:
project_config_data = yaml.safe_load(f)
project_root = project_config_data["project_root"]
shutil.move(str(path), str(Path(project_root) / ProjectConfig.rel_path_to_project_yml()))
return Path(project_root).resolve()
except Exception as e:
log.error(f"Error migrating configuration file: {e}")
return None
def save(self) -> None:
loaded_original_yaml = deepcopy(self.loaded_commented_yaml)
# projects are unique absolute paths
# we also canonicalize them before saving
loaded_original_yaml["projects"] = sorted({str(Path(project.project_root).resolve()) for project in self.projects})
save_yaml(self.get_config_file_path(), loaded_original_yaml, preserve_comments=True)
@override
def _add_new_project(self, project: Project) -> None:
super()._add_new_project(project)
self.save()
@override
def remove_project(self, project_name: str) -> None:
super().remove_project(project_name)
self.save()
class LinesRead:
@@ -450,7 +462,7 @@ class SerenaAgent:
self,
project: str | None = None,
project_activation_callback: Callable[[], None] | None = None,
serena_config: SerenaConfig | None = None,
serena_config: SerenaConfigBase | None = None,
context: SerenaAgentContext | None = None,
modes: list[SerenaAgentMode] | None = None,
):
@@ -622,7 +634,7 @@ class SerenaAgent:
"""
new_project_generated = False
new_project_config_generated = False
project_instance: Project | None = self.serena_config.find_project(project_root_or_name)
project_instance: Project | None = self.serena_config.get_project(project_root_or_name)
if project_instance is not None:
log.info(f"Found registered project {project_instance.project_name} at path {project_instance.project_root}.")
else:
@@ -631,9 +643,7 @@ class SerenaAgent:
f"Project '{project_root_or_name}' not found: Not a valid project name or directory. "
f"Existing project names: {self.serena_config.project_names}"
)
project_instance, new_project_config_generated = self.serena_config.add_project_from_path(
project_root_or_name, update_config_file=True
)
project_instance, new_project_config_generated = self.serena_config.add_project_from_path(project_root_or_name)
new_project_generated = True
log.info(f"Added new project {project_instance.project_name} for path {project_instance.project_root}.")
if new_project_config_generated:
@@ -1745,7 +1755,7 @@ class RemoveProjectTool(Tool, ToolMarkerDoesNotRequireActiveProject):
:param project_name: Name of the project to remove
"""
self.agent.serena_config.remove_project(project_name, update_config_file=True)
self.agent.serena_config.remove_project(project_name)
return f"Successfully removed project '{project_name}' from configuration."
+13 -10
View File
@@ -16,22 +16,22 @@ log = logging.getLogger(__name__)
class MemoryLogHandler(logging.Handler):
def __init__(self, level=logging.NOTSET):
def __init__(self, level: int = logging.NOTSET) -> None:
super().__init__(level=level)
self.setFormatter(logging.Formatter(logging.LOG_DEFAULT_FORMAT))
self._log_buffer = LogBuffer()
self._log_queue = queue.Queue()
self._log_queue: queue.Queue[str] = queue.Queue()
self._stop_event = threading.Event()
# start background thread to process logs
self.worker_thread = threading.Thread(target=self._process_queue, daemon=True)
self.worker_thread.start()
def emit(self, record):
def emit(self, record: logging.LogRecord) -> None:
msg = self.format(record)
self._log_queue.put_nowait(msg)
def _process_queue(self):
def _process_queue(self) -> None:
while not self._stop_event.is_set():
try:
msg = self._log_queue.get(timeout=1)
@@ -45,10 +45,10 @@ class MemoryLogHandler(logging.Handler):
class LogBuffer:
def __init__(self):
self.logs = []
def __init__(self) -> None:
self.logs: list[str] = []
def append(self, msg: str):
def append(self, msg: str) -> None:
self.logs.append(msg)
@@ -68,13 +68,13 @@ class ResponseToolNames(BaseModel):
class SerenaDashboardAPI:
log = logging.getLogger(__qualname__)
def __init__(self, memory_log_handler: MemoryLogHandler, tool_names: list[str]):
def __init__(self, memory_log_handler: MemoryLogHandler, tool_names: list[str]) -> None:
self._memory_log_handler = memory_log_handler
self._tool_names = tool_names
self._app = FastAPI(title="Serena Dashboard")
self._setup_routes()
def _setup_routes(self):
def _setup_routes(self) -> None:
static_dir = os.path.join(serena_root_path(), "dashboard")
self._app.mount("/dashboard", StaticFiles(directory=static_dir), name="dashboard")
@@ -110,7 +110,10 @@ class SerenaDashboardAPI:
raise RuntimeError(f"No free ports found starting from {start_port}")
def run(self, host="127.0.0.1", port=0x5EDA):
def run(self, host: str = "127.0.0.1", port: int = 0x5EDA) -> int:
"""
Runs the dashboard on the given host and port and returns the port number.
"""
uvicorn.run(self._app, host=host, port=port, workers=1, log_config=None, log_level="critical")
return port
+10 -4
View File
@@ -1,8 +1,10 @@
from typing import Literal, overload
from ruamel.yaml import YAML
from ruamel.yaml.comments import CommentedMap
def create_YAML(preserve_comments: bool = False) -> YAML:
def _create_YAML(preserve_comments: bool = False) -> YAML:
"""
Creates a YAML that can load/save with comments if preserve_comments is True.
"""
@@ -12,13 +14,17 @@ def create_YAML(preserve_comments: bool = False) -> YAML:
return result
def load_yaml(path: str, preserve_comments: bool = False) -> dict:
@overload
def load_yaml(path: str, preserve_comments: Literal[False]) -> dict: ...
@overload
def load_yaml(path: str, preserve_comments: Literal[True]) -> CommentedMap: ...
def load_yaml(path: str, preserve_comments: bool = False) -> dict | CommentedMap:
with open(path, encoding="utf-8") as f:
yaml = create_YAML(preserve_comments)
yaml = _create_YAML(preserve_comments)
return yaml.load(f)
def save_yaml(path: str, data: dict | CommentedMap, preserve_comments: bool = False) -> None:
yaml = create_YAML(preserve_comments)
yaml = _create_YAML(preserve_comments)
with open(path, "w", encoding="utf-8") as f:
yaml.dump(data, f)
+13 -2
View File
@@ -1,17 +1,28 @@
import json
import os
import time
from dataclasses import dataclass
import pytest
from multilspy.multilspy_config import Language
from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, SerenaAgent, SerenaConfig
from serena.agent import FindReferencingSymbolsTool, FindSymbolTool, SerenaAgent, SerenaConfigBase
from test.conftest import LanguageParamRequest, get_repo_path
@dataclass
class SerenaConfigForTests(SerenaConfigBase):
"""
In-memory implementation of Serena configuration with the GUI disabled.
"""
gui_log_window_enabled: bool = False
web_dashboard: bool = False
@pytest.fixture
def serena_config():
return SerenaConfig()
return SerenaConfigForTests()
@pytest.fixture