<s> from tests . compat import mock , unittest <EOL> from httpretty import HTTPretty <EOL> import json <EOL> import requests <EOL> from boto . cloudsearch . search import SearchConnection , SearchServiceException <EOL> from boto . compat import six , map <EOL> HOSTNAME = "<STR_LIT>" <EOL> FULL_URL = '<STR_LIT>' % HOSTNAME <EOL> class CloudSearchSearchBaseTest ( unittest . TestCase ) : <EOL> hits = [ <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> } , <EOL> ] <EOL> content_type = "<STR_LIT>" <EOL> response_status = <NUM_LIT:200> <EOL> def get_args ( self , requestline ) : <EOL> ( _ , request , _ ) = requestline . split ( b"<STR_LIT:U+0020>" ) <EOL> ( _ , request ) = request . split ( b"<STR_LIT:?>" , <NUM_LIT:1> ) <EOL> args = six . moves . urllib . parse . parse_qs ( request ) <EOL> return args <EOL> def setUp ( self ) : <EOL> HTTPretty . enable ( ) <EOL> body = self . response <EOL> if not isinstance ( body , bytes ) : <EOL> body = json . dumps ( body ) . encode ( '<STR_LIT:utf-8>' ) <EOL> HTTPretty . register_uri ( HTTPretty . GET , FULL_URL , <EOL> body = body , <EOL> content_type = self . content_type , <EOL> status = self . response_status ) <EOL> def tearDown ( self ) : <EOL> HTTPretty . disable ( ) <EOL> class CloudSearchSearchTest ( CloudSearchSearchBaseTest ) : <EOL> response = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT:30> , <EOL> '<STR_LIT:start>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : CloudSearchSearchBaseTest . hits <EOL> } , <EOL> '<STR_LIT:info>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> } <EOL> def test_cloudsearch_qsearch ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT:q>' ] , [ b"<STR_LIT>" ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT:start>' ] , [ b"<STR_LIT:0>" ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT:size>' ] , [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_bqsearch ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( bq = "<STR_LIT>" ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_search_details ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , size = <NUM_LIT:50> , start = <NUM_LIT:20> ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT:q>' ] , [ b"<STR_LIT>" ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT:size>' ] , [ b"<STR_LIT>" ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT:start>' ] , [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_facet_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , facet = [ "<STR_LIT>" ] ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_facet_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , facet = [ "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_facet_constraint_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( <EOL> q = '<STR_LIT>' , <EOL> facet_constraints = { '<STR_LIT>' : "<STR_LIT>" } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , <EOL> [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_facet_constraint_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( <EOL> q = '<STR_LIT>' , <EOL> facet_constraints = { '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , <EOL> [ b"<STR_LIT>" ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , <EOL> [ b"<STR_LIT>" ] ) <EOL> def test_cloudsearch_facet_sort_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , facet_sort = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_facet_sort_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , facet_sort = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:count>' } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT:count>' ] ) <EOL> def test_cloudsearch_top_n_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , facet_top_n = { '<STR_LIT>' : <NUM_LIT:5> } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT:5>' ] ) <EOL> def test_cloudsearch_top_n_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , facet_top_n = { '<STR_LIT>' : <NUM_LIT:5> , '<STR_LIT>' : <NUM_LIT:10> } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT:5>' ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_rank_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , rank = [ "<STR_LIT:date>" ] ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT:date>' ] ) <EOL> def test_cloudsearch_rank_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , rank = [ "<STR_LIT:date>" , "<STR_LIT>" ] ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_result_fields_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , return_fields = [ '<STR_LIT>' ] ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_result_fields_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , return_fields = [ '<STR_LIT>' , '<STR_LIT:title>' ] ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_t_field_single ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , t = { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_t_field_multiple ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> search . search ( q = '<STR_LIT>' , t = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : '<STR_LIT>' } ) <EOL> args = self . get_args ( HTTPretty . last_request . raw_requestline ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> self . assertEqual ( args [ b'<STR_LIT>' ] , [ b'<STR_LIT>' ] ) <EOL> def test_cloudsearch_results_meta ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> results = search . search ( q = '<STR_LIT>' ) <EOL> self . assertEqual ( results . rank , "<STR_LIT>" ) <EOL> self . assertEqual ( results . match_expression , "<STR_LIT>" ) <EOL> def test_cloudsearch_results_info ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> results = search . search ( q = '<STR_LIT>' ) <EOL> self . assertEqual ( results . num_pages_needed , <NUM_LIT> ) <EOL> def test_cloudsearch_results_matched ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> query = search . build_query ( q = '<STR_LIT>' ) <EOL> results = search ( query ) <EOL> self . assertEqual ( results . search_service , search ) <EOL> self . assertEqual ( results . query , query ) <EOL> def test_cloudsearch_results_hits ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> results = search . search ( q = '<STR_LIT>' ) <EOL> hits = list ( map ( lambda x : x [ '<STR_LIT:id>' ] , results . docs ) ) <EOL> self . assertEqual ( <EOL> hits , [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> def test_cloudsearch_results_iterator ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> results = search . search ( q = '<STR_LIT>' ) <EOL> results_correct = iter ( [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" , <EOL> "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] ) <EOL> for x in results : <EOL> self . assertEqual ( x [ '<STR_LIT:id>' ] , next ( results_correct ) ) <EOL> def test_cloudsearch_results_internal_consistancy ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> results = search . search ( q = '<STR_LIT>' ) <EOL> self . assertEqual ( len ( results ) , len ( results . docs ) ) <EOL> def test_cloudsearch_search_nextpage ( self ) : <EOL> """<STR_LIT>""" <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> query1 = search . build_query ( q = '<STR_LIT>' ) <EOL> query2 = search . build_query ( q = '<STR_LIT>' ) <EOL> results = search ( query2 ) <EOL> self . assertEqual ( results . next_page ( ) . query . start , <EOL> query1 . start + query1 . size ) <EOL> self . assertEqual ( query1 . q , query2 . q ) <EOL> class CloudSearchSearchFacetTest ( CloudSearchSearchBaseTest ) : <EOL> response = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : <NUM_LIT:30> , <EOL> '<STR_LIT:start>' : <NUM_LIT:0> , <EOL> '<STR_LIT>' : CloudSearchSearchBaseTest . hits <EOL> } , <EOL> '<STR_LIT:info>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : { '<STR_LIT>' : [ { '<STR_LIT:count>' : '<STR_LIT:2>' , '<STR_LIT:value>' : '<STR_LIT>' } , { '<STR_LIT:count>' : '<STR_LIT:1>' , '<STR_LIT:value>' : '<STR_LIT>' } ] } , <EOL> } <EOL> } <EOL> def test_cloudsearch_search_facets ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> results = search . search ( q = '<STR_LIT>' , facet = [ '<STR_LIT>' ] ) <EOL> self . assertTrue ( '<STR_LIT>' not in results . facets ) <EOL> self . assertEqual ( results . facets [ '<STR_LIT>' ] , { u'<STR_LIT>' : u'<STR_LIT:1>' , u'<STR_LIT>' : u'<STR_LIT:2>' } ) <EOL> class CloudSearchNonJsonTest ( CloudSearchSearchBaseTest ) : <EOL> response = b'<STR_LIT>' <EOL> response_status = <NUM_LIT> <EOL> content_type = '<STR_LIT>' <EOL> def test_response ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> with self . assertRaises ( SearchServiceException ) : <EOL> search . search ( q = '<STR_LIT>' ) <EOL> class CloudSearchUnauthorizedTest ( CloudSearchSearchBaseTest ) : <EOL> response = b'<STR_LIT>' <EOL> response_status = <NUM_LIT> <EOL> content_type = '<STR_LIT>' <EOL> def test_response ( self ) : <EOL> search = SearchConnection ( endpoint = HOSTNAME ) <EOL> with self . assertRaisesRegexp ( SearchServiceException , '<STR_LIT>' ) : <EOL> search . search ( q = '<STR_LIT>' ) <EOL> class FakeResponse ( object ) : <EOL> status_code = <NUM_LIT> <EOL> content = b'<STR_LIT>' <EOL> class CloudSearchConnectionTest ( unittest . TestCase ) : <EOL> cloudsearch = True <EOL> def setUp ( self ) : <EOL> super ( CloudSearchConnectionTest , self ) . setUp ( ) <EOL> self . conn = SearchConnection ( <EOL> endpoint = '<STR_LIT>' <EOL> ) <EOL> def test_expose_additional_error_info ( self ) : <EOL> mpo = mock . patch . object <EOL> fake = FakeResponse ( ) <EOL> fake . content = b'<STR_LIT>' <EOL> with mpo ( requests , '<STR_LIT>' , return_value = fake ) as mock_request : <EOL> with self . assertRaises ( SearchServiceException ) as cm : <EOL> self . conn . search ( q = '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in str ( cm . exception ) ) <EOL> self . assertTrue ( '<STR_LIT>' in str ( cm . exception ) ) <EOL> fake . content = json . dumps ( { <EOL> '<STR_LIT:error>' : "<STR_LIT>" <EOL> } ) . encode ( '<STR_LIT:utf-8>' ) <EOL> with mpo ( requests , '<STR_LIT>' , return_value = fake ) as mock_request : <EOL> with self . assertRaises ( SearchServiceException ) as cm : <EOL> self . conn . search ( q = '<STR_LIT>' ) <EOL> self . assertTrue ( '<STR_LIT>' in str ( cm . exception ) ) <EOL> self . assertTrue ( '<STR_LIT>' in str ( cm . exception ) ) </s>
