Compare commits

...

5 Commits

Author SHA1 Message Date
openhands
7b305d8e9d Fix pr #8099: Write configuration 2025-05-04 00:23:33 +00:00
openhands
a0a2fdfddb Fix pr #8099: Write configuration 2025-05-04 00:05:04 +00:00
OpenHands Bot
75428c1060 🤖 Auto-fix Python linting issues 2025-05-03 23:11:13 +00:00
Engel Nyst
70c013e46f Merge branch 'main' into enyst/write-config 2025-05-04 01:10:08 +02:00
Engel Nyst
6e575277bd write config file 2025-04-25 18:50:27 +02:00
9 changed files with 922 additions and 94 deletions

View File

@@ -0,0 +1,40 @@
# Status: Implementing User Configuration Persistence
This PR implements a new configuration persistence system that allows saving user-specific settings to a dedicated TOML file (`~/.openhands/config.toml`), while keeping secrets in a separate JSON file (`~/.openhands/secrets.json`). This split architecture provides better separation of concerns and more appropriate storage formats for each type of data.
## Current Status
### Completed
1. **Core TOML Writing Mechanism**
* Implemented `save_setting_to_user_toml` in `config_save.py` with comprehensive validation and error handling
* Added snapshot mechanism to track TOML-sourced values vs. env/cli overrides
* Implemented helpers for safe value access and default value retrieval
* Added extensive unit tests covering all key scenarios
2. **Settings Store Integration**
* Updated `FileSettingsStore` to use the new TOML writing mechanism
* Implemented mapping between Settings model and AppConfig paths
* Added unit tests for store operations
3. **Secrets Handling**
* Secrets (API keys, tokens) are now stored separately in `secrets.json`
* Implemented `FileSecretsStore` for dedicated secrets management
* Secrets remain in JSON format for better compatibility with existing code
### Remaining Tasks
1. **Runtime Config Updates**
* Implement mechanism to update runtime `AppConfig` instance when settings change
* Add callbacks or direct access to global config object
* Add tests for runtime updates
2. **CLI Integration**
* Add CLI commands for viewing/modifying user TOML settings
* Ensure CLI respects the same validation rules as the API
* Add CLI integration tests
3. **Documentation & Cleanup**
* Update configuration documentation to reflect the new architecture
* Document the split between settings (TOML) and secrets (JSON)
* Remove references to old `settings.json` mechanism

View File

@@ -1,6 +1,6 @@
from typing import Any, ClassVar
from typing import Any, ClassVar, Optional
from pydantic import BaseModel, Field, SecretStr
from pydantic import BaseModel, Field, PrivateAttr, SecretStr
from openhands.core import logger
from openhands.core.config.agent_config import AgentConfig
@@ -92,10 +92,23 @@ class AppConfig(BaseModel):
) # Maximum number of concurrent agent loops allowed per user
mcp: MCPConfig = Field(default_factory=MCPConfig)
# Private attribute to store the configuration state after loading TOML files
# but before environment variables and CLI args are applied.
# Used to determine the source of a setting when saving back to user TOML.
_toml_snapshot: Optional['AppConfig'] = PrivateAttr(default=None)
defaults_dict: ClassVar[dict] = {}
model_config = {'extra': 'forbid'}
def set_toml_snapshot(self, snapshot: 'AppConfig'):
"""Stores a deep copy of the config state after TOML loading."""
self._toml_snapshot = snapshot
def get_toml_snapshot(self) -> Optional['AppConfig']:
"""Retrieves the stored TOML snapshot."""
return self._toml_snapshot
def get_llm_config(self, name: str = 'llm') -> LLMConfig:
"""'llm' is the name for default config (for backward compatibility prior to 0.8)."""
if name in self.llms:

View File

