Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use test_search method in localstack

Best Python code snippet using localstack_python

helpers.py

Source:helpers.py

1try:2    import unittest2 as unittest3except:4    import unittest5from server.search import helpers6es_client = helpers.es_client7helpers.search_client = es_client.test_search8helpers.ac_client = es_client.test_autocomplete9helpers.history_client = es_client.test_history10helpers.search_index = 'test_search'11helpers.history_index = 'test_history'12from server import schemas13class TestResetIndex(unittest.TestCase):14    def test_expect_creates_index(self):15        cut = helpers.reset_index16        cut('test_search')17        actual = es_client._cat.indices.get().text18        self.assertIn('test_search', actual)19    def test_expect_adds_appropriate_mapping(self): 20        cut = helpers.reset_index21        cut('test_search')22        actual = es_client.test_search._mapping.get().json()['test_search']['mappings'].keys()23        actual.sort()24        schema_keys = schemas.test_search['mappings'].keys()25        schema_keys.sort()26        self.assertEqual(actual, schema_keys)27    def test_expect_destroys_existing_index(self):28        cut = helpers.reset_index29        cut('test_search')30        es_client.test_search.test_type.test_item.put(data={'hello': 'world'})31        cut('test_search')32        actual = es_client.test_search.test_type.test_item.get()33        self.assertGreater(actual.status_code, 400)34class TestDeleteIndexSubset(unittest.TestCase):35    def setUp(self):36        helpers.reset_index('test_search')37    def test_expect_delete_all_repo_docs(self):38        es_client.test_search.wiki.test1.put(data={'path':'/a/b', 'title': 'test1', 'source': 'github'})39        cut = helpers.delete_index_subset40        cut('GH', 'wiki', '/a/b')41        actual = es_client.test_search.wiki.test1.get()42        self.assertGreater(actual.status_code, 400)43    def test_expect_not_delete_docs_in_other_repo(self):44        es_client.test_search.wiki.test1.put(data={'path':'/a/c', 'title': 'test1', 'source': 'github'})45        cut = helpers.delete_index_subset46        cut('GH', 'wiki', '/a/b')47        actual = es_client.test_search.wiki.test1.get()48        self.assertEqual(actual.status_code, 200)49    def test_expect_not_delete_docs_of_other_type(self):50        es_client.test_search.readme.test1.put(data={'path':'/a/b', 'title': 'test1', 'source': 'github'})51        cut = helpers.delete_index_subset52        cut('GH', 'wiki', '/a/b')53        actual = es_client.test_search.readme.test1.get()54        self.assertEqual(actual.status_code, 200)55    def test_expect_not_delete_docs_of_other_gh_type(self):56        es_client.test_search.wiki.test1.put(data={'path':'/a/b', 'title': 'test1', 'source': 'enterprise'})57        cut = helpers.delete_index_subset58        cut('GH', 'wiki', '/a/b')59        actual = es_client.test_search.wiki.test1.get()60        self.assertEqual(actual.status_code, 200)61class TestGetIndexedVersion(unittest.TestCase):62    def setUp(self):63        helpers.reset_index('test_history')64    def test_expect_returns_None_if_no_indexed_version(self):65        repo_name = '/a/b'66        cut = helpers.get_indexed_version67        actual = cut('GH', repo_name, 'wiki')68        self.assertIsNone(actual)69    def test_expect_returns_indexed_version_if_exists(self):70        repo_name = '/a/b'71        helpers.save_indexed_version('GH', repo_name, 'wiki', 'xxx')72        cut = helpers.get_indexed_version73        actual = cut('GH', repo_name, 'wiki')74        self.assertEqual(actual, 'xxx')75class TestSaveIndexedVersion(unittest.TestCase):76    def setUp(self):77        helpers.reset_index('test_history')78    def test_expect_create_doc_with_new_verison_if_not_exist(self):79        repo_name = '/a/b'80        doc_id = ('GH/' + repo_name).replace('/', '%2F')81        cut = helpers.save_indexed_version82        cut('GH', repo_name, 'wiki', 'xxx')83        actual = helpers.history_client.wiki._(doc_id).get().json()84        self.assertEqual(actual['_source'], {u'version': u'xxx'})85    def test_expect_update_doc_with_new_verison_if_already_exist(self):86        repo_name = '/a/b'87        doc_id = ('GH/' + repo_name).replace('/', '%2F')88        cut = helpers.save_indexed_version89        cut('GH', repo_name, 'wiki', 'xxx')90        cut('GH', repo_name, 'wiki', 'yyy')91        actual = helpers.history_client.wiki._(doc_id).get().json()92        self.assertEqual(actual['_source'], {u'version': u'yyy'})93class TestWriteBulkData(unittest.TestCase):94    def setUp(self):95        helpers.reset_index('test_search')96    bulk_data = [97      {98        "index": {99          "_type": "wiki",100          "_id": "a%2Fb",101          "_index": "test_search"102        }103      },104      {105        "url": "http://example.com/a/b",106        "content": "The quick brown fox",107        "path": "/a/b",108        "title": "Alpha Bravo"109      },110      {111        "index": {112          "_type": "wiki",113          "_id": "c%2Fd",114          "_index": "test_search"115        }116      },117      {118        "url": "http://example.com/c/d",119        "content": "jumped over the lazy dog",120        "path": "/c/d",121        "title": "Charly Delta"122      },123    ]124    def test_expect_writes_bulk_rows(self):125        cut = helpers.write_bulk_data126        cut(self.bulk_data)127        actual = es_client.test_search._count.get().json()...

