"""Provider credentials and model selections reach mocked inference calls.""" import asyncio import os import tempfile import unittest from pathlib import Path from unittest.mock import AsyncMock, Mock, patch os.environ.setdefault('CONFIG_DIR', tempfile.mkdtemp(prefix='ttvc-engine-tests-')) from fastapi import HTTPException from routes import docker, stt, settings as settings_routes from core import tts_helpers from core.provider_health import provider_key_for_url, provider_request_params, validate_provider_params class EngineConnectionTests(unittest.TestCase): def test_extra_params_are_role_scoped_and_required_fields_protected(self): settings = {'custom_engine_cards': [ {'id': 'chat', 'url': 'http://gateway/v1', 'role': 'llm', 'params': {'temperature': 0.7}}, {'id': 'speech', 'url': 'http://gateway/v1', 'role': 'tts', 'params': {'speed': 1.2}}, ]} payload = {'model': 'chosen', 'messages': [], 'temperature': 0.1} merged = provider_request_params('http://gateway/v1/chat/completions', payload, settings) self.assertEqual(merged['temperature'], 0.7) self.assertNotIn('speed', merged) self.assertEqual(merged['model'], 'chosen') settings['custom_engine_cards'].append({'id': 'other-chat', 'url': 'http://gateway/v1', 'role': 'llm', 'params': {'top_p': 0.1}}) self.assertEqual(provider_request_params('http://gateway/v1/chat/completions', payload, settings), payload) def test_params_follow_saved_endpoint_override(self): settings = {'custom_engine_cards': [{'id': 'card', 'url': 'http://old', 'role': 'tts', 'params': {'speed': 1.2}}], 'engine_local_urls': {'dc-card': 'http://new/v1'}} self.assertEqual(provider_request_params('http://new/v1/audio/speech', {'input': 'hello'}, settings)['speed'], 1.2) self.assertEqual(provider_request_params('http://old/v1/audio/speech', {'input': 'hello'}, settings), {'input': 'hello'}) @patch('routes.settings._save_settings') def test_invalid_or_secret_extra_params_are_rejected_before_save(self, save): invalid = [[], {'model': 'replace'}, {'headers': {'x-api-key': 'dummy'}}, {'oauth': {'client_secret': 'dummy'}}, {'llm_api_key': 'dummy'}, {'nested': [{'BearerToken': 'dummy'}]}] for params in invalid: request = Mock(json=AsyncMock(return_value={'custom_engine_cards': [{'id': 'card', 'params': params}]})) with self.assertRaises(HTTPException) as caught: asyncio.run(settings_routes.post_settings(request)) self.assertEqual(caught.exception.status_code, 400) save.assert_not_called() validate_provider_params({'max_tokens': 100, 'temperature': 0.3}) @patch('routes.docker._save_settings') @patch('routes.docker._load_settings') def test_explicit_activation_selects_custom_stt_and_enables_ocr(self, load, save): for role, setting_key in [('stt', 'faster_whisper_url'), ('ocr', 'vision_ocr_url'), ('tts', 'tts_url')]: load.return_value = {'custom_engine_cards': [{'id': 'card', 'role': role}]} request = Mock(json=AsyncMock(return_value={'url': 'http://engine', 'role': role, 'setting_key': setting_key, 'activate': True})) asyncio.run(docker.use_as_engine('card', request, '')) saved = save.call_args.args[0] if role == 'stt': self.assertEqual(saved['stt_preferred_backend'], 'custom:card') elif role == 'ocr': self.assertTrue(saved['vision_ocr_enabled']) else: self.assertEqual(saved['tts_backend'], 'openai') @patch('routes.settings._save_settings') @patch('routes.settings._load_settings') def test_custom_rename_migrates_legacy_maps_without_exposing_key(self, load, save): load.return_value = {'custom_engine_cards': [{'id': 'custom-1', 'name': 'Old'}], 'engine_api_keys': {'Old': 'dummy-secret'}, 'engine_models': {'Old': 'selected'}, 'engine_local_urls': {'dc-Old': 'http://engine'}} request = Mock(json=AsyncMock(return_value={'custom_engine_cards': [{'id': 'custom-1', 'name': 'Renamed'}], 'engine_api_keys': {'custom-1': '__configured__'}})) result = asyncio.run(settings_routes.post_settings(request)) saved = save.call_args.args[0] self.assertEqual(saved['engine_api_keys']['custom-1'], 'dummy-secret') self.assertEqual(saved['engine_models']['custom-1'], 'selected') self.assertEqual(saved['engine_local_urls']['dc-custom-1'], 'http://engine') self.assertNotIn('dummy-secret', str(result)) @patch('routes.settings._save_settings') @patch('routes.settings._load_settings') def test_custom_migration_preserves_id_values_and_normal_cards(self, load, save): load.return_value = {'custom_engine_cards': [{'id': 'custom-1', 'name': 'Old'}], 'engine_api_keys': {'Old': 'legacy', 'custom-1': 'current', 'ollama': 'normal'}, 'engine_models': {'Old': 'legacy', 'custom-1': 'current'}} request = Mock(json=AsyncMock(return_value={'custom_engine_cards': [{'id': 'custom-1', 'name': 'Renamed'}]})) asyncio.run(settings_routes.post_settings(request)) saved = save.call_args.args[0] self.assertEqual(saved['engine_api_keys']['custom-1'], 'current') self.assertEqual(saved['engine_api_keys']['ollama'], 'normal') self.assertEqual(saved['engine_models']['custom-1'], 'current') def test_card_credentials_do_not_use_other_provider_key(self): settings = {'llm_api_key': 'unrelated', 'engine_api_keys': {'card': 'saved'}} self.assertEqual(docker._engine_api_key(settings, 'card', 'llm', 'typed'), 'typed') self.assertEqual(docker._engine_api_key(settings, 'card', 'llm'), 'saved') self.assertEqual(docker._engine_api_key(settings, 'empty', 'llm'), 'sk-dummy-key') self.assertEqual(docker._engine_headers('typed', 'anthropic')['x-api-key'], 'typed') @patch('routes.docker.requests.get') @patch('routes.docker._load_settings') def test_model_list_uses_saved_card_key_and_root_fallback(self, load, get): load.return_value = {'engine_api_keys': {'card': 'saved'}, 'llm_api_key': 'other'} get.side_effect = [Mock(status_code=404), Mock(status_code=200, json=Mock(return_value={'data': [{'id': 'selected'}]}))] result = asyncio.run(docker.engine_models('http://engine/v1', 'stt', 'card', '')) self.assertEqual(result['models'], ['selected']) self.assertEqual(get.call_args.args[0], 'http://engine/models') self.assertEqual(get.call_args.kwargs['headers']['Authorization'], 'Bearer saved') @patch('routes.docker._save_settings') @patch('routes.docker._load_settings') def test_activation_scopes_key_and_clears_previous_model(self, load, save): load.return_value = {'whisper_api_key': 'default', 'stt_models': {'faster_whisper': 'old'}} request = Mock(json=AsyncMock(return_value={'url': 'http://engine', 'role': 'stt', 'setting_key': 'faster_whisper_url', 'model': ''})) asyncio.run(docker.use_as_engine('card', request, 'typed')) saved = save.call_args.args[0] self.assertEqual(saved['whisper_api_key'], 'default') self.assertEqual(saved['engine_api_keys']['card'], 'typed') self.assertEqual(saved['stt_models']['faster_whisper'], '') self.assertEqual(stt._stt_backend_api_key(saved, 'faster_whisper'), 'typed') def test_activation_rejects_non_object_and_cross_role_setting(self): for body in ([], {'role': 'ocr', 'setting_key': 'tts_url', 'url': 'http://engine'}): request = Mock(json=AsyncMock(return_value=body)) with self.assertRaises(HTTPException) as caught: asyncio.run(docker.use_as_engine('card', request, '')) self.assertEqual(caught.exception.status_code, 400) def test_endpoint_key_resolution_handles_ui_prefix(self): settings = {'engine_local_urls': {'dc-card': 'http://engine/v1'}, 'engine_api_keys': {'card': 'saved'}} self.assertEqual(provider_key_for_url(settings, 'http://engine', 'other'), 'saved') settings['engine_api_keys'] = {} self.assertEqual(provider_key_for_url(settings, 'http://engine', 'other'), '') @patch('routes.stt.requests.post') def test_stt_inference_uses_custom_card_key_and_model(self, post): settings = {'custom_engine_cards': [{'id': 'card', 'url': 'http://engine/v1'}], 'engine_api_keys': {'card': 'saved'}, 'engine_models': {'card': 'selected'}, 'custom_engine_cards': [{'id': 'card', 'url': 'http://engine/v1', 'role': 'stt', 'params': {'language': 'en'}}]} post.return_value = Mock(status_code=200, json=Mock(return_value={'text': 'heard'})) with tempfile.TemporaryDirectory() as directory: audio = Path(directory) / 'audio.wav' audio.write_bytes(b'test') stt._transcribe_audio(audio, settings, 'custom:card') self.assertEqual(post.call_args.kwargs['headers']['Authorization'], 'Bearer saved') self.assertEqual(post.call_args.kwargs['data']['model'], 'selected') self.assertEqual(post.call_args.kwargs['data']['language'], 'en') def test_tts_model_key_and_v1_base_reach_request_config(self): settings = {'tts_url': 'http://engine/v1', 'tts_models': {'tts_url': 'selected'}, 'engine_local_urls': {'dc-card': 'http://engine/v1'}, 'engine_api_keys': {'card': 'saved'}, 'custom_engine_cards': [{'id': 'card', 'url': 'http://engine/v1', 'role': 'tts', 'params': {'speed': 1.2}}]} endpoint, payload, headers = tts_helpers._tts_request_config('hello', 'voice', settings) self.assertEqual(endpoint, 'http://engine/v1/audio/speech') self.assertEqual(payload['model'], 'selected') self.assertEqual(payload['speed'], 1.2) self.assertEqual(headers['Authorization'], 'Bearer saved') def test_explicit_tts_key_override_is_preserved(self): settings = {'tts_url': 'http://primary', 'tts_api_key': 'default'} _, _, headers = tts_helpers._tts_request_config('hello', 'voice', settings, url_override='http://other', api_key_override='default') self.assertEqual(headers['Authorization'], 'Bearer default') def test_sibling_backend_does_not_receive_default_credentials(self): settings = {'tts_url': 'http://primary', 'tts_api_key': 'primary-key', 'whisper_url': 'http://primary', 'whisper_api_key': 'primary-key', 'faster_whisper_url': 'http://sibling'} _, _, headers = tts_helpers._tts_request_config('hello', 'voice', settings, url_override='http://sibling') self.assertEqual(headers, {}) self.assertEqual(stt._stt_backend_api_key(settings, 'faster_whisper'), '') if __name__ == '__main__': unittest.main()