From 33b3388b11352b1dd5757e24599aaec12046ede7 Mon Sep 17 00:00:00 2001 From: matdev83 <211248003+matdev83@users.noreply.github.com> Date: Mon, 13 Oct 2025 00:58:18 +0200 Subject: [PATCH] Prevent global command settings mutation from auto-registered commands --- src/core/app/stages/command.py | 81 +++++++++++-------- .../app/test_default_command_state_service.py | 40 +++++++++ 2 files changed, 88 insertions(+), 33 deletions(-) create mode 100644 tests/unit/core/app/test_default_command_state_service.py diff --git a/src/core/app/stages/command.py b/src/core/app/stages/command.py index 9bddbc85..fcba40f0 100644 --- a/src/core/app/stages/command.py +++ b/src/core/app/stages/command.py @@ -10,17 +10,64 @@ from __future__ import annotations import logging +from typing import Any from src.core.config.app_config import AppConfig from src.core.di.container import ServiceCollection from src.core.interfaces.application_state_interface import IApplicationState +from src.core.interfaces.command_settings_interface import ICommandSettingsService from src.core.interfaces.di_interface import IServiceProvider +from src.core.interfaces.state_provider_interface import ( + ISecureStateAccess, + ISecureStateModification, +) from .base import InitializationStage logger = logging.getLogger(__name__) +class DefaultCommandStateService(ISecureStateAccess, ISecureStateModification): + """Lightweight state holder used when auto-registering domain commands.""" + + def __init__(self, settings_service: ICommandSettingsService) -> None: + self._settings = settings_service + self._routes: list[dict[str, Any]] = [] + self._command_prefix_override: str | None = None + self._api_key_redaction_override: bool | None = None + self._disable_interactive_override: bool | None = None + + def get_command_prefix(self) -> str: + if self._command_prefix_override is not None: + return self._command_prefix_override + return self._settings.get_command_prefix() + + def get_failover_routes(self) -> list[dict[str, Any]] | None: + return self._routes + + def update_failover_routes(self, routes: list[dict[str, Any]]) -> None: + self._routes = routes + + def get_api_key_redaction_enabled(self) -> bool: + if self._api_key_redaction_override is not None: + return self._api_key_redaction_override + return self._settings.get_api_key_redaction_enabled() + + def get_disable_interactive_commands(self) -> bool: + if self._disable_interactive_override is not None: + return self._disable_interactive_override + return self._settings.get_disable_interactive_commands() + + def update_command_prefix(self, prefix: str) -> None: + if isinstance(prefix, str) and prefix: + self._command_prefix_override = prefix + + def update_api_key_redaction(self, enabled: bool) -> None: + self._api_key_redaction_override = bool(enabled) + + def update_interactive_commands(self, enabled: bool) -> None: + self._disable_interactive_override = bool(enabled) + class CommandStage(InitializationStage): """ Stage for registering command-related services. @@ -188,39 +235,7 @@ def populate_commands_factory(provider: IServiceProvider) -> None: ICommandSettingsService # type: ignore[type-abstract] ) - # Create a simple state service for commands - class DefaultStateService( - ISecureStateAccess, ISecureStateModification - ): - def __init__(self, settings_service): - self._settings = settings_service - self._routes = [] - - def get_command_prefix(self): - return self._settings.get_command_prefix() - - def get_failover_routes(self): - return self._routes - - def update_failover_routes(self, routes): - self._routes = routes - - def get_api_key_redaction_enabled(self): - return self._settings.get_api_key_redaction_enabled() - - def get_disable_interactive_commands(self): - return self._settings.get_disable_interactive_commands() - - def update_command_prefix(self, prefix: str) -> None: - self._settings.command_prefix = prefix - - def update_api_key_redaction(self, enabled: bool) -> None: - self._settings.api_key_redaction_enabled = enabled - - def update_interactive_commands(self, enabled: bool) -> None: - pass - - state_service = DefaultStateService(settings_service) + state_service = DefaultCommandStateService(settings_service) # Auto-register all commands from the domain command registry for ( diff --git a/tests/unit/core/app/test_default_command_state_service.py b/tests/unit/core/app/test_default_command_state_service.py new file mode 100644 index 00000000..14b370c9 --- /dev/null +++ b/tests/unit/core/app/test_default_command_state_service.py @@ -0,0 +1,40 @@ +from src.core.app.stages.command import DefaultCommandStateService +from src.core.services.command_settings_service import CommandSettingsService + + +def _make_settings() -> CommandSettingsService: + return CommandSettingsService( + default_command_prefix="!/", + default_api_key_redaction=True, + default_disable_interactive_commands=False, + ) + + +def test_command_prefix_override_is_session_local() -> None: + settings = _make_settings() + state_service = DefaultCommandStateService(settings) + + state_service.update_command_prefix("$/") + + assert state_service.get_command_prefix() == "$/" + assert settings.get_command_prefix() == "!/" + + +def test_api_key_redaction_override_is_session_local() -> None: + settings = _make_settings() + state_service = DefaultCommandStateService(settings) + + state_service.update_api_key_redaction(False) + + assert state_service.get_api_key_redaction_enabled() is False + assert settings.get_api_key_redaction_enabled() is True + + +def test_disable_interactive_commands_override_is_session_local() -> None: + settings = _make_settings() + state_service = DefaultCommandStateService(settings) + + state_service.update_interactive_commands(True) + + assert state_service.get_disable_interactive_commands() is True + assert settings.get_disable_interactive_commands() is False