<s> """<STR_LIT>""" <EOL> __authors__ = [ '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> import contextlib <EOL> import time <EOL> from tornado . ioloop import IOLoop <EOL> from viewfinder . backend . base import util <EOL> from viewfinder . backend . db . accounting import Accounting <EOL> from viewfinder . backend . db . activity import Activity <EOL> from viewfinder . backend . db . comment import Comment <EOL> from viewfinder . backend . db . db_client import DBKey <EOL> from viewfinder . backend . db . notification import Notification <EOL> from viewfinder . backend . op . notification_manager import NotificationManager <EOL> from viewfinder . backend . www . test import service_base_test <EOL> class QueryNotificationsTestCase ( service_base_test . ServiceBaseTestCase ) : <EOL> def setUp ( self ) : <EOL> self . _time_adjustment = <NUM_LIT:0> <EOL> super ( QueryNotificationsTestCase , self ) . setUp ( ) <EOL> self . _CreateSimpleTestAssets ( ) <EOL> def get_new_ioloop ( self ) : <EOL> """<STR_LIT>""" <EOL> return IOLoop ( time_func = lambda : time . time ( ) + self . _time_adjustment ) <EOL> @ contextlib . contextmanager <EOL> def _AccelerateTime ( self ) : <EOL> """<STR_LIT>""" <EOL> done = [ False ] <EOL> def _Adjust ( ) : <EOL> if not done [ <NUM_LIT:0> ] : <EOL> self . _time_adjustment += <NUM_LIT:1> <EOL> self . io_loop . add_callback ( _Adjust ) <EOL> _Adjust ( ) <EOL> yield <EOL> done [ <NUM_LIT:0> ] = True <EOL> def testQueryNotificationsEmpty ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _validate = False <EOL> notifications = self . _RunAsync ( Notification . RangeQuery , self . _client , self . _user . user_id , None , None , None ) <EOL> for n in notifications : <EOL> self . _RunAsync ( n . Delete , self . _client ) <EOL> response_dict = self . _tester . SendRequest ( '<STR_LIT>' , self . _cookie , { } ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> response_dict = self . _tester . SendRequest ( '<STR_LIT>' , self . _cookie , { '<STR_LIT>' : False } ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> with self . _AccelerateTime ( ) : <EOL> real_start_time = time . time ( ) <EOL> ioloop_start_time = self . io_loop . time ( ) <EOL> self . _tester . SendRequestAsync ( '<STR_LIT>' , self . _cookie , { '<STR_LIT>' : <NUM_LIT> } , callback = self . stop ) <EOL> response_dict = self . wait ( timeout = <NUM_LIT> ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> real_end_time = time . time ( ) <EOL> ioloop_end_time = self . io_loop . time ( ) <EOL> self . assertGreater ( ioloop_end_time - ioloop_start_time , <NUM_LIT:50> ) <EOL> self . assertLess ( real_end_time - real_start_time , <NUM_LIT:1> ) <EOL> def testQueryNotificationsNotifications ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _tester . ShareNew ( self . _cookie , <EOL> [ ( self . _episode_id , self . _photo_ids ) ] , <EOL> [ self . _user2 . user_id ] ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:4> ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , max_long_poll = <NUM_LIT> ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:4> ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , limit = <NUM_LIT:2> ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:2> ) <EOL> vp_id , ep_ids = self . _tester . ShareNew ( self . _cookie , <EOL> [ ( self . _episode_id , self . _photo_ids [ : <NUM_LIT:1> ] ) ] , <EOL> [ self . _user3 . user_id ] ) <EOL> self . _PostCommentChain ( [ self . _cookie , self . _cookie3 ] , vp_id , <NUM_LIT:3> ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , start_key = response_dict [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:6> ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , start_key = response_dict [ '<STR_LIT>' ] , limit = <NUM_LIT:1> ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( response_dict [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , start_key = response_dict [ '<STR_LIT>' ] , <EOL> scan_forward = True ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:0> ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , scan_forward = False ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:9> ) <EOL> self . assertEqual ( response_dict [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ '<STR_LIT:name>' ] , '<STR_LIT>' ) <EOL> start_key = str ( response_dict [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , start_key = start_key , <EOL> limit = <NUM_LIT:1> , scan_forward = False ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:1> ) <EOL> def testQueryNotificationsLongPoll ( self ) : <EOL> """<STR_LIT>""" <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie ) <EOL> start_key = response_dict [ '<STR_LIT>' ] <EOL> with self . _AccelerateTime ( ) : <EOL> start_time = self . io_loop . time ( ) <EOL> self . _tester . SendRequestAsync ( '<STR_LIT>' , self . _cookie , <EOL> { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : start_key } , <EOL> callback = self . stop ) <EOL> self . io_loop . add_timeout ( self . io_loop . time ( ) + <NUM_LIT:10> , lambda : self . stop ( '<STR_LIT>' ) ) <EOL> response = self . wait ( timeout = <NUM_LIT:12> ) <EOL> self . assertEqual ( response , '<STR_LIT>' ) <EOL> self . _tester . ShareNew ( self . _cookie , <EOL> [ ( self . _episode_id , self . _photo_ids ) ] , <EOL> [ self . _user2 . user_id ] ) <EOL> with self . _AccelerateTime ( ) : <EOL> response_dict = self . wait ( timeout = <NUM_LIT> ) <EOL> end_time = self . io_loop . time ( ) <EOL> self . assertLess ( end_time - start_time , <NUM_LIT:30> ) <EOL> self . assertEqual ( len ( response_dict [ '<STR_LIT>' ] ) , <NUM_LIT:1> ) <EOL> def testQueryNotificationsMultipleDevices ( self ) : <EOL> """<STR_LIT>""" <EOL> web_cookie = self . _GetSecureUserCookie ( device_id = self . _webapp_device_id ) <EOL> ep_ph_ids = self . _UploadOneEpisode ( self . _cookie3 , <NUM_LIT:2> ) <EOL> self . _tester . ShareNew ( self . _cookie3 , [ ep_ph_ids ] , [ self . _user . user_id ] ) <EOL> self . _tester . QueryNotifications ( self . _cookie ) <EOL> self . _tester . QueryNotifications ( web_cookie ) <EOL> self . _tester . ShareNew ( self . _cookie3 , [ ep_ph_ids ] , [ self . _user . user_id ] ) <EOL> self . _tester . QueryNotifications ( web_cookie ) <EOL> self . _tester . QueryNotifications ( self . _cookie ) <EOL> def testQueryNotificationsCoverage ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _tester . LinkFacebookUser ( { '<STR_LIT:id>' : <NUM_LIT:100> } , user_cookie = self . _cookie ) <EOL> self . _tester . UnlinkIdentity ( self . _cookie , '<STR_LIT>' ) <EOL> ep_ph_ids = self . _UploadOneEpisode ( self . _cookie , <NUM_LIT:2> ) <EOL> vp_id , ep_ids = self . _tester . ShareNew ( self . _cookie , [ ( self . _episode_id , self . _photo_ids ) , ep_ph_ids ] , <EOL> [ self . _user2 . user_id , self . _user3 . user_id ] ) <EOL> self . _tester . UpdateEpisode ( self . _cookie , ep_ids [ <NUM_LIT:0> ] , description = '<STR_LIT>' ) <EOL> self . _tester . PostComment ( self . _cookie2 , vp_id , message = '<STR_LIT>' ) <EOL> self . _tester . AddFollowers ( self . _cookie , self . _user . private_vp_id , [ '<STR_LIT>' ] ) <EOL> self . _tester . UpdateViewpoint ( self . _cookie , self . _user . private_vp_id , title = '<STR_LIT>' ) <EOL> self . _tester . UpdateViewpoint ( self . _cookie , self . _user . private_vp_id , viewed_seq = <NUM_LIT:2> ) <EOL> self . _tester . HidePhotos ( self . _cookie , [ ( self . _episode_id , self . _photo_ids [ <NUM_LIT:1> : ] ) ] ) <EOL> self . _tester . RemovePhotos ( self . _cookie , [ ( self . _episode_id , self . _photo_ids [ <NUM_LIT:1> : ] ) ] ) <EOL> self . _tester . Unshare ( self . _cookie , vp_id , [ ( ep_ids [ <NUM_LIT:0> ] , self . _photo_ids [ : <NUM_LIT:1> ] ) ] ) <EOL> self . _tester . QueryNotifications ( self . _cookie ) <EOL> self . _tester . QueryNotifications ( self . _cookie2 ) <EOL> self . _tester . QueryNotifications ( self . _cookie3 ) <EOL> def testCommentInlining ( self ) : <EOL> """<STR_LIT>""" <EOL> message = '<STR_LIT:a>' * NotificationManager . MAX_INLINE_COMMENT_LEN <EOL> self . _tester . PostComment ( self . _cookie , self . _user . private_vp_id , message = message ) <EOL> self . _tester . PostComment ( self . _cookie , self . _user . private_vp_id , message + '<STR_LIT:b>' ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie , limit = <NUM_LIT:2> , scan_forward = False ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_dict [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' in response_dict [ '<STR_LIT>' ] [ <NUM_LIT:0> ] ) <EOL> self . assertTrue ( '<STR_LIT>' in response_dict [ '<STR_LIT>' ] [ <NUM_LIT:1> ] [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' not in response_dict [ '<STR_LIT>' ] [ <NUM_LIT:1> ] ) <EOL> def testPostCommentNotification ( self ) : <EOL> """<STR_LIT>""" <EOL> vp_id , ep_ids = self . _tester . ShareNew ( self . _cookie , <EOL> [ ( self . _episode_id , self . _photo_ids ) ] , <EOL> [ self . _user2 . user_id ] ) <EOL> timestamp = time . time ( ) <EOL> message = '<STR_LIT:a>' * ( NotificationManager . MAX_INLINE_COMMENT_LEN + <NUM_LIT:1> ) <EOL> comment_id1 = self . _tester . PostComment ( self . _cookie , vp_id , timestamp = timestamp , message = message ) <EOL> comment_id2 = self . _tester . PostComment ( self . _cookie , vp_id , timestamp = timestamp + <NUM_LIT:1> , message = message ) <EOL> comment_id3 = self . _tester . PostComment ( self . _cookie , vp_id , timestamp = timestamp + <NUM_LIT:2> , message = message ) <EOL> response_dict = self . _tester . QueryNotifications ( self . _cookie ) <EOL> for n in response_dict [ '<STR_LIT>' ] : <EOL> for vp in n [ '<STR_LIT>' ] . get ( '<STR_LIT>' , [ ] ) : <EOL> if '<STR_LIT>' in vp : <EOL> start_key = vp [ '<STR_LIT>' ] <EOL> self . assertLess ( start_key , comment_id1 ) <EOL> self . assertLess ( start_key , comment_id2 ) <EOL> self . assertLess ( start_key , comment_id3 ) <EOL> return <EOL> def _TestQueryNotifications ( tester , user_cookie , request_dict ) : <EOL> """<STR_LIT>""" <EOL> validator = tester . validator <EOL> user_id , device_id = tester . GetIdsFromCookie ( user_cookie ) <EOL> def _MakeNotificationDict ( notification ) : <EOL> """<STR_LIT>""" <EOL> notification_dict = { '<STR_LIT>' : notification . notification_id , <EOL> '<STR_LIT:name>' : notification . name , <EOL> '<STR_LIT>' : notification . sender_id , <EOL> '<STR_LIT>' : notification . timestamp } <EOL> util . SetIfNotNone ( notification_dict , '<STR_LIT>' , notification . op_id ) <EOL> if notification . update_seq is not None or notification . viewed_seq is not None : <EOL> vp_dict = notification_dict . setdefault ( '<STR_LIT>' , { } ) . setdefault ( '<STR_LIT>' , { } ) <EOL> vp_dict [ '<STR_LIT>' ] = notification . viewpoint_id <EOL> util . SetIfNotNone ( vp_dict , '<STR_LIT>' , notification . update_seq ) <EOL> util . SetIfNotNone ( vp_dict , '<STR_LIT>' , notification . viewed_seq ) <EOL> if notification . activity_id is not None : <EOL> viewpoint_id = notification . viewpoint_id <EOL> activity = validator . GetModelObject ( Activity , DBKey ( viewpoint_id , notification . activity_id ) ) <EOL> activity_dict = activity . MakeMetadataDict ( ) <EOL> notification_dict . setdefault ( '<STR_LIT>' , { } ) [ '<STR_LIT>' ] = activity_dict <EOL> if activity . name == '<STR_LIT>' and notification . invalidate is None : <EOL> comment_id = activity_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> comment = validator . GetModelObject ( Comment , DBKey ( viewpoint_id , comment_id ) ) <EOL> notification_dict [ '<STR_LIT>' ] [ '<STR_LIT>' ] = comment . _asdict ( ) <EOL> invalidate = notification . GetInvalidate ( ) <EOL> if invalidate is not None : <EOL> invalidate . pop ( '<STR_LIT>' ) <EOL> notification_dict [ '<STR_LIT>' ] = invalidate <EOL> return notification_dict <EOL> def _MakeUsageDict ( ) : <EOL> """<STR_LIT>""" <EOL> validator . ValidateAccounting ( ) <EOL> user_hash = '<STR_LIT>' % ( Accounting . USER_SIZE , user_id ) <EOL> def _AccountingAsDict ( sort_key ) : <EOL> act = validator . GetModelObject ( Accounting , DBKey ( user_hash , sort_key ) , must_exist = False ) <EOL> if act is None : <EOL> return None <EOL> act_dict = act . _asdict ( ) <EOL> act_dict . pop ( '<STR_LIT>' , None ) <EOL> act_dict . pop ( '<STR_LIT>' , None ) <EOL> act_dict . pop ( '<STR_LIT>' , None ) <EOL> return act_dict <EOL> usage_dict = { } <EOL> util . SetIfNotNone ( usage_dict , '<STR_LIT>' , _AccountingAsDict ( Accounting . OWNED_BY ) ) <EOL> util . SetIfNotNone ( usage_dict , '<STR_LIT>' , _AccountingAsDict ( Accounting . SHARED_BY ) ) <EOL> util . SetIfNotNone ( usage_dict , '<STR_LIT>' , _AccountingAsDict ( Accounting . VISIBLE_TO ) ) <EOL> if len ( usage_dict . keys ( ) ) > <NUM_LIT:0> : <EOL> return usage_dict <EOL> return None <EOL> actual_dict = tester . SendRequest ( '<STR_LIT>' , user_cookie , request_dict ) <EOL> start_key = int ( request_dict [ '<STR_LIT>' ] ) if '<STR_LIT>' in request_dict else None <EOL> limit = request_dict . get ( '<STR_LIT>' , None ) <EOL> scan_forward = request_dict . get ( '<STR_LIT>' , True ) <EOL> notifications = validator . QueryModelObjects ( Notification , <EOL> user_id , <EOL> start_key = start_key , <EOL> limit = limit , <EOL> query_forward = scan_forward ) <EOL> expected_dict = { '<STR_LIT>' : [ _MakeNotificationDict ( n ) for n in notifications ] } <EOL> if len ( notifications ) > <NUM_LIT:0> : <EOL> expected_dict [ '<STR_LIT>' ] = '<STR_LIT>' % notifications [ - <NUM_LIT:1> ] . notification_id <EOL> if len ( expected_dict [ '<STR_LIT>' ] ) > <NUM_LIT:0> : <EOL> usage_dict = _MakeUsageDict ( ) <EOL> if usage_dict is not None : <EOL> last_notification = expected_dict [ '<STR_LIT>' ] [ - <NUM_LIT:1> ] <EOL> last_notification . setdefault ( '<STR_LIT>' , { } ) [ '<STR_LIT:user>' ] = { '<STR_LIT>' : usage_dict } <EOL> notifications = validator . QueryModelObjects ( Notification , user_id ) <EOL> last_notification = notifications [ - <NUM_LIT:1> ] if notifications else None <EOL> if last_notification is not None and last_notification . badge != <NUM_LIT:0> and scan_forward : <EOL> validator . ValidateCreateDBObject ( Notification , <EOL> notification_id = last_notification . notification_id + <NUM_LIT:1> , <EOL> user_id = user_id , <EOL> name = '<STR_LIT>' , <EOL> timestamp = util . _TEST_TIME , <EOL> sender_id = user_id , <EOL> sender_device_id = device_id , <EOL> badge = <NUM_LIT:0> ) <EOL> tester . _CompareResponseDicts ( '<STR_LIT>' , user_id , request_dict , expected_dict , actual_dict ) <EOL> return actual_dict </s>