@@ -0,0 +1,237 @@
import traceback
from typing import Any, Optional
import tomlkit
from pydantic import BaseModel
from openhands.core import logger
from openhands.core.config.app_config import AppConfig
from openhands.core.config.utils import USER_CONFIG_DIR, USER_CONFIG_PATH
def _get_value_from_path(obj: Any, path: str) -> Any:
"""Gets a value from a potentially nested object using dot notation."""
keys = path.split('.')
value = obj
for key in keys:
if isinstance(value, BaseModel):
if hasattr(value, key):
value = getattr(value, key)
else:
# Handle nested models like llms['llama3'].model
if '.' in key and hasattr(value, key.split('.')[0]):
# This part needs refinement if we support saving nested dict items like llms
raise NotImplementedError(
f'Accessing nested dict items like {key} not fully implemented yet.'
)
raise AttributeError(
f"Attribute '{key}' not found in object path '{path}'"
)
elif isinstance(value, dict):
if key in value:
value = value[key]
else:
raise KeyError(f"Key '{key}' not found in dict path '{path}'")
else:
raise TypeError(
f'Cannot traverse path "{path}". Object is not a BaseModel or dict at key "{key}".'
)
return value
def _get_default_from_path(model: BaseModel, path: str) -> Any:
"""Gets the Pydantic default value for a field specified by dot notation."""
keys = path.split('.')
current_model = model
field = None
for i, key in enumerate(keys):
if not isinstance(current_model, BaseModel):
raise TypeError(f'Expected Pydantic model at path segment "{key}"')
field = current_model.model_fields.get(key)
if field is None:
# Handle nested models like llms['llama3'].model - needs refinement
raise AttributeError(
f"Field '{key}' not found in model path '{path}'"
)
if i < len(keys) - 1:
# If not the last key, get the nested model type
# This assumes the field annotation is the nested model type
nested_model_type = field.annotation
if hasattr(nested_model_type, 'model_fields'):
# Instantiate the nested model to access its defaults (if needed)
# This might be inefficient if defaults are complex
current_model = nested_model_type()
else:
raise TypeError(f'Field "{key}" is not a nested Pydantic model.')
else:
# Last key, get the default
return field.get_default()
# Should not be reached if path is valid
return None
def _ensure_nested_tables(doc: tomlkit.TOMLDocument, path_keys: list[str]):
"""Ensures nested tables exist in the TOML document for a given path."""
current_item = doc
for i, key in enumerate(path_keys[:-1]): # Iterate up to the second-to-last key
if key not in current_item:
current_item[key] = tomlkit.table()
logger.openhands_logger.debug(f'Created table [{key}] in user TOML.')
elif not isinstance(current_item[key], tomlkit.items.Table):
raise TypeError(
f'Expected a table at key "{key}", found {type(current_item[key])}. Cannot overwrite.'
)
current_item = current_item[key]
return current_item
def save_setting_to_user_toml(
app_config: AppConfig, setting_path: str, new_value: Any
) -> bool:
"""
Saves a specific setting to the user's config TOML file (~/.openhands/config.toml).
Args:
app_config: The current runtime AppConfig instance (containing the snapshot).
setting_path: The dot-notation path to the setting (e.g., 'llm.model', 'sandbox.timeout').
new_value: The new value to save.
Returns:
True if the value was successfully saved or removed (if set to default),
False otherwise (e.g., if overridden by env/cli or no change needed).
"""
logger.openhands_logger.info(
f"Attempting to save setting '{setting_path}' with value '{new_value}' to user TOML."
)
snapshot = app_config.get_toml_snapshot()
if snapshot is None:
logger.openhands_logger.error(
'TOML snapshot not found in AppConfig. Cannot determine setting source.'
)
return False
try:
# Get values from runtime config, snapshot, and defaults
runtime_value = _get_value_from_path(app_config, setting_path)
snapshot_value = _get_value_from_path(snapshot, setting_path)
pydantic_default = _get_default_from_path(AppConfig(), setting_path)
logger.openhands_logger.debug(f"Runtime value: {runtime_value}")
logger.openhands_logger.debug(f"Snapshot value: {snapshot_value}")
logger.openhands_logger.debug(f"Pydantic default: {pydantic_default}")
logger.openhands_logger.debug(f"New value: {new_value}")
# Decision Logic
if new_value == runtime_value:
logger.openhands_logger.info(
f"No change detected for '{setting_path}'. Skipping save."
)
return False # No change needed
if runtime_value != snapshot_value:
logger.openhands_logger.warning(
f"Setting '{setting_path}' is currently overridden by environment variable "
f"or CLI argument (runtime: {runtime_value}, TOMLs: {snapshot_value}). "
f"Change will not be persisted to user TOML file."
)
# TODO: Update runtime app_config instance if needed, even if not saving to TOML?
# This depends on desired behavior for UI changes vs env/cli overrides.
# For now, we only save if the source was TOML/default.
return False
# Ensure user config directory exists
try:
USER_CONFIG_DIR.mkdir(parents=True, exist_ok=True)
except OSError as e:
logger.openhands_logger.error(
f"Failed to create user config directory {USER_CONFIG_DIR}: {e}"
)
return False
# Load user TOML document
try:
with open(USER_CONFIG_PATH, 'r', encoding='utf-8') as f:
user_toml_doc = tomlkit.load(f)
logger.openhands_logger.debug(f"Loaded user TOML: {USER_CONFIG_PATH}")
except FileNotFoundError:
user_toml_doc = tomlkit.document()
logger.openhands_logger.debug(
f"User TOML not found, creating new document: {USER_CONFIG_PATH}"
)
except Exception as e:
logger.openhands_logger.error(
f"Failed to load user TOML file {USER_CONFIG_PATH}: {e}\n{traceback.format_exc()}"
)
return False
# Navigate and update/remove
path_keys = setting_path.split('.')
target_key = path_keys[-1]
try:
target_section = _ensure_nested_tables(user_toml_doc, path_keys)
if new_value == pydantic_default:
# Setting back to default: remove the key if it exists
if target_key in target_section:
del target_section[target_key]
logger.openhands_logger.info(
f"Setting '{setting_path}' matches default. Removed from user TOML."
)
modified = True
else:
logger.openhands_logger.info(
f"Setting '{setting_path}' matches default and is not present in user TOML. No change needed."
)
modified = False # No actual change to the file needed
else:
# Setting to a non-default value: update or add the key
target_section[target_key] = new_value
logger.openhands_logger.info(
f"Updated setting '{setting_path}' in user TOML."
)
modified = True
# Write back to file only if modified
if modified:
try:
with open(USER_CONFIG_PATH, 'w', encoding='utf-8') as f:
tomlkit.dump(user_toml_doc, f)
logger.openhands_logger.info(
f"Successfully saved user TOML: {USER_CONFIG_PATH}"
)
# TODO: Update the runtime app_config instance with the new_value
# so the change takes effect immediately without restart.
# This needs careful handling, potentially involving callbacks
# or direct access to the global config object.
return True
except Exception as e:
logger.openhands_logger.error(
f"Failed to write user TOML file {USER_CONFIG_PATH}: {e}\n{traceback.format_exc()}"
)
return False
else:
return True # Indicate success even if no file write was needed (e.g., removing a non-existent key)
except (AttributeError, KeyError, TypeError, NotImplementedError) as e:
logger.openhands_logger.error(
f"Failed to navigate or update setting '{setting_path}' in user TOML: {e}"
)
return False
except Exception as e:
logger.openhands_logger.error(
f"An unexpected error occurred during saving setting '{setting_path}': {e}\n{traceback.format_exc()}"
)
return False
except (AttributeError, KeyError, TypeError, NotImplementedError) as e:
logger.openhands_logger.error(
f"Failed to retrieve values for setting path '{setting_path}': {e}"
)
return False
except Exception as e:
logger.openhands_logger.error(
f"An unexpected error occurred processing setting '{setting_path}': {e}\n{traceback.format_exc()}"
)
return False

