<s> import base64 <EOL> import json <EOL> import os <EOL> import tempfile <EOL> import pytest <EOL> import six <EOL> from sparkpost . tornado import SparkPost , SparkPostAPIException <EOL> from tornado import ioloop <EOL> from . utils import AsyncClientMock <EOL> responses = AsyncClientMock ( ) <EOL> @ responses . activate <EOL> def test_success_send ( ) : <EOL> responses . add ( <EOL> responses . POST , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT:200> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> results = ioloop . IOLoop ( ) . run_sync ( sp . transmission . send ) <EOL> assert results == '<STR_LIT>' <EOL> @ responses . activate <EOL> def test_success_send_with_attachments ( ) : <EOL> try : <EOL> test_content = six . u ( "<STR_LIT>" ) <EOL> ( _ , temp_file_path ) = tempfile . mkstemp ( ) <EOL> with open ( temp_file_path , "<STR_LIT:w>" ) as temp_file : <EOL> temp_file . write ( test_content ) <EOL> responses . add ( <EOL> responses . POST , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT:200> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> attachment = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:filename>" : temp_file_path <EOL> } <EOL> def send ( ) : <EOL> return sp . transmission . send ( attachments = [ attachment ] ) <EOL> results = ioloop . IOLoop ( ) . run_sync ( send ) <EOL> request_params = json . loads ( responses . calls [ <NUM_LIT:0> ] . request . body ) <EOL> content = base64 . b64decode ( <EOL> request_params [ "<STR_LIT:content>" ] [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:data>" ] ) <EOL> assert test_content == content . decode ( "<STR_LIT:ascii>" ) <EOL> assert results == '<STR_LIT>' <EOL> attachment = { <EOL> "<STR_LIT:name>" : "<STR_LIT>" , <EOL> "<STR_LIT:type>" : "<STR_LIT>" , <EOL> "<STR_LIT:data>" : base64 . b64encode ( <EOL> test_content . encode ( "<STR_LIT:ascii>" ) ) . decode ( "<STR_LIT:ascii>" ) <EOL> } <EOL> def send ( ) : <EOL> return sp . transmission . send ( attachments = [ attachment ] ) <EOL> results = ioloop . IOLoop ( ) . run_sync ( send ) <EOL> request_params = json . loads ( responses . calls [ <NUM_LIT:1> ] . request . body ) <EOL> content = base64 . b64decode ( <EOL> request_params [ "<STR_LIT:content>" ] [ "<STR_LIT>" ] [ <NUM_LIT:0> ] [ "<STR_LIT:data>" ] ) <EOL> assert test_content == content . decode ( "<STR_LIT:ascii>" ) <EOL> assert results == '<STR_LIT>' <EOL> finally : <EOL> os . unlink ( temp_file_path ) <EOL> @ responses . activate <EOL> def test_fail_send ( ) : <EOL> responses . add ( <EOL> responses . POST , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = """<STR_LIT>""" <EOL> ) <EOL> with pytest . raises ( SparkPostAPIException ) : <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> ioloop . IOLoop ( ) . run_sync ( sp . transmission . send ) <EOL> @ responses . activate <EOL> def test_success_get ( ) : <EOL> responses . add ( <EOL> responses . GET , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT:200> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> def send ( ) : <EOL> return sp . transmission . get ( '<STR_LIT>' ) <EOL> results = ioloop . IOLoop ( ) . run_sync ( send , timeout = <NUM_LIT:3> ) <EOL> assert results == { } <EOL> @ responses . activate <EOL> def test_fail_get ( ) : <EOL> responses . add ( <EOL> responses . GET , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = """<STR_LIT>""" <EOL> ) <EOL> with pytest . raises ( SparkPostAPIException ) : <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> def send ( ) : <EOL> return sp . transmission . get ( '<STR_LIT>' ) <EOL> ioloop . IOLoop ( ) . run_sync ( send , timeout = <NUM_LIT:3> ) <EOL> @ responses . activate <EOL> def test_nocontent_get ( ) : <EOL> responses . add ( <EOL> responses . GET , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> response = ioloop . IOLoop ( ) . run_sync ( sp . transmission . list ) <EOL> assert response is True <EOL> @ responses . activate <EOL> def test_brokenjson_get ( ) : <EOL> responses . add ( <EOL> responses . GET , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT:200> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> with pytest . raises ( SparkPostAPIException ) : <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> ioloop . IOLoop ( ) . run_sync ( sp . transmission . list ) <EOL> @ responses . activate <EOL> def test_noresults_get ( ) : <EOL> responses . add ( <EOL> responses . GET , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT:200> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> response = ioloop . IOLoop ( ) . run_sync ( sp . transmission . list ) <EOL> assert response == { "<STR_LIT>" : False } <EOL> @ responses . activate <EOL> def test_success_list ( ) : <EOL> responses . add ( <EOL> responses . GET , <EOL> '<STR_LIT>' , <EOL> status = <NUM_LIT:200> , <EOL> content_type = '<STR_LIT:application/json>' , <EOL> body = '<STR_LIT>' <EOL> ) <EOL> sp = SparkPost ( '<STR_LIT>' ) <EOL> response = ioloop . IOLoop ( ) . run_sync ( sp . transmission . list ) <EOL> assert response == [ ] </s>
