(self)
| 90 | self.nparts = len(split_fqdn) |
| 91 | |
| 92 | async def get_options(self) -> Optional[str]: |
| 93 | from dns import resolver |
| 94 | |
| 95 | try: |
| 96 | results = await _resolve(self.__fqdn, "TXT", lifetime=self.__connect_timeout) |
| 97 | except (resolver.NoAnswer, resolver.NXDOMAIN): |
| 98 | # No TXT records |
| 99 | return None |
| 100 | except Exception as exc: |
| 101 | raise ConfigurationError(str(exc)) from exc |
| 102 | if len(results) > 1: |
| 103 | raise ConfigurationError("Only one TXT record is supported") |
| 104 | return (b"&".join([b"".join(res.strings) for res in results])).decode("utf-8") # type: ignore[attr-defined] |
| 105 | |
| 106 | async def _resolve_uri(self, encapsulate_errors: bool) -> resolver.Answer: |
| 107 | try: |
no test coverage detected