Files
DDNS/tests/test_config_env.py
3c8578e9f4 feat(mcp): add Streamable HTTP transport (#736)
* feat(mcp): add Streamable HTTP transport

Add a modern MCP 2026-07-28 HTTP endpoint for standalone and Web modes, with shared listener configuration, authentication, schema, tests, and bilingual documentation.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(tests): support Python 2 HTTP assertions

Normalize response header names and call handler methods through their underlying functions so the HTTP tests behave consistently on Python 2.7 and Python 3.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): harden HTTP request handling

Validate mirrored request IDs, align Config Studio listener checks, bound socket reads, and accept the binary wrapper's documented SIGINT exit status.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): align HTTP negotiation validation

Reject unavailable Accept media types by quality and keep schema host validation consistent with runtime listener checks.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(tests): validate host schema behavior

Assert accepted and rejected host samples instead of comparing Unicode escape representations that differ on Python 2.7.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(tests): avoid cross-engine schema regex

Keep the schema pattern contract without executing its ECMA-262 Unicode escape through Python 2's incompatible regex engine.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): validate HTTP request boundaries

Reject empty tokens and malformed JSON-RPC request shapes, and keep Config Studio bracket validation aligned with runtime listener handling.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): tighten HTTP input validation

Reject legacy notifications, nested host brackets, and document the required JSON content type so validation remains consistent across runtime, schema, and Config Studio.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): validate Origin before method errors

Apply the existing Origin policy to non-POST MCP requests and return CORS headers for configured allowed origins.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* docs(mcp): clarify proxy Origin allowlist

Explain that browser clients behind TLS termination must allowlist the public HTTPS origin even when the public page appears same-origin.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* docs(mcp): add standalone server guide

Move MCP client setup, transports, tools, security, and protocol limits into dedicated Chinese and English pages with matching navigation and index links.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): reject duplicate routing headers

Reject repeated MCP protocol, method, and name headers before body consistency checks across Python 2 and Python 3.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

* fix(mcp): validate named request headers

Validate Mcp-Name for tools/call, prompts/get, and resources/read before dispatching unsupported methods.

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: NewFuture <6290356+NewFuture@users.noreply.github.com>
Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>
2026-08-23 21:36:26 +08:00

307 lines
12 KiB
Python

