File Manager

📂 FileMgr

📍/opt/gsutil/gslib/vendored/boto/tests/unit/cloudsearch

✏️ ch.py

⬅ Kembali

#!/usr/bin env python

from tests.compat import mock, unittest
from httpretty import HTTPretty

import json
import requests

from boto.cloudsearch.search import SearchConnection, SearchServiceException
from boto.compat import six, map

HOSTNAME = "search-demo-userdomain.us-east-1.cloudsearch.amazonaws.com"
FULL_URL = 'http://%s/2011-02-01/search' % HOSTNAME

class CloudSearchSearchBaseTest(unittest.TestCase):

hits = [
        {
            'id': '12341',
            'title': 'Document 1',
        },
        {
            'id': '12342',
            'title': 'Document 2',
        },
        {
            'id': '12343',
            'title': 'Document 3',
        },
        {
            'id': '12344',
            'title': 'Document 4',
        },
        {
            'id': '12345',
            'title': 'Document 5',
        },
        {
            'id': '12346',
            'title': 'Document 6',
        },
        {
            'id': '12347',
            'title': 'Document 7',
        },
    ]

content_type = "text/xml"
    response_status = 200

def get_args(self, requestline):
        (_, request, _) = requestline.split(b" ")
        (_, request) = request.split(b"?", 1)
        args = six.moves.urllib.parse.parse_qs(request)
        return args

def setUp(self):
        HTTPretty.enable()
        body = self.response

if not isinstance(body, bytes):
            body = json.dumps(body).encode('utf-8')

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
            },
        'info': {
            '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[b'q'], [b"Test"])
        self.assertEqual(args[b'start'], [b"0"])
        self.assertEqual(args[b'size'], [b"10"])

def test_cloudsearch_bqsearch(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(bq="'Test'")

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'bq'], [b"'Test'"])

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[b'q'], [b"Test"])
        self.assertEqual(args[b'size'], [b"50"])
        self.assertEqual(args[b'start'], [b"20"])

def test_cloudsearch_facet_single(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', facet=["Author"])

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet'], [b"Author"])

def test_cloudsearch_facet_multiple(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', facet=["author", "cat"])

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet'], [b"author,cat"])

def test_cloudsearch_facet_constraint_single(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(
            q='Test',
            facet_constraints={'author': "'John Smith','Mark Smith'"})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet-author-constraints'],
                         [b"'John Smith','Mark Smith'"])

def test_cloudsearch_facet_constraint_multiple(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(
            q='Test',
            facet_constraints={'author': "'John Smith','Mark Smith'",
                               'category': "'News','Reviews'"})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet-author-constraints'],
                         [b"'John Smith','Mark Smith'"])
        self.assertEqual(args[b'facet-category-constraints'],
                         [b"'News','Reviews'"])

def test_cloudsearch_facet_sort_single(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', facet_sort={'author': 'alpha'})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet-author-sort'], [b'alpha'])

def test_cloudsearch_facet_sort_multiple(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', facet_sort={'author': 'alpha',
                                            'cat': 'count'})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet-author-sort'], [b'alpha'])
        self.assertEqual(args[b'facet-cat-sort'], [b'count'])

def test_cloudsearch_top_n_single(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', facet_top_n={'author': 5})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet-author-top-n'], [b'5'])

def test_cloudsearch_top_n_multiple(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', facet_top_n={'author': 5, 'cat': 10})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'facet-author-top-n'], [b'5'])
        self.assertEqual(args[b'facet-cat-top-n'], [b'10'])

def test_cloudsearch_rank_single(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', rank=["date"])

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'rank'], [b'date'])

def test_cloudsearch_rank_multiple(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', rank=["date", "score"])

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b'rank'], [b'date,score'])

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[b'return-fields'], [b'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[b'return-fields'], [b'author,title'])

def test_cloudsearch_t_field_single(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', t={'year': '2001..2007'})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b't-year'], [b'2001..2007'])

def test_cloudsearch_t_field_multiple(self):
        search = SearchConnection(endpoint=HOSTNAME)

search.search(q='Test', t={'year': '2001..2007', 'score': '10..50'})

args = self.get_args(HTTPretty.last_request.raw_requestline)

self.assertEqual(args[b't-year'], [b'2001..2007'])
        self.assertEqual(args[b't-score'], [b'10..50'])

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.rank, "-text_relevance")
        self.assertEqual(results.match_expression, "Test")

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 = list(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'], next(results_correct))

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
            },
        'info': {
            'rid': 'b7c167f6c2da6d93531b9a7b314ad030b3a74803b4b7797edb905ba5a6a08',
            'time-ms': 2,
            'cpu-time-ms': 0
        },
        'facets': {
            'tags': {},
            'animals': {'constraints': [{'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 = b'<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 = b'<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.assertRaisesRegex(SearchServiceException, 'foo bar baz'):
            search.search(q='Test')

class FakeResponse(object):
    status_code = 405
    content = b''

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 = b'Nopenopenope'

# First, in the case of a non-JSON, non-403 error.
        with mpo(requests, '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."
        }).encode('utf-8')

with mpo(requests, '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))

Batal

⬅ unit 5 item

Nama	Tipe	Ukuran	Diubah	Aksi
🐍__init__.py	py	1 B	2025-06-20 07:13	✏️ 👁️
🐍test_connection.py	py	8.3 KB	2025-06-20 07:13	✏️ 👁️
🐍test_document.py	py	11.7 KB	2025-06-20 07:13	✏️ 👁️
🐍test_exceptions.py	py	1.4 KB	2025-06-20 07:13	✏️ 👁️
🐍test_search.py	py	13.4 KB	2025-06-20 07:13	✏️ 👁️