yaoyao/integrations/hermes-bots-bridge/tests/test_model_settings.py

import importlib.util
import json
from contextlib import contextmanager, nullcontext
from contextvars import ContextVar
from concurrent.futures import ThreadPoolExecutor
from threading import Barrier
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock, patch
import unittest

spec = importlib.util.spec_from_file_location('bot_model_settings_test', Path(__file__).parents[1] / 'model_settings.py')
models = importlib.util.module_from_spec(spec)
spec.loader.exec_module(models)


class BotModelSettingsTests(unittest.TestCase):
    def setUp(self):
        self.config = {"agent": {"reasoning_effort": "medium", "service_tier": "auto"}, "model": {"api_key": "fixture-secret"}}
        self.ctx = SimpleNamespace(current_provider='openai', current_model='model-a', current_base_url='https://api.openai.com/v1', user_providers={}, custom_providers=[])
        self.switch = Mock(side_effect=lambda **kw: SimpleNamespace(success=True, target_provider=kw['explicit_provider'], new_model=kw['raw_input'], base_url=self.ctx.current_base_url, api_key='fixture-secret', model_info=None))
        self.warning = Mock(return_value=None)
        self.meta = Mock(return_value=SimpleNamespace(supports_reasoning=True))
        self.fast = Mock(side_effect=lambda model, **kw: {'service_tier': 'priority'} if kw['base_url'].startswith('https://api.openai.com/') else None)
        self.reader = Mock(return_value=None)
        modules = {
            'hermes_constants': SimpleNamespace(VALID_REASONING_EFFORTS=['low','medium','high','ultra'], parse_reasoning_effort=lambda value: {} if value in ['none','low','medium','high','ultra'] else None),
            'hermes_cli.config': SimpleNamespace(load_config=lambda: self.config),
            'hermes_cli.inventory': SimpleNamespace(load_picker_context=lambda: self.ctx, _reasoning_catalog_reader=lambda _: self.reader,
                build_model_options_payload=lambda *a, **kw: {'providers': [{'slug':'openai','models':['model-a'], 'capabilities': {'model-a': {'reasoning': True}}}]}),
            'hermes_cli.models': SimpleNamespace(resolve_fast_mode_overrides=self.fast, github_model_reasoning_efforts=lambda _: ['low','high']),
            'hermes_cli.model_switch': SimpleNamespace(switch_model=self.switch),
            'hermes_cli.model_selection_guards': SimpleNamespace(combined_selection_warning=self.warning),
            'agent.models_dev': SimpleNamespace(get_model_capabilities=self.meta),
        }
        self.scoped = patch.object(models, 'profile_scope', side_effect=lambda _: nullcontext())
        self.modules = patch.dict('sys.modules', modules)
        self.scoped.start(); self.modules.start()
        self.addCleanup(self.scoped.stop); self.addCleanup(self.modules.stop)

    def native_scope_fixture(self):
        self.scoped.stop()
        home = ContextVar('test_model_home', default='caller')
        secret = ContextVar('test_model_secret', default=None)
        calls = []

        @contextmanager
        def scope(profile):
            calls.append(profile)
            home_token = home.set(profile)
            secret_token = secret.set(profile + '-credential')
            try:
                yield
            finally:
                secret.reset(secret_token)
                home.reset(home_token)

        def picker():
            if secret.get() != home.get() + '-credential':
                raise RuntimeError('UnscopedSecretError: missing Profile secret scope')
            return self.ctx

        import sys
        constants = sys.modules['hermes_constants']
        constants.set_hermes_home_override = home.set
        constants.reset_hermes_home_override = home.reset
        sys.modules['hermes_cli.inventory'].load_picker_context = picker
        modules = {
            'hermes_cli.profiles': SimpleNamespace(profile_exists=lambda p: p in ('default', 'writer'), get_profile_dir=lambda p: p),
            'hermes_cli.web_server_profiles': SimpleNamespace(_config_profile_scope=scope),
        }
        patcher = patch.dict('sys.modules', modules)
        patcher.start(); self.addCleanup(patcher.stop)
        return home, secret, calls

    def test_catalogue_and_resolution_bind_profile_secrets_in_multiplex_mode(self):
        home, secret, calls = self.native_scope_fixture()
        for profile in ('default', 'writer'):
            self.assertEqual(models.options(profile)['version'], 1)
            self.assertEqual(models.resolve(profile, None)['version'], 1)
        self.assertEqual(calls, ['default', 'default', 'writer', 'writer'])
        self.assertEqual(home.get(), 'caller')
        self.assertIsNone(secret.get())

    def test_scopes_are_thread_local_and_restore_the_caller_even_on_error(self):
        home, secret, _ = self.native_scope_fixture()
        barrier = Barrier(2)

        def read(profile):
            with models.profile_scope(profile):
                barrier.wait(timeout=5)
                self.assertEqual(home.get(), profile)
                self.assertEqual(secret.get(), profile + '-credential')
                with models.profile_scope('default'):
                    self.assertEqual(secret.get(), 'default-credential')
                self.assertEqual(secret.get(), profile + '-credential')
            return home.get(), secret.get()

        with ThreadPoolExecutor(max_workers=2) as pool:
            self.assertEqual(list(pool.map(read, ('default', 'writer'))), [('caller', None), ('caller', None)])
        with self.assertRaisesRegex(ValueError, 'fixture failure'):
            with models.profile_scope('writer'):
                raise ValueError('fixture failure')
        self.assertEqual(home.get(), 'caller')
        self.assertIsNone(secret.get())

    def test_invalid_profiles_never_enter_a_credential_scope(self):
        _, _, calls = self.native_scope_fixture()
        for profile in ('../writer', 'missing', None, ''):
            with self.subTest(profile=profile), self.assertRaises(models.ModelSettingsError):
                models.options(profile)
        self.assertEqual(calls, [])

    def test_older_hermes_uses_its_native_scope_without_home_only_fallback(self):
        home, secret, calls = self.native_scope_fixture()
        import sys
        native = sys.modules['hermes_cli.web_server_profiles']
        with patch.dict('sys.modules', {'hermes_cli.web_server_profiles': None, 'hermes_cli.web_server': native}):
            self.assertEqual(models.options('writer')['version'], 1)
        self.assertEqual(calls, ['writer'])
        self.assertEqual(home.get(), 'caller')
        self.assertIsNone(secret.get())

    def test_catalogue_preserves_defaults_modes_and_does_not_disclose_credentials(self):
        value = models.options('default')
        self.assertEqual(value['defaults']['fastMode'], 'auto')
        self.assertEqual(value['models'][0]['fastModes'], ['normal','fast','auto','cold'])
        self.assertNotIn('fixture-secret', json.dumps(value))

    def test_inheritance_and_explicit_none_are_distinct_and_resolution_never_writes(self):
        value = models.resolve('default', {'provider': None, 'model': None, 'reasoningEffort': 'none', 'fastMode': 'cold'})
        self.assertEqual(value['effective'], {'provider':'openai','model':'model-a','reasoningEffort':'none','fastMode':'cold'})
        self.assertFalse(self.switch.call_args.kwargs['is_global'])
        self.assertEqual(self.switch.call_args.kwargs['current_api_key'], 'fixture-secret')
        self.assertNotIn('fixture-secret', json.dumps(value))
        self.config['agent'].update(reasoning_effort=False, service_tier='normal')
        self.assertEqual(models.resolve('default', None)['effective']['reasoningEffort'], 'none')

    def test_proxy_route_rejects_all_accelerated_modes(self):
        self.ctx.current_base_url = 'https://proxy.invalid/v1'
        for speed in ['fast','auto','cold']:
            with self.subTest(speed=speed), self.assertRaises(models.ModelSettingsError):
                models.resolve('default', {'fastMode': speed})
        self.assertEqual(models.resolve('default', {'fastMode':'normal'})['effective']['fastMode'], 'normal')

    def test_copilot_and_mandatory_reasoning_restrict_levels(self):
        with self.assertRaises(models.ModelSettingsError):
            models.resolve('default', {'provider':'copilot','model':'model-a','reasoningEffort':'ultra'})
        self.reader.return_value = {'supports_reasoning':True,'mandatory':True}
        with self.assertRaises(models.ModelSettingsError):
            models.resolve('default', {'reasoningEffort':'none'})

    def test_unknown_capabilities_remain_unknown(self):
        self.meta.return_value = None
        value = models.options('default')['models'][0]
        self.assertFalse(value['reasoningKnown'])
        self.assertIn('ultra', value['reasoningEfforts'])

    def test_invalid_pair_and_switch_failure_stop_resolution_without_secrets(self):
        for settings in [{'model':'model-b'}, {'provider':'openai','model':'bad --global'}, {'fastMode':'turbo'}, {'config':{}}]:
            with self.subTest(settings=settings), self.assertRaises(models.ModelSettingsError): models.resolve('default',settings)
        self.switch.side_effect = None
        self.switch.return_value = SimpleNamespace(success=False, error_message='fixture-secret')
        with self.assertRaises(models.ModelSettingsError) as exc: models.resolve('default',None)
        self.assertNotIn('fixture-secret', str(exc.exception))

    def test_confirmation_is_returned_without_applying_any_change(self):
        self.warning.return_value = SimpleNamespace(message='模型费用较高')
        self.assertEqual(models.resolve('default',{})['confirmationMessage'], '模型费用较高')


if __name__ == '__main__': unittest.main()