Update checked in version of gsutil to version 4.6

third_party/boto/tests/unit/cloudsearch2/test_search.py

+#!/usr/bin env python
+
+from tests.unit import unittest
+from httpretty import HTTPretty
+
+import urlparse
+import json
+import mock
+import requests
+
+from boto.cloudsearch2.search import SearchConnection, SearchServiceException
+
+HOSTNAME = "search-demo-userdomain.us-east-1.cloudsearch.amazonaws.com"
+FULL_URL = 'http://%s/2013-01-01/search' % HOSTNAME
+
+
+class CloudSearchSearchBaseTest(unittest.TestCase):
+
+    hits = [
+        {
+            'id': '12341',
+            'fields': {
+                'title': 'Document 1',
+                'rank': 1
+            }
+        },
+        {
+            'id': '12342',
+            'fields': {
+                'title': 'Document 2',
+                'rank': 2
+            }
+        },
+        {
+            'id': '12343',
+            'fields': {
+                'title': 'Document 3',
+                'rank': 3
+            }
+        },
+        {
+            'id': '12344',
+            'fields': {
+                'title': 'Document 4',
+                'rank': 4
+            }
+        },
+        {
+            'id': '12345',
+            'fields': {
+                'title': 'Document 5',
+                'rank': 5
+            }
+        },
+        {
+            'id': '12346',
+            'fields': {
+                'title': 'Document 6',
+                'rank': 6
+            }
+        },
+        {
+            'id': '12347',
+            'fields': {
+                'title': 'Document 7',
+                'rank': 7
+            }
+        },
+    ]
+
+    content_type = "text/xml"
+    response_status = 200
+
+    def get_args(self, requestline):
+        (_, request, _) = requestline.split(" ")
+        (_, request) = request.split("?", 1)
+        args = urlparse.parse_qs(request)
+        return args
+
+    def setUp(self):
+        HTTPretty.enable()
+        body = self.response
+
+        if not isinstance(body, basestring):
+            body = json.dumps(body)
+
+        HTTPretty.register_uri(HTTPretty.GET, FULL_URL,
+                               body=body,
+                               content_type=self.content_type,
+                               status=self.response_status)
+
+    def tearDown(self):
+        HTTPretty.disable()
+
+class CloudSearchSearchTest(CloudSearchSearchBaseTest):
+    response = {
+        'rank': '-text_relevance',
+        'match-expr':"Test",
+        'hits': {
+            'found': 30,
+            'start': 0,
+            'hit':CloudSearchSearchBaseTest.hits
+            },
+        'status': {
+            'rid':'b7c167f6c2da6d93531b9a7b314ad030b3a74803b4b7797edb905ba5a6a08',
+            'time-ms': 2,
+            'cpu-time-ms': 0
+        }
+
+    }
+
+    def test_cloudsearch_qsearch(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(q='Test')
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['q'], ["Test"])
+        self.assertEqual(args['start'], ["0"])
+        self.assertEqual(args['size'], ["10"])
+
+
+    def test_cloudsearch_search_details(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(q='Test', size=50, start=20)
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['q'], ["Test"])
+        self.assertEqual(args['size'], ["50"])
+        self.assertEqual(args['start'], ["20"])
+
+    def test_cloudsearch_facet_constraint_single(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(
+            q='Test',
+            facet={'author': "'John Smith','Mark Smith'"})
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['facet.author'],
+                         ["'John Smith','Mark Smith'"])
+
+    def test_cloudsearch_facet_constraint_multiple(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(
+            q='Test',
+            facet={'author': "'John Smith','Mark Smith'",
+                   'category': "'News','Reviews'"})
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['facet.author'],
+                         ["'John Smith','Mark Smith'"])
+        self.assertEqual(args['facet.category'],
+                         ["'News','Reviews'"])
+
+    def test_cloudsearch_facet_sort_single(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(q='Test', facet={'author': {'sort':'alpha'}})
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        print args
+
+        self.assertEqual(args['facet.author'], ['{"sort": "alpha"}'])
+
+    def test_cloudsearch_facet_sort_multiple(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(q='Test', facet={'author': {'sort': 'alpha'},
+                                       'cat': {'sort': 'count'}})
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['facet.author'], ['{"sort": "alpha"}'])
+        self.assertEqual(args['facet.cat'], ['{"sort": "count"}'])
+
+    def test_cloudsearch_result_fields_single(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(q='Test', return_fields=['author'])
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['return'], ['author'])
+
+    def test_cloudsearch_result_fields_multiple(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        search.search(q='Test', return_fields=['author', 'title'])
+
+        args = self.get_args(HTTPretty.last_request.raw_requestline)
+
+        self.assertEqual(args['return'], ['author,title'])
+
+    def test_cloudsearch_results_meta(self):
+        """Check returned metadata is parsed correctly"""
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        results = search.search(q='Test')
+
+        # These rely on the default response which is fed into HTTPretty
+        self.assertEqual(results.hits, 30)
+        self.assertEqual(results.docs[0]['fields']['rank'], 1)
+
+    def test_cloudsearch_results_info(self):
+        """Check num_pages_needed is calculated correctly"""
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        results = search.search(q='Test')
+
+        # This relies on the default response which is fed into HTTPretty
+        self.assertEqual(results.num_pages_needed, 3.0)
+
+    def test_cloudsearch_results_matched(self):
+        """
+        Check that information objects are passed back through the API
+        correctly.
+        """
+        search = SearchConnection(endpoint=HOSTNAME)
+        query = search.build_query(q='Test')
+
+        results = search(query)
+
+        self.assertEqual(results.search_service, search)
+        self.assertEqual(results.query, query)
+
+    def test_cloudsearch_results_hits(self):
+        """Check that documents are parsed properly from AWS"""
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        results = search.search(q='Test')
+
+        hits = map(lambda x: x['id'], results.docs)
+
+        # This relies on the default response which is fed into HTTPretty
+        self.assertEqual(
+            hits, ["12341", "12342", "12343", "12344",
+                   "12345", "12346", "12347"])
+
+    def test_cloudsearch_results_iterator(self):
+        """Check the results iterator"""
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        results = search.search(q='Test')
+        results_correct = iter(["12341", "12342", "12343", "12344",
+                                "12345", "12346", "12347"])
+        for x in results:
+            self.assertEqual(x['id'], results_correct.next())
+
+
+    def test_cloudsearch_results_internal_consistancy(self):
+        """Check the documents length matches the iterator details"""
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        results = search.search(q='Test')
+
+        self.assertEqual(len(results), len(results.docs))
+
+    def test_cloudsearch_search_nextpage(self):
+        """Check next page query is correct"""
+        search = SearchConnection(endpoint=HOSTNAME)
+        query1 = search.build_query(q='Test')
+        query2 = search.build_query(q='Test')
+
+        results = search(query2)
+
+        self.assertEqual(results.next_page().query.start,
+                         query1.start + query1.size)
+        self.assertEqual(query1.q, query2.q)
+
+class CloudSearchSearchFacetTest(CloudSearchSearchBaseTest):
+    response = {
+        'rank': '-text_relevance',
+        'match-expr':"Test",
+        'hits': {
+            'found': 30,
+            'start': 0,
+            'hit':CloudSearchSearchBaseTest.hits
+            },
+        'status': {
+            'rid':'b7c167f6c2da6d93531b9a7b314ad030b3a74803b4b7797edb905ba5a6a08',
+            'time-ms': 2,
+            'cpu-time-ms': 0
+        },
+        'facets': {
+            'tags': {},
+            'animals': {'buckets': [{'count': '2', 'value': 'fish'}, {'count': '1', 'value':'lions'}]},
+        }
+    }
+
+    def test_cloudsearch_search_facets(self):
+        #self.response['facets'] = {'tags': {}}
+
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        results = search.search(q='Test', facet={'tags': {}})
+
+        self.assertTrue('tags' not in results.facets)
+        self.assertEqual(results.facets['animals'], {u'lions': u'1', u'fish': u'2'})
+
+
+class CloudSearchNonJsonTest(CloudSearchSearchBaseTest):
+    response = '<html><body><h1>500 Internal Server Error</h1></body></html>'
+    response_status = 500
+    content_type = 'text/xml'
+
+    def test_response(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        with self.assertRaises(SearchServiceException):
+            search.search(q='Test')
+
+
+class CloudSearchUnauthorizedTest(CloudSearchSearchBaseTest):
+    response = '<html><body><h1>403 Forbidden</h1>foo bar baz</body></html>'
+    response_status = 403
+    content_type = 'text/html'
+
+    def test_response(self):
+        search = SearchConnection(endpoint=HOSTNAME)
+
+        with self.assertRaisesRegexp(SearchServiceException, 'foo bar baz'):
+            search.search(q='Test')
+
+
+class FakeResponse(object):
+    status_code = 405
+    content = ''
+
+
+class CloudSearchConnectionTest(unittest.TestCase):
+    cloudsearch = True
+
+    def setUp(self):
+        super(CloudSearchConnectionTest, self).setUp()
+        self.conn = SearchConnection(
+            endpoint='test-domain.cloudsearch.amazonaws.com'
+        )
+
+    def test_expose_additional_error_info(self):
+        mpo = mock.patch.object
+        fake = FakeResponse()
+        fake.content = 'Nopenopenope'
+
+        # First, in the case of a non-JSON, non-403 error.
+        with mpo(self.conn.session, 'get', return_value=fake) as mock_request:
+            with self.assertRaises(SearchServiceException) as cm:
+                self.conn.search(q='not_gonna_happen')
+
+            self.assertTrue('non-json response' in str(cm.exception))
+            self.assertTrue('Nopenopenope' in str(cm.exception))
+
+        # Then with JSON & an 'error' key within.
+        fake.content = json.dumps({
+            'error': "Something went wrong. Oops."
+        })
+
+        with mpo(self.conn.session, 'get', return_value=fake) as mock_request:
+            with self.assertRaises(SearchServiceException) as cm:
+                self.conn.search(q='no_luck_here')
+
+            self.assertTrue('Unknown error' in str(cm.exception))
+            self.assertTrue('went wrong. Oops' in str(cm.exception))