View File

@@ -1,5 +1,6 @@
import argparse
import os
from pathlib import Path
import pathlib
import platform
import sys
@@ -9,6 +10,7 @@ from typing import MutableMapping, get_args, get_origin
from uuid import uuid4
import toml
import tomlkit
from dotenv import load_dotenv
from pydantic import BaseModel, SecretStr, ValidationError
@@ -30,6 +32,8 @@ from openhands.storage import get_file_store
from openhands.storage.files import FileStore
JWT_SECRET = '.jwt_secret'
USER_CONFIG_DIR = Path.home() / '.openhands'
USER_CONFIG_PATH = USER_CONFIG_DIR / 'config.toml'
load_dotenv()
@@ -572,31 +576,60 @@ def parse_arguments() -> argparse.Namespace:
def load_app_config(
set_logging_levels: bool = True, config_file: str = 'config.toml'
) -> AppConfig:
"""Load the configuration from the specified config file and environment variables.
"""Load the configuration from TOML files and environment variables.
Stores a snapshot within the AppConfig instance after loading TOML files
but before environment variables are applied.
Args:
set_logging_levels: Whether to set the global variables for logging levels.
config_file: Path to the config file. Defaults to 'config.toml' in the current directory.
config_file: Path to the project config file. Defaults to 'config.toml'.
Returns:
The final AppConfig object with all layers applied (TOMLs, Env) and
the TOML snapshot stored internally.
"""
config = AppConfig()
# Load project config
logger.openhands_logger.info(f'Loading project config from: {config_file}')
load_from_toml(config, config_file)
# Load user config
logger.openhands_logger.info(f'Loading user config from: {USER_CONFIG_PATH}')
load_from_toml(config, str(USER_CONFIG_PATH))
# Create and store snapshot after TOMLs are loaded
toml_config_snapshot = config.model_copy(deep=True)
config.set_toml_snapshot(toml_config_snapshot)
logger.openhands_logger.info('Stored TOML config snapshot internally.')
# Load environment variables (overrides TOMLs)
logger.openhands_logger.info('Loading config from environment variables.')
load_from_env(config, os.environ)
# Final adjustments
finalize_config(config)
if set_logging_levels:
logger.DEBUG = config.debug
logger.DISABLE_COLOR_PRINTING = config.disable_color
logger.openhands_logger.info('Config loading complete.')
return config
def setup_config_from_args(args: argparse.Namespace) -> AppConfig:
"""Load config from toml and override with command line arguments.
"""Load config from toml/env and override with command line arguments.
The returned config object contains the TOML snapshot internally.
Common setup used by both CLI and main.py entry points.
"""
# Load base config from toml and env vars
# Load base config from toml and env vars, snapshot is stored internally
config = load_app_config(config_file=args.config_file)
# Override with command line arguments if provided
# Note: These CLI overrides affect the final 'config' but not the internal snapshot
if args.llm_config:
# if we didn't already load it, get it from the toml file
if args.llm_config not in config.llms:
@@ -606,19 +639,34 @@ def setup_config_from_args(args: argparse.Namespace) -> AppConfig:
if llm_config is None:
raise ValueError(f'Invalid toml file, cannot read {args.llm_config}')
config.set_llm_config(llm_config)
logger.openhands_logger.info(
f'Overriding LLM config with CLI argument: [{args.llm_config}]'
)
# Override default agent if provided
if args.agent_cls:
config.default_agent = args.agent_cls
logger.openhands_logger.info(
f'Overriding default agent with CLI argument: {args.agent_cls}'
)
# Set max iterations and max budget per task if provided, otherwise fall back to config values
if args.max_iterations is not None:
config.max_iterations = args.max_iterations
logger.openhands_logger.info(
f'Overriding max iterations with CLI argument: {args.max_iterations}'
)
if args.max_budget_per_task is not None:
config.max_budget_per_task = args.max_budget_per_task
logger.openhands_logger.info(
f'Overriding max budget per task with CLI argument: {args.max_budget_per_task}'
)
# Read selected repository in config for use by CLI and main.py
if args.selected_repo is not None:
config.sandbox.selected_repo = args.selected_repo
logger.openhands_logger.info(
f'Setting selected repo from CLI argument: {args.selected_repo}'
)
return config

