From 8a088183e6e05a63eddbdb39f7e0d2d5d29b5b76 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C5=A0t=C4=9Bp=C3=A1n=20Bal=C3=A1=C5=BEik?= Date: Tue, 23 Dec 2025 14:36:56 +0100 Subject: [PATCH] Add SwitchControlCommand for ControllableAsyncServer To provide feature parity with `bin/tests/system/ans.pl` add a control command to allow easy switching between different sequences of ResponseHandlers. (cherry picked from commit 2302fe1235f64691620834922eda33397e1f0157) --- bin/tests/system/isctest/asyncserver.py | 36 +++++++++++++++++++++++++ 1 file changed, 36 insertions(+) diff --git a/bin/tests/system/isctest/asyncserver.py b/bin/tests/system/isctest/asyncserver.py index 1eaf2a4ec7..85254800e9 100644 --- a/bin/tests/system/isctest/asyncserver.py +++ b/bin/tests/system/isctest/asyncserver.py @@ -21,6 +21,7 @@ from typing import ( List, Optional, Set, + Sequence, Tuple, Union, cast, @@ -891,6 +892,14 @@ class AsyncDnsServer(AsyncServer): for handler in handlers: self.install_response_handler(handler) + def replace_response_handlers(self, *new_handlers: ResponseHandler) -> None: + """ + Uninstall all currently installed handlers and install the provided ones. + """ + logging.info("Uninstalling response handlers: %s", str(self._response_handlers)) + self._response_handlers.clear() + self.install_response_handlers(*new_handlers) + def uninstall_response_handler(self, handler: ResponseHandler) -> None: """ Remove the specified handler from the list of response handlers. @@ -1556,3 +1565,30 @@ class ToggleResponsesCommand(ControlCommand): logging.error("Unrecognized response sending mode '%s'", mode) qctx.response.set_rcode(dns.rcode.SERVFAIL) return f"unrecognized response sending mode '{mode}'" + + +class SwitchControlCommand(ControlCommand): + """ + Switch the server's response handlers based on the control query. + + A sequence of response handlers is associated with each key. When a + control query is received, the server's response handlers are replaced + with the sequence associated with the key extracted from the control + query. + """ + + control_subdomain = "switch" + + def __init__(self, handler_mapping: Dict[str, Sequence[ResponseHandler]]): + self._handler_mapping = handler_mapping + + def handle( + self, args: List[str], server: ControllableAsyncDnsServer, qctx: QueryContext + ) -> Optional[str]: + if len(args) != 1 or args[0] not in self._handler_mapping: + logging.error("Invalid %s query %s", self, qctx.qname) + qctx.response.set_rcode(dns.rcode.SERVFAIL) + return f"invalid query; exactly one of {list(self._handler_mapping.keys())} is expected in QNAME" + + server.replace_response_handlers(*self._handler_mapping[args[0]]) + return f"switched to handler set '{args[0]}'"