diff --git a/tests/http/federation/test_matrix_federation_agent.py b/tests/http/federation/test_matrix_federation_agent.py
index 992d8f94fd..acfdcd3bca 100644
--- a/tests/http/federation/test_matrix_federation_agent.py
+++ b/tests/http/federation/test_matrix_federation_agent.py
@@ -14,7 +14,7 @@
import base64
import logging
import os
-from typing import Iterable, Optional
+from typing import Any, Awaitable, Callable, Generator, List, Optional, cast
from unittest.mock import Mock, patch
import treq
@@ -24,14 +24,19 @@ from zope.interface import implementer
from twisted.internet import defer
from twisted.internet._sslverify import ClientTLSOptions, OpenSSLCertificateOptions
-from twisted.internet.interfaces import IProtocolFactory
+from twisted.internet.defer import Deferred
+from twisted.internet.endpoints import _WrappingProtocol
+from twisted.internet.interfaces import (
+ IOpenSSLClientConnectionCreator,
+ IProtocolFactory,
+)
from twisted.internet.protocol import Factory
from twisted.protocols.tls import TLSMemoryBIOFactory, TLSMemoryBIOProtocol
from twisted.web._newclient import ResponseNeverReceived
from twisted.web.client import Agent
from twisted.web.http import HTTPChannel, Request
from twisted.web.http_headers import Headers
-from twisted.web.iweb import IPolicyForHTTPS
+from twisted.web.iweb import IPolicyForHTTPS, IResponse
from synapse.config.homeserver import HomeServerConfig
from synapse.crypto.context_factory import FederationPolicyForHTTPS
@@ -42,11 +47,21 @@ from synapse.http.federation.well_known_resolver import (
WellKnownResolver,
_cache_period_from_headers,
)
-from synapse.logging.context import SENTINEL_CONTEXT, LoggingContext, current_context
+from synapse.logging.context import (
+ SENTINEL_CONTEXT,
+ LoggingContext,
+ LoggingContextOrSentinel,
+ current_context,
+)
+from synapse.types import ISynapseReactor
from synapse.util.caches.ttlcache import TTLCache
from tests import unittest
-from tests.http import TestServerTLSConnectionFactory, get_test_ca_cert_file
+from tests.http import (
+ TestServerTLSConnectionFactory,
+ dummy_address,
+ get_test_ca_cert_file,
+)
from tests.server import FakeTransport, ThreadedMemoryReactorClock
from tests.utils import default_config
@@ -54,15 +69,17 @@ logger = logging.getLogger(__name__)
# Once Async Mocks or lambdas are supported this can go away.
-def generate_resolve_service(result):
- async def resolve_service(_):
+def generate_resolve_service(
+ result: List[Server],
+) -> Callable[[Any], Awaitable[List[Server]]]:
+ async def resolve_service(_: Any) -> List[Server]:
return result
return resolve_service
class MatrixFederationAgentTests(unittest.TestCase):
- def setUp(self):
+ def setUp(self) -> None:
self.reactor = ThreadedMemoryReactorClock()
self.mock_resolver = Mock()
@@ -75,8 +92,12 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.tls_factory = FederationPolicyForHTTPS(config)
- self.well_known_cache = TTLCache("test_cache", timer=self.reactor.seconds)
- self.had_well_known_cache = TTLCache("test_cache", timer=self.reactor.seconds)
+ self.well_known_cache: TTLCache[bytes, Optional[bytes]] = TTLCache(
+ "test_cache", timer=self.reactor.seconds
+ )
+ self.had_well_known_cache: TTLCache[bytes, bool] = TTLCache(
+ "test_cache", timer=self.reactor.seconds
+ )
self.well_known_resolver = WellKnownResolver(
self.reactor,
Agent(self.reactor, contextFactory=self.tls_factory),
@@ -89,8 +110,8 @@ class MatrixFederationAgentTests(unittest.TestCase):
self,
client_factory: IProtocolFactory,
ssl: bool = True,
- expected_sni: bytes = None,
- tls_sanlist: Optional[Iterable[bytes]] = None,
+ expected_sni: Optional[bytes] = None,
+ tls_sanlist: Optional[List[bytes]] = None,
) -> HTTPChannel:
"""Builds a test server, and completes the outgoing client connection
Args:
@@ -116,8 +137,8 @@ class MatrixFederationAgentTests(unittest.TestCase):
if ssl:
server_factory = _wrap_server_factory_for_tls(server_factory, tls_sanlist)
- server_protocol = server_factory.buildProtocol(None)
-
+ server_protocol = server_factory.buildProtocol(dummy_address)
+ assert server_protocol is not None
# now, tell the client protocol factory to build the client protocol (it will be a
# _WrappingProtocol, around a TLSMemoryBIOProtocol, around an
# HTTP11ClientProtocol) and wire the output of said protocol up to the server via
@@ -125,7 +146,8 @@ class MatrixFederationAgentTests(unittest.TestCase):
#
# Normally this would be done by the TCP socket code in Twisted, but we are
# stubbing that out here.
- client_protocol = client_factory.buildProtocol(None)
+ client_protocol = client_factory.buildProtocol(dummy_address)
+ assert isinstance(client_protocol, _WrappingProtocol)
client_protocol.makeConnection(
FakeTransport(server_protocol, self.reactor, client_protocol)
)
@@ -136,6 +158,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
)
if ssl:
+ assert isinstance(server_protocol, TLSMemoryBIOProtocol)
# fish the test server back out of the server-side TLS protocol.
http_protocol = server_protocol.wrappedProtocol
# grab a hold of the TLS connection, in case it gets torn down
@@ -144,6 +167,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
http_protocol = server_protocol
tls_connection = None
+ assert isinstance(http_protocol, HTTPChannel)
# give the reactor a pump to get the TLS juices flowing (if needed)
self.reactor.advance(0)
@@ -159,12 +183,14 @@ class MatrixFederationAgentTests(unittest.TestCase):
return http_protocol
@defer.inlineCallbacks
- def _make_get_request(self, uri: bytes):
+ def _make_get_request(
+ self, uri: bytes
+ ) -> Generator["Deferred[object]", object, IResponse]:
"""
Sends a simple GET request via the agent, and checks its logcontext management
"""
with LoggingContext("one") as context:
- fetch_d = self.agent.request(b"GET", uri)
+ fetch_d: Deferred[IResponse] = self.agent.request(b"GET", uri)
# Nothing happened yet
self.assertNoResult(fetch_d)
@@ -172,8 +198,9 @@ class MatrixFederationAgentTests(unittest.TestCase):
# should have reset logcontext to the sentinel
_check_logcontext(SENTINEL_CONTEXT)
+ fetch_res: IResponse
try:
- fetch_res = yield fetch_d
+ fetch_res = yield fetch_d # type: ignore[misc, assignment]
return fetch_res
except Exception as e:
logger.info("Fetch of %s failed: %s", uri.decode("ascii"), e)
@@ -216,7 +243,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
request: Request,
content: bytes,
headers: Optional[dict] = None,
- ):
+ ) -> None:
"""Check that an incoming request looks like a valid .well-known request, and
send back the response.
"""
@@ -237,16 +264,16 @@ class MatrixFederationAgentTests(unittest.TestCase):
because it is created too early during setUp
"""
return MatrixFederationAgent(
- reactor=self.reactor,
+ reactor=cast(ISynapseReactor, self.reactor),
tls_client_options_factory=self.tls_factory,
- user_agent="test-agent", # Note that this is unused since _well_known_resolver is provided.
+ user_agent=b"test-agent", # Note that this is unused since _well_known_resolver is provided.
ip_whitelist=IPSet(),
ip_blacklist=IPSet(),
_srv_resolver=self.mock_resolver,
_well_known_resolver=self.well_known_resolver,
)
- def test_get(self):
+ def test_get(self) -> None:
"""happy-path test of a GET request with an explicit port"""
self._do_get()
@@ -254,11 +281,11 @@ class MatrixFederationAgentTests(unittest.TestCase):
os.environ,
{"https_proxy": "proxy.com", "no_proxy": "testserv"},
)
- def test_get_bypass_proxy(self):
+ def test_get_bypass_proxy(self) -> None:
"""test of a GET request with an explicit port and bypass proxy"""
self._do_get()
- def _do_get(self):
+ def _do_get(self) -> None:
"""test of a GET request with an explicit port"""
self.agent = self._make_agent()
@@ -318,7 +345,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
@patch.dict(
os.environ, {"https_proxy": "http://proxy.com", "no_proxy": "unused.com"}
)
- def test_get_via_http_proxy(self):
+ def test_get_via_http_proxy(self) -> None:
"""test for federation request through a http proxy"""
self._do_get_via_proxy(expect_proxy_ssl=False, expected_auth_credentials=None)
@@ -326,7 +353,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
os.environ,
{"https_proxy": "http://user:pass@proxy.com", "no_proxy": "unused.com"},
)
- def test_get_via_http_proxy_with_auth(self):
+ def test_get_via_http_proxy_with_auth(self) -> None:
"""test for federation request through a http proxy with authentication"""
self._do_get_via_proxy(
expect_proxy_ssl=False, expected_auth_credentials=b"user:pass"
@@ -335,7 +362,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
@patch.dict(
os.environ, {"https_proxy": "https://proxy.com", "no_proxy": "unused.com"}
)
- def test_get_via_https_proxy(self):
+ def test_get_via_https_proxy(self) -> None:
"""test for federation request through a https proxy"""
self._do_get_via_proxy(expect_proxy_ssl=True, expected_auth_credentials=None)
@@ -343,7 +370,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
os.environ,
{"https_proxy": "https://user:pass@proxy.com", "no_proxy": "unused.com"},
)
- def test_get_via_https_proxy_with_auth(self):
+ def test_get_via_https_proxy_with_auth(self) -> None:
"""test for federation request through a https proxy with authentication"""
self._do_get_via_proxy(
expect_proxy_ssl=True, expected_auth_credentials=b"user:pass"
@@ -353,7 +380,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self,
expect_proxy_ssl: bool = False,
expected_auth_credentials: Optional[bytes] = None,
- ):
+ ) -> None:
"""Send a https federation request via an agent and check that it is correctly
received at the proxy and client. The proxy can use either http or https.
Args:
@@ -418,10 +445,12 @@ class MatrixFederationAgentTests(unittest.TestCase):
# now we make another test server to act as the upstream HTTP server.
server_ssl_protocol = _wrap_server_factory_for_tls(
_get_test_protocol_factory()
- ).buildProtocol(None)
+ ).buildProtocol(dummy_address)
+ assert isinstance(server_ssl_protocol, TLSMemoryBIOProtocol)
# Tell the HTTP server to send outgoing traffic back via the proxy's transport.
proxy_server_transport = proxy_server.transport
+ assert proxy_server_transport is not None
server_ssl_protocol.makeConnection(proxy_server_transport)
# ... and replace the protocol on the proxy's transport with the
@@ -451,6 +480,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
# now there should be a pending request
http_server = server_ssl_protocol.wrappedProtocol
+ assert isinstance(http_server, HTTPChannel)
self.assertEqual(len(http_server.requests), 1)
request = http_server.requests[0]
@@ -491,7 +521,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
json = self.successResultOf(treq.json_content(response))
self.assertEqual(json, {"a": 1})
- def test_get_ip_address(self):
+ def test_get_ip_address(self) -> None:
"""
Test the behaviour when the server name contains an explicit IP (with no port)
"""
@@ -526,7 +556,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_get_ipv6_address(self):
+ def test_get_ipv6_address(self) -> None:
"""
Test the behaviour when the server name contains an explicit IPv6 address
(with no port)
@@ -562,7 +592,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_get_ipv6_address_with_port(self):
+ def test_get_ipv6_address_with_port(self) -> None:
"""
Test the behaviour when the server name contains an explicit IPv6 address
(with explicit port)
@@ -598,7 +628,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_get_hostname_bad_cert(self):
+ def test_get_hostname_bad_cert(self) -> None:
"""
Test the behaviour when the certificate on the server doesn't match the hostname
"""
@@ -651,7 +681,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
failure_reason = e.value.reasons[0]
self.assertIsInstance(failure_reason.value, VerificationError)
- def test_get_ip_address_bad_cert(self):
+ def test_get_ip_address_bad_cert(self) -> None:
"""
Test the behaviour when the server name contains an explicit IP, but
the server cert doesn't cover it
@@ -684,7 +714,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
failure_reason = e.value.reasons[0]
self.assertIsInstance(failure_reason.value, VerificationError)
- def test_get_no_srv_no_well_known(self):
+ def test_get_no_srv_no_well_known(self) -> None:
"""
Test the behaviour when the server name has no port, no SRV, and no well-known
"""
@@ -740,7 +770,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_get_well_known(self):
+ def test_get_well_known(self) -> None:
"""Test the behaviour when the .well-known delegates elsewhere"""
self.agent = self._make_agent()
@@ -802,7 +832,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.well_known_cache.expire()
self.assertNotIn(b"testserv", self.well_known_cache)
- def test_get_well_known_redirect(self):
+ def test_get_well_known_redirect(self) -> None:
"""Test the behaviour when the server name has no port and no SRV record, but
the .well-known has a 300 redirect
"""
@@ -892,7 +922,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.well_known_cache.expire()
self.assertNotIn(b"testserv", self.well_known_cache)
- def test_get_invalid_well_known(self):
+ def test_get_invalid_well_known(self) -> None:
"""
Test the behaviour when the server name has an *invalid* well-known (and no SRV)
"""
@@ -945,7 +975,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_get_well_known_unsigned_cert(self):
+ def test_get_well_known_unsigned_cert(self) -> None:
"""Test the behaviour when the .well-known server presents a cert
not signed by a CA
"""
@@ -969,7 +999,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
ip_blacklist=IPSet(),
_srv_resolver=self.mock_resolver,
_well_known_resolver=WellKnownResolver(
- self.reactor,
+ cast(ISynapseReactor, self.reactor),
Agent(self.reactor, contextFactory=tls_factory),
b"test-agent",
well_known_cache=self.well_known_cache,
@@ -999,7 +1029,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
b"_matrix._tcp.testserv"
)
- def test_get_hostname_srv(self):
+ def test_get_hostname_srv(self) -> None:
"""
Test the behaviour when there is a single SRV record
"""
@@ -1041,7 +1071,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_get_well_known_srv(self):
+ def test_get_well_known_srv(self) -> None:
"""Test the behaviour when the .well-known redirects to a place where there
is a SRV.
"""
@@ -1101,7 +1131,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_idna_servername(self):
+ def test_idna_servername(self) -> None:
"""test the behaviour when the server name has idna chars in"""
self.agent = self._make_agent()
@@ -1163,7 +1193,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_idna_srv_target(self):
+ def test_idna_srv_target(self) -> None:
"""test the behaviour when the target of a SRV record has idna chars"""
self.agent = self._make_agent()
@@ -1206,7 +1236,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
self.reactor.pump((0.1,))
self.successResultOf(test_d)
- def test_well_known_cache(self):
+ def test_well_known_cache(self) -> None:
self.reactor.lookups["testserv"] = "1.2.3.4"
fetch_d = defer.ensureDeferred(
@@ -1262,7 +1292,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
r = self.successResultOf(fetch_d)
self.assertEqual(r.delegated_server, b"other-server")
- def test_well_known_cache_with_temp_failure(self):
+ def test_well_known_cache_with_temp_failure(self) -> None:
"""Test that we refetch well-known before the cache expires, and that
it ignores transient errors.
"""
@@ -1341,7 +1371,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
r = self.successResultOf(fetch_d)
self.assertEqual(r.delegated_server, None)
- def test_well_known_too_large(self):
+ def test_well_known_too_large(self) -> None:
"""A well-known query that returns a result which is too large should be rejected."""
self.reactor.lookups["testserv"] = "1.2.3.4"
@@ -1367,7 +1397,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
r = self.successResultOf(fetch_d)
self.assertIsNone(r.delegated_server)
- def test_srv_fallbacks(self):
+ def test_srv_fallbacks(self) -> None:
"""Test that other SRV results are tried if the first one fails."""
self.agent = self._make_agent()
@@ -1427,7 +1457,7 @@ class MatrixFederationAgentTests(unittest.TestCase):
class TestCachePeriodFromHeaders(unittest.TestCase):
- def test_cache_control(self):
+ def test_cache_control(self) -> None:
# uppercase
self.assertEqual(
_cache_period_from_headers(
@@ -1464,7 +1494,7 @@ class TestCachePeriodFromHeaders(unittest.TestCase):
0,
)
- def test_expires(self):
+ def test_expires(self) -> None:
self.assertEqual(
_cache_period_from_headers(
Headers({b"Expires": [b"Wed, 30 Jan 2019 07:35:33 GMT"]}),
@@ -1491,14 +1521,14 @@ class TestCachePeriodFromHeaders(unittest.TestCase):
self.assertEqual(_cache_period_from_headers(Headers({b"Expires": [b"0"]})), 0)
-def _check_logcontext(context):
+def _check_logcontext(context: LoggingContextOrSentinel) -> None:
current = current_context()
if current is not context:
raise AssertionError("Expected logcontext %s but was %s" % (context, current))
def _wrap_server_factory_for_tls(
- factory: IProtocolFactory, sanlist: Iterable[bytes] = None
+ factory: IProtocolFactory, sanlist: Optional[List[bytes]] = None
) -> IProtocolFactory:
"""Wrap an existing Protocol Factory with a test TLSMemoryBIOFactory
The resultant factory will create a TLS server which presents a certificate
@@ -1537,7 +1567,7 @@ def _get_test_protocol_factory() -> IProtocolFactory:
return server_factory
-def _log_request(request: str):
+def _log_request(request: str) -> None:
"""Implements Factory.log, which is expected by Request.finish"""
logger.info(f"Completed request {request}")
@@ -1547,6 +1577,8 @@ class TrustingTLSPolicyForHTTPS:
"""An IPolicyForHTTPS which checks that the certificate belongs to the
right server, but doesn't check the certificate chain."""
- def creatorForNetloc(self, hostname, port):
+ def creatorForNetloc(
+ self, hostname: bytes, port: int
+ ) -> IOpenSSLClientConnectionCreator:
certificateOptions = OpenSSLCertificateOptions()
return ClientTLSOptions(hostname, certificateOptions.getContext())
diff --git a/tests/http/federation/test_srv_resolver.py b/tests/http/federation/test_srv_resolver.py
index 77ce8432ac..7748f56ee6 100644
--- a/tests/http/federation/test_srv_resolver.py
+++ b/tests/http/federation/test_srv_resolver.py
@@ -12,7 +12,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
-
+from typing import Dict, Generator, List, Tuple, cast
from unittest.mock import Mock
from twisted.internet import defer
@@ -20,7 +20,7 @@ from twisted.internet.defer import Deferred
from twisted.internet.error import ConnectError
from twisted.names import dns, error
-from synapse.http.federation.srv_resolver import SrvResolver
+from synapse.http.federation.srv_resolver import Server, SrvResolver
from synapse.logging.context import LoggingContext, current_context
from tests import unittest
@@ -28,7 +28,7 @@ from tests.utils import MockClock
class SrvResolverTestCase(unittest.TestCase):
- def test_resolve(self):
+ def test_resolve(self) -> None:
dns_client_mock = Mock()
service_name = b"test_service.example.com"
@@ -38,18 +38,19 @@ class SrvResolverTestCase(unittest.TestCase):
type=dns.SRV, payload=dns.Record_SRV(target=host_name)
)
- result_deferred = Deferred()
+ result_deferred: "Deferred[Tuple[List[dns.RRHeader], None, None]]" = Deferred()
dns_client_mock.lookupService.return_value = result_deferred
- cache = {}
+ cache: Dict[bytes, List[Server]] = {}
resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
@defer.inlineCallbacks
- def do_lookup():
+ def do_lookup() -> Generator["Deferred[object]", object, List[Server]]:
with LoggingContext("one") as ctx:
resolve_d = resolver.resolve_service(service_name)
- result = yield defer.ensureDeferred(resolve_d)
+ result: List[Server]
+ result = yield defer.ensureDeferred(resolve_d) # type: ignore[assignment]
# should have restored our context
self.assertIs(current_context(), ctx)
@@ -70,7 +71,9 @@ class SrvResolverTestCase(unittest.TestCase):
self.assertEqual(servers[0].host, host_name)
@defer.inlineCallbacks
- def test_from_cache_expired_and_dns_fail(self):
+ def test_from_cache_expired_and_dns_fail(
+ self,
+ ) -> Generator["Deferred[object]", object, None]:
dns_client_mock = Mock()
dns_client_mock.lookupService.return_value = defer.fail(error.DNSServerError())
@@ -81,10 +84,13 @@ class SrvResolverTestCase(unittest.TestCase):
entry.priority = 0
entry.weight = 0
- cache = {service_name: [entry]}
+ cache = {service_name: [cast(Server, entry)]}
resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
- servers = yield defer.ensureDeferred(resolver.resolve_service(service_name))
+ servers: List[Server]
+ servers = yield defer.ensureDeferred(
+ resolver.resolve_service(service_name)
+ ) # type: ignore[assignment]
dns_client_mock.lookupService.assert_called_once_with(service_name)
@@ -92,7 +98,7 @@ class SrvResolverTestCase(unittest.TestCase):
self.assertEqual(servers, cache[service_name])
@defer.inlineCallbacks
- def test_from_cache(self):
+ def test_from_cache(self) -> Generator["Deferred[object]", object, None]:
clock = MockClock()
dns_client_mock = Mock(spec_set=["lookupService"])
@@ -105,12 +111,15 @@ class SrvResolverTestCase(unittest.TestCase):
entry.priority = 0
entry.weight = 0
- cache = {service_name: [entry]}
+ cache = {service_name: [cast(Server, entry)]}
resolver = SrvResolver(
dns_client=dns_client_mock, cache=cache, get_time=clock.time
)
- servers = yield defer.ensureDeferred(resolver.resolve_service(service_name))
+ servers: List[Server]
+ servers = yield defer.ensureDeferred(
+ resolver.resolve_service(service_name)
+ ) # type: ignore[assignment]
self.assertFalse(dns_client_mock.lookupService.called)
@@ -118,45 +127,48 @@ class SrvResolverTestCase(unittest.TestCase):
self.assertEqual(servers, cache[service_name])
@defer.inlineCallbacks
- def test_empty_cache(self):
+ def test_empty_cache(self) -> Generator["Deferred[object]", object, None]:
dns_client_mock = Mock()
dns_client_mock.lookupService.return_value = defer.fail(error.DNSServerError())
service_name = b"test_service.example.com"
- cache = {}
+ cache: Dict[bytes, List[Server]] = {}
resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
with self.assertRaises(error.DNSServerError):
yield defer.ensureDeferred(resolver.resolve_service(service_name))
@defer.inlineCallbacks
- def test_name_error(self):
+ def test_name_error(self) -> Generator["Deferred[object]", object, None]:
dns_client_mock = Mock()
dns_client_mock.lookupService.return_value = defer.fail(error.DNSNameError())
service_name = b"test_service.example.com"
- cache = {}
+ cache: Dict[bytes, List[Server]] = {}
resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
- servers = yield defer.ensureDeferred(resolver.resolve_service(service_name))
+ servers: List[Server]
+ servers = yield defer.ensureDeferred(
+ resolver.resolve_service(service_name)
+ ) # type: ignore[assignment]
self.assertEqual(len(servers), 0)
self.assertEqual(len(cache), 0)
- def test_disabled_service(self):
+ def test_disabled_service(self) -> None:
"""
test the behaviour when there is a single record which is ".".
"""
service_name = b"test_service.example.com"
- lookup_deferred = Deferred()
+ lookup_deferred: "Deferred[Tuple[List[dns.RRHeader], None, None]]" = Deferred()
dns_client_mock = Mock()
dns_client_mock.lookupService.return_value = lookup_deferred
- cache = {}
+ cache: Dict[bytes, List[Server]] = {}
resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
# Old versions of Twisted don't have an ensureDeferred in failureResultOf.
@@ -173,16 +185,16 @@ class SrvResolverTestCase(unittest.TestCase):
self.failureResultOf(resolve_d, ConnectError)
- def test_non_srv_answer(self):
+ def test_non_srv_answer(self) -> None:
"""
test the behaviour when the dns server gives us a spurious non-SRV response
"""
service_name = b"test_service.example.com"
- lookup_deferred = Deferred()
+ lookup_deferred: "Deferred[Tuple[List[dns.RRHeader], None, None]]" = Deferred()
dns_client_mock = Mock()
dns_client_mock.lookupService.return_value = lookup_deferred
- cache = {}
+ cache: Dict[bytes, List[Server]] = {}
resolver = SrvResolver(dns_client=dns_client_mock, cache=cache)
# Old versions of Twisted don't have an ensureDeferred in successResultOf.
|