View File

@@ -1,23 +1,15 @@
from __future__ import annotations
from pydantic import (
BaseModel,
Field,
SecretStr,
SerializationInfo,
field_serializer,
model_validator,
)
from pydantic.json import pydantic_encoder
from pydantic import BaseModel
from openhands.core.config.llm_config import LLMConfig
from openhands.core.config.utils import load_app_config
from openhands.storage.data_models.user_secrets import UserSecrets
class Settings(BaseModel):
"""
Persisted settings for OpenHands sessions
Persisted settings for OpenHands sessions.
Secrets are handled separately by the SecretsStore.
"""
language: str | None = None
@@ -26,11 +18,8 @@ class Settings(BaseModel):
security_analyzer: str | None = None
confirmation_mode: bool | None = None
llm_model: str | None = None
llm_api_key: SecretStr | None = None
llm_base_url: str | None = None
remote_runtime_resource_factor: int | None = None
# Planned to be removed from settings
secrets_store: UserSecrets = Field(default_factory=UserSecrets, frozen=True)
enable_default_condenser: bool = True
enable_sound_notifications: bool = False
user_consents_to_analytics: bool | None = None
@@ -41,70 +30,10 @@ class Settings(BaseModel):
'validate_assignment': True,
}
@field_serializer('llm_api_key')
def llm_api_key_serializer(self, llm_api_key: SecretStr, info: SerializationInfo):
"""Custom serializer for the LLM API key.
To serialize the API key instead of ********, set expose_secrets to True in the serialization context.
"""
context = info.context
if context and context.get('expose_secrets', False):
return llm_api_key.get_secret_value()
return pydantic_encoder(llm_api_key) if llm_api_key else None
@model_validator(mode='before')
@classmethod
def convert_provider_tokens(cls, data: dict | object) -> dict | object:
"""Convert provider tokens from JSON format to UserSecrets format."""
if not isinstance(data, dict):
return data
secrets_store = data.get('secrets_store')
if not isinstance(secrets_store, dict):
return data
custom_secrets = secrets_store.get('custom_secrets')
tokens = secrets_store.get('provider_tokens')
secret_store = UserSecrets(provider_tokens={}, custom_secrets={})
if isinstance(tokens, dict):
converted_store = UserSecrets(provider_tokens=tokens)
secret_store = secret_store.model_copy(
update={'provider_tokens': converted_store.provider_tokens}
)
else:
secret_store.model_copy(update={'provider_tokens': tokens})
if isinstance(custom_secrets, dict):
converted_store = UserSecrets(custom_secrets=custom_secrets)
secret_store = secret_store.model_copy(
update={'custom_secrets': converted_store.custom_secrets}
)
else:
secret_store = secret_store.model_copy(
update={'custom_secrets': custom_secrets}
)
data['secret_store'] = secret_store
return data
@field_serializer('secrets_store')
def secrets_store_serializer(self, secrets: UserSecrets, info: SerializationInfo):
"""Custom serializer for secrets store."""
"""Force invalidate secret store"""
return {
'provider_tokens': {}
}
@staticmethod
def from_config() -> Settings | None:
app_config = load_app_config()
llm_config: LLMConfig = app_config.get_llm_config()
if llm_config.api_key is None:
# If no api key has been set, we take this to mean that there is no reasonable default
return None
security = app_config.security
settings = Settings(
language='en',
@@ -113,7 +42,6 @@ class Settings(BaseModel):
security_analyzer=security.security_analyzer,
confirmation_mode=security.confirmation_mode,
llm_model=llm_config.model,
llm_api_key=llm_config.api_key,
llm_base_url=llm_config.base_url,
remote_runtime_resource_factor=app_config.sandbox.remote_runtime_resource_factor,
)

View File

