KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run pytest-benchmark automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
