Move STRATO MCP to standalone repository
This commit is contained in:
@@ -1,116 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
SOURCE = Path(__file__).parents[1] / "services/strato-dns-mcp/strato_client.py"
|
||||
spec = importlib.util.spec_from_file_location("strato_client", SOURCE)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
assert spec.loader
|
||||
sys.modules[spec.name] = module
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
|
||||
class FakeResponse:
|
||||
def __init__(self, url: str, body: str) -> None:
|
||||
self.url = url
|
||||
self.body = body.encode()
|
||||
|
||||
def read(self, _limit: int) -> bytes:
|
||||
return self.body
|
||||
|
||||
def geturl(self) -> str:
|
||||
return self.url
|
||||
|
||||
|
||||
class FakeOpener:
|
||||
def __init__(self, responses: list[FakeResponse]) -> None:
|
||||
self.responses = responses
|
||||
self.requests: list[tuple[object, float]] = []
|
||||
|
||||
def open(self, request: object, timeout: float) -> FakeResponse:
|
||||
self.requests.append((request, timeout))
|
||||
return self.responses.pop(0)
|
||||
|
||||
|
||||
class StratoParserTests(unittest.TestCase):
|
||||
def test_dns_form_is_parsed_without_script_or_markup(self) -> None:
|
||||
html = """
|
||||
<select name="type"><option value="TXT">TXT</option><option value="CNAME" selected>CNAME</option></select>
|
||||
<input name="prefix" value="media">
|
||||
<textarea name="value">proxy.example.net.</textarea>
|
||||
<select name="type"><option value="TXT" selected>TXT</option></select>
|
||||
<input name="prefix" value="_acme-challenge">
|
||||
<textarea name="value">public-validation-value</textarea>
|
||||
"""
|
||||
records = module.parse_records(html)
|
||||
self.assertEqual(records[0].as_dict(), {
|
||||
"type": "CNAME", "prefix": "media", "value": "proxy.example.net."
|
||||
})
|
||||
self.assertEqual(records[1].type, "TXT")
|
||||
|
||||
def test_changed_form_fails_closed(self) -> None:
|
||||
with self.assertRaisesRegex(module.StratoParseError, "field counts"):
|
||||
module.parse_records('<input name="prefix" value="orphan">')
|
||||
|
||||
def test_package_is_selected_by_domain_not_fallback(self) -> None:
|
||||
html = """
|
||||
<table id="package_list"><tr><td>other.example</td><td><a href="?cID=1">open</a></td></tr>
|
||||
<tr><td>example.de Hosting</td><td><a href="?node=x&cID=42">open</a></td></tr></table>
|
||||
"""
|
||||
self.assertEqual(module.parse_package_id(html, "example.de"), "42")
|
||||
with self.assertRaises(module.StratoParseError):
|
||||
module.parse_package_id(html, "missing.de")
|
||||
|
||||
def test_totp_matches_rfc_vector_truncated_to_six_digits(self) -> None:
|
||||
# RFC 6238 secret, SHA-1, at t=59 gives 94287082 (therefore 287082 for 6 digits).
|
||||
secret = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"
|
||||
self.assertEqual(module._totp(secret, at_time=59), "287082")
|
||||
|
||||
def test_environment_errors_do_not_echo_values(self) -> None:
|
||||
old = dict(os.environ)
|
||||
try:
|
||||
for key in ("STRATO_USERNAME", "STRATO_PASSWORD", "STRATO_DOMAIN"):
|
||||
os.environ.pop(key, None)
|
||||
with self.assertRaises(module.StratoError) as caught:
|
||||
module.StratoConfig.from_env()
|
||||
message = str(caught.exception)
|
||||
self.assertIn("STRATO_PASSWORD", message)
|
||||
self.assertNotIn("secret-value", message)
|
||||
finally:
|
||||
os.environ.clear()
|
||||
os.environ.update(old)
|
||||
|
||||
def test_complete_read_path_has_no_dns_write_request(self) -> None:
|
||||
package_html = """
|
||||
<table id="package_list"><tr><td>example.de Hosting</td>
|
||||
<td><a href="?node=x&cID=42">open</a></td></tr></table>
|
||||
"""
|
||||
records_html = """
|
||||
<select name="type"><option value="CNAME" selected>CNAME</option></select>
|
||||
<input name="prefix" value="media">
|
||||
<textarea name="value">proxy.example.net.</textarea>
|
||||
"""
|
||||
opener = FakeOpener([
|
||||
FakeResponse(module.STRATO_URL, "login"),
|
||||
FakeResponse(module.STRATO_URL + "?sessionID=test-session", "welcome"),
|
||||
FakeResponse(module.STRATO_URL, package_html),
|
||||
FakeResponse(module.STRATO_URL, records_html),
|
||||
])
|
||||
config = module.StratoConfig("customer", "secret-value", "example.de")
|
||||
records = module.StratoClient(
|
||||
config, opener=opener, sleep=lambda _seconds: None
|
||||
).list_cnames()
|
||||
self.assertEqual([record.prefix for record in records], ["media"])
|
||||
methods = [request.method for request, _timeout in opener.requests]
|
||||
self.assertEqual(methods, ["GET", "POST", "GET", "GET"])
|
||||
self.assertNotIn("secret-value", opener.requests[1][0].full_url)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user