61 lines
2.7 KiB
Python
61 lines
2.7 KiB
Python
import unittest
|
|
|
|
from mcp_calculator.config import load_settings
|
|
|
|
TOKEN = "0123456789abcdef"
|
|
|
|
|
|
class SettingsTest(unittest.TestCase):
|
|
def test_defaults(self):
|
|
s = load_settings([], env={})
|
|
self.assertEqual(s.transport, "stdio")
|
|
self.assertEqual(s.executor, "process")
|
|
self.assertEqual(s.auth_tokens, [])
|
|
self.assertEqual(s.clients_file, "")
|
|
self.assertEqual(s.admin_tokens, [])
|
|
self.assertTrue(s.metrics)
|
|
self.assertEqual(s.max_queue, 20)
|
|
self.assertEqual(s.max_queue_wait_seconds, 30)
|
|
self.assertEqual(s.client_max_concurrent, 10)
|
|
self.assertEqual(s.trusted_proxies, [])
|
|
|
|
def test_env_and_cli_precedence(self):
|
|
env = {"MCP_CALC_TRANSPORT": "http", "MCP_CALC_PORT": "9000", "MCP_CALC_AUTH_TOKENS": f"a:{TOKEN}, b:{TOKEN}x"}
|
|
s = load_settings(["--port", "9100"], env=env)
|
|
self.assertEqual(s.transport, "http")
|
|
self.assertEqual(s.port, 9100)
|
|
self.assertEqual(s.auth_tokens, [f"a:{TOKEN}", f"b:{TOKEN}x"])
|
|
|
|
def test_limits_from_env(self):
|
|
s = load_settings([], env={"MCP_CALC_MAX_MATRIX_SIZE": "5", "MCP_CALC_TIMEOUT_SECONDS": "2.5"})
|
|
self.assertEqual(s.limits.max_matrix_size, 5)
|
|
self.assertEqual(s.timeout_seconds, 2.5)
|
|
|
|
def test_booleans(self):
|
|
self.assertFalse(load_settings([], env={"MCP_CALC_METRICS": "false"}).metrics)
|
|
self.assertTrue(load_settings([], env={"MCP_CALC_METRICS": "On"}).metrics)
|
|
|
|
def test_admin_settings(self):
|
|
s = load_settings([], env={"MCP_CALC_CLIENTS_FILE": "/data/clients.json", "MCP_CALC_ADMIN_TOKENS": f"ops:{TOKEN}"})
|
|
self.assertEqual(s.admin_tokens, [f"ops:{TOKEN}"])
|
|
self.assertEqual(s.admin_port, 9090)
|
|
self.assertEqual(s.admin_host, "127.0.0.1")
|
|
|
|
def test_invalid_settings(self):
|
|
for env in (
|
|
{"MCP_CALC_TRANSPORT": "websocket"},
|
|
{"MCP_CALC_PORT": "abc"},
|
|
{"MCP_CALC_WORKERS": "0"},
|
|
{"MCP_CALC_MAX_QUEUE": "-1"},
|
|
{"MCP_CALC_MAX_QUEUE_WAIT_SECONDS": "0"},
|
|
{"MCP_CALC_CLIENT_MAX_CONCURRENT": "-1"},
|
|
{"MCP_CALC_MAX_CONNECTIONS": "0"},
|
|
{"MCP_CALC_METRICS": "maybe"},
|
|
{"MCP_CALC_PATH": "mcp"},
|
|
{"MCP_CALC_AUTH_TOKENS": "no-name-token-0123456789"},
|
|
{"MCP_CALC_AUTH_TOKENS": "name:short"},
|
|
{"MCP_CALC_ADMIN_TOKENS": f"ops:{TOKEN}"}, # admin API without clients file
|
|
{"MCP_CALC_CLIENTS_FILE": "/x.json", "MCP_CALC_ADMIN_TOKENS": f"ops:{TOKEN}", "MCP_CALC_ADMIN_PORT": "8080"},
|
|
):
|
|
with self.subTest(env=env), self.assertRaises(SystemExit):
|
|
load_settings([], env=env)
|