# -*- coding:utf-8 -*-
"""
Configuration loader tests for environment variables.
"""
import os
import io
import unittest
import ddns.config.env as env_module
from ddns.config.env import load_config
class TestConfigEnv(unittest.TestCase):
"""Test configuration loading from environment variables"""
def setUp(self):
"""Set up test environment"""
self._clear_env_prefix("DDNS_TEST_")
self._clear_env_prefix("DDNS_")
self._clear_standard_env()
def tearDown(self):
"""Clean up after tests"""
self._clear_env_prefix("DDNS_TEST_")
self._clear_env_prefix("DDNS_")
self._clear_standard_env()
def _clear_env_prefix(self, prefix):
# type: (str) -> None
"""Clear environment variables with a specific prefix"""
test_prefixes = [prefix.lower(), prefix.upper()]
for key in list(os.environ.keys()):
if any(key.startswith(prefix) for prefix in test_prefixes):
del os.environ[key]
def _clear_standard_env(self):
# type: () -> None
"""Clear standard environment variables used in tests"""
keys = ["PYTHONHTTPSVERIFY"]
for key in keys:
if key in os.environ:
del os.environ[key]
def test_basic_string_values(self):
"""Test that basic string values are preserved"""
os.environ["DDNS_DNS"] = "cloudflare"
os.environ["DDNS_ID"] = "test@example.com"
os.environ["DDNS_TOKEN"] = "secret123"
config = load_config()
self.assertEqual(config.get("dns"), "cloudflare")
self.assertEqual(config.get("id"), "test@example.com")
self.assertEqual(config.get("token"), "secret123")
def test_json_array_conversion(self):
"""Test JSON array format conversion"""
os.environ["DDNS_IPV4"] = '["domain1.com", "domain2.com"]'
os.environ["DDNS_INDEX4"] = '["public", 0]'
config = load_config()
self.assertEqual(config.get("ipv4"), ["domain1.com", "domain2.com"])
self.assertEqual(config.get("index4"), ["public", 0])
def test_key_normalization(self):
"""Test key normalization (uppercase to lowercase)"""
os.environ["DDNS_DNS"] = "cloudflare"
os.environ["ddns_id"] = "test@example.com"
config = load_config()
self.assertEqual(config.get("dns"), "cloudflare")
self.assertEqual(config.get("id"), "test@example.com")
def test_dot_to_underscore_conversion(self):
"""Test conversion of dots to underscores in keys"""
os.environ["DDNS_LOG.LEVEL"] = "DEBUG"
os.environ["DDNS_LOG.FILE"] = "/var/log/ddns.log"
config = load_config()
self.assertEqual(config.get("log_level"), "DEBUG")
self.assertEqual(config.get("log_file"), "/var/log/ddns.log")
def test_pythonhttpsverify_values(self):
"""Test PYTHONHTTPSVERIFY with different values"""
test_cases = [("0", "0"), ("1", "1"), ("false", "false"), ("true", "true"), ("anything", "anything")]
for env_value, expected in test_cases:
os.environ["PYTHONHTTPSVERIFY"] = env_value
config = load_config()
self.assertEqual(config.get("ssl"), expected)
del os.environ["PYTHONHTTPSVERIFY"]
def test_ddns_proxy_basic(self):
"""Test basic DDNS_PROXY functionality"""
os.environ["DDNS_PROXY"] = "http://ddns.proxy.com:9090"
config = load_config()
self.assertEqual(config.get("proxy"), "http://ddns.proxy.com:9090")
del os.environ["DDNS_PROXY"]
def test_ddns_cache_max_age(self):
"""Test DDNS_CACHE_MAX_AGE is loaded by generic environment handling."""
os.environ["DDNS_CACHE_MAX_AGE"] = "86400"
config = load_config()
self.assertEqual(config.get("cache_max_age"), "86400")
del os.environ["DDNS_CACHE_MAX_AGE"]
def test_http_environment_settings(self):
"""Load the shared listener settings without treating origins as a scalar."""
os.environ["DDNS_HTTP_HOST"] = "0.0.0.0"
os.environ["DDNS_HTTP_PORT"] = "9877"
os.environ["DDNS_HTTP_TOKEN"] = "secret"
os.environ["DDNS_HTTP_ORIGINS"] = '["https://one.example", "https://two.example"]'
config = load_config()
self.assertEqual(config["http_host"], "0.0.0.0")
self.assertEqual(config["http_port"], "9877")
self.assertEqual(config["http_token"], "secret")
self.assertEqual(config["http_origins"], ["https://one.example", "https://two.example"])
def test_token_values_are_never_array_parsed_or_logged(self):
"""Preserve bracket-prefixed secrets without writing them to stderr."""
os.environ["DDNS_TOKEN"] = "[provider-secret]"
os.environ["DDNS_HTTP_TOKEN"] = "[http-secret]"
capture = io.StringIO()
original_stderr = env_module.stderr
env_module.stderr = capture
try:
config = load_config()
finally:
env_module.stderr = original_stderr
self.assertEqual(config["token"], "[provider-secret]")
self.assertEqual(config["http_token"], "[http-secret]")
self.assertEqual(capture.getvalue(), "")
def test_ddns_variables_override_standard_vars(self):
"""Test that DDNS variables take precedence over standard environment variables"""
self._clear_standard_env()
# Set standard environment variables
os.environ["PYTHONHTTPSVERIFY"] = "0"
# Set conflicting DDNS variables
os.environ["DDNS_SSL"] = "auto"
config = load_config()
# DDNS should take precedence
self.assertEqual(config.get("ssl"), "auto")
# Clean up
del os.environ["DDNS_SSL"]
self._clear_standard_env()
def test_custom_prefix(self):
"""Test custom prefix functionality"""
os.environ["CUSTOM_DNS"] = "cloudflare"
os.environ["CUSTOM_ID"] = "test@example.com"
config = load_config(prefix="CUSTOM_")
self.assertEqual(config.get("dns"), "cloudflare")
self.assertEqual(config.get("id"), "test@example.com")
# Clean up
del os.environ["CUSTOM_DNS"]
del os.environ["CUSTOM_ID"]
def test_empty_prefix(self):
"""Test empty prefix (loads all environment variables)"""
os.environ["TEST_DNS"] = "cloudflare"
config = load_config(prefix="")
self.assertEqual(config.get("test_dns"), "cloudflare")
del os.environ["TEST_DNS"]
def test_no_matching_environment_variables(self):
"""Test behavior when no matching environment variables exist"""
config = load_config()
self.assertEqual(config, {})
def test_whitespace_stripping(self):
"""Test that leading and trailing whitespace is stripped from values"""
test_cases = [
(" value ", "value"),
("\tvalue\t", "value"),
("\nvalue\n", "value"),
(" value with spaces ", "value with spaces"),
]
for env_value, expected in test_cases:
os.environ["DDNS_TEST_STRIP"] = env_value
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("strip"), expected, "Failed for value with whitespace: %r" % env_value)
def test_case_insensitive_prefix_matching(self):
"""Test case-insensitive prefix matching"""
os.environ["ddns_dns"] = "cloudflare"
os.environ["DDNS_ID"] = "test@example.com"
config = load_config()
self.assertEqual(config.get("dns"), "cloudflare")
self.assertEqual(config.get("id"), "test@example.com")
def test_invalid_json_array(self):
"""Test handling of invalid JSON array format"""
os.environ["DDNS_TEST_INVALID"] = "[invalid"
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("invalid"), "[invalid") # Should remain as string
def test_malformed_json_arrays(self):
"""Test handling of malformed JSON arrays"""
test_cases = [
("[,]", "[,]"), # Empty elements
("[unclosed", "[unclosed"), # Unclosed bracket
("[1,2,3", "[1,2,3"), # Missing closing bracket
]
for malformed, expected in test_cases:
os.environ["DDNS_TEST_MALFORMED"] = malformed
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("malformed"), expected)
def test_json_array_edge_cases(self):
"""Test JSON array edge cases"""
test_cases = [
("[]", []), # Empty array
("[1]", [1]), # Single element
('["string"]', ["string"]), # Single string
('[1, "two", 3]', [1, "two", 3]), # Mixed types
]
for json_array, expected in test_cases:
os.environ["DDNS_TEST_ARRAY"] = json_array
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("array"), expected)
def test_empty_values_remain_as_strings(self):
"""Test that empty values remain as strings"""
os.environ["DDNS_TEST_EMPTY"] = ""
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("empty"), "")
def test_numeric_and_boolean_strings(self):
"""Test that numeric and boolean strings are preserved"""
test_cases = [("123", "123"), ("true", "true"), ("false", "false"), ("3.14", "3.14")]
for value, expected in test_cases:
os.environ["DDNS_TEST_VALUE"] = value
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("value"), expected)
def test_special_characters_in_values(self):
"""Test handling of special characters in values"""
test_cases = [
("value with spaces", "value with spaces"),
("value@with#special$chars%", "value@with#special$chars%"),
("http://example.com:8080", "http://example.com:8080"),
]
for value, expected in test_cases:
os.environ["DDNS_TEST_SPECIAL"] = value
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("special"), expected)
def test_regular_comma_values_kept_as_string(self):
"""Test that regular comma-separated values are kept as strings"""
os.environ["DDNS_TEST_COMMA"] = "value1,value2,value3"
config = load_config(prefix="DDNS_TEST_")
self.assertEqual(config.get("comma"), "value1,value2,value3")
def test_edge_cases(self):
"""Test edge cases"""
# Test with only prefix (should be ignored)
os.environ["DDNS_"] = "ignored"
config = load_config()
self.assertNotIn("", config)
# Clean up
del os.environ["DDNS_"]
def test_mixed_params_config(self):
"""Test mixed configuration with different parameter types"""
os.environ["DDNS_DNS"] = "cloudflare"
os.environ["DDNS_IPV4"] = '["domain1.com", "domain2.com"]'
os.environ["DDNS_TTL"] = "300"
os.environ["DDNS_SSL"] = "auto"
config = load_config()
self.assertEqual(config.get("dns"), "cloudflare")
self.assertEqual(config.get("ipv4"), ["domain1.com", "domain2.com"])
self.assertEqual(config.get("ttl"), "300")
self.assertEqual(config.get("ssl"), "auto")
def test_configuration_consistency(self):
"""Test that configuration parsing is consistent across different input formats"""
# Test array consistency
os.environ["DDNS_TEST1"] = '["item1", "item2"]'
os.environ["DDNS_TEST2"] = "single_item"
config = load_config(prefix="DDNS_")
self.assertEqual(config.get("test1"), ["item1", "item2"])
self.assertEqual(config.get("test2"), "single_item")
if __name__ == "__main__":
unittest.main()