156 lines
8.1 KiB
Python
156 lines
8.1 KiB
Python
"""Configuration boundaries; credentials and model calls use synthetic fixtures."""
|
|
|
|
import os
|
|
from pathlib import Path
|
|
import tempfile
|
|
import unittest
|
|
from unittest.mock import patch, sentinel
|
|
|
|
from logistics_chat import config, create_app
|
|
|
|
|
|
class ConfigurationTests(unittest.TestCase):
|
|
def setUp(self):
|
|
directory = tempfile.TemporaryDirectory()
|
|
self.addCleanup(directory.cleanup)
|
|
self.root = Path(directory.name)
|
|
root_patch = patch.object(config, 'ROOT_DIR', self.root)
|
|
root_patch.start()
|
|
self.addCleanup(root_patch.stop)
|
|
environment = patch.dict(os.environ)
|
|
environment.start()
|
|
self.addCleanup(environment.stop)
|
|
os.environ.pop('DEEPSEEK_API_KEY', None)
|
|
os.environ.pop('RATE_LIMIT_PER_MIN', None)
|
|
|
|
def write_env(self, text):
|
|
(self.root / '.env').write_text(text, encoding='utf-8')
|
|
|
|
def application(self, **overrides):
|
|
return create_app({'TESTING': True, 'INITIALIZE_RUNTIME': False, **overrides})
|
|
|
|
def test_missing_file_leaves_ai_disabled_without_creating_files(self):
|
|
self.assertEqual(config.default_config()['DEEPSEEK_API_KEY'], '')
|
|
self.assertEqual(list(self.root.iterdir()), [])
|
|
|
|
def test_file_accepts_bom_quotes_comments_and_whitespace(self):
|
|
self.write_env('\ufeff# Local configuration\nexport DEEPSEEK_API_KEY=" file-key " # comment\n')
|
|
self.assertEqual(self.application().config['DEEPSEEK_API_KEY'], 'file-key')
|
|
self.assertNotIn('DEEPSEEK_API_KEY', os.environ)
|
|
|
|
def test_empty_or_unassigned_file_value_leaves_ai_disabled(self):
|
|
for line in ('DEEPSEEK_API_KEY=', 'DEEPSEEK_API_KEY', 'OTHER_SETTING=unused'):
|
|
with self.subTest(line=line):
|
|
self.write_env(line + '\n')
|
|
self.assertEqual(config.default_config()['DEEPSEEK_API_KEY'], '')
|
|
|
|
def test_environment_overrides_file_including_an_explicit_empty_value(self):
|
|
self.write_env('DEEPSEEK_API_KEY=file-key\n')
|
|
for value, expected in ((' environment-key ', 'environment-key'), ('', '')):
|
|
with self.subTest(value=value):
|
|
os.environ['DEEPSEEK_API_KEY'] = value
|
|
self.assertEqual(self.application().config['DEEPSEEK_API_KEY'], expected)
|
|
|
|
def test_file_values_are_literal_and_do_not_expand_environment_variables(self):
|
|
os.environ['SYNTHETIC_CONFIG_VALUE'] = 'expanded'
|
|
self.write_env("DEEPSEEK_API_KEY='literal-${SYNTHETIC_CONFIG_VALUE}'\n")
|
|
self.assertEqual(config.default_config()['DEEPSEEK_API_KEY'], 'literal-${SYNTHETIC_CONFIG_VALUE}')
|
|
|
|
def test_file_is_resolved_from_repository_root_instead_of_working_directory(self):
|
|
self.write_env('DEEPSEEK_API_KEY=repository-key\n')
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
(Path(directory) / '.env').write_text('DEEPSEEK_API_KEY=wrong-key\n', encoding='utf-8')
|
|
previous = Path.cwd()
|
|
try:
|
|
os.chdir(directory)
|
|
self.assertEqual(self.application().config['DEEPSEEK_API_KEY'], 'repository-key')
|
|
finally:
|
|
os.chdir(previous)
|
|
|
|
def test_explicit_factory_configuration_has_highest_priority(self):
|
|
self.write_env('DEEPSEEK_API_KEY=file-key\n')
|
|
os.environ['DEEPSEEK_API_KEY'] = 'environment-key'
|
|
for value in ('override-key', ''):
|
|
with self.subTest(value=value):
|
|
self.assertEqual(self.application(DEEPSEEK_API_KEY=value).config['DEEPSEEK_API_KEY'], value)
|
|
|
|
def test_new_factory_reads_updated_file_without_changing_an_existing_app(self):
|
|
self.write_env('DEEPSEEK_API_KEY=first-key\n')
|
|
first = self.application()
|
|
self.write_env('DEEPSEEK_API_KEY=second-key\n')
|
|
self.assertEqual(self.application().config['DEEPSEEK_API_KEY'], 'second-key')
|
|
self.assertEqual(first.config['DEEPSEEK_API_KEY'], 'first-key')
|
|
self.assertNotIn('DEEPSEEK_API_KEY', os.environ)
|
|
|
|
def test_model_client_sends_file_key_without_a_real_http_request(self):
|
|
from logistics_chat.ai_client import post_completion
|
|
|
|
self.write_env('DEEPSEEK_API_KEY=file-key\n')
|
|
app = self.application()
|
|
with app.app_context(), patch('logistics_chat.ai_client.requests.post', return_value=sentinel.response) as post:
|
|
response = post_completion({'model': 'deepseek-chat'}, timeout=(5, 90), stream=True)
|
|
self.assertIs(response, sentinel.response)
|
|
self.assertEqual(post.call_args.kwargs['headers']['Authorization'], 'Bearer file-key')
|
|
|
|
def test_common_model_configuration_can_select_another_compatible_provider(self):
|
|
from logistics_chat.ai_client import post_completion
|
|
|
|
self.write_env('CHAT_API_KEY=common-key\nDEEPSEEK_API_KEY=legacy-key\n'
|
|
'MODEL_URL=https://provider.example.invalid/chat/completions\nMODEL_NAME=other-model\n')
|
|
app = self.application()
|
|
with app.app_context(), patch('logistics_chat.ai_client.requests.post', return_value=sentinel.response) as post:
|
|
post_completion({'model': app.config['MODEL_NAME']}, timeout=(5, 30))
|
|
self.assertEqual(post.call_args.args[0], 'https://provider.example.invalid/chat/completions')
|
|
self.assertEqual(post.call_args.kwargs['headers']['Authorization'], 'Bearer common-key')
|
|
self.assertEqual(post.call_args.kwargs['json']['model'], 'other-model')
|
|
|
|
def test_model_provider_environment_overrides_the_local_file(self):
|
|
self.write_env('CHAT_API_KEY=file-key\nMODEL_URL=https://file.example.invalid/chat/completions\nMODEL_NAME=file-model\n')
|
|
with patch.dict(os.environ, {'CHAT_API_KEY': 'environment-key',
|
|
'MODEL_URL': 'https://environment.example.invalid/chat/completions', 'MODEL_NAME': 'environment-model'}):
|
|
settings = self.application().config
|
|
self.assertEqual(settings['CHAT_API_KEY'], 'environment-key')
|
|
self.assertEqual(settings['MODEL_NAME'], 'environment-model')
|
|
self.assertEqual(settings['MODEL_URL'], 'https://environment.example.invalid/chat/completions')
|
|
|
|
def test_order_scope_is_fixed_by_server_file_or_environment(self):
|
|
self.write_env('ORDER_DATA_TENANT_ID=tenant-file\n')
|
|
os.environ.pop('ORDER_DATA_TENANT_ID', None)
|
|
self.assertEqual(self.application().config['ORDER_DATA_TENANT_ID'], 'tenant-file')
|
|
os.environ['ORDER_DATA_TENANT_ID'] = 'tenant-environment'
|
|
self.assertEqual(self.application().config['ORDER_DATA_TENANT_ID'], 'tenant-environment')
|
|
|
|
def test_rate_limit_defaults_to_600_requests_per_minute(self):
|
|
self.assertEqual(self.application().config['RATE_LIMIT_PER_MIN'], 600)
|
|
|
|
def test_rate_limit_can_be_configured_in_the_repository_env_file(self):
|
|
self.write_env('RATE_LIMIT_PER_MIN=" 1200 " # Local request budget\n')
|
|
self.assertEqual(self.application().config['RATE_LIMIT_PER_MIN'], 1200)
|
|
self.assertNotIn('RATE_LIMIT_PER_MIN', os.environ)
|
|
|
|
def test_rate_limit_environment_and_factory_overrides_take_priority(self):
|
|
self.write_env('RATE_LIMIT_PER_MIN=1200\n')
|
|
os.environ['RATE_LIMIT_PER_MIN'] = '600'
|
|
self.assertEqual(self.application().config['RATE_LIMIT_PER_MIN'], 600)
|
|
self.assertEqual(self.application(RATE_LIMIT_PER_MIN=900).config['RATE_LIMIT_PER_MIN'], 900)
|
|
|
|
def test_rate_limit_rejects_invalid_file_values(self):
|
|
for line in ('RATE_LIMIT_PER_MIN=', 'RATE_LIMIT_PER_MIN', 'RATE_LIMIT_PER_MIN=0',
|
|
'RATE_LIMIT_PER_MIN=-1', 'RATE_LIMIT_PER_MIN=600.5', 'RATE_LIMIT_PER_MIN=invalid'):
|
|
with self.subTest(line=line):
|
|
self.write_env(line + '\n')
|
|
with self.assertRaisesRegex(ValueError, 'RATE_LIMIT_PER_MIN must be a positive integer'):
|
|
self.application()
|
|
|
|
def test_rate_limit_rejects_invalid_environment_values_over_a_valid_file(self):
|
|
self.write_env('RATE_LIMIT_PER_MIN=600\n')
|
|
for value in ('', '0', '-1', '600.5', 'invalid'):
|
|
with self.subTest(value=value):
|
|
os.environ['RATE_LIMIT_PER_MIN'] = value
|
|
with self.assertRaisesRegex(ValueError, 'RATE_LIMIT_PER_MIN must be a positive integer'):
|
|
self.application()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|