mirror of
https://github.com/All-Hands-AI/OpenHands.git
synced 2026-04-29 03:00:45 -04:00
Compare commits
5 Commits
fix/git-di
...
enyst/writ
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7b305d8e9d | ||
|
|
a0a2fdfddb | ||
|
|
75428c1060 | ||
|
|
70c013e46f | ||
|
|
6e575277bd |
40
.openhands/microagents/config_write_task.md
Normal file
40
.openhands/microagents/config_write_task.md
Normal 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
|
||||
|
||||
|
||||
@@ -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:
|
||||
|
||||
237
openhands/core/config/config_save.py
Normal file
237
openhands/core/config/config_save.py
Normal 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
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -24,6 +24,7 @@ termcolor = "*"
|
||||
docker = "*"
|
||||
fastapi = "*"
|
||||
toml = "*"
|
||||
tomlkit = "*"
|
||||
uvicorn = "*"
|
||||
types-toml = "*"
|
||||
numpy = "*"
|
||||
|
||||
314
tests/unit/test_config_write.py
Normal file
314
tests/unit/test_config_write.py
Normal 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)
|
||||
@@ -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',
|
||||
|
||||
Reference in New Issue
Block a user