<s> import json <EOL> import mock <EOL> from tests . unit import AWSMockServiceTestCase <EOL> from boto . cloudsearch2 . domain import Domain <EOL> from boto . cloudsearch2 . layer1 import CloudSearchConnection <EOL> from boto . cloudsearchdomain . layer1 import CloudSearchDomainConnection <EOL> class CloudSearchDomainConnectionTest ( AWSMockServiceTestCase ) : <EOL> connection_class = CloudSearchDomainConnection <EOL> domain_status = """<STR_LIT>""" <EOL> def create_service_connection ( self , ** kwargs ) : <EOL> if kwargs . get ( '<STR_LIT:host>' , None ) is None : <EOL> kwargs [ '<STR_LIT:host>' ] = '<STR_LIT>' <EOL> return super ( CloudSearchDomainConnectionTest , self ) . create_service_connection ( ** kwargs ) <EOL> def test_get_search_service ( self ) : <EOL> layer1 = CloudSearchConnection ( aws_access_key_id = '<STR_LIT>' , <EOL> aws_secret_access_key = '<STR_LIT>' , <EOL> sign_request = True ) <EOL> domain = Domain ( layer1 = layer1 , data = json . loads ( self . domain_status ) ) <EOL> search_service = domain . get_search_service ( ) <EOL> self . assertEqual ( search_service . sign_request , True ) <EOL> def test_get_document_service ( self ) : <EOL> layer1 = CloudSearchConnection ( aws_access_key_id = '<STR_LIT>' , <EOL> aws_secret_access_key = '<STR_LIT>' , <EOL> sign_request = True ) <EOL> domain = Domain ( layer1 = layer1 , data = json . loads ( self . domain_status ) ) <EOL> document_service = domain . get_document_service ( ) <EOL> self . assertEqual ( document_service . sign_request , True ) <EOL> def test_search_with_auth ( self ) : <EOL> layer1 = CloudSearchConnection ( aws_access_key_id = '<STR_LIT>' , <EOL> aws_secret_access_key = '<STR_LIT>' , <EOL> sign_request = True ) <EOL> domain = Domain ( layer1 = layer1 , data = json . loads ( self . domain_status ) ) <EOL> search_service = domain . get_search_service ( ) <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>' : { <EOL> '<STR_LIT:id>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:title>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> <EOL> } <EOL> } <EOL> } , <EOL> '<STR_LIT:status>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:2> , <EOL> '<STR_LIT>' : <NUM_LIT:0> <EOL> } <EOL> } <EOL> self . set_http_response ( status_code = <NUM_LIT:200> , <EOL> body = json . dumps ( response ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> search_service . domain_connection = self . service_connection <EOL> resp = search_service . search ( ) <EOL> headers = self . actual_request . headers <EOL> self . assertIsNotNone ( headers . get ( '<STR_LIT>' ) ) <EOL> def test_upload_documents_with_auth ( self ) : <EOL> layer1 = CloudSearchConnection ( aws_access_key_id = '<STR_LIT>' , <EOL> aws_secret_access_key = '<STR_LIT>' , <EOL> sign_request = True ) <EOL> domain = Domain ( layer1 = layer1 , data = json . loads ( self . domain_status ) ) <EOL> document_service = domain . get_document_service ( ) <EOL> response = { <EOL> '<STR_LIT:status>' : '<STR_LIT:success>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : <NUM_LIT:0> , <EOL> } <EOL> document = { <EOL> "<STR_LIT:id>" : "<STR_LIT>" , <EOL> "<STR_LIT:title>" : "<STR_LIT>" , <EOL> "<STR_LIT>" : [ "<STR_LIT>" , "<STR_LIT>" , "<STR_LIT>" ] <EOL> } <EOL> self . set_http_response ( status_code = <NUM_LIT:200> , <EOL> body = json . dumps ( response ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> document_service . domain_connection = self . service_connection <EOL> document_service . add ( "<STR_LIT>" , document ) <EOL> resp = document_service . commit ( ) <EOL> headers = self . actual_request . headers <EOL> self . assertIsNotNone ( headers . get ( '<STR_LIT>' ) ) <EOL> def test_no_host_provided ( self ) : <EOL> with self . assertRaises ( ValueError ) : <EOL> CloudSearchDomainConnection ( <EOL> aws_access_key_id = '<STR_LIT>' , <EOL> aws_secret_access_key = '<STR_LIT>' <EOL> ) </s>