@@ -1,37 +1,129 @@
from __future__ import annotations
import json
from dataclasses import dataclass
import traceback
from dataclasses import dataclass, field
from typing import Any
from pydantic import SecretStr
from openhands.core import logger
from openhands.core.config.app_config import AppConfig
from openhands.core.config.config_save import (
_get_value_from_path, # Use helper to read from AppConfig
save_setting_to_user_toml,
)
from openhands.server.settings import Settings
from openhands.storage import get_file_store
from openhands.storage.data_models.settings import Settings
from openhands.storage.files import FileStore
from openhands.storage.settings.settings_store import SettingsStore
from openhands.utils.async_utils import call_sync_from_async
# Maps fields from the Settings model (used by UI/API) to AppConfig paths for user TOML saving
SETTINGS_TO_APPCONFIG_MAP = {
'agent': 'default_agent',
'max_iterations': 'max_iterations',
'security_analyzer': 'security.security_analyzer',
'confirmation_mode': 'security.confirmation_mode',
'llm_model': 'llm.model', # Assumes default [llm] section
'llm_base_url': 'llm.base_url', # Assumes default [llm] section
'enable_default_condenser': 'enable_default_condenser',
'sandbox_base_container_image': 'sandbox.base_container_image',
'sandbox_runtime_container_image': 'sandbox.runtime_container_image',
# Fields NOT mapped (likely UI/Server specific or need separate handling):
# language, enable_sound_notifications, user_consents_to_analytics, remote_runtime_resource_factor
# llm_api_key - moved to secrets store
}
@dataclass
class FileSettingsStore(SettingsStore):
file_store: FileStore
path: str = 'settings.json'
"""
Stores and retrieves settings by writing to/reading from the user's
~/.openhands/config.toml file.
Requires the runtime AppConfig instance to access the TOML snapshot
for safe saving and to reconstruct the Settings object.
"""
app_config: AppConfig = field(init=False) # Set by get_instance
async def load(self) -> Settings | None:
"""
Constructs a Settings object based on the current runtime AppConfig.
Does NOT read the user TOML file directly here, as that's part of the initial load.
"""
logger.openhands_logger.info("Loading settings from runtime AppConfig for FileSettingsStore")
if not hasattr(self, 'app_config'):
logger.openhands_logger.error("AppConfig not set in FileSettingsStore instance. Cannot load settings.")
return None
settings_data = {}
try:
json_str = await call_sync_from_async(self.file_store.read, self.path)
kwargs = json.loads(json_str)
settings = Settings(**kwargs)
# Populate settings_data using the mapping
for settings_key, appconfig_path in SETTINGS_TO_APPCONFIG_MAP.items():
try:
# Use helper to safely get potentially nested values
settings_data[settings_key] = _get_value_from_path(self.app_config, appconfig_path)
except (AttributeError, KeyError, TypeError, NotImplementedError):
logger.openhands_logger.debug(f"Path '{appconfig_path}' not found in AppConfig for setting '{settings_key}'. Skipping.")
settings_data[settings_key] = None # Or use Settings model default?
# Instantiate the Settings object
settings = Settings(**settings_data)
return settings
except FileNotFoundError:
except Exception as e:
logger.openhands_logger.error(f"Unexpected error loading settings from AppConfig: {e}\n{traceback.format_exc()}")
return None
async def store(self, settings: Settings) -> None:
json_str = settings.model_dump_json(context={'expose_secrets': True})
await call_sync_from_async(self.file_store.write, self.path, json_str)
"""
Saves relevant fields from the Settings object to the user's config TOML file.
Uses the save_setting_to_user_toml function which handles snapshot comparison.
"""
logger.openhands_logger.info("Storing settings via FileSettingsStore to user TOML")
if not hasattr(self, 'app_config'):
logger.openhands_logger.error("AppConfig not set in FileSettingsStore instance. Cannot store settings.")
return
tasks = []
# Use model_dump to iterate through fields present in the input 'settings' object
settings_dict = settings.model_dump(exclude_none=True)
for field_name, value in settings_dict.items():
if field_name in SETTINGS_TO_APPCONFIG_MAP:
setting_path = SETTINGS_TO_APPCONFIG_MAP[field_name]
tasks.append(
call_sync_from_async(
save_setting_to_user_toml,
self.app_config,
setting_path,
value
)
)
else:
logger.openhands_logger.debug(f"No AppConfig path mapping found for setting '{field_name}'. Skipping save.")
# Execute all save tasks (could potentially run in parallel if save_setting is thread-safe)
# For simplicity, run sequentially for now.
for task in tasks:
try:
await task
except Exception as e:
# Error is logged within save_setting_to_user_toml, but maybe log context here too
logger.openhands_logger.error(f"Error occurred during async save task: {e}")
@classmethod
async def get_instance(
cls, config: AppConfig, user_id: str | None
cls, config: AppConfig, user_id: str | None # user_id currently unused by FileSettingsStore
) -> FileSettingsStore:
file_store = get_file_store(config.file_store, config.file_store_path)
return FileSettingsStore(file_store)
# Removed file_store dependency as we write directly via tomlkit now
instance = cls()
instance.app_config = config # Store the runtime config instance
logger.openhands_logger.debug("Created FileSettingsStore instance")
return instance

View File

@@ -24,6 +24,7 @@ termcolor = "*"
docker = "*"
fastapi = "*"
toml = "*"
tomlkit = "*"
uvicorn = "*"
types-toml = "*"
numpy = "*"

View File

