tcp_handler: Optional[_TcpHandler],
pidfile: Optional[str] = None,
) -> None:
+ self._abort_if_on_dnspython_version_less_than_2_0_0()
logging.basicConfig(
format="%(asctime)s %(levelname)8s %(message)s",
level=os.environ.get("ANS_LOG_LEVEL", "INFO").upper(),
self._pidfile: Optional[str] = pidfile
self._work_done: Optional[asyncio.Future] = None
+ @classmethod
+ def _abort_if_on_dnspython_version_less_than_2_0_0(cls) -> None:
+ if dns.version.MAJOR < 2:
+ error = f"Using {cls.__name__} requires dnspython >= 2.0.0; "
+ error += 'add `pytest.importorskip("dns", minversion="2.0.0")` '
+ error += "to the test module to skip this test."
+ raise RuntimeError(error)
+
def _get_ipv4_address_from_directory_name(self) -> str:
containing_directory = pathlib.Path().absolute().stem
match_result = re.match(r"ans(?P<index>\d+)", containing_directory)
try:
query = dns.message.from_wire(wire)
except dns.message.UnknownTSIGKey:
- self._abort_if_on_dnspython_version_less_than_2_0_0()
self._abort_if_tsig_signed_query_received_unless_acknowledged()
query = _DnsMessageWithTsigDisabled.from_wire(wire)
except dns.exception.DNSException as exc:
response_length = struct.pack("!H", len(response))
yield response_length + response
- def _abort_if_on_dnspython_version_less_than_2_0_0(self) -> None:
- if dns.version.MAJOR < 2:
- error = "Receiving TSIG signed queries requires dnspython >= 2.0.0; "
- error += 'add `pytest.importorskip("dns", minversion="2.0.0")` '
- error += "to the test module to skip this test."
- raise RuntimeError(error)
-
def _abort_if_tsig_signed_query_received_unless_acknowledged(self) -> None:
if self._acknowledge_tsig_dnspython_hacks:
return