Unlock 30% off on Manual Testing Annual Plans this Holiday Season.

Copied to Clipboard!

How to use _search method in pytest-benchmark

Best Python code snippet using pytest-benchmark

test_search.py

Source:test_search.py

...129            response = self.client.post('/alert', json=alert, content_type='application/json')130            self.assertEqual(response.status_code, 201)131    def tearDown(self):132        db.destroy()133    def _search(self, q):134        response = self.client.get('/alerts?q={}'.format(q))135        self.assertEqual(response.status_code, 200)136        data = json.loads(response.data.decode('utf-8'))137        return data['total']138    def test_single_word_terms(self):139        self.assertEqual(self._search(q='foo'), 3)140        self.assertEqual(self._search(q='bar'), 4)141        self.assertEqual(self._search(q='foo bar'), 5)142        self.assertEqual(self._search(q='foo baz'), 4)143        self.assertEqual(self._search(q='foo quux'), 4)144    def test_phrase_terms(self):145        self.assertEqual(self._search(q='"foo bar"'), 2)146        self.assertEqual(self._search(q='"foo quux"'), 0)147        self.assertEqual(self._search(q='"fu bar baz"'), 1)148        self.assertEqual(self._search(q='"bar baz"'), 2)149    def test_field_names(self):150        self.assertEqual(self._search(q='status:ack'), 1)151        self.assertEqual(self._search(q='severity:major'), 2)152        self.assertEqual(self._search(q='group:(Network OR Performance)'), 4)153        self.assertEqual(self._search(q='group:(Network Performance)'), 4)154        self.assertEqual(self._search(q='text:"kernel panic"'), 2)155        self.assertEqual(self._search(q='_exists_:region'), 4)156        self.assertEqual(self._search(q='service:Shared'), 4)157        self.assertEqual(self._search(q='service:"Shared"'), 4)158        self.assertEqual(self._search(q='tags:aaa'), 3)159        self.assertEqual(self._search(q='tags:"aaa"'), 3)160        self.assertEqual(self._search(q='attributes.region:EMEA'), 2)161        self.assertEqual(self._search(q='_.region:EMEA'), 2)162        self.assertEqual(self._search(q='_.region:(EMEA LATAM)'), 3)163        self.assertEqual(self._search(q='_.region:(EMEA OR LATAM)'), 3)164        self.assertEqual(self._search(q='attributes.partition:7'), 4)165        self.assertEqual(self._search(q='_.partition:7'), 4)166        self.assertEqual(self._search(q='attributes.partition:"7"'), 1)167        self.assertEqual(self._search(q='_.partition:"7"'), 1)168    def test_wildcards(self):169        self.assertEqual(self._search(q='f*'), 4)170        self.assertEqual(self._search(q='f* ba?'), 5)171        self.assertEqual(self._search(q='te?t'), 2)172        self.assertEqual(self._search(q='test*'), 3)173    def test_regex(self):174        self.assertEqual(self._search(q='/[mb]oat/'), 2)175        self.assertEqual(self._search(q='value:/joh?n(ath[oa]n)/'), 2)176        self.assertEqual(self._search(q='resource:/net(wo?rk)?[0-9]/'), 5)177        self.assertEqual(self._search(q='/f(oo|u) ba.?/'), 3)178    def test_ranges(self):179        self.assertEqual(self._search(q='timeout:[100 TO 500]'), 5)180        self.assertEqual(self._search(q='origin:{alpha TO zulu}'), 3)181        self.assertEqual(self._search(q='timeout:{* TO 300}'), 2)182        self.assertEqual(self._search(q='timeout:[500 TO *]'), 1)183        self.assertEqual(self._search(q='timeout:>500'), 0)184        self.assertEqual(self._search(q='timeout:>=500'), 1)185        self.assertEqual(self._search(q='timeout:<500'), 4)186        self.assertEqual(self._search(q='timeout:<=500'), 5)187    def test_boolean_operators(self):188        self.assertEqual(self._search(q='"foo bar" foo'), 3)189        self.assertEqual(self._search(q='"foo bar" OR foo'), 3)190        self.assertEqual(self._search(q='"foo bar" || foo'), 3)191        self.assertEqual(self._search(q='"foo bar" AND "bar baz"'), 1)192        self.assertEqual(self._search(q='"foo bar" %26%26 "bar baz"'), 1)  # URL encode ampersands i.e &=%26193        self.assertEqual(self._search(q='"foo bar" NOT "bar baz"'), 1)194        self.assertEqual(self._search(q='"foo bar" !"bar baz"'), 1)195        self.assertEqual(self._search(q='"foo bar" AND NOT "bar baz"'), 1)196        self.assertEqual(self._search(q='NOT "foo bar"'), 3)197        self.assertEqual(self._search(q='foo resource:net01'), 3)198        self.assertEqual(self._search(q='foo OR resource:net01'), 3)199        self.assertEqual(self._search(q='foo AND resource:net01'), 1)200        self.assertEqual(self._search(q='foo !resource:net01'), 2)201        self.assertEqual(self._search(q='foo NOT resource:net01'), 2)202        self.assertEqual(self._search(q='foo AND !resource:net01'), 2)203        self.assertEqual(self._search(q='foo AND NOT resource:net01'), 2)204    def test_grouping(self):205        self.assertEqual(self._search(q='(foo OR bar) AND baz'), 2)206        self.assertEqual(self._search(q='status:(open OR ack)'), 2)...

query.py

Source:query.py

