Mirrors
/
synapse-old
mirror of https://github.com/matrix-org/synapse.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Use inline type hints in `tests/` (#10350) 

This PR is tantamount to running:

    python3.8 -m com2ann -v 6 tests/

(com2ann requires python 3.8 to run)
This commit is contained in:
Jonathan de Jong 2021-07-13 12:52:58 +02:00 committed by GitHub
parent 89cfc3dd98
commit 93729719b8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
18 changed files with 62 additions and 63 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
changelog.d/10350.misc Normal file
View File

@ -0,0 +1 @@
Convert internal type variable syntax to reflect wider ecosystem use.

6
tests/events/test_presence_router.py
View File

@ -152,7 +152,7 @@ class PresenceRouterTestCase(FederatingHomeserverTestCase):
) )
self.assertEqual(len(presence_updates), 1) self.assertEqual(len(presence_updates), 1)
presence_update = presence_updates[0] # type: UserPresenceState presence_update: UserPresenceState = presence_updates[0]
self.assertEqual(presence_update.user_id, self.other_user_one_id) self.assertEqual(presence_update.user_id, self.other_user_one_id)
self.assertEqual(presence_update.state, "online") self.assertEqual(presence_update.state, "online")
self.assertEqual(presence_update.status_msg, "boop") self.assertEqual(presence_update.status_msg, "boop")
@ -274,7 +274,7 @@ class PresenceRouterTestCase(FederatingHomeserverTestCase):
presence_updates, _ = sync_presence(self, self.other_user_id) presence_updates, _ = sync_presence(self, self.other_user_id)
self.assertEqual(len(presence_updates), 1) self.assertEqual(len(presence_updates), 1)
presence_update = presence_updates[0] # type: UserPresenceState presence_update: UserPresenceState = presence_updates[0]
self.assertEqual(presence_update.user_id, self.other_user_id) self.assertEqual(presence_update.user_id, self.other_user_id)
self.assertEqual(presence_update.state, "online") self.assertEqual(presence_update.state, "online")
self.assertEqual(presence_update.status_msg, "I'm online!") self.assertEqual(presence_update.status_msg, "I'm online!")
@ -320,7 +320,7 @@ class PresenceRouterTestCase(FederatingHomeserverTestCase):
) )
for call in calls: for call in calls:
call_args = call[0] call_args = call[0]
federation_transaction = call_args[0] # type: Transaction federation_transaction: Transaction = call_args[0]
# Get the sent EDUs in this transaction # Get the sent EDUs in this transaction
edus = federation_transaction.get_dict()["edus"] edus = federation_transaction.get_dict()["edus"]

16
tests/module_api/test_api.py
View File