search_quality.py

Source:search_quality.py

1from src.queries.search_es import search_es_full2def test_search(args):3    print("\n\n==========", args)4    found = search_es_full(args)5    search_type = args.get("kind", "all")6    def print_entity(title, entities):7        if not entities:8            return9        print(f"\n[ {title} ]")10        for entity in entities:11            print(12                "   ",13                [14                    entity["user"]["handle"],15                    entity["user"]["name"],16                    entity.get("title") or entity.get("playlist_name"),17                    f"{entity['repost_count']} reposts",18                    f"{entity['user']['follower_count']} followers",19                    f"{entity.get('_score')} score",20                ],21            )22    def print_users(title, users):23        if not users:24            return25        print(f"\n[ {title} ]")26        for user in users:27            print(28                "   ",29                [30                    user["handle"],31                    user["name"],32                    f"{user.get('follower_count')} followers",33                    f"{user.get('is_verified')} verified",34                    f"{user.get('_score')} score",35                ],36            )37    if search_type == "tracks" or search_type == "all":38        print_entity("tracks", found["tracks"])39        print_entity("saved tracks", found["saved_tracks"])40    if search_type == "users" or search_type == "all":41        print_users("users", found["users"])42        print_users("followed_users", found["followed_users"])43    if search_type == "playlists" or search_type == "all":44        print_entity("playlists", found["playlists"])45        print_entity("saved_playlists", found["saved_playlists"])46    if search_type == "albums" or search_type == "all":47        print_entity("albums", found["albums"])48        print_entity("saved_albums", found["saved_albums"])49test_search({"query": "space fm lido", "limit": 3, "kind": "tracks"})50test_search({"query": "issac solo", "limit": 3, "kind": "users"})  # misspell51test_search(52    {53        "query": "the cycle of change",54        "limit": 4,55        "is_auto_complete": True,56    }57)58test_search(59    {60        "query": "isaac pho",61        "limit": 4,62        "is_auto_complete": True,63    }64)65test_search(66    {67        "query": "RAC wat",68        "limit": 4,69        "current_user_id": 1,70        "is_auto_complete": True,71    }72)73test_search(74    {75        "query": "RAC water",76        "limit": 4,77        "current_user_id": 1,78        "is_auto_complete": False,79    }80)81test_search(82    {83        "query": "deadmau",84        "limit": 4,85        "current_user_id": 1,86        "is_auto_complete": False,87    }88)89# should have disclosure at the top90test_search(91    {92        "query": "waterfal",93        "limit": 10,94        "current_user_id": 1,95        "is_auto_complete": True,96    }97)98test_search(99    {100        "query": "closer 2 u ray",101        "limit": 4,102        "current_user_id": 1,103        "is_auto_complete": True,104    }105)106test_search(107    {108        "query": "raymont",109        "limit": 4,110        "current_user_id": 1,111        "is_auto_complete": True,112    }113)114test_search(115    {116        "query": "low",117        "limit": 4,118        "current_user_id": 14,119        "is_auto_complete": True,120    }121)122test_search(123    {124        "query": "stereosteve guitar",125        "limit": 4,126        "current_user_id": 1,127        "is_auto_complete": True,128    }129)130test_search(131    {132        "query": "skrillex",133        "limit": 4,134        "current_user_id": 1,135        "is_auto_complete": True,136    }137)138test_search(139    {140        "query": "camo",141        "limit": 4,142        "is_auto_complete": True,143    }144)145test_search(146    {147        "query": "zouai",148        "limit": 4,149        "is_auto_complete": True,150    }151)...

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.