<s> import pytest <EOL> from pushkin import pushkin_cli <EOL> import tornado . web <EOL> from pushkin import context <EOL> from pushkin . database import database <EOL> from pushkin . request . request_processor import RequestProcessor <EOL> from pushkin . requesthandlers . events import JsonEventHandler <EOL> from pushkin . requesthandlers . notifications import JsonNotificationHandler <EOL> from pushkin import test_config_ini_path <EOL> from pushkin import config <EOL> @ pytest . fixture <EOL> def setup_database ( ) : <EOL> database . create_database ( ) <EOL> @ pytest . fixture <EOL> def mock_processor ( mocker ) : <EOL> '''<STR_LIT>''' <EOL> mocker . patch ( '<STR_LIT>' ) <EOL> mocker . patch ( '<STR_LIT>' ) <EOL> @ pytest . fixture <EOL> def app ( ) : <EOL> pushkin_cli . CONFIGURATION_FILENAME = test_config_ini_path <EOL> pushkin_cli . init ( ) <EOL> return pushkin_cli . create_app ( ) <EOL> @ pytest . fixture <EOL> def notification_batch_json ( ) : <EOL> '''<STR_LIT>''' <EOL> return '''<STR_LIT>''' <EOL> @ pytest . fixture <EOL> def post_notification_url ( base_url ) : <EOL> return base_url + config . json_notification_handler_url <EOL> @ pytest . fixture <EOL> def event_batch_json ( ) : <EOL> '''<STR_LIT>''' <EOL> return '''<STR_LIT>''' <EOL> @ pytest . fixture <EOL> def post_event_url ( base_url ) : <EOL> return base_url + config . json_event_handler_url <EOL> @ pytest . mark . gen_test <EOL> @ pytest . mark . parametrize ( "<STR_LIT:input>" , [ <EOL> ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' ) , <EOL> ] ) <EOL> def test_post_notification_empty_request ( setup_database , mock_processor , http_client , post_notification_url , input ) : <EOL> '''<STR_LIT>''' <EOL> request = tornado . httpclient . HTTPRequest ( post_notification_url , method = '<STR_LIT:POST>' , body = input ) <EOL> with pytest . raises ( tornado . httpclient . HTTPError ) : <EOL> yield http_client . fetch ( request ) <EOL> assert not context . request_processor . submit . called <EOL> @ pytest . mark . gen_test <EOL> def test_post_notification ( setup_database , mock_processor , http_client , post_notification_url , <EOL> notification_batch_json ) : <EOL> '''<STR_LIT>''' <EOL> request = tornado . httpclient . HTTPRequest ( post_notification_url , method = '<STR_LIT:POST>' , body = notification_batch_json ) <EOL> response = yield http_client . fetch ( request ) <EOL> assert response . code == <NUM_LIT:200> <EOL> assert context . request_processor . submit . called <EOL> @ pytest . mark . gen_test <EOL> @ pytest . mark . parametrize ( "<STR_LIT:input>" , [ <EOL> ( '<STR_LIT>' ) , <EOL> ( '<STR_LIT>' ) , <EOL> ] ) <EOL> def test_post_event_empty_request ( setup_database , mock_processor , http_client , post_event_url , input ) : <EOL> '''<STR_LIT>''' <EOL> request = tornado . httpclient . HTTPRequest ( post_event_url , method = '<STR_LIT:POST>' , body = input ) <EOL> with pytest . raises ( tornado . httpclient . HTTPError ) : <EOL> yield http_client . fetch ( request ) <EOL> assert not context . request_processor . submit . called <EOL> @ pytest . mark . gen_test <EOL> def test_post_event ( setup_database , mock_processor , http_client , post_event_url , event_batch_json ) : <EOL> '''<STR_LIT>''' <EOL> context . request_processor . submit . return_value = True <EOL> request = tornado . httpclient . HTTPRequest ( post_event_url , method = '<STR_LIT:POST>' , body = event_batch_json ) <EOL> response = yield http_client . fetch ( request ) <EOL> assert response . code == <NUM_LIT:200> <EOL> assert context . request_processor . submit . called <EOL> @ pytest . mark . gen_test <EOL> def test_post_event_service_unavailable ( setup_database , mock_processor , http_client , post_event_url , event_batch_json , <EOL> app ) : <EOL> '''<STR_LIT>''' <EOL> context . request_processor . submit . return_value = False <EOL> request = tornado . httpclient . HTTPRequest ( post_event_url , method = '<STR_LIT:POST>' , body = event_batch_json ) <EOL> RequestProcessor . submit . return_value = False <EOL> with pytest . raises ( tornado . httpclient . HTTPError ) : <EOL> yield http_client . fetch ( request ) </s>
