<s> from collections import OrderedDict <EOL> import mock <EOL> from nose . tools import * <EOL> from website . notifications . events . base import Event , register , event_registry <EOL> from website . notifications . events . files import ( <EOL> FileAdded , FileRemoved , FolderCreated , FileUpdated , <EOL> AddonFileCopied , AddonFileMoved , AddonFileRenamed , <EOL> ) <EOL> from website . notifications . events import utils <EOL> from website . addons . base import signals <EOL> from framework . auth import Auth <EOL> from tests import factories <EOL> from tests . base import OsfTestCase <EOL> email_transactional = '<STR_LIT>' <EOL> email_digest = '<STR_LIT>' <EOL> class TestEventNotImplemented ( OsfTestCase ) : <EOL> """<STR_LIT>""" <EOL> @ register ( "<STR_LIT>" ) <EOL> class NotImplementedEvent ( Event ) : <EOL> pass <EOL> def setUp ( self ) : <EOL> super ( TestEventNotImplemented , self ) . setUp ( ) <EOL> self . user = factories . UserFactory ( ) <EOL> self . auth = Auth ( user = self . user ) <EOL> self . node = factories . ProjectFactory ( creator = self . user ) <EOL> self . event = self . NotImplementedEvent ( self . user , self . node , '<STR_LIT>' ) <EOL> @ raises ( NotImplementedError ) <EOL> def test_text ( self ) : <EOL> text = self . event . text_message <EOL> @ raises ( NotImplementedError ) <EOL> def test_html ( self ) : <EOL> html = self . event . html_message <EOL> @ raises ( NotImplementedError ) <EOL> def test_url ( self ) : <EOL> url = self . event . url <EOL> @ raises ( NotImplementedError ) <EOL> def test_event ( self ) : <EOL> event = self . event . event_type <EOL> class TestListOfFiles ( OsfTestCase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TestListOfFiles , self ) . setUp ( ) <EOL> self . tree = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : '<STR_LIT:a>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:path>' : '<STR_LIT:b>' , <EOL> '<STR_LIT>' : [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT:file>' , <EOL> '<STR_LIT:path>' : '<STR_LIT:e>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT:file>' , <EOL> '<STR_LIT:path>' : '<STR_LIT:f>' <EOL> } <EOL> ] <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT:file>' , <EOL> '<STR_LIT:path>' : '<STR_LIT:c>' <EOL> } , <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT:file>' , <EOL> '<STR_LIT:path>' : '<STR_LIT:d>' <EOL> } <EOL> ] <EOL> } <EOL> def test_list_of_files ( self ) : <EOL> assert_equal ( [ '<STR_LIT:e>' , '<STR_LIT:f>' , '<STR_LIT:c>' , '<STR_LIT:d>' ] , utils . list_of_files ( self . tree ) ) <EOL> class TestEventExists ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestEventExists , self ) . setUp ( ) <EOL> self . user = factories . UserFactory ( ) <EOL> self . consolidate_auth = Auth ( user = self . user ) <EOL> self . node = factories . ProjectFactory ( creator = self . user ) <EOL> def test_get_file_updated ( self ) : <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , payload = file_payload ) <EOL> assert_is_instance ( event , FileUpdated ) <EOL> def test_get_file_added ( self ) : <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , payload = file_payload ) <EOL> assert_is_instance ( event , FileAdded ) <EOL> def test_get_file_removed ( self ) : <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , payload = file_deleted_payload ) <EOL> assert_is_instance ( event , FileRemoved ) <EOL> def test_get_folder_created ( self ) : <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , payload = folder_created_payload ) <EOL> assert_is_instance ( event , FolderCreated ) <EOL> def test_get_file_moved ( self ) : <EOL> file_moved_payload = file_move_payload ( self . node , self . node ) <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , payload = file_moved_payload ) <EOL> assert_is_instance ( event , AddonFileMoved ) <EOL> def test_get_file_copied ( self ) : <EOL> file_copied_payload = file_copy_payload ( self . node , self . node ) <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , <EOL> payload = file_copied_payload ) <EOL> assert_is_instance ( event , AddonFileCopied ) <EOL> def test_get_file_renamed ( self ) : <EOL> file_rename_payload = file_renamed_payload ( ) <EOL> event = event_registry [ '<STR_LIT>' ] ( self . user , self . node , '<STR_LIT>' , <EOL> payload = file_rename_payload ) <EOL> assert_is_instance ( event , AddonFileRenamed ) <EOL> class TestSignalEvent ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestSignalEvent , self ) . setUp ( ) <EOL> self . user = factories . UserFactory ( ) <EOL> self . auth = Auth ( user = self . user ) <EOL> self . node = factories . ProjectFactory ( creator = self . user ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_event_signal ( self , mock_perform ) : <EOL> signals . file_updated . send ( <EOL> user = self . user , node = self . node , event_type = '<STR_LIT>' , payload = file_payload <EOL> ) <EOL> assert_true ( mock_perform . called ) <EOL> class TestFileUpdated ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFileUpdated , self ) . setUp ( ) <EOL> self . user_1 = factories . AuthUserFactory ( ) <EOL> self . auth = Auth ( user = self . user_1 ) <EOL> self . user_2 = factories . AuthUserFactory ( ) <EOL> self . project = factories . ProjectFactory ( creator = self . user_1 ) <EOL> self . sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' , <EOL> ) <EOL> self . sub . save ( ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( self . user_2 , self . project , '<STR_LIT>' , payload = file_payload ) <EOL> def test_info_formed_correct ( self ) : <EOL> assert_equal ( '<STR_LIT>' . format ( wb_path ) , self . event . event_type ) <EOL> assert_equal ( '<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . html_message ) <EOL> assert_equal ( '<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . text_message ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_file_updated ( self , mock_notify ) : <EOL> self . event . perform ( ) <EOL> assert_true ( mock_notify . called ) <EOL> class TestFileAdded ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFileAdded , self ) . setUp ( ) <EOL> self . user = factories . UserFactory ( ) <EOL> self . consolidate_auth = Auth ( user = self . user ) <EOL> self . project = factories . ProjectFactory ( ) <EOL> self . project_subscription = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . project_subscription . save ( ) <EOL> self . user2 = factories . UserFactory ( ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( self . user2 , self . project , '<STR_LIT>' , payload = file_payload ) <EOL> def test_info_formed_correct ( self ) : <EOL> assert_equal ( '<STR_LIT>' . format ( wb_path ) , self . event . event_type ) <EOL> assert_equal ( '<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . html_message ) <EOL> assert_equal ( '<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . text_message ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_file_added ( self , mock_notify ) : <EOL> self . event . perform ( ) <EOL> assert_true ( mock_notify . called ) <EOL> class TestFileRemoved ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFileRemoved , self ) . setUp ( ) <EOL> self . user = factories . UserFactory ( ) <EOL> self . consolidate_auth = Auth ( user = self . user ) <EOL> self . project = factories . ProjectFactory ( ) <EOL> self . project_subscription = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . project_subscription . save ( ) <EOL> self . user2 = factories . UserFactory ( ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( <EOL> self . user2 , self . project , '<STR_LIT>' , payload = file_deleted_payload <EOL> ) <EOL> def test_info_formed_correct_file ( self ) : <EOL> assert_equal ( '<STR_LIT>' , self . event . event_type ) <EOL> assert_equal ( '<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . html_message ) <EOL> assert_equal ( '<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . text_message ) <EOL> def test_info_formed_correct_folder ( self ) : <EOL> assert_equal ( '<STR_LIT>' , self . event . event_type ) <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] += u'<STR_LIT:/>' <EOL> assert_equal ( u'<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . html_message ) <EOL> assert_equal ( u'<STR_LIT>' . format ( materialized . lstrip ( '<STR_LIT:/>' ) ) , self . event . text_message ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_file_removed ( self , mock_notify ) : <EOL> self . event . perform ( ) <EOL> assert_true ( mock_notify . called ) <EOL> class TestFolderCreated ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFolderCreated , self ) . setUp ( ) <EOL> self . user = factories . UserFactory ( ) <EOL> self . consolidate_auth = Auth ( user = self . user ) <EOL> self . project = factories . ProjectFactory ( ) <EOL> self . project_subscription = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . project_subscription . save ( ) <EOL> self . user2 = factories . UserFactory ( ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( <EOL> self . user2 , self . project , '<STR_LIT>' , payload = folder_created_payload <EOL> ) <EOL> def test_info_formed_correct ( self ) : <EOL> assert_equal ( '<STR_LIT>' , self . event . event_type ) <EOL> assert_equal ( '<STR_LIT>' , self . event . html_message ) <EOL> assert_equal ( '<STR_LIT>' , self . event . text_message ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_folder_added ( self , mock_notify ) : <EOL> self . event . perform ( ) <EOL> assert_true ( mock_notify . called ) <EOL> class TestFolderFileRenamed ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFolderFileRenamed , self ) . setUp ( ) <EOL> self . user_1 = factories . AuthUserFactory ( ) <EOL> self . auth = Auth ( user = self . user_1 ) <EOL> self . user_2 = factories . AuthUserFactory ( ) <EOL> self . project = factories . ProjectFactory ( creator = self . user_1 ) <EOL> self . sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' , <EOL> ) <EOL> self . sub . save ( ) <EOL> file_renamed_payload = file_move_payload ( self . project , self . project ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( <EOL> self . user_1 , self . project , '<STR_LIT>' , <EOL> payload = file_renamed_payload <EOL> ) <EOL> self . sub . email_digest . append ( self . user_2 ) <EOL> self . sub . save ( ) <EOL> def test_rename_file_html ( self ) : <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> assert_equal ( self . event . html_message , '<STR_LIT>' ) <EOL> def test_rename_folder_html ( self ) : <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> self . event . payload [ '<STR_LIT:source>' ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> assert_equal ( self . event . html_message , '<STR_LIT>' ) <EOL> def test_rename_file_text ( self ) : <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> assert_equal ( self . event . text_message , '<STR_LIT>' ) <EOL> def test_rename_folder_text ( self ) : <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . event . payload [ '<STR_LIT>' ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> self . event . payload [ '<STR_LIT:source>' ] [ '<STR_LIT>' ] = "<STR_LIT>" <EOL> assert_equal ( self . event . text_message , '<STR_LIT>' ) <EOL> class TestFileMoved ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFileMoved , self ) . setUp ( ) <EOL> self . user_1 = factories . AuthUserFactory ( ) <EOL> self . auth = Auth ( user = self . user_1 ) <EOL> self . user_2 = factories . AuthUserFactory ( ) <EOL> self . user_3 = factories . AuthUserFactory ( ) <EOL> self . user_4 = factories . AuthUserFactory ( ) <EOL> self . project = factories . ProjectFactory ( creator = self . user_1 ) <EOL> self . private_node = factories . NodeFactory ( parent = self . project , is_public = False , creator = self . user_1 ) <EOL> file_moved_payload = file_move_payload ( self . private_node , self . project ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( <EOL> self . user_2 , self . private_node , '<STR_LIT>' , payload = file_moved_payload <EOL> ) <EOL> self . sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . sub . save ( ) <EOL> self . private_sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . private_node . _id + '<STR_LIT>' , <EOL> owner = self . private_node , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . private_sub . save ( ) <EOL> self . file_sub = factories . NotificationSubscriptionFactory ( <EOL> _id = '<STR_LIT>' . format ( <EOL> pid = self . project . _id , <EOL> wbid = self . event . waterbutler_id <EOL> ) , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . file_sub . save ( ) <EOL> def test_info_formed_correct ( self ) : <EOL> assert_equal ( '<STR_LIT>' . format ( wb_path ) , self . event . event_type ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_user_performing_action_no_email ( self , mock_store ) : <EOL> self . sub . email_digest . append ( self . user_2 ) <EOL> self . sub . save ( ) <EOL> self . event . perform ( ) <EOL> assert_equal ( <NUM_LIT:0> , mock_store . call_count ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_perform_store_called_once ( self , mock_store ) : <EOL> self . sub . email_transactional . append ( self . user_1 ) <EOL> self . sub . save ( ) <EOL> self . event . perform ( ) <EOL> assert_equal ( <NUM_LIT:1> , mock_store . call_count ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_perform_store_one_of_each ( self , mock_store ) : <EOL> self . sub . email_transactional . append ( self . user_1 ) <EOL> self . project . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . private_node . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . private_node . save ( ) <EOL> self . sub . email_digest . append ( self . user_3 ) <EOL> self . sub . save ( ) <EOL> self . project . add_contributor ( self . user_4 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . file_sub . email_digest . append ( self . user_4 ) <EOL> self . file_sub . save ( ) <EOL> self . event . perform ( ) <EOL> assert_equal ( <NUM_LIT:3> , mock_store . call_count ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_remove_user_sent_once ( self , mock_store ) : <EOL> self . project . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . file_sub . email_digest . append ( self . user_3 ) <EOL> self . file_sub . save ( ) <EOL> self . event . perform ( ) <EOL> assert_equal ( <NUM_LIT:1> , mock_store . call_count ) <EOL> class TestFileCopied ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestFileCopied , self ) . setUp ( ) <EOL> self . user_1 = factories . AuthUserFactory ( ) <EOL> self . auth = Auth ( user = self . user_1 ) <EOL> self . user_2 = factories . AuthUserFactory ( ) <EOL> self . user_3 = factories . AuthUserFactory ( ) <EOL> self . user_4 = factories . AuthUserFactory ( ) <EOL> self . project = factories . ProjectFactory ( creator = self . user_1 ) <EOL> self . private_node = factories . NodeFactory ( parent = self . project , is_public = False , creator = self . user_1 ) <EOL> file_copied_payload = file_copy_payload ( self . private_node , self . project ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( <EOL> self . user_2 , self . private_node , '<STR_LIT>' , <EOL> payload = file_copied_payload <EOL> ) <EOL> self . sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . sub . save ( ) <EOL> self . private_sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . private_node . _id + '<STR_LIT>' , <EOL> owner = self . private_node , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . private_sub . save ( ) <EOL> self . file_sub = factories . NotificationSubscriptionFactory ( <EOL> _id = '<STR_LIT>' . format ( <EOL> pid = self . project . _id , <EOL> wbid = self . event . waterbutler_id <EOL> ) , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . file_sub . save ( ) <EOL> def test_info_correct ( self ) : <EOL> assert_equal ( '<STR_LIT>' . format ( wb_path ) , self . event . event_type ) <EOL> assert_equal ( ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , self . event . html_message ) <EOL> assert_equal ( ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) , self . event . text_message ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_copied_one_of_each ( self , mock_store ) : <EOL> self . sub . email_transactional . append ( self . user_1 ) <EOL> self . project . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . private_node . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . private_node . save ( ) <EOL> self . sub . email_digest . append ( self . user_3 ) <EOL> self . sub . save ( ) <EOL> self . project . add_contributor ( self . user_4 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . file_sub . email_digest . append ( self . user_4 ) <EOL> self . file_sub . save ( ) <EOL> self . event . perform ( ) <EOL> assert_equal ( <NUM_LIT:2> , mock_store . call_count ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_user_performing_action_no_email ( self , mock_store ) : <EOL> self . sub . email_digest . append ( self . user_2 ) <EOL> self . sub . save ( ) <EOL> self . event . perform ( ) <EOL> assert_equal ( <NUM_LIT:0> , mock_store . call_count ) <EOL> class TestCategorizeUsers ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestCategorizeUsers , self ) . setUp ( ) <EOL> self . user_1 = factories . AuthUserFactory ( ) <EOL> self . auth = Auth ( user = self . user_1 ) <EOL> self . user_2 = factories . AuthUserFactory ( ) <EOL> self . user_3 = factories . AuthUserFactory ( ) <EOL> self . user_4 = factories . AuthUserFactory ( ) <EOL> self . project = factories . ProjectFactory ( creator = self . user_1 ) <EOL> self . private_node = factories . NodeFactory ( <EOL> parent = self . project , is_public = False , creator = self . user_1 <EOL> ) <EOL> file_moved_payload = file_move_payload ( self . private_node , self . project ) <EOL> self . event = event_registry [ '<STR_LIT>' ] ( <EOL> self . user_2 , self . private_node , '<STR_LIT>' , <EOL> payload = file_moved_payload <EOL> ) <EOL> self . sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . project . _id + '<STR_LIT>' , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . sub . save ( ) <EOL> self . private_sub = factories . NotificationSubscriptionFactory ( <EOL> _id = self . private_node . _id + '<STR_LIT>' , <EOL> owner = self . private_node , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . private_sub . save ( ) <EOL> self . file_sub = factories . NotificationSubscriptionFactory ( <EOL> _id = '<STR_LIT>' . format ( <EOL> pid = self . project . _id , <EOL> wbid = self . event . waterbutler_id <EOL> ) , <EOL> owner = self . project , <EOL> event_name = '<STR_LIT>' <EOL> ) <EOL> self . file_sub . save ( ) <EOL> def test_warn_user ( self ) : <EOL> self . sub . email_transactional . append ( self . user_1 ) <EOL> self . project . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . private_node . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . private_node . save ( ) <EOL> self . sub . email_digest . append ( self . user_3 ) <EOL> self . sub . save ( ) <EOL> self . private_sub . none . append ( self . user_3 ) <EOL> self . private_sub . save ( ) <EOL> moved , warn , removed = utils . categorize_users ( <EOL> self . event . user , self . event . event_type , self . event . source_node , <EOL> self . event . event_type , self . event . node <EOL> ) <EOL> assert_equal ( { email_transactional : [ ] , email_digest : [ self . user_3 . _id ] , '<STR_LIT:none>' : [ ] } , warn ) <EOL> assert_equal ( { email_transactional : [ self . user_1 . _id ] , email_digest : [ ] , '<STR_LIT:none>' : [ ] } , moved ) <EOL> def test_moved_user ( self ) : <EOL> self . project . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . private_node . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . private_node . save ( ) <EOL> self . sub . email_digest . append ( self . user_3 ) <EOL> self . sub . save ( ) <EOL> self . private_sub . email_transactional . append ( self . user_3 ) <EOL> self . private_sub . save ( ) <EOL> moved , warn , removed = utils . categorize_users ( <EOL> self . event . user , self . event . event_type , self . event . source_node , <EOL> self . event . event_type , self . event . node <EOL> ) <EOL> assert_equal ( { email_transactional : [ ] , email_digest : [ ] , '<STR_LIT:none>' : [ ] } , warn ) <EOL> assert_equal ( { email_transactional : [ self . user_3 . _id ] , email_digest : [ ] , '<STR_LIT:none>' : [ ] } , moved ) <EOL> def test_remove_user ( self ) : <EOL> self . project . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] , auth = self . auth ) <EOL> self . project . save ( ) <EOL> self . file_sub . email_transactional . append ( self . user_3 ) <EOL> self . file_sub . save ( ) <EOL> moved , warn , removed = utils . categorize_users ( <EOL> self . event . user , self . event . event_type , self . event . source_node , <EOL> self . event . event_type , self . event . node <EOL> ) <EOL> assert_equal ( { email_transactional : [ self . user_3 . _id ] , email_digest : [ ] , '<STR_LIT:none>' : [ ] } , removed ) <EOL> def test_node_permissions ( self ) : <EOL> self . private_node . add_contributor ( self . user_3 , permissions = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . private_sub . email_digest . extend ( [ self . user_3 , self . user_4 ] ) <EOL> remove = { email_transactional : [ ] , email_digest : [ ] , '<STR_LIT:none>' : [ ] } <EOL> warn = { email_transactional : [ ] , email_digest : [ self . user_3 . _id , self . user_4 . _id ] , '<STR_LIT:none>' : [ ] } <EOL> subbed , remove = utils . subscriptions_node_permissions ( <EOL> self . private_node , <EOL> warn , <EOL> remove <EOL> ) <EOL> assert_equal ( { email_transactional : [ ] , email_digest : [ self . user_3 . _id ] , '<STR_LIT:none>' : [ ] } , subbed ) <EOL> assert_equal ( { email_transactional : [ ] , email_digest : [ self . user_4 . _id ] , '<STR_LIT:none>' : [ ] } , remove ) <EOL> class TestSubscriptionManipulations ( OsfTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestSubscriptionManipulations , self ) . setUp ( ) <EOL> self . emails_1 = { <EOL> email_digest : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> email_transactional : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT:none>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> } <EOL> self . emails_2 = { <EOL> email_digest : [ '<STR_LIT>' ] , <EOL> email_transactional : [ '<STR_LIT>' ] , <EOL> '<STR_LIT:none>' : [ '<STR_LIT>' ] <EOL> } <EOL> self . emails_3 = { <EOL> email_digest : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> email_transactional : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT:none>' : [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> } <EOL> self . emails_4 = { <EOL> email_digest : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> email_transactional : [ '<STR_LIT>' ] , <EOL> '<STR_LIT:none>' : [ ] <EOL> } <EOL> self . diff_1_3 = { email_transactional : [ '<STR_LIT>' ] , '<STR_LIT:none>' : [ '<STR_LIT>' ] , email_digest : [ '<STR_LIT>' ] } <EOL> self . union_1_2 = { '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT:none>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] } <EOL> self . dup_1_3 = { email_transactional : [ '<STR_LIT>' , '<STR_LIT>' ] , '<STR_LIT:none>' : [ '<STR_LIT>' , '<STR_LIT>' ] , <EOL> '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } <EOL> def test_subscription_user_difference ( self ) : <EOL> result = utils . subscriptions_users_difference ( self . emails_1 , self . emails_3 ) <EOL> assert_equal ( self . diff_1_3 , result ) <EOL> def test_subscription_user_union ( self ) : <EOL> result = utils . subscriptions_users_union ( self . emails_1 , self . emails_2 ) <EOL> assert_equal ( self . union_1_2 , result ) <EOL> def test_remove_duplicates ( self ) : <EOL> result = utils . subscriptions_users_remove_duplicates ( <EOL> self . emails_1 , self . emails_4 , remove_same = False <EOL> ) <EOL> assert_equal ( self . dup_1_3 , result ) <EOL> def test_remove_duplicates_true ( self ) : <EOL> result = utils . subscriptions_users_remove_duplicates ( <EOL> self . emails_1 , self . emails_1 , remove_same = True <EOL> ) <EOL> assert_equal ( { email_digest : [ ] , email_transactional : [ ] , '<STR_LIT:none>' : [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] } , result ) <EOL> wb_path = u'<STR_LIT>' <EOL> materialized = u'<STR_LIT>' <EOL> provider = u'<STR_LIT>' <EOL> name = u'<STR_LIT>' <EOL> file_payload = OrderedDict ( [ <EOL> ( u'<STR_LIT:action>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT:email>' , u'<STR_LIT>' ) , ( u'<STR_LIT:id>' , u'<STR_LIT>' ) , ( u'<STR_LIT:name>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , None ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , <NUM_LIT:0> ) , <EOL> ( u'<STR_LIT:version>' , <NUM_LIT:30> ) ] ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:file>' ) , <EOL> ( u'<STR_LIT>' , materialized ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , name ) , <EOL> ( u'<STR_LIT:path>' , wb_path ) , <EOL> ( u'<STR_LIT>' , provider ) , <EOL> ( u'<STR_LIT:size>' , <NUM_LIT> ) ] ) ) , <EOL> ( u'<STR_LIT>' , provider ) , <EOL> ( u'<STR_LIT:time>' , <NUM_LIT> ) ] ) <EOL> file_deleted_payload = OrderedDict ( [ <EOL> ( u'<STR_LIT:action>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT:email>' , u'<STR_LIT>' ) , ( u'<STR_LIT:id>' , u'<STR_LIT>' ) , ( u'<STR_LIT:name>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , materialized ) , <EOL> ( u'<STR_LIT:path>' , materialized ) ] ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:time>' , <NUM_LIT> ) ] ) <EOL> folder_created_payload = OrderedDict ( [ <EOL> ( u'<STR_LIT:action>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT:email>' , u'<STR_LIT>' ) , ( u'<STR_LIT:id>' , u'<STR_LIT>' ) , ( u'<STR_LIT:name>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:path>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:time>' , <NUM_LIT> ) ] ) <EOL> def file_move_payload ( new_node , old_node ) : <EOL> return OrderedDict ( [ <EOL> ( u'<STR_LIT:action>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT:email>' , '<STR_LIT>' ) , ( u'<STR_LIT:id>' , '<STR_LIT>' ) , ( u'<STR_LIT:name>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , None ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , <NUM_LIT:0> ) , <EOL> ( u'<STR_LIT:version>' , <NUM_LIT:30> ) ] ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:file>' ) , <EOL> ( u'<STR_LIT>' , materialized ) , <EOL> ( u'<STR_LIT>' , None ) , <EOL> ( u'<STR_LIT:name>' , name ) , <EOL> ( u'<STR_LIT>' , str ( new_node ) ) , <EOL> ( u'<STR_LIT:path>' , wb_path ) , <EOL> ( u'<STR_LIT>' , provider ) , <EOL> ( u'<STR_LIT:size>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:url>' : '<STR_LIT>' . format ( new_node . _id ) , '<STR_LIT>' : new_node . _id , '<STR_LIT:title>' : u'<STR_LIT>' } ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:source>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , materialized ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , str ( old_node ) ) , <EOL> ( u'<STR_LIT:path>' , materialized ) , <EOL> ( u'<STR_LIT>' , provider ) , <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:url>' : '<STR_LIT>' . format ( old_node . _id ) , '<STR_LIT>' : old_node . _id , '<STR_LIT:title>' : u'<STR_LIT>' } ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:time>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , None ) ] ) <EOL> def file_copy_payload ( new_node , old_node ) : <EOL> return OrderedDict ( [ <EOL> ( u'<STR_LIT:action>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT:email>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:id>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , None ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , <NUM_LIT:0> ) , <EOL> ( u'<STR_LIT:version>' , <NUM_LIT:30> ) ] ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:file>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , None ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:path>' , wb_path ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:size>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:url>' : '<STR_LIT>' , '<STR_LIT>' : old_node . _id , '<STR_LIT:title>' : u'<STR_LIT>' } ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:source>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:path>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:url>' : '<STR_LIT>' , '<STR_LIT>' : new_node . _id , '<STR_LIT:title>' : u'<STR_LIT>' } ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:time>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , None ) ] ) <EOL> def file_renamed_payload ( ) : <EOL> return OrderedDict ( [ <EOL> ( u'<STR_LIT:action>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT:email>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:id>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , None ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , <NUM_LIT:0> ) , <EOL> ( u'<STR_LIT>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:version>' , <NUM_LIT:1> ) ] ) ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT:file>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:path>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:size>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:url>' : '<STR_LIT>' , '<STR_LIT>' : u'<STR_LIT>' , '<STR_LIT:title>' : u'<STR_LIT>' } ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:source>' , OrderedDict ( [ <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:name>' , u'<STR_LIT>' ) , ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT:path>' , u'<STR_LIT>' ) , <EOL> ( u'<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( '<STR_LIT:url>' , '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , { '<STR_LIT:url>' : '<STR_LIT>' , '<STR_LIT>' : u'<STR_LIT>' , '<STR_LIT:title>' : u'<STR_LIT>' } ) , <EOL> ( '<STR_LIT>' , '<STR_LIT>' ) ] ) ) , <EOL> ( u'<STR_LIT:time>' , <NUM_LIT> ) , <EOL> ( '<STR_LIT>' , u'<STR_LIT>' ) , <EOL> ( '<STR_LIT>' , None ) ] ) </s>
