<s> from tests import unittest <EOL> from mock import Mock , call <EOL> from synapse . api . constants import PresenceState <EOL> from synapse . handlers . presence import ( <EOL> handle_update , handle_timeout , <EOL> IDLE_TIMER , SYNC_ONLINE_TIMEOUT , LAST_ACTIVE_GRANULARITY , FEDERATION_TIMEOUT , <EOL> FEDERATION_PING_INTERVAL , <EOL> ) <EOL> from synapse . storage . presence import UserPresenceState <EOL> class PresenceUpdateTestCase ( unittest . TestCase ) : <EOL> def test_offline_to_online ( self ) : <EOL> wheel_timer = Mock ( ) <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> prev_state = UserPresenceState . default ( user_id ) <EOL> new_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> ) <EOL> state , persist_and_notify , federation_ping = handle_update ( <EOL> prev_state , new_state , is_mine = True , wheel_timer = wheel_timer , now = now <EOL> ) <EOL> self . assertTrue ( persist_and_notify ) <EOL> self . assertTrue ( state . currently_active ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( new_state . status_msg , state . status_msg ) <EOL> self . assertEquals ( state . last_federation_update_ts , now ) <EOL> self . assertEquals ( wheel_timer . insert . call_count , <NUM_LIT:3> ) <EOL> wheel_timer . insert . assert_has_calls ( [ <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_active_ts + IDLE_TIMER <EOL> ) , <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_user_sync_ts + SYNC_ONLINE_TIMEOUT <EOL> ) , <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_active_ts + LAST_ACTIVE_GRANULARITY <EOL> ) , <EOL> ] , any_order = True ) <EOL> def test_online_to_online ( self ) : <EOL> wheel_timer = Mock ( ) <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> prev_state = UserPresenceState . default ( user_id ) <EOL> prev_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> currently_active = True , <EOL> ) <EOL> new_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> ) <EOL> state , persist_and_notify , federation_ping = handle_update ( <EOL> prev_state , new_state , is_mine = True , wheel_timer = wheel_timer , now = now <EOL> ) <EOL> self . assertFalse ( persist_and_notify ) <EOL> self . assertTrue ( federation_ping ) <EOL> self . assertTrue ( state . currently_active ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( new_state . status_msg , state . status_msg ) <EOL> self . assertEquals ( state . last_federation_update_ts , now ) <EOL> self . assertEquals ( wheel_timer . insert . call_count , <NUM_LIT:3> ) <EOL> wheel_timer . insert . assert_has_calls ( [ <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_active_ts + IDLE_TIMER <EOL> ) , <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_user_sync_ts + SYNC_ONLINE_TIMEOUT <EOL> ) , <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_active_ts + LAST_ACTIVE_GRANULARITY <EOL> ) , <EOL> ] , any_order = True ) <EOL> def test_online_to_online_last_active ( self ) : <EOL> wheel_timer = Mock ( ) <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> prev_state = UserPresenceState . default ( user_id ) <EOL> prev_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now - LAST_ACTIVE_GRANULARITY - <NUM_LIT:1> , <EOL> currently_active = True , <EOL> ) <EOL> new_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> ) <EOL> state , persist_and_notify , federation_ping = handle_update ( <EOL> prev_state , new_state , is_mine = True , wheel_timer = wheel_timer , now = now <EOL> ) <EOL> self . assertTrue ( persist_and_notify ) <EOL> self . assertFalse ( state . currently_active ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( new_state . status_msg , state . status_msg ) <EOL> self . assertEquals ( state . last_federation_update_ts , now ) <EOL> self . assertEquals ( wheel_timer . insert . call_count , <NUM_LIT:2> ) <EOL> wheel_timer . insert . assert_has_calls ( [ <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_active_ts + IDLE_TIMER <EOL> ) , <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_user_sync_ts + SYNC_ONLINE_TIMEOUT <EOL> ) <EOL> ] , any_order = True ) <EOL> def test_remote_ping_timer ( self ) : <EOL> wheel_timer = Mock ( ) <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> prev_state = UserPresenceState . default ( user_id ) <EOL> prev_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> ) <EOL> new_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> ) <EOL> state , persist_and_notify , federation_ping = handle_update ( <EOL> prev_state , new_state , is_mine = False , wheel_timer = wheel_timer , now = now <EOL> ) <EOL> self . assertFalse ( persist_and_notify ) <EOL> self . assertFalse ( federation_ping ) <EOL> self . assertFalse ( state . currently_active ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( new_state . status_msg , state . status_msg ) <EOL> self . assertEquals ( wheel_timer . insert . call_count , <NUM_LIT:1> ) <EOL> wheel_timer . insert . assert_has_calls ( [ <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_federation_update_ts + FEDERATION_TIMEOUT <EOL> ) , <EOL> ] , any_order = True ) <EOL> def test_online_to_offline ( self ) : <EOL> wheel_timer = Mock ( ) <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> prev_state = UserPresenceState . default ( user_id ) <EOL> prev_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> currently_active = True , <EOL> ) <EOL> new_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . OFFLINE , <EOL> ) <EOL> state , persist_and_notify , federation_ping = handle_update ( <EOL> prev_state , new_state , is_mine = True , wheel_timer = wheel_timer , now = now <EOL> ) <EOL> self . assertTrue ( persist_and_notify ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( state . last_federation_update_ts , now ) <EOL> self . assertEquals ( wheel_timer . insert . call_count , <NUM_LIT:0> ) <EOL> def test_online_to_idle ( self ) : <EOL> wheel_timer = Mock ( ) <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> prev_state = UserPresenceState . default ( user_id ) <EOL> prev_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> currently_active = True , <EOL> ) <EOL> new_state = prev_state . copy_and_replace ( <EOL> state = PresenceState . UNAVAILABLE , <EOL> ) <EOL> state , persist_and_notify , federation_ping = handle_update ( <EOL> prev_state , new_state , is_mine = True , wheel_timer = wheel_timer , now = now <EOL> ) <EOL> self . assertTrue ( persist_and_notify ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( state . last_federation_update_ts , now ) <EOL> self . assertEquals ( new_state . state , state . state ) <EOL> self . assertEquals ( new_state . status_msg , state . status_msg ) <EOL> self . assertEquals ( wheel_timer . insert . call_count , <NUM_LIT:1> ) <EOL> wheel_timer . insert . assert_has_calls ( [ <EOL> call ( <EOL> now = now , <EOL> obj = user_id , <EOL> then = new_state . last_user_sync_ts + SYNC_ONLINE_TIMEOUT <EOL> ) <EOL> ] , any_order = True ) <EOL> class PresenceTimeoutTestCase ( unittest . TestCase ) : <EOL> def test_idle_timer ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now - IDLE_TIMER - <NUM_LIT:1> , <EOL> last_user_sync_ts = now , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = True , user_to_num_current_syncs = { } , now = now <EOL> ) <EOL> self . assertIsNotNone ( new_state ) <EOL> self . assertEquals ( new_state . state , PresenceState . UNAVAILABLE ) <EOL> def test_sync_timeout ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> last_user_sync_ts = now - SYNC_ONLINE_TIMEOUT - <NUM_LIT:1> , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = True , user_to_num_current_syncs = { } , now = now <EOL> ) <EOL> self . assertIsNotNone ( new_state ) <EOL> self . assertEquals ( new_state . state , PresenceState . OFFLINE ) <EOL> def test_sync_online ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now - SYNC_ONLINE_TIMEOUT - <NUM_LIT:1> , <EOL> last_user_sync_ts = now - SYNC_ONLINE_TIMEOUT - <NUM_LIT:1> , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = True , user_to_num_current_syncs = { <EOL> user_id : <NUM_LIT:1> , <EOL> } , now = now <EOL> ) <EOL> self . assertIsNotNone ( new_state ) <EOL> self . assertEquals ( new_state . state , PresenceState . ONLINE ) <EOL> def test_federation_ping ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> last_user_sync_ts = now , <EOL> last_federation_update_ts = now - FEDERATION_PING_INTERVAL - <NUM_LIT:1> , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = True , user_to_num_current_syncs = { } , now = now <EOL> ) <EOL> self . assertIsNotNone ( new_state ) <EOL> self . assertEquals ( new_state , new_state ) <EOL> def test_no_timeout ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> last_user_sync_ts = now , <EOL> last_federation_update_ts = now , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = True , user_to_num_current_syncs = { } , now = now <EOL> ) <EOL> self . assertIsNone ( new_state ) <EOL> def test_federation_timeout ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now , <EOL> last_user_sync_ts = now , <EOL> last_federation_update_ts = now - FEDERATION_TIMEOUT - <NUM_LIT:1> , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = False , user_to_num_current_syncs = { } , now = now <EOL> ) <EOL> self . assertIsNotNone ( new_state ) <EOL> self . assertEquals ( new_state . state , PresenceState . OFFLINE ) <EOL> def test_last_active ( self ) : <EOL> user_id = "<STR_LIT>" <EOL> now = <NUM_LIT> <EOL> state = UserPresenceState . default ( user_id ) <EOL> state = state . copy_and_replace ( <EOL> state = PresenceState . ONLINE , <EOL> last_active_ts = now - LAST_ACTIVE_GRANULARITY - <NUM_LIT:1> , <EOL> last_user_sync_ts = now , <EOL> last_federation_update_ts = now , <EOL> ) <EOL> new_state = handle_timeout ( <EOL> state , is_mine = True , user_to_num_current_syncs = { } , now = now <EOL> ) <EOL> self . assertIsNotNone ( new_state ) <EOL> self . assertEquals ( state , new_state ) </s>
