<s> from collections import namedtuple <EOL> import json <EOL> import sharknado <EOL> import unittest <EOL> from bson import ObjectId <EOL> from datetime import datetime , timedelta <EOL> from tornado import escape , gen <EOL> from tornado . escape import to_unicode <EOL> from tornado . ioloop import IOLoop <EOL> from tornado . testing import AsyncHTTPTestCase <EOL> from six . moves . urllib . parse import urlencode <EOL> from six import iteritems <EOL> class TestHelpers ( unittest . TestCase ) : <EOL> def test_mongo_encoder_datetime ( self ) : <EOL> now = datetime . utcnow ( ) <EOL> data = { '<STR_LIT>' : now } <EOL> json_data = sharknado . MongoEncoder ( ) . encode ( data ) <EOL> self . assertEqual ( now . isoformat ( ) , json . loads ( json_data ) [ '<STR_LIT>' ] ) <EOL> def test_mongo_encoder_objectid ( self ) : <EOL> _id = ObjectId ( ) <EOL> data = { '<STR_LIT>' : _id } <EOL> json_data = sharknado . MongoEncoder ( ) . encode ( data ) <EOL> self . assertEqual ( str ( _id ) , json . loads ( json_data ) [ '<STR_LIT>' ] ) <EOL> def test_tornado_encoder_patch ( self ) : <EOL> self . assertEqual ( sharknado . json_encode , escape . json_encode ) <EOL> def test_evt_response_content ( self ) : <EOL> expected = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } <EOL> result = sharknado . make_evt_response ( expected [ '<STR_LIT>' ] , expected [ '<STR_LIT>' ] , expected [ '<STR_LIT>' ] ) <EOL> self . assertDictEqual ( expected , result ) <EOL> def test_evt_response_order ( self ) : <EOL> expected_items = [ ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , '<STR_LIT>' ) , ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) ] <EOL> expected = dict ( expected_items ) <EOL> result = sharknado . make_evt_response ( expected [ '<STR_LIT>' ] , expected [ '<STR_LIT>' ] , expected [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( expected_items , list ( iteritems ( result ) ) ) <EOL> def test_args_parser ( self ) : <EOL> args = { '<STR_LIT>' : [ '<STR_LIT>' ] , '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } <EOL> parsed = sharknado . parse_args ( args ) <EOL> self . assertEqual ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : [ '<STR_LIT>' , '<STR_LIT>' ] } , parsed ) <EOL> class TestServices ( AsyncHTTPTestCase ) : <EOL> def setUp ( self ) : <EOL> self . _options = sharknado . options <EOL> sharknado . options = namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) . _make ( <EOL> [ '<STR_LIT>' , <NUM_LIT:0> , '<STR_LIT:*>' ] ) <EOL> self . db = sharknado . make_mongo_db ( ) <EOL> super ( TestServices , self ) . setUp ( ) <EOL> def tearDown ( self ) : <EOL> sharknado . options = self . _options <EOL> self . db . connection . drop_database ( self . db . delegate . name ) <EOL> super ( TestServices , self ) . tearDown ( ) <EOL> def get_app ( self ) : <EOL> app = sharknado . make_app ( ) <EOL> app . settings [ '<STR_LIT>' ] = self . db <EOL> return app <EOL> def get_new_ioloop ( self ) : <EOL> return IOLoop . instance ( ) <EOL> def _get ( self , url , params = None , ** kwargs ) : <EOL> if params : <EOL> url += ( '<STR_LIT>' % urlencode ( params ) ) <EOL> return self . fetch ( url , ** kwargs ) <EOL> def test_send_empty_message ( self ) : <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> self . assertDictContainsSubset ( { '<STR_LIT:content>' : { } , '<STR_LIT>' : '<STR_LIT:test>' } , json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] ) <EOL> @ gen . coroutine <EOL> def check_count ( ) : <EOL> count = yield self . db . messages . count ( ) <EOL> self . assertEqual ( <NUM_LIT:1> , count ) <EOL> self . io_loop . run_sync ( check_count ) <EOL> def test_send_message_with_data ( self ) : <EOL> params = { '<STR_LIT>' : '<STR_LIT>' } <EOL> resp = self . _get ( '<STR_LIT>' , params = params ) <EOL> self . assertEqual ( params , json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] [ '<STR_LIT:content>' ] ) <EOL> @ gen . coroutine <EOL> def check_stored ( ) : <EOL> stored = yield self . db . messages . find_one ( ) <EOL> self . assertEqual ( params , stored [ '<STR_LIT:content>' ] ) <EOL> self . io_loop . run_sync ( check_stored ) <EOL> def test_send_message_json_body ( self ) : <EOL> params = { '<STR_LIT>' : '<STR_LIT>' } <EOL> resp = self . fetch ( '<STR_LIT>' , method = '<STR_LIT:POST>' , body = json . dumps ( params ) ) <EOL> self . assertEqual ( params , json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] [ '<STR_LIT:content>' ] ) <EOL> def test_send_message_formencoded_failure ( self ) : <EOL> params = { '<STR_LIT>' : '<STR_LIT>' } <EOL> resp = self . fetch ( '<STR_LIT>' , method = '<STR_LIT:POST>' , body = urlencode ( params ) ) <EOL> self . assertEqual ( '<STR_LIT>' , json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] ) <EOL> def test_get_latest_message ( self ) : <EOL> latest_params = None <EOL> for idx in range ( <NUM_LIT:3> ) : <EOL> latest_params = { '<STR_LIT>' : '<STR_LIT>' % idx } <EOL> self . _get ( '<STR_LIT>' , params = latest_params ) <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> resp_json = json . loads ( to_unicode ( resp . body ) ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( resp_json [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( latest_params , resp_json [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:content>' ] ) <EOL> def test_get_all_messages ( self ) : <EOL> latest_params = None <EOL> for idx in range ( <NUM_LIT:3> ) : <EOL> latest_params = { '<STR_LIT>' : '<STR_LIT>' % idx } <EOL> self . _get ( '<STR_LIT>' , params = latest_params ) <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> resp_json = json . loads ( to_unicode ( resp . body ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( resp_json [ '<STR_LIT>' ] ) ) <EOL> self . assertEqual ( latest_params , resp_json [ '<STR_LIT>' ] [ <NUM_LIT:0> ] [ '<STR_LIT:content>' ] ) <EOL> def test_filter_message_days ( self ) : <EOL> @ gen . coroutine <EOL> def store_messages ( ) : <EOL> delta_days = [ <NUM_LIT:1> , <NUM_LIT:1> , <NUM_LIT:2> , <NUM_LIT:3> , <NUM_LIT:5> , <NUM_LIT:8> ] <EOL> for days in delta_days : <EOL> created = datetime . utcnow ( ) - timedelta ( days = days ) <EOL> message = { '<STR_LIT>' : '<STR_LIT:test>' , '<STR_LIT>' : created , '<STR_LIT:content>' : { } } <EOL> yield self . db . messages . insert ( message ) <EOL> self . io_loop . run_sync ( store_messages ) <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:4> , len ( json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] ) ) <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:5> , len ( json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] ) ) <EOL> def test_default_filter_30_days ( self ) : <EOL> @ gen . coroutine <EOL> def store_messages ( ) : <EOL> message = { '<STR_LIT>' : '<STR_LIT:test>' , '<STR_LIT>' : datetime . utcnow ( ) , '<STR_LIT:content>' : { } } <EOL> yield self . db . messages . insert ( message ) <EOL> message = { '<STR_LIT>' : '<STR_LIT:test>' , '<STR_LIT>' : datetime . utcnow ( ) - timedelta ( days = <NUM_LIT> ) , '<STR_LIT:content>' : { } } <EOL> yield self . db . messages . insert ( message ) <EOL> self . io_loop . run_sync ( store_messages ) <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] ) ) <EOL> def test_message_counter ( self ) : <EOL> for _ in range ( <NUM_LIT:3> ) : <EOL> self . _get ( '<STR_LIT>' ) <EOL> resp = self . _get ( '<STR_LIT>' ) <EOL> self . assertEqual ( <NUM_LIT:3> , json . loads ( to_unicode ( resp . body ) ) [ '<STR_LIT>' ] [ '<STR_LIT:count>' ] ) <EOL> def test_cors ( self ) : <EOL> resp = self . _get ( '<STR_LIT>' , headers = { '<STR_LIT>' : '<STR_LIT:localhost>' } ) <EOL> self . assertEqual ( '<STR_LIT:*>' , resp . headers [ '<STR_LIT>' ] ) </s>
