<s> import types <EOL> import urllib <EOL> from contextlib import nested <EOL> import mock <EOL> import testify as T <EOL> from pushmanager . core import db <EOL> from pushmanager . core . util import get_servlet_urlspec <EOL> from pushmanager . servlets . addrequest import AddRequestServlet <EOL> from pushmanager . testing . mocksettings import MockedSettings <EOL> from pushmanager . testing . testservlet import ServletTestMixin <EOL> class AddRequestServletTest ( T . TestCase , ServletTestMixin ) : <EOL> @ T . class_setup_teardown <EOL> def mock_servlet_env ( self ) : <EOL> self . results = [ ] <EOL> with nested ( <EOL> mock . patch . dict ( db . Settings , MockedSettings ) , <EOL> mock . patch . object ( <EOL> AddRequestServlet , <EOL> "<STR_LIT>" , <EOL> return_value = "<STR_LIT>" <EOL> ) <EOL> ) : <EOL> yield <EOL> def record_pushcontents ( self , success , db_results ) : <EOL> assert success <EOL> self . results = [ ] <EOL> self . results . extend ( db_results . fetchall ( ) ) <EOL> def get_handlers ( self ) : <EOL> return [ get_servlet_urlspec ( AddRequestServlet ) ] <EOL> def test_add_existing_request ( self ) : <EOL> db . execute_cb ( db . push_pushcontents . select ( ) , self . record_pushcontents ) <EOL> num_results_before = len ( self . results ) <EOL> request = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> } <EOL> response = self . fetch ( <EOL> '<STR_LIT>' , <EOL> method = '<STR_LIT:POST>' , <EOL> body = urllib . urlencode ( request ) <EOL> ) <EOL> T . assert_equal ( response . error , None ) <EOL> db . execute_cb ( db . push_pushcontents . select ( ) , self . record_pushcontents ) <EOL> num_results_after = len ( self . results ) <EOL> T . assert_equal ( num_results_after , num_results_before , "<STR_LIT>" ) <EOL> def test_add_new_request ( self ) : <EOL> db . execute_cb ( db . push_pushcontents . select ( ) , self . record_pushcontents ) <EOL> num_results_before = len ( self . results ) <EOL> request = { '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : <NUM_LIT:1> } <EOL> response = self . fetch ( <EOL> '<STR_LIT>' , <EOL> method = '<STR_LIT:POST>' , <EOL> body = urllib . urlencode ( request ) <EOL> ) <EOL> T . assert_equal ( response . error , None ) <EOL> db . execute_cb ( db . push_pushcontents . select ( ) , self . record_pushcontents ) <EOL> num_results_after = len ( self . results ) <EOL> T . assert_equal ( num_results_after , num_results_before + <NUM_LIT:1> , "<STR_LIT>" ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_pushcontent_insert_ignore ( self , mock_transaction ) : <EOL> request = { '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> } <EOL> response = self . fetch ( <EOL> '<STR_LIT>' , <EOL> method = '<STR_LIT:POST>' , <EOL> body = urllib . urlencode ( request ) <EOL> ) <EOL> T . assert_equal ( response . error , None ) <EOL> T . assert_equal ( mock_transaction . call_count , <NUM_LIT:1> ) <EOL> insert_ignore_clause = mock_transaction . call_args [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> T . assert_is ( type ( insert_ignore_clause ) , db . InsertIgnore ) <EOL> def call_on_db_complete ( self ) : <EOL> mocked_self = mock . Mock ( ) <EOL> mocked_self . current_user = '<STR_LIT>' <EOL> mocked_self . pushid = <NUM_LIT:0> <EOL> mocked_self . check_db_results = mock . Mock ( return_value = None ) <EOL> mocked_self . on_db_complete = types . MethodType ( AddRequestServlet . on_db_complete . im_func , mocked_self ) <EOL> no_watcher_req = { <EOL> '<STR_LIT:user>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT:title>' , <EOL> } <EOL> watched_req = { <EOL> '<STR_LIT:user>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT:title>' , <EOL> } <EOL> reqs = [ no_watcher_req , watched_req ] <EOL> mocked_self . on_db_complete ( '<STR_LIT:success>' , [ None , reqs ] ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_mailqueue_on_db_complete ( self , mailq , _ ) : <EOL> self . call_on_db_complete ( ) <EOL> no_watcher_call_args = mailq . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> T . assert_equal ( [ '<STR_LIT>' ] , no_watcher_call_args [ <NUM_LIT:0> ] ) <EOL> T . assert_in ( '<STR_LIT>' , no_watcher_call_args [ <NUM_LIT:1> ] ) <EOL> T . assert_in ( '<STR_LIT>' , no_watcher_call_args [ <NUM_LIT:1> ] ) <EOL> T . assert_in ( '<STR_LIT>' , no_watcher_call_args [ <NUM_LIT:2> ] ) <EOL> watched_call_args = mailq . call_args_list [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> T . assert_equal ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , watched_call_args [ <NUM_LIT:0> ] ) <EOL> T . assert_in ( '<STR_LIT>' , watched_call_args [ <NUM_LIT:1> ] ) <EOL> T . assert_in ( '<STR_LIT>' , watched_call_args [ <NUM_LIT:1> ] ) <EOL> T . assert_in ( '<STR_LIT>' , watched_call_args [ <NUM_LIT:2> ] ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_xmppqueue_on_db_complete ( self , xmppq , _ ) : <EOL> self . call_on_db_complete ( ) <EOL> no_watcher_call_args = xmppq . call_args_list [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> T . assert_equal ( [ '<STR_LIT>' ] , no_watcher_call_args [ <NUM_LIT:0> ] ) <EOL> T . assert_in ( '<STR_LIT>' , no_watcher_call_args [ <NUM_LIT:1> ] ) <EOL> watched_call_args = xmppq . call_args_list [ <NUM_LIT:1> ] [ <NUM_LIT:0> ] <EOL> T . assert_equal ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , watched_call_args [ <NUM_LIT:0> ] ) <EOL> T . assert_in ( '<STR_LIT>' , watched_call_args [ <NUM_LIT:1> ] ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> T . run ( ) </s>