@@ -0,0 +1,314 @@
from pathlib import Path
from unittest import mock
import pytest
import tomlkit
# Assuming AppConfig and other necessary classes/functions are importable
# Adjust imports based on actual project structure if needed
from openhands.core.config.app_config import AppConfig
from openhands.core.config.config_save import save_setting_to_user_toml
from openhands.core.config.llm_config import LLMConfig
from openhands.core.config.utils import (
load_app_config,
)
# Mock USER_CONFIG_PATH before it's used at module level if necessary,
# or ensure tests mock it appropriately using fixtures like tmp_path.
# Fixture to provide a clean AppConfig instance for tests
@pytest.fixture
def base_app_config():
return AppConfig()
# Fixture to provide a clean snapshot (can be modified per test)
@pytest.fixture
def base_snapshot(base_app_config):
# In a real scenario, this would be populated after loading TOMLs
# For testing save logic, we often set specific states
return base_app_config.model_copy(deep=True)
# Fixture to mock Path.home() to use tmp_path
@pytest.fixture(autouse=True)
def mock_home_dir(tmp_path, monkeypatch):
"""Ensure USER_CONFIG_DIR/PATH point to tmp_path for test isolation."""
mock_home = tmp_path / 'mock_home'
mock_user_config_dir = mock_home / '.openhands'
mock_user_config_path = mock_user_config_dir / 'config.toml'
monkeypatch.setattr(Path, 'home', lambda: mock_home)
# Important: Patch the constants directly where they are defined or used
monkeypatch.setattr(
'openhands.core.config.utils.USER_CONFIG_DIR', mock_user_config_dir
)
monkeypatch.setattr(
'openhands.core.config.utils.USER_CONFIG_PATH', mock_user_config_path
)
monkeypatch.setattr(
'openhands.core.config.config_save.USER_CONFIG_DIR', mock_user_config_dir
)
monkeypatch.setattr(
'openhands.core.config.config_save.USER_CONFIG_PATH', mock_user_config_path
)
# Ensure the mocked directory exists for tests that write
# mock_user_config_dir.mkdir(parents=True, exist_ok=True) # Let the save function create it
return mock_user_config_path # Return the path for convenience
# =============================================
# Tests for save_setting_to_user_toml
# =============================================
def test_save_simple_value_success(mock_home_dir, base_app_config, base_snapshot):
"""Test saving a basic value that should be persisted."""
# Arrange
user_toml_path = mock_home_dir
setting_path = 'llm.model'
new_value = 'new-model-123'
expected_toml_content = f'[llm]\nmodel = "{new_value}"\n'
# Ensure runtime and snapshot values allow saving
# Let's assume the default model is different
base_app_config.llm.model = 'old-model' # Simulate current runtime value from TOML
base_snapshot.llm.model = 'old-model' # Simulate snapshot value from TOML
base_app_config.set_toml_snapshot(base_snapshot) # Set the snapshot
# Act
saved = save_setting_to_user_toml(base_app_config, setting_path, new_value)
# Assert
assert saved is True
assert user_toml_path.exists()
content = user_toml_path.read_text()
assert content == expected_toml_content
def test_save_value_matches_runtime_no_change(
mock_home_dir, base_app_config, base_snapshot
):
"""Test that saving is skipped if the new value matches the current runtime value."""
# Arrange
user_toml_path = mock_home_dir
setting_path = 'llm.model'
current_value = 'current-model-456'
new_value = current_value # Same as runtime
base_app_config.llm.model = current_value
base_snapshot.llm.model = 'snapshot-model' # Snapshot differs, but shouldn't matter
base_app_config.set_toml_snapshot(base_snapshot)
# Act
saved = save_setting_to_user_toml(base_app_config, setting_path, new_value)
# Assert
assert saved is False
assert not user_toml_path.exists() # File shouldn't be created if no change
def test_save_value_overridden_by_env_cli(
mock_home_dir, base_app_config, base_snapshot
):
"""Test that saving is skipped if the runtime value was overridden by env/cli."""
# Arrange
user_toml_path = mock_home_dir
setting_path = 'llm.model'
new_value = 'user-wants-this'
runtime_value = 'env-or-cli-set-this'
snapshot_value = 'toml-value' # Different from runtime
base_app_config.llm.model = runtime_value # Simulate override
base_snapshot.llm.model = snapshot_value
base_app_config.set_toml_snapshot(base_snapshot)
# Act
saved = save_setting_to_user_toml(base_app_config, setting_path, new_value)
# Assert
assert saved is False
assert not user_toml_path.exists() # File shouldn't be created
def test_save_value_set_to_default_removes_key(
mock_home_dir, base_app_config, base_snapshot
):
"""Test that setting a value back to its default removes it from the user TOML."""
# Arrange
user_toml_path = mock_home_dir
setting_path = 'llm.temperature'
# Assume default temperature is 0.0 from LLMConfig definition
default_value = LLMConfig().temperature
new_value = default_value
# Pre-populate user TOML with a non-default value
initial_toml_content = '[llm]\ntemperature = 0.7\n'
user_toml_path.parent.mkdir(parents=True, exist_ok=True)
user_toml_path.write_text(initial_toml_content)
# Set runtime and snapshot to reflect the non-default value initially
base_app_config.llm.temperature = 0.7
base_snapshot.llm.temperature = 0.7
base_app_config.set_toml_snapshot(base_snapshot)
# Act
saved = save_setting_to_user_toml(base_app_config, setting_path, new_value)
# Assert
assert saved is True
assert user_toml_path.exists()
content = user_toml_path.read_text()
# The key should be removed, leaving an empty or non-existent [llm] section
# tomlkit might leave the section header if other keys exist, or remove it if empty.
# Let's check if the specific key is gone.
doc = tomlkit.parse(content)
assert 'llm' not in doc or 'temperature' not in doc['llm']
def test_save_nested_value_creates_tables(
mock_home_dir, base_app_config, base_snapshot
):
"""Test saving a nested value creates necessary tables."""
# Arrange
user_toml_path = mock_home_dir
setting_path = 'sandbox.timeout'
new_value = 120
expected_toml_content = f'[sandbox]\ntimeout = {new_value}\n'
# Ensure runtime and snapshot allow saving (assume default is different)
base_app_config.sandbox.timeout = 60 # Simulate current runtime from TOML
base_snapshot.sandbox.timeout = 60 # Simulate snapshot from TOML
base_app_config.set_toml_snapshot(base_snapshot)
# Act
saved = save_setting_to_user_toml(base_app_config, setting_path, new_value)
# Assert
assert saved is True
assert user_toml_path.exists()
content = user_toml_path.read_text()
assert content == expected_toml_content
def test_save_preserves_comments_and_other_values(
mock_home_dir, base_app_config, base_snapshot
):
"""Test that saving preserves existing comments and unrelated values."""
# Arrange
user_toml_path = mock_home_dir
setting_path = 'llm.model'
new_value = 'new-model-for-test'
initial_toml_content = """
# This is a comment
[core]
max_iterations = 100 # Another comment
[llm]
model = "old-model" # Will be changed
api_key = "keep-this-key"
"""
expected_toml_content = f"""
# This is a comment
[core]
max_iterations = 100 # Another comment
[llm]
model = "{new_value}" # Will be changed
api_key = "keep-this-key"
"""
user_toml_path.parent.mkdir(parents=True, exist_ok=True)
user_toml_path.write_text(initial_toml_content)
# Set runtime and snapshot to allow saving
base_app_config.llm.model = 'old-model'
base_snapshot.llm.model = 'old-model'
# Set other values to match initial TOML for simplicity in this test
base_app_config.max_iterations = 100
base_snapshot.max_iterations = 100
base_app_config.llm.api_key = 'keep-this-key'
base_snapshot.llm.api_key = 'keep-this-key'
base_app_config.set_toml_snapshot(base_snapshot)
# Act
saved = save_setting_to_user_toml(base_app_config, setting_path, new_value)
# Assert
assert saved is True
content_after_save = user_toml_path.read_text()
# Parse both to ignore minor whitespace differences from tomlkit
doc_expected = tomlkit.parse(expected_toml_content)
doc_actual = tomlkit.parse(content_after_save)
assert doc_actual.as_string() == doc_expected.as_string()
# =============================================
# Tests for load_app_config snapshot creation
# =============================================
# Mock load_from_toml and load_from_env to isolate snapshot creation
@mock.patch('openhands.core.config.utils.load_from_env')
@mock.patch('openhands.core.config.utils.load_from_toml')
def test_load_app_config_creates_snapshot(mock_load_toml, mock_load_env, mock_home_dir):
"""Verify snapshot is taken after TOML loads and before ENV load."""
# Arrange
project_config_path = 'config.toml' # Default path used by load_app_config
user_config_path_str = str(mock_home_dir) # Mocked user path
# Define side effects for mocked load_from_toml
# First call (project): sets model='project-model'
# Second call (user): sets model='user-model', timeout=100
# Snapshot should capture model='user-model', timeout=100
# Env call: sets model='env-model'
def load_toml_side_effect(config, path):
if path == project_config_path:
config.llm.model = 'project-model'
print(f'Mock load_from_toml({path}): set model=project-model')
elif path == user_config_path_str:
config.llm.model = 'user-model' # Overrides project
config.sandbox.timeout = 100
print(f'Mock load_from_toml({path}): set model=user-model, timeout=100')
else:
print(f'Mock load_from_toml({path}): unexpected path')
def load_env_side_effect(config, env):
config.llm.model = 'env-model' # Overrides user
config.max_iterations = 50 # New value from env
print('Mock load_from_env: set model=env-model, max_iterations=50')
mock_load_toml.side_effect = load_toml_side_effect
mock_load_env.side_effect = load_env_side_effect
# Act
final_config = load_app_config(config_file=project_config_path)
snapshot = final_config.get_toml_snapshot()
# Assert
assert snapshot is not None
# Snapshot should reflect state after user TOML load
assert snapshot.llm.model == 'user-model'
assert snapshot.sandbox.timeout == 100
# max_iterations wasn't set by TOML, should have default in snapshot
assert snapshot.max_iterations == AppConfig().max_iterations
# Final config should reflect state after ENV load
assert final_config.llm.model == 'env-model'
assert final_config.sandbox.timeout == 100 # Unchanged by env
assert final_config.max_iterations == 50 # Changed by env
# Check mock calls order (optional but good)
assert mock_load_toml.call_count == 2
mock_load_env.assert_called_once()
# TODO: Add more tests:
# - Test save_setting_to_user_toml with invalid setting_path
# - Test save_setting_to_user_toml error handling for file I/O
# - Test _get_value_from_path and _get_default_from_path helpers directly? (Maybe less critical if covered by save tests)
# - Test setup_config_from_args correctly passes snapshot through (if needed later)