@ -100,9 +100,9 @@ class ModuleApiTestCase(HomeserverTestCase):
"content": content, "content": content,
"sender": user_id, "sender": user_id,
} }
event = self.get_success( event: EventBase = self.get_success(
self.module_api.create_and_send_event_into_room(event_dict) self.module_api.create_and_send_event_into_room(event_dict)
) # type: EventBase )
self.assertEqual(event.sender, user_id) self.assertEqual(event.sender, user_id)
self.assertEqual(event.type, "m.room.message") self.assertEqual(event.type, "m.room.message")
self.assertEqual(event.room_id, room_id) self.assertEqual(event.room_id, room_id)
@ -136,9 +136,9 @@ class ModuleApiTestCase(HomeserverTestCase):
"sender": user_id, "sender": user_id,
"state_key": "", "state_key": "",
} }
event = self.get_success( event: EventBase = self.get_success(
self.module_api.create_and_send_event_into_room(event_dict) self.module_api.create_and_send_event_into_room(event_dict)
) # type: EventBase )
self.assertEqual(event.sender, user_id) self.assertEqual(event.sender, user_id)
self.assertEqual(event.type, "m.room.power_levels") self.assertEqual(event.type, "m.room.power_levels")
self.assertEqual(event.room_id, room_id) self.assertEqual(event.room_id, room_id)
@ -281,7 +281,7 @@ class ModuleApiTestCase(HomeserverTestCase):
) )
for call in calls: for call in calls:
call_args = call[0] call_args = call[0]
federation_transaction = call_args[0] # type: Transaction federation_transaction: Transaction = call_args[0]
# Get the sent EDUs in this transaction # Get the sent EDUs in this transaction
edus = federation_transaction.get_dict()["edus"] edus = federation_transaction.get_dict()["edus"]
@ -390,7 +390,7 @@ def _test_sending_local_online_presence_to_local_user(
) )
test_case.assertEqual(len(presence_updates), 1) test_case.assertEqual(len(presence_updates), 1)
presence_update = presence_updates[0] # type: UserPresenceState presence_update: UserPresenceState = presence_updates[0]
test_case.assertEqual(presence_update.user_id, test_case.presence_sender_id) test_case.assertEqual(presence_update.user_id, test_case.presence_sender_id)
test_case.assertEqual(presence_update.state, "online") test_case.assertEqual(presence_update.state, "online")
@ -443,7 +443,7 @@ def _test_sending_local_online_presence_to_local_user(
) )
test_case.assertEqual(len(presence_updates), 1) test_case.assertEqual(len(presence_updates), 1)
presence_update = presence_updates[0] # type: UserPresenceState presence_update: UserPresenceState = presence_updates[0]
test_case.assertEqual(presence_update.user_id, test_case.presence_sender_id) test_case.assertEqual(presence_update.user_id, test_case.presence_sender_id)
test_case.assertEqual(presence_update.state, "online") test_case.assertEqual(presence_update.state, "online")
@ -454,7 +454,7 @@ def _test_sending_local_online_presence_to_local_user(
) )
test_case.assertEqual(len(presence_updates), 1) test_case.assertEqual(len(presence_updates), 1)
presence_update = presence_updates[0] # type: UserPresenceState presence_update: UserPresenceState = presence_updates[0]
test_case.assertEqual(presence_update.user_id, test_case.presence_sender_id) test_case.assertEqual(presence_update.user_id, test_case.presence_sender_id)
test_case.assertEqual(presence_update.state, "online") test_case.assertEqual(presence_update.state, "online")

12
tests/replication/_base.py
View File

