diff --git a/tests/handlers/test_register.py b/tests/handlers/test_register.py
index 782ef09cf4..1db99b3c00 100644
--- a/tests/handlers/test_register.py
+++ b/tests/handlers/test_register.py
@@ -62,7 +62,7 @@ class TestSpamChecker:
request_info: Collection[Tuple[str, str]],
auth_provider_id: Optional[str],
) -> RegistrationBehaviour:
- pass
+ return RegistrationBehaviour.ALLOW
class DenyAll(TestSpamChecker):
@@ -111,7 +111,7 @@ class TestLegacyRegistrationSpamChecker:
username: Optional[str],
request_info: Collection[Tuple[str, str]],
) -> RegistrationBehaviour:
- pass
+ return RegistrationBehaviour.ALLOW
class LegacyAllowAll(TestLegacyRegistrationSpamChecker):
diff --git a/tests/http/federation/test_matrix_federation_agent.py b/tests/http/federation/test_matrix_federation_agent.py
index acfdcd3bca..d27422515c 100644
--- a/tests/http/federation/test_matrix_federation_agent.py
+++ b/tests/http/federation/test_matrix_federation_agent.py
@@ -63,7 +63,7 @@ from tests.http import (
get_test_ca_cert_file,
)
from tests.server import FakeTransport, ThreadedMemoryReactorClock
-from tests.utils import default_config
+from tests.utils import checked_cast, default_config
logger = logging.getLogger(__name__)
@@ -146,8 +146,10 @@ 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(dummy_address)
- assert isinstance(client_protocol, _WrappingProtocol)
+ # NB: we use a checked_cast here to workaround https://github.com/Shoobx/mypy-zope/issues/91)
+ client_protocol = checked_cast(
+ _WrappingProtocol, client_factory.buildProtocol(dummy_address)
+ )
client_protocol.makeConnection(
FakeTransport(server_protocol, self.reactor, client_protocol)
)
@@ -446,7 +448,6 @@ class MatrixFederationAgentTests(unittest.TestCase):
server_ssl_protocol = _wrap_server_factory_for_tls(
_get_test_protocol_factory()
).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
@@ -1529,7 +1530,7 @@ def _check_logcontext(context: LoggingContextOrSentinel) -> None:
def _wrap_server_factory_for_tls(
factory: IProtocolFactory, sanlist: Optional[List[bytes]] = None
-) -> IProtocolFactory:
+) -> TLSMemoryBIOFactory:
"""Wrap an existing Protocol Factory with a test TLSMemoryBIOFactory
The resultant factory will create a TLS server which presents a certificate
signed by our test CA, valid for the domains in `sanlist`
diff --git a/tests/http/test_proxyagent.py b/tests/http/test_proxyagent.py
index a817940730..22fdc7f5f2 100644
--- a/tests/http/test_proxyagent.py
+++ b/tests/http/test_proxyagent.py
@@ -43,6 +43,7 @@ from tests.http import (
)
from tests.server import FakeTransport, ThreadedMemoryReactorClock
from tests.unittest import TestCase
+from tests.utils import checked_cast
logger = logging.getLogger(__name__)
@@ -620,7 +621,6 @@ class MatrixFederationAgentTests(TestCase):
server_ssl_protocol = _wrap_server_factory_for_tls(
_get_test_protocol_factory()
).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
@@ -757,12 +757,14 @@ class MatrixFederationAgentTests(TestCase):
assert isinstance(proxy_server, HTTPChannel)
# fish the transports back out so that we can do the old switcheroo
- s2c_transport = proxy_server.transport
- assert isinstance(s2c_transport, FakeTransport)
- client_protocol = s2c_transport.other
- assert isinstance(client_protocol, _WrappingProtocol)
- c2s_transport = client_protocol.transport
- assert isinstance(c2s_transport, FakeTransport)
+ # To help mypy out with the various Protocols and wrappers and mocks, we do
+ # some explicit casting. Without the casts, we hit the bug I reported at
+ # https://github.com/Shoobx/mypy-zope/issues/91 .
+ # We also double-checked these casts at runtime (test-time) because I found it
+ # quite confusing to deduce these types in the first place!
+ s2c_transport = checked_cast(FakeTransport, proxy_server.transport)
+ client_protocol = checked_cast(_WrappingProtocol, s2c_transport.other)
+ c2s_transport = checked_cast(FakeTransport, client_protocol.transport)
# the FakeTransport is async, so we need to pump the reactor
self.reactor.advance(0)
@@ -822,9 +824,9 @@ class MatrixFederationAgentTests(TestCase):
@patch.dict(os.environ, {"http_proxy": "proxy.com:8888"})
def test_proxy_with_no_scheme(self) -> None:
http_proxy_agent = ProxyAgent(self.reactor, use_proxy=True)
- assert isinstance(http_proxy_agent.http_proxy_endpoint, HostnameEndpoint)
- self.assertEqual(http_proxy_agent.http_proxy_endpoint._hostStr, "proxy.com")
- self.assertEqual(http_proxy_agent.http_proxy_endpoint._port, 8888)
+ proxy_ep = checked_cast(HostnameEndpoint, http_proxy_agent.http_proxy_endpoint)
+ self.assertEqual(proxy_ep._hostStr, "proxy.com")
+ self.assertEqual(proxy_ep._port, 8888)
@patch.dict(os.environ, {"http_proxy": "socks://proxy.com:8888"})
def test_proxy_with_unsupported_scheme(self) -> None:
@@ -834,25 +836,21 @@ class MatrixFederationAgentTests(TestCase):
@patch.dict(os.environ, {"http_proxy": "http://proxy.com:8888"})
def test_proxy_with_http_scheme(self) -> None:
http_proxy_agent = ProxyAgent(self.reactor, use_proxy=True)
- assert isinstance(http_proxy_agent.http_proxy_endpoint, HostnameEndpoint)
- self.assertEqual(http_proxy_agent.http_proxy_endpoint._hostStr, "proxy.com")
- self.assertEqual(http_proxy_agent.http_proxy_endpoint._port, 8888)
+ proxy_ep = checked_cast(HostnameEndpoint, http_proxy_agent.http_proxy_endpoint)
+ self.assertEqual(proxy_ep._hostStr, "proxy.com")
+ self.assertEqual(proxy_ep._port, 8888)
@patch.dict(os.environ, {"http_proxy": "https://proxy.com:8888"})
def test_proxy_with_https_scheme(self) -> None:
https_proxy_agent = ProxyAgent(self.reactor, use_proxy=True)
- assert isinstance(https_proxy_agent.http_proxy_endpoint, _WrapperEndpoint)
- self.assertEqual(
- https_proxy_agent.http_proxy_endpoint._wrappedEndpoint._hostStr, "proxy.com"
- )
- self.assertEqual(
- https_proxy_agent.http_proxy_endpoint._wrappedEndpoint._port, 8888
- )
+ proxy_ep = checked_cast(_WrapperEndpoint, https_proxy_agent.http_proxy_endpoint)
+ self.assertEqual(proxy_ep._wrappedEndpoint._hostStr, "proxy.com")
+ self.assertEqual(proxy_ep._wrappedEndpoint._port, 8888)
def _wrap_server_factory_for_tls(
factory: IProtocolFactory, sanlist: Optional[List[bytes]] = None
-) -> IProtocolFactory:
+) -> TLSMemoryBIOFactory:
"""Wrap an existing Protocol Factory with a test TLSMemoryBIOFactory
The resultant factory will create a TLS server which presents a certificate
diff --git a/tests/logging/test_remote_handler.py b/tests/logging/test_remote_handler.py
index c08954d887..5191e31a8a 100644
--- a/tests/logging/test_remote_handler.py
+++ b/tests/logging/test_remote_handler.py
@@ -21,6 +21,7 @@ from synapse.logging import RemoteHandler
from tests.logging import LoggerCleanupMixin
from tests.server import FakeTransport, get_clock
from tests.unittest import TestCase
+from tests.utils import checked_cast
def connect_logging_client(
@@ -56,8 +57,8 @@ class RemoteHandlerTestCase(LoggerCleanupMixin, TestCase):
client, server = connect_logging_client(self.reactor, 0)
# Trigger data being sent
- assert isinstance(client.transport, FakeTransport)
- client.transport.flush()
+ client_transport = checked_cast(FakeTransport, client.transport)
+ client_transport.flush()
# One log message, with a single trailing newline
logs = server.data.decode("utf8").splitlines()
@@ -89,8 +90,8 @@ class RemoteHandlerTestCase(LoggerCleanupMixin, TestCase):
# Allow the reconnection
client, server = connect_logging_client(self.reactor, 0)
- assert isinstance(client.transport, FakeTransport)
- client.transport.flush()
+ client_transport = checked_cast(FakeTransport, client.transport)
+ client_transport.flush()
# Only the 7 infos made it through, the debugs were elided
logs = server.data.splitlines()
@@ -123,8 +124,8 @@ class RemoteHandlerTestCase(LoggerCleanupMixin, TestCase):
# Allow the reconnection
client, server = connect_logging_client(self.reactor, 0)
- assert isinstance(client.transport, FakeTransport)
- client.transport.flush()
+ client_transport = checked_cast(FakeTransport, client.transport)
+ client_transport.flush()
# The 10 warnings made it through, the debugs and infos were elided
logs = server.data.splitlines()
@@ -148,8 +149,8 @@ class RemoteHandlerTestCase(LoggerCleanupMixin, TestCase):
# Allow the reconnection
client, server = connect_logging_client(self.reactor, 0)
- assert isinstance(client.transport, FakeTransport)
- client.transport.flush()
+ client_transport = checked_cast(FakeTransport, client.transport)
+ client_transport.flush()
# The first five and last five warnings made it through, the debugs and
# infos were elided
diff --git a/tests/rest/client/test_auth.py b/tests/rest/client/test_auth.py
index 208ec44829..f4e1e7de43 100644
--- a/tests/rest/client/test_auth.py
+++ b/tests/rest/client/test_auth.py
@@ -43,6 +43,9 @@ class DummyRecaptchaChecker(UserInteractiveAuthChecker):
super().__init__(hs)
self.recaptcha_attempts: List[Tuple[dict, str]] = []
+ def is_enabled(self) -> bool:
+ return True
+
def check_auth(self, authdict: dict, clientip: str) -> Any:
self.recaptcha_attempts.append((authdict, clientip))
return succeed(True)
diff --git a/tests/rest/client/test_third_party_rules.py b/tests/rest/client/test_third_party_rules.py
index 3325d43a2f..5fa3440691 100644
--- a/tests/rest/client/test_third_party_rules.py
+++ b/tests/rest/client/test_third_party_rules.py
@@ -425,7 +425,7 @@ class ThirdPartyRulesTestCase(unittest.FederatingHomeserverTestCase):
async def test_fn(
event: EventBase, state_events: StateMap[EventBase]
) -> Tuple[bool, Optional[JsonDict]]:
- if event.is_state and event.type == EventTypes.PowerLevels:
+ if event.is_state() and event.type == EventTypes.PowerLevels:
await api.create_and_send_event_into_room(
{
"room_id": event.room_id,
diff --git a/tests/utils.py b/tests/utils.py
index 15fabbc2d0..a0ac11bc5c 100644
--- a/tests/utils.py
+++ b/tests/utils.py
@@ -15,7 +15,7 @@
import atexit
import os
-from typing import Any, Callable, Dict, List, Tuple, Union, overload
+from typing import Any, Callable, Dict, List, Tuple, Type, TypeVar, Union, overload
import attr
from typing_extensions import Literal, ParamSpec
@@ -341,3 +341,27 @@ async def create_room(hs: HomeServer, room_id: str, creator_id: str) -> None:
context = await unpersisted_context.persist(event)
await persistence_store.persist_event(event, context)
+
+
+T = TypeVar("T")
+
+
+def checked_cast(type: Type[T], x: object) -> T:
+ """A version of typing.cast that is checked at runtime.
+
+ We have our own function for this for two reasons:
+
+ 1. typing.cast itself is deliberately a no-op at runtime, see
+ https://docs.python.org/3/library/typing.html#typing.cast
+ 2. To help workaround a mypy-zope bug https://github.com/Shoobx/mypy-zope/issues/91
+ where mypy would erroneously consider `isinstance(x, type)` to be false in all
+ circumstances.
+
+ For this to make sense, `T` needs to be something that `isinstance` can check; see
+ https://docs.python.org/3/library/functions.html?highlight=isinstance#isinstance
+ https://docs.python.org/3/glossary.html#term-abstract-base-class
+ https://docs.python.org/3/library/typing.html#typing.runtime_checkable
+ for more details.
+ """
+ assert isinstance(x, type)
+ return x
|