View File

@@ -1,13 +1,23 @@
from unittest.mock import patch
from unittest.mock import call, mock, patch
import pytest
from pydantic import SecretStr
from pytest import mark
from openhands.core.config.app_config import AppConfig
from openhands.core.config.config_save import save_setting_to_user_toml
from openhands.core.config.llm_config import LLMConfig
from openhands.core.config.sandbox_config import SandboxConfig
from openhands.core.config.security_config import SecurityConfig
from openhands.server.routes.settings import convert_to_settings
from openhands.server.settings import Settings
from openhands.storage.data_models.settings import Settings
from openhands.storage.settings.file_settings_store import (
SECRETS_STORE_TO_APPCONFIG_MAP,
SETTINGS_TO_APPCONFIG_MAP,
FileSettingsStore,
)
from openhands.utils.async_utils import call_sync_from_async
def test_settings_from_config():
@@ -88,6 +98,151 @@ def test_settings_handles_sensitive_data():
assert settings.llm_api_key.get_secret_value() == 'test-key'
# =============================================
# Tests for FileSettingsStore (New)
# =============================================
@pytest.fixture
def mock_app_config():
"""Provides a mock AppConfig instance for testing FileSettingsStore."""
# Create a base config with some values
config = AppConfig(
default_agent='CodeActAgent',
max_iterations=50,
llms={
'llm': LLMConfig(
model='gpt-4',
api_key=SecretStr('existing_key'),
base_url='existing_url',
)
},
security=SecurityConfig(confirmation_mode=False),
sandbox=SandboxConfig(base_container_image='image1'),
)
# Set a dummy snapshot (tests for save logic are in test_config_write.py)
config.set_toml_snapshot(config.model_copy(deep=True))
return config
@pytest.fixture
async def file_settings_store_instance(mock_app_config):
"""Provides an initialized FileSettingsStore instance."""
# Note: user_id is currently unused by FileSettingsStore.get_instance
return await FileSettingsStore.get_instance(mock_app_config, user_id='test_user')
@mark.asyncio
async def test_file_settings_store_load(file_settings_store_instance, mock_app_config):
"""Test loading settings reconstructs Settings from AppConfig."""
# Arrange (AppConfig is already set in the fixture)
# Act
loaded_settings = await file_settings_store_instance.load()
# Assert
assert loaded_settings is not None
assert loaded_settings.agent == mock_app_config.default_agent
assert loaded_settings.max_iterations == mock_app_config.max_iterations
assert loaded_settings.llm_model == mock_app_config.llm.model
assert loaded_settings.llm_api_key == mock_app_config.llm.api_key
assert loaded_settings.llm_base_url == mock_app_config.llm.base_url
assert (
loaded_settings.confirmation_mode == mock_app_config.security.confirmation_mode
)
assert (
loaded_settings.sandbox_base_container_image
== mock_app_config.sandbox.base_container_image
)
# Check a field not present in mock_app_config (should be None or default)
assert loaded_settings.security_analyzer is None # Assuming it wasn't set
# Secrets store is currently basic in load
assert isinstance(loaded_settings.secrets_store, SecretStore)
@mark.asyncio
@patch('openhands.storage.settings.file_settings_store.save_setting_to_user_toml')
async def test_file_settings_store_store_general_settings(
mock_save_setting, file_settings_store_instance, mock_app_config
):
"""Test storing general settings calls save_setting_to_user_toml correctly."""
# Arrange
settings_to_store = Settings(
agent='NewAgent', # Mapped
llm_model='new-model', # Mapped
llm_api_key=SecretStr('new-key'), # Mapped
language='fr', # Not mapped
secrets_store=SecretStore(), # Excluded from general loop
)
# Act
await file_settings_store_instance.store(settings_to_store)
# Assert
expected_calls = [
call(mock_app_config, SETTINGS_TO_APPCONFIG_MAP['agent'], 'NewAgent'),
call(mock_app_config, SETTINGS_TO_APPCONFIG_MAP['llm_model'], 'new-model'),
call(
mock_app_config,
SETTINGS_TO_APPCONFIG_MAP['llm_api_key'],
'new-key', # Raw string
),
# Note: language is not mapped, so it shouldn't be called
]
# We need to wrap the sync function call for mocking async context
call_sync_from_async(save_setting_to_user_toml)
# Check calls made via call_sync_from_async
# This requires inspecting the arguments passed to call_sync_from_async
# A direct mock on save_setting_to_user_toml might be simpler if call_sync_from_async is reliable
# Let's assume direct mock works for simplicity here, adjust if needed based on execution flow.
mock_save_setting.assert_has_calls(expected_calls, any_order=True)
assert mock_save_setting.call_count == 3 # Only mapped fields
@mark.asyncio
@patch('openhands.storage.settings.file_settings_store.save_setting_to_user_toml')
async def test_file_settings_store_store_secrets(
mock_save_setting, file_settings_store_instance, mock_app_config
):
"""Test storing secrets calls save_setting_to_user_toml correctly."""
# Arrange
github_token = 'ghp_test123'
custom_key = 'custom_value_abc'
custom_secret_name = 'MY_CUSTOM_API_KEY'
settings_to_store = Settings(
secrets_store=SecretStore(
provider_tokens={
ProviderType.GITHUB: ProviderToken(token=SecretStr(github_token))
},
custom_secrets={custom_secret_name: SecretStr(custom_key)},
)
)
# Act
await file_settings_store_instance.store(settings_to_store)
# Assert
token_base_path = SECRETS_STORE_TO_APPCONFIG_MAP['provider_tokens']
custom_base_path = SECRETS_STORE_TO_APPCONFIG_MAP['custom_secrets']
expected_calls = [
mock.call(
mock_app_config,
f'{token_base_path}.{ProviderType.GITHUB.value}',
github_token, # Raw string
),
mock.call(
mock_app_config,
f'{custom_base_path}.{custom_secret_name}',
custom_key, # Raw string
),
]
mock_save_setting.assert_has_calls(expected_calls, any_order=True)
assert mock_save_setting.call_count == 2 # Only secrets in this test case
def test_convert_to_settings():
settings_with_token_data = Settings(
llm_api_key='test-key',