@ -53,9 +53,9 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
# build a replication server # build a replication server
server_factory = ReplicationStreamProtocolFactory(hs) server_factory = ReplicationStreamProtocolFactory(hs)
self.streamer = hs.get_replication_streamer() self.streamer = hs.get_replication_streamer()
self.server = server_factory.buildProtocol( self.server: ServerReplicationStreamProtocol = server_factory.buildProtocol(
None None
) # type: ServerReplicationStreamProtocol )
# Make a new HomeServer object for the worker # Make a new HomeServer object for the worker
self.reactor.lookups["testserv"] = "1.2.3.4" self.reactor.lookups["testserv"] = "1.2.3.4"
@ -195,7 +195,7 @@ class BaseStreamTestCase(unittest.HomeserverTestCase):
fetching updates for given stream. fetching updates for given stream.
""" """
path = request.path # type: bytes # type: ignore path: bytes = request.path # type: ignore
self.assertRegex( self.assertRegex(
path, path,
br"^/_synapse/replication/get_repl_stream_updates/%s/[^/]+$" br"^/_synapse/replication/get_repl_stream_updates/%s/[^/]+$"
@ -212,7 +212,7 @@ class BaseMultiWorkerStreamTestCase(unittest.HomeserverTestCase):
unlike `BaseStreamTestCase`. unlike `BaseStreamTestCase`.
""" """
servlets = [] # type: List[Callable[[HomeServer, JsonResource], None]] servlets: List[Callable[[HomeServer, JsonResource], None]] = []
def setUp(self): def setUp(self):
super().setUp() super().setUp()
@ -448,7 +448,7 @@ class TestReplicationDataHandler(ReplicationDataHandler):
super().__init__(hs) super().__init__(hs)
# list of received (stream_name, token, row) tuples # list of received (stream_name, token, row) tuples
self.received_rdata_rows = [] # type: List[Tuple[str, int, Any]] self.received_rdata_rows: List[Tuple[str, int, Any]] = []
async def on_rdata(self, stream_name, instance_name, token, rows): async def on_rdata(self, stream_name, instance_name, token, rows):
await super().on_rdata(stream_name, instance_name, token, rows) await super().on_rdata(stream_name, instance_name, token, rows)
@ -484,7 +484,7 @@ class FakeRedisPubSubServer:
class FakeRedisPubSubProtocol(Protocol): class FakeRedisPubSubProtocol(Protocol):
"""A connection from a client talking to the fake Redis server.""" """A connection from a client talking to the fake Redis server."""
transport = None # type: Optional[FakeTransport] transport: Optional[FakeTransport] = None
def __init__(self, server: FakeRedisPubSubServer): def __init__(self, server: FakeRedisPubSubServer):
self._server = server self._server = server

14
tests/replication/tcp/streams/test_events.py
View File

@ -135,9 +135,9 @@ class EventsStreamTestCase(BaseStreamTestCase):
) )
# this is the point in the DAG where we make a fork # this is the point in the DAG where we make a fork
fork_point = self.get_success( fork_point: List[str] = self.get_success(
self.hs.get_datastore().get_latest_event_ids_in_room(self.room_id) self.hs.get_datastore().get_latest_event_ids_in_room(self.room_id)
) # type: List[str] )
events = [ events = [
self._inject_state_event(sender=OTHER_USER) self._inject_state_event(sender=OTHER_USER)
@ -238,7 +238,7 @@ class EventsStreamTestCase(BaseStreamTestCase):
self.assertEqual(row.data.event_id, pl_event.event_id) self.assertEqual(row.data.event_id, pl_event.event_id)
# the state rows are unsorted # the state rows are unsorted
state_rows = [] # type: List[EventsStreamCurrentStateRow] state_rows: List[EventsStreamCurrentStateRow] = []
for stream_name, _, row in received_rows: for stream_name, _, row in received_rows:
self.assertEqual("events", stream_name) self.assertEqual("events", stream_name)
self.assertIsInstance(row, EventsStreamRow) self.assertIsInstance(row, EventsStreamRow)
@ -290,11 +290,11 @@ class EventsStreamTestCase(BaseStreamTestCase):
) )
# this is the point in the DAG where we make a fork # this is the point in the DAG where we make a fork
fork_point = self.get_success( fork_point: List[str] = self.get_success(
self.hs.get_datastore().get_latest_event_ids_in_room(self.room_id) self.hs.get_datastore().get_latest_event_ids_in_room(self.room_id)
) # type: List[str] )
events = [] # type: List[EventBase] events: List[EventBase] = []
for user in user_ids: for user in user_ids:
events.extend( events.extend(
self._inject_state_event(sender=user) for _ in range(STATES_PER_USER) self._inject_state_event(sender=user) for _ in range(STATES_PER_USER)
@ -355,7 +355,7 @@ class EventsStreamTestCase(BaseStreamTestCase):
self.assertEqual(row.data.event_id, pl_events[i].event_id) self.assertEqual(row.data.event_id, pl_events[i].event_id)
# the state rows are unsorted # the state rows are unsorted
state_rows = [] # type: List[EventsStreamCurrentStateRow] state_rows: List[EventsStreamCurrentStateRow] = []
for _ in range(STATES_PER_USER + 1): for _ in range(STATES_PER_USER + 1):
stream_name, token, row = received_rows.pop(0) stream_name, token, row = received_rows.pop(0)
self.assertEqual("events", stream_name) self.assertEqual("events", stream_name)

4
tests/replication/tcp/streams/test_receipts.py
View File

@ -43,7 +43,7 @@ class ReceiptsStreamTestCase(BaseStreamTestCase):
stream_name, _, token, rdata_rows = self.test_handler.on_rdata.call_args[0] stream_name, _, token, rdata_rows = self.test_handler.on_rdata.call_args[0]
self.assertEqual(stream_name, "receipts") self.assertEqual(stream_name, "receipts")
self.assertEqual(1, len(rdata_rows)) self.assertEqual(1, len(rdata_rows))
row = rdata_rows[0] # type: ReceiptsStream.ReceiptsStreamRow row: ReceiptsStream.ReceiptsStreamRow = rdata_rows[0]
self.assertEqual("!room:blue", row.room_id) self.assertEqual("!room:blue", row.room_id)
self.assertEqual("m.read", row.receipt_type) self.assertEqual("m.read", row.receipt_type)
self.assertEqual(USER_ID, row.user_id) self.assertEqual(USER_ID, row.user_id)
@ -75,7 +75,7 @@ class ReceiptsStreamTestCase(BaseStreamTestCase):
self.assertEqual(token, 3) self.assertEqual(token, 3)
self.assertEqual(1, len(rdata_rows)) self.assertEqual(1, len(rdata_rows))
row = rdata_rows[0] # type: ReceiptsStream.ReceiptsStreamRow row: ReceiptsStream.ReceiptsStreamRow = rdata_rows[0]
self.assertEqual("!room2:blue", row.room_id) self.assertEqual("!room2:blue", row.room_id)
self.assertEqual("m.read", row.receipt_type) self.assertEqual("m.read", row.receipt_type)
self.assertEqual(USER_ID, row.user_id) self.assertEqual(USER_ID, row.user_id)

4
tests/replication/tcp/streams/test_typing.py
View File

@ -47,7 +47,7 @@ class TypingStreamTestCase(BaseStreamTestCase):
stream_name, _, token, rdata_rows = self.test_handler.on_rdata.call_args[0] stream_name, _, token, rdata_rows = self.test_handler.on_rdata.call_args[0]
self.assertEqual(stream_name, "typing") self.assertEqual(stream_name, "typing")
self.assertEqual(1, len(rdata_rows)) self.assertEqual(1, len(rdata_rows))
row = rdata_rows[0] # type: TypingStream.TypingStreamRow row: TypingStream.TypingStreamRow = rdata_rows[0]
self.assertEqual(ROOM_ID, row.room_id) self.assertEqual(ROOM_ID, row.room_id)
self.assertEqual([USER_ID], row.user_ids) self.assertEqual([USER_ID], row.user_ids)
@ -102,7 +102,7 @@ class TypingStreamTestCase(BaseStreamTestCase):
stream_name, _, token, rdata_rows = self.test_handler.on_rdata.call_args[0] stream_name, _, token, rdata_rows = self.test_handler.on_rdata.call_args[0]
self.assertEqual(stream_name, "typing") self.assertEqual(stream_name, "typing")
self.assertEqual(1, len(rdata_rows)) self.assertEqual(1, len(rdata_rows))
row = rdata_rows[0] # type: TypingStream.TypingStreamRow row: TypingStream.TypingStreamRow = rdata_rows[0]
self.assertEqual(ROOM_ID, row.room_id) self.assertEqual(ROOM_ID, row.room_id)
self.assertEqual([USER_ID], row.user_ids) self.assertEqual([USER_ID], row.user_ids)

2
tests/replication/test_multi_media_repo.py
View File

@ -31,7 +31,7 @@ from tests.server import FakeChannel, FakeSite, FakeTransport, make_request
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
test_server_connection_factory = None # type: Optional[TestServerTLSConnectionFactory] test_server_connection_factory: Optional[TestServerTLSConnectionFactory] = None
class MediaRepoShardTestCase(BaseMultiWorkerStreamTestCase): class MediaRepoShardTestCase(BaseMultiWorkerStreamTestCase):

4
tests/rest/client/test_third_party_rules.py
View File

@ -233,11 +233,11 @@ class ThirdPartyRulesTestCase(unittest.HomeserverTestCase):
"content": content, "content": content,
"sender": self.user_id, "sender": self.user_id,
} }
event = self.get_success( event: EventBase = self.get_success(
current_rules_module().module_api.create_and_send_event_into_room( current_rules_module().module_api.create_and_send_event_into_room(
event_dict event_dict
) )
) # type: EventBase )
self.assertEquals(event.sender, self.user_id) self.assertEquals(event.sender, self.user_id)
self.assertEquals(event.room_id, self.room_id) self.assertEquals(event.room_id, self.room_id)

14
tests/rest/client/v1/test_login.py
View File

@ -453,7 +453,7 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
self.assertEqual(channel.code, 200, channel.result) self.assertEqual(channel.code, 200, channel.result)
# stick the flows results in a dict by type # stick the flows results in a dict by type
flow_results = {} # type: Dict[str, Any] flow_results: Dict[str, Any] = {}
for f in channel.json_body["flows"]: for f in channel.json_body["flows"]:
flow_type = f["type"] flow_type = f["type"]
self.assertNotIn( self.assertNotIn(
@ -501,7 +501,7 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
p.close() p.close()
# there should be a link for each href # there should be a link for each href
returned_idps = [] # type: List[str] returned_idps: List[str] = []
for link in p.links: for link in p.links:
path, query = link.split("?", 1) path, query = link.split("?", 1)
self.assertEqual(path, "pick_idp") self.assertEqual(path, "pick_idp")
@ -582,7 +582,7 @@ class MultiSSOTestCase(unittest.HomeserverTestCase):
# ... and should have set a cookie including the redirect url # ... and should have set a cookie including the redirect url
cookie_headers = channel.headers.getRawHeaders("Set-Cookie") cookie_headers = channel.headers.getRawHeaders("Set-Cookie")
assert cookie_headers assert cookie_headers
cookies = {} # type: Dict[str, str] cookies: Dict[str, str] = {}
for h in cookie_headers: for h in cookie_headers:
key, value = h.split(";")[0].split("=", maxsplit=1) key, value = h.split(";")[0].split("=", maxsplit=1)
cookies[key] = value cookies[key] = value
@ -874,9 +874,7 @@ class JWTTestCase(unittest.HomeserverTestCase):
def jwt_encode(self, payload: Dict[str, Any], secret: str = jwt_secret) -> str: def jwt_encode(self, payload: Dict[str, Any], secret: str = jwt_secret) -> str:
# PyJWT 2.0.0 changed the return type of jwt.encode from bytes to str. # PyJWT 2.0.0 changed the return type of jwt.encode from bytes to str.
result = jwt.encode( result: Union[str, bytes] = jwt.encode(payload, secret, self.jwt_algorithm)
payload, secret, self.jwt_algorithm
) # type: Union[str, bytes]
if isinstance(result, bytes): if isinstance(result, bytes):
return result.decode("ascii") return result.decode("ascii")
return result return result
@ -1084,7 +1082,7 @@ class JWTPubKeyTestCase(unittest.HomeserverTestCase):
def jwt_encode(self, payload: Dict[str, Any], secret: str = jwt_privatekey) -> str: def jwt_encode(self, payload: Dict[str, Any], secret: str = jwt_privatekey) -> str:
# PyJWT 2.0.0 changed the return type of jwt.encode from bytes to str. # PyJWT 2.0.0 changed the return type of jwt.encode from bytes to str.
result = jwt.encode(payload, secret, "RS256") # type: Union[bytes,str] result: Union[bytes, str] = jwt.encode(payload, secret, "RS256")
if isinstance(result, bytes): if isinstance(result, bytes):
return result.decode("ascii") return result.decode("ascii")
return result return result
@ -1272,7 +1270,7 @@ class UsernamePickerTestCase(HomeserverTestCase):
self.assertEqual(picker_url, "/_synapse/client/pick_username/account_details") self.assertEqual(picker_url, "/_synapse/client/pick_username/account_details")
# ... with a username_mapping_session cookie # ... with a username_mapping_session cookie
cookies = {} # type: Dict[str,str] cookies: Dict[str, str] = {}
channel.extract_cookies(cookies) channel.extract_cookies(cookies)
self.assertIn("username_mapping_session", cookies) self.assertIn("username_mapping_session", cookies)
session_id = cookies["username_mapping_session"] session_id = cookies["username_mapping_session"]

8
tests/server.py
View File

@ -52,7 +52,7 @@ class FakeChannel:
_reactor = attr.ib() _reactor = attr.ib()
result = attr.ib(type=dict, default=attr.Factory(dict)) result = attr.ib(type=dict, default=attr.Factory(dict))
_ip = attr.ib(type=str, default="127.0.0.1") _ip = attr.ib(type=str, default="127.0.0.1")
_producer = None # type: Optional[Union[IPullProducer, IPushProducer]] _producer: Optional[Union[IPullProducer, IPushProducer]] = None
@property @property
def json_body(self): def json_body(self):
@ -316,8 +316,10 @@ class ThreadedMemoryReactorClock(MemoryReactorClock):
self._tcp_callbacks = {} self._tcp_callbacks = {}
self._udp = [] self._udp = []
lookups = self.lookups = {} # type: Dict[str, str] self.lookups: Dict[str, str] = {}
self._thread_callbacks = deque() # type: Deque[Callable[[], None]] self._thread_callbacks: Deque[Callable[[], None]] = deque()
lookups = self.lookups
@implementer(IResolverSimple) @implementer(IResolverSimple)
class FakeResolver: class FakeResolver:

4
tests/storage/test_background_update.py
View File

@ -7,9 +7,7 @@ from tests import unittest
class BackgroundUpdateTestCase(unittest.HomeserverTestCase): class BackgroundUpdateTestCase(unittest.HomeserverTestCase):
def prepare(self, reactor, clock, homeserver): def prepare(self, reactor, clock, homeserver):
self.updates = ( self.updates: BackgroundUpdater = self.hs.get_datastore().db_pool.updates
self.hs.get_datastore().db_pool.updates
) # type: BackgroundUpdater
# the base test class should have run the real bg updates for us # the base test class should have run the real bg updates for us
self.assertTrue( self.assertTrue(
self.get_success(self.updates.has_completed_background_updates()) self.get_success(self.updates.has_completed_background_updates())

6
tests/storage/test_id_generators.py
View File

@ -27,7 +27,7 @@ class MultiWriterIdGeneratorTestCase(HomeserverTestCase):
def prepare(self, reactor, clock, hs): def prepare(self, reactor, clock, hs):
self.store = hs.get_datastore() self.store = hs.get_datastore()
self.db_pool = self.store.db_pool # type: DatabasePool self.db_pool: DatabasePool = self.store.db_pool
self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db)) self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db))
@ -460,7 +460,7 @@ class BackwardsMultiWriterIdGeneratorTestCase(HomeserverTestCase):
def prepare(self, reactor, clock, hs): def prepare(self, reactor, clock, hs):
self.store = hs.get_datastore() self.store = hs.get_datastore()
self.db_pool = self.store.db_pool # type: DatabasePool self.db_pool: DatabasePool = self.store.db_pool
self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db)) self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db))
@ -586,7 +586,7 @@ class MultiTableMultiWriterIdGeneratorTestCase(HomeserverTestCase):
def prepare(self, reactor, clock, hs): def prepare(self, reactor, clock, hs):
self.store = hs.get_datastore() self.store = hs.get_datastore()
self.db_pool = self.store.db_pool # type: DatabasePool self.db_pool: DatabasePool = self.store.db_pool
self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db)) self.get_success(self.db_pool.runInteraction("_setup_db", self._setup_db))

2
tests/test_state.py
View File

@ -199,7 +199,7 @@ class StateTestCase(unittest.TestCase):
self.store.register_events(graph.walk()) self.store.register_events(graph.walk())
context_store = {} # type: dict[str, EventContext] context_store: dict[str, EventContext] = {}
for event in graph.walk(): for event in graph.walk():
context = yield defer.ensureDeferred( context = yield defer.ensureDeferred(

6
tests/test_utils/html_parsers.py
View File

@ -23,13 +23,13 @@ class TestHtmlParser(HTMLParser):
super().__init__() super().__init__()
# a list of links found in the doc # a list of links found in the doc
self.links = [] # type: List[str] self.links: List[str] = []
# the values of any hidden <input>s: map from name to value # the values of any hidden <input>s: map from name to value
self.hiddens = {} # type: Dict[str, Optional[str]] self.hiddens: Dict[str, Optional[str]] = {}
# the values of any radio buttons: map from name to list of values # the values of any radio buttons: map from name to list of values
self.radios = {} # type: Dict[str, List[Optional[str]]] self.radios: Dict[str, List[Optional[str]]] = {}
def handle_starttag( def handle_starttag(
self, tag: str, attrs: Iterable[Tuple[str, Optional[str]]] self, tag: str, attrs: Iterable[Tuple[str, Optional[str]]]

2
tests/unittest.py
View File

@ -520,7 +520,7 @@ class HomeserverTestCase(TestCase):
if not isinstance(deferred, Deferred): if not isinstance(deferred, Deferred):
return d return d
results = [] # type: list results: list = []
deferred.addBoth(results.append) deferred.addBoth(results.append)
self.pump(by=by) self.pump(by=by)

2
tests/util/caches/test_descriptors.py
View File

@ -174,7 +174,7 @@ class DescriptorTestCase(unittest.TestCase):
return self.result return self.result
obj = Cls() obj = Cls()
callbacks = set() # type: Set[str] callbacks: Set[str] = set()
# set off an asynchronous request # set off an asynchronous request
obj.result = origin_d = defer.Deferred() obj.result = origin_d = defer.Deferred()

18
tests/util/test_itertools.py
View File

@ -44,7 +44,7 @@ class ChunkSeqTests(TestCase):
) )
def test_empty_input(self): def test_empty_input(self):
parts = chunk_seq([], 5) # type: Iterable[Sequence] parts: Iterable[Sequence] = chunk_seq([], 5)
self.assertEqual( self.assertEqual(
list(parts), list(parts),
@ -56,13 +56,13 @@ class SortTopologically(TestCase):
def test_empty(self): def test_empty(self):
"Test that an empty graph works correctly" "Test that an empty graph works correctly"
graph = {} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {}
self.assertEqual(list(sorted_topologically([], graph)), []) self.assertEqual(list(sorted_topologically([], graph)), [])
def test_handle_empty_graph(self): def test_handle_empty_graph(self):
"Test that a graph where a node doesn't have an entry is treated as empty" "Test that a graph where a node doesn't have an entry is treated as empty"
graph = {} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {}
# For disconnected nodes the output is simply sorted. # For disconnected nodes the output is simply sorted.
self.assertEqual(list(sorted_topologically([1, 2], graph)), [1, 2]) self.assertEqual(list(sorted_topologically([1, 2], graph)), [1, 2])
@ -70,7 +70,7 @@ class SortTopologically(TestCase):
def test_disconnected(self): def test_disconnected(self):
"Test that a graph with no edges work" "Test that a graph with no edges work"
graph = {1: [], 2: []} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {1: [], 2: []}
# For disconnected nodes the output is simply sorted. # For disconnected nodes the output is simply sorted.
self.assertEqual(list(sorted_topologically([1, 2], graph)), [1, 2]) self.assertEqual(list(sorted_topologically([1, 2], graph)), [1, 2])
@ -78,19 +78,19 @@ class SortTopologically(TestCase):
def test_linear(self): def test_linear(self):
"Test that a simple `4 -> 3 -> 2 -> 1` graph works" "Test that a simple `4 -> 3 -> 2 -> 1` graph works"
graph = {1: [], 2: [1], 3: [2], 4: [3]} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {1: [], 2: [1], 3: [2], 4: [3]}
self.assertEqual(list(sorted_topologically([4, 3, 2, 1], graph)), [1, 2, 3, 4]) self.assertEqual(list(sorted_topologically([4, 3, 2, 1], graph)), [1, 2, 3, 4])
def test_subset(self): def test_subset(self):
"Test that only sorting a subset of the graph works" "Test that only sorting a subset of the graph works"
graph = {1: [], 2: [1], 3: [2], 4: [3]} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {1: [], 2: [1], 3: [2], 4: [3]}
self.assertEqual(list(sorted_topologically([4, 3], graph)), [3, 4]) self.assertEqual(list(sorted_topologically([4, 3], graph)), [3, 4])
def test_fork(self): def test_fork(self):
"Test that a forked graph works" "Test that a forked graph works"
graph = {1: [], 2: [1], 3: [1], 4: [2, 3]} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {1: [], 2: [1], 3: [1], 4: [2, 3]}
# Valid orderings are `[1, 3, 2, 4]` or `[1, 2, 3, 4]`, but we should # Valid orderings are `[1, 3, 2, 4]` or `[1, 2, 3, 4]`, but we should
# always get the same one. # always get the same one.
@ -98,12 +98,12 @@ class SortTopologically(TestCase):
def test_duplicates(self): def test_duplicates(self):
"Test that a graph with duplicate edges work" "Test that a graph with duplicate edges work"
graph = {1: [], 2: [1, 1], 3: [2, 2], 4: [3]} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {1: [], 2: [1, 1], 3: [2, 2], 4: [3]}
self.assertEqual(list(sorted_topologically([4, 3, 2, 1], graph)), [1, 2, 3, 4]) self.assertEqual(list(sorted_topologically([4, 3, 2, 1], graph)), [1, 2, 3, 4])
def test_multiple_paths(self): def test_multiple_paths(self):
"Test that a graph with multiple paths between two nodes work" "Test that a graph with multiple paths between two nodes work"
graph = {1: [], 2: [1], 3: [2], 4: [3, 2, 1]} # type: Dict[int, List[int]] graph: Dict[int, List[int]] = {1: [], 2: [1], 3: [2], 4: [3, 2, 1]}
self.assertEqual(list(sorted_topologically([4, 3, 2, 1], graph)), [1, 2, 3, 4]) self.assertEqual(list(sorted_topologically([4, 3, 2, 1], graph)), [1, 2, 3, 4])