From 9d493d8aabf0962fd98a7247632f5e320130df85 Mon Sep 17 00:00:00 2001 From: Michael Panchenko Date: Wed, 28 May 2025 15:11:33 +0200 Subject: [PATCH] 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 --- src/serena/agent.py | 238 ++++++++++++++++--------------- src/serena/dashboard.py | 23 +-- src/serena/util/general.py | 14 +- test/serena/test_serena_agent.py | 15 +- 4 files changed, 160 insertions(+), 130 deletions(-) diff --git a/src/serena/agent.py b/src/serena/agent.py index a8c938f3..1a69da45 100644 --- a/src/serena/agent.py +++ b/src/serena/agent.py @@ -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." diff --git a/src/serena/dashboard.py b/src/serena/dashboard.py index f6f129a4..d52040b5 100644 --- a/src/serena/dashboard.py +++ b/src/serena/dashboard.py @@ -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 diff --git a/src/serena/util/general.py b/src/serena/util/general.py index 8cabcd82..bfddaf47 100644 --- a/src/serena/util/general.py +++ b/src/serena/util/general.py @@ -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) diff --git a/test/serena/test_serena_agent.py b/test/serena/test_serena_agent.py index 604b9021..691c75d7 100644 --- a/test/serena/test_serena_agent.py +++ b/test/serena/test_serena_agent.py @@ -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