<s> from tests import unittest <EOL> from twisted . internet import defer <EOL> from mock import Mock , call , ANY <EOL> import json <EOL> from . . utils import ( <EOL> MockHttpResource , MockClock , DeferredMockCallable , setup_test_homeserver <EOL> ) <EOL> from synapse . api . errors import AuthError <EOL> from synapse . handlers . typing import TypingNotificationHandler <EOL> from synapse . types import UserID <EOL> def _expect_edu ( destination , edu_type , content , origin = "<STR_LIT:test>" ) : <EOL> return { <EOL> "<STR_LIT>" : origin , <EOL> "<STR_LIT>" : <NUM_LIT> , <EOL> "<STR_LIT>" : [ ] , <EOL> "<STR_LIT>" : [ <EOL> { <EOL> "<STR_LIT>" : edu_type , <EOL> "<STR_LIT:content>" : content , <EOL> } <EOL> ] , <EOL> "<STR_LIT>" : [ ] , <EOL> } <EOL> def _make_edu_json ( origin , edu_type , content ) : <EOL> return json . dumps ( _expect_edu ( "<STR_LIT:test>" , edu_type , content , origin = origin ) ) <EOL> class JustTypingNotificationHandlers ( object ) : <EOL> def __init__ ( self , hs ) : <EOL> self . typing_notification_handler = TypingNotificationHandler ( hs ) <EOL> class TypingNotificationsTestCase ( unittest . TestCase ) : <EOL> """<STR_LIT>""" <EOL> @ defer . inlineCallbacks <EOL> def setUp ( self ) : <EOL> self . clock = MockClock ( ) <EOL> self . mock_http_client = Mock ( spec = [ ] ) <EOL> self . mock_http_client . put_json = DeferredMockCallable ( ) <EOL> self . mock_federation_resource = MockHttpResource ( ) <EOL> mock_notifier = Mock ( spec = [ "<STR_LIT>" ] ) <EOL> self . on_new_event = mock_notifier . on_new_event <EOL> self . auth = Mock ( spec = [ ] ) <EOL> hs = yield setup_test_homeserver ( <EOL> auth = self . auth , <EOL> clock = self . clock , <EOL> datastore = Mock ( spec = [ <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> ] ) , <EOL> handlers = None , <EOL> notifier = mock_notifier , <EOL> resource_for_client = Mock ( ) , <EOL> resource_for_federation = self . mock_federation_resource , <EOL> http_client = self . mock_http_client , <EOL> keyring = Mock ( ) , <EOL> ) <EOL> hs . handlers = JustTypingNotificationHandlers ( hs ) <EOL> self . handler = hs . get_handlers ( ) . typing_notification_handler <EOL> self . event_source = hs . get_event_sources ( ) . sources [ "<STR_LIT>" ] <EOL> self . datastore = hs . get_datastore ( ) <EOL> retry_timings_res = { <EOL> "<STR_LIT>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> "<STR_LIT>" : <NUM_LIT:0> , <EOL> } <EOL> self . datastore . get_destination_retry_timings . return_value = ( <EOL> defer . succeed ( retry_timings_res ) <EOL> ) <EOL> def get_received_txn_response ( * args ) : <EOL> return defer . succeed ( None ) <EOL> self . datastore . get_received_txn_response = get_received_txn_response <EOL> self . room_id = "<STR_LIT>" <EOL> hs . handlers . room_member_handler = Mock ( spec = [ ] ) <EOL> self . room_member_handler = hs . handlers . room_member_handler <EOL> self . room_members = [ ] <EOL> def get_rooms_for_user ( user ) : <EOL> if user in self . room_members : <EOL> return defer . succeed ( [ self . room_id ] ) <EOL> else : <EOL> return defer . succeed ( [ ] ) <EOL> self . room_member_handler . get_rooms_for_user = get_rooms_for_user <EOL> def get_room_members ( room_id ) : <EOL> if room_id == self . room_id : <EOL> return defer . succeed ( self . room_members ) <EOL> else : <EOL> return defer . succeed ( [ ] ) <EOL> self . room_member_handler . get_room_members = get_room_members <EOL> def get_joined_rooms_for_user ( user ) : <EOL> if user in self . room_members : <EOL> return defer . succeed ( [ self . room_id ] ) <EOL> else : <EOL> return defer . succeed ( [ ] ) <EOL> self . room_member_handler . get_joined_rooms_for_user = get_joined_rooms_for_user <EOL> @ defer . inlineCallbacks <EOL> def fetch_room_distributions_into ( <EOL> room_id , localusers = None , remotedomains = None , ignore_user = None <EOL> ) : <EOL> members = yield get_room_members ( room_id ) <EOL> for member in members : <EOL> if ignore_user is not None and member == ignore_user : <EOL> continue <EOL> if hs . is_mine ( member ) : <EOL> if localusers is not None : <EOL> localusers . add ( member ) <EOL> else : <EOL> if remotedomains is not None : <EOL> remotedomains . add ( member . domain ) <EOL> self . room_member_handler . fetch_room_distributions_into = ( <EOL> fetch_room_distributions_into <EOL> ) <EOL> def check_joined_room ( room_id , user_id ) : <EOL> if user_id not in [ u . to_string ( ) for u in self . room_members ] : <EOL> raise AuthError ( <NUM_LIT> , "<STR_LIT>" ) <EOL> self . auth . check_joined_room = check_joined_room <EOL> self . u_apple = UserID . from_string ( "<STR_LIT>" ) <EOL> self . u_banana = UserID . from_string ( "<STR_LIT>" ) <EOL> self . u_onion = UserID . from_string ( "<STR_LIT>" ) <EOL> @ defer . inlineCallbacks <EOL> def test_started_typing_local ( self ) : <EOL> self . room_members = [ self . u_apple , self . u_banana ] <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:0> ) <EOL> yield self . handler . started_typing ( <EOL> target_user = self . u_apple , <EOL> auth_user = self . u_apple , <EOL> room_id = self . room_id , <EOL> timeout = <NUM_LIT> , <EOL> ) <EOL> self . on_new_event . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' , <NUM_LIT:1> , rooms = [ self . room_id ] ) , <EOL> ] ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:1> ) <EOL> events = yield self . event_source . get_new_events ( <EOL> room_ids = [ self . room_id ] , <EOL> from_key = <NUM_LIT:0> , <EOL> ) <EOL> self . assertEquals ( <EOL> events [ <NUM_LIT:0> ] , <EOL> [ <EOL> { "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT:content>" : { <EOL> "<STR_LIT>" : [ self . u_apple . to_string ( ) ] , <EOL> } } , <EOL> ] <EOL> ) <EOL> @ defer . inlineCallbacks <EOL> def test_started_typing_remote_send ( self ) : <EOL> self . room_members = [ self . u_apple , self . u_onion ] <EOL> put_json = self . mock_http_client . put_json <EOL> put_json . expect_call_and_return ( <EOL> call ( <EOL> "<STR_LIT>" , <EOL> path = "<STR_LIT>" , <EOL> data = _expect_edu ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> content = { <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT>" : self . u_apple . to_string ( ) , <EOL> "<STR_LIT>" : True , <EOL> } <EOL> ) , <EOL> json_data_callback = ANY , <EOL> long_retries = True , <EOL> ) , <EOL> defer . succeed ( ( <NUM_LIT:200> , "<STR_LIT:OK>" ) ) <EOL> ) <EOL> yield self . handler . started_typing ( <EOL> target_user = self . u_apple , <EOL> auth_user = self . u_apple , <EOL> room_id = self . room_id , <EOL> timeout = <NUM_LIT> , <EOL> ) <EOL> yield put_json . await_calls ( ) <EOL> @ defer . inlineCallbacks <EOL> def test_started_typing_remote_recv ( self ) : <EOL> self . room_members = [ self . u_apple , self . u_onion ] <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:0> ) <EOL> yield self . mock_federation_resource . trigger ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> _make_edu_json ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> content = { <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT>" : self . u_onion . to_string ( ) , <EOL> "<STR_LIT>" : True , <EOL> } <EOL> ) <EOL> ) <EOL> self . on_new_event . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' , <NUM_LIT:1> , rooms = [ self . room_id ] ) , <EOL> ] ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:1> ) <EOL> events = yield self . event_source . get_new_events ( <EOL> room_ids = [ self . room_id ] , <EOL> from_key = <NUM_LIT:0> <EOL> ) <EOL> self . assertEquals ( events [ <NUM_LIT:0> ] , [ { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT:content>" : { <EOL> "<STR_LIT>" : [ self . u_onion . to_string ( ) ] , <EOL> } , <EOL> } ] ) <EOL> @ defer . inlineCallbacks <EOL> def test_stopped_typing ( self ) : <EOL> self . room_members = [ self . u_apple , self . u_banana , self . u_onion ] <EOL> put_json = self . mock_http_client . put_json <EOL> put_json . expect_call_and_return ( <EOL> call ( <EOL> "<STR_LIT>" , <EOL> path = "<STR_LIT>" , <EOL> data = _expect_edu ( <EOL> "<STR_LIT>" , <EOL> "<STR_LIT>" , <EOL> content = { <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT>" : self . u_apple . to_string ( ) , <EOL> "<STR_LIT>" : False , <EOL> } <EOL> ) , <EOL> json_data_callback = ANY , <EOL> long_retries = True , <EOL> ) , <EOL> defer . succeed ( ( <NUM_LIT:200> , "<STR_LIT:OK>" ) ) <EOL> ) <EOL> from synapse . handlers . typing import RoomMember <EOL> member = RoomMember ( self . room_id , self . u_apple ) <EOL> self . handler . _member_typing_until [ member ] = <NUM_LIT> <EOL> self . handler . _member_typing_timer [ member ] = ( <EOL> self . clock . call_later ( <NUM_LIT> , lambda : <NUM_LIT:0> ) <EOL> ) <EOL> self . handler . _room_typing [ self . room_id ] = set ( ( self . u_apple , ) ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:0> ) <EOL> yield self . handler . stopped_typing ( <EOL> target_user = self . u_apple , <EOL> auth_user = self . u_apple , <EOL> room_id = self . room_id , <EOL> ) <EOL> self . on_new_event . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' , <NUM_LIT:1> , rooms = [ self . room_id ] ) , <EOL> ] ) <EOL> yield put_json . await_calls ( ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:1> ) <EOL> events = yield self . event_source . get_new_events ( <EOL> room_ids = [ self . room_id ] , <EOL> from_key = <NUM_LIT:0> , <EOL> ) <EOL> self . assertEquals ( events [ <NUM_LIT:0> ] , [ { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT:content>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> } , <EOL> } ] ) <EOL> @ defer . inlineCallbacks <EOL> def test_typing_timeout ( self ) : <EOL> self . room_members = [ self . u_apple , self . u_banana ] <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:0> ) <EOL> yield self . handler . started_typing ( <EOL> target_user = self . u_apple , <EOL> auth_user = self . u_apple , <EOL> room_id = self . room_id , <EOL> timeout = <NUM_LIT> , <EOL> ) <EOL> self . on_new_event . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' , <NUM_LIT:1> , rooms = [ self . room_id ] ) , <EOL> ] ) <EOL> self . on_new_event . reset_mock ( ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:1> ) <EOL> events = yield self . event_source . get_new_events ( <EOL> room_ids = [ self . room_id ] , <EOL> from_key = <NUM_LIT:0> , <EOL> ) <EOL> self . assertEquals ( events [ <NUM_LIT:0> ] , [ { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT:content>" : { <EOL> "<STR_LIT>" : [ self . u_apple . to_string ( ) ] , <EOL> } , <EOL> } ] ) <EOL> self . clock . advance_time ( <NUM_LIT:11> ) <EOL> self . on_new_event . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' , <NUM_LIT:2> , rooms = [ self . room_id ] ) , <EOL> ] ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:2> ) <EOL> events = yield self . event_source . get_new_events ( <EOL> room_ids = [ self . room_id ] , <EOL> from_key = <NUM_LIT:1> , <EOL> ) <EOL> self . assertEquals ( events [ <NUM_LIT:0> ] , [ { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT:content>" : { <EOL> "<STR_LIT>" : [ ] , <EOL> } , <EOL> } ] ) <EOL> yield self . handler . started_typing ( <EOL> target_user = self . u_apple , <EOL> auth_user = self . u_apple , <EOL> room_id = self . room_id , <EOL> timeout = <NUM_LIT> , <EOL> ) <EOL> self . on_new_event . assert_has_calls ( [ <EOL> call ( '<STR_LIT>' , <NUM_LIT:3> , rooms = [ self . room_id ] ) , <EOL> ] ) <EOL> self . on_new_event . reset_mock ( ) <EOL> self . assertEquals ( self . event_source . get_current_key ( ) , <NUM_LIT:3> ) <EOL> events = yield self . event_source . get_new_events ( <EOL> room_ids = [ self . room_id ] , <EOL> from_key = <NUM_LIT:0> , <EOL> ) <EOL> self . assertEquals ( events [ <NUM_LIT:0> ] , [ { <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : self . room_id , <EOL> "<STR_LIT:content>" : { <EOL> "<STR_LIT>" : [ self . u_apple . to_string ( ) ] , <EOL> } , <EOL> } ] ) </s>