1import json2from elasticsearch_dsl import Search3from elasticsearch_dsl.query import Fuzzy, Match, MultiMatch, Wildcard, Bool4class MemberSearchClient:5    extra = {6        'min_score': 0,7        'timeout': '30s',8    }9    def __init__(self, app):10        self.app = app11        self._search = Search(12            using=app.es, index=app.config['ELASTICSEARCH_MEMBERS_INDEX'])13        self._search = self._search.extra(**self.extra)14    def execute(self):15        return self._search.execute()16    def debug(self):17        return json.dumps(self._search.to_dict(), indent=4, sort_keys=False)18    def set_from(self, from_value):19        if str(from_value):20            self._search._extra['from'] = from_value21    def set_size(self, size):22        if str(size):23            self._search._extra['size'] = size24    def last_name(self, term):25        self._search._extra['min_score'] += 226        self._search = self._search.query(Bool(27            should=[Fuzzy(last_name=term)],28            must=[Match(last_name__phonetic={29                        'query': term, 'boost': 10, 'fuzziness': 1})]30        ))31    def ssn(self, term):32        self._search._extra['min_score'] += 233        self._search = self._search.query(Bool(should=[34            Wildcard(demographics__ssn='*%s*' % term),35            Fuzzy(demographics__ssn=term)36        ]))37    def should_match(self, **kwargs):38        return self._search.query(Bool(should=[Match(**kwargs)]))39    def first_name(self, term):40        self._search._extra['min_score'] += 641        self._search = self.should_match(first_name__phonetic=term)42    def employer_id(self, term):43        self._search._extra['min_score'] += 144        self._search = self.should_match(employers__employer_id=term)45    def employer_name(self, term):46        self._search._extra['min_score'] += 147        self._search = self.should_match(employers__employer_name=term)48    def classification(self, term):49        self._search._extra['min_score'] += 150        self._search = self.should_match(employers__classification=term)51    def state_province(self, term):52        self._search._extra['min_score'] += 253        self._search = self.should_match(addresses__state_province=term)54class HistorySearchClient:55    extra = {56        'timeout': '30s',57    }58    def __init__(self, app):59        self.app = app60        self._search = Search(61            using=app.es, index=app.config['ELASTICSEARCH_HISTORY_INDEX'])62        self._search = self._search.extra(**self.extra)63    def execute(self):64        return self._search.execute()65    def debug(self):66        return json.dumps(self._search.to_dict(), indent=4, sort_keys=False)67    def set_from(self, from_value):68        if str(from_value):69            self._search._extra['from'] = from_value70    def set_size(self, size):71        if str(size):72            self._search._extra['size'] = size73    def ssn(self, term):74        self._search = self._search.query(Bool(must=[75            Match(ssn_decrypted='true'),76            Match(entity='UH_DEMO'),77            Match(property='SSN'),78            MultiMatch(query=term, fields=['old_value', 'new_value'])79        ]))80    def should_match(self, **kwargs):...

base.py

Source:base.py

1from elasticsearch import TransportError, ConnectionTimeout2class BaseSearchManager(object):3    """ Wraps elastic search result object4        This class wraps the elasticsearch result object5        to add extra functionality"""6    def __init__(self, doc_class, search, page_size=100):7        self._doc_class = doc_class8        self._search = search9        self._page_size = page_size10    def count(self):11        return self._search.count()12    def source(self, **kwargs):13        self._search = self._search.source(**kwargs)14        return self._search15    def filter(self, *args, **kwargs):16        self._search = self._search.filter(*args, **kwargs)17        return self._search18    def query(self, *args, **kwargs):19        self._search = self._search.query(*args, **kwargs)20        return self._search21    def sort(self, *keys):22        self._search = self._search.sort(*keys)23        return self._search24    def extra(self, *args, **kwargs):25        self._search = self._search.extra(*args, **kwargs)26        return self._search27    def highlight(self, *args, **kwargs):28        self._search = self._search.highlight(*args, **kwargs)29        return self._search30    def execute(self):31        try:32            res = self._search.execute()33        except (TransportError, ConnectionTimeout) as err:34            if getattr(err, 'status_code', 500) == 404:35                raise36            res = self._search.execute()37        return res38    def all(self):39        res = self._search.execute()40        if res.success() and res.hits.total.value:41            res_offset = 042            page_size = len(res)43            res_limit = page_size44            while res_limit <= res.hits.total.value:45                for doc in self._search[res_offset:res_limit]:46                    yield self._doc_class(doc)47                res_limit += page_size48                res_offset += page_size49            res_limit = res.hits.total.value - \50                ((res.hits.total.value / page_size) * page_size)51            if res_limit > 0:52                res_offset -= page_size53                res_limit += res_offset54                for doc in self._search[res_offset:res_limit]:55                    yield self._doc_class(doc)56    def results(self, offset):57        res = self._search.execute()58        limit = offset + self._page_size59        if res.hits.total.value < limit:60            limit = res.hits.total.value61        if offset > limit:62            offset = 063            limit = 064        for doc in self._search[offset:limit]:65            yield self._doc_class(doc)66    def __iter__(self):67        for doc in self._search.execute():68            yield self._doc_class(doc)69    def scan(self):70        self._search.execute()71        for doc in self._search.scan():72            yield self._doc_class(doc)73    def __getitem__(self, index):74        return self._search.__getitem__(index)75    def __getattr__(self, name):76        search = self._search77        val = getattr(search, name, None)78        if val:79            return val80        else:81            raise AttributeError(...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.