KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _es_url method in localstack

Best Python code snippet using localstack_python

init_elasticsearch.py

Source:init_elasticsearch.py

1import requests2import json3from src.utils.config import config4# TODO use a util for creating index names5narrative_index_name = ''.join([6    config['index_prefix'],7    config['prefix_delimiter'],8    config['global']['ws_type_to_indexes']['KBaseNarrative.Narrative'],9])10index_names = [11    config['index_prefix'] + config['prefix_delimiter'] + 'index1',12    config['index_prefix'] + config['prefix_delimiter'] + 'index2',13]14_ES_URL = 'http://localhost:9200'15# Simple run once semaphore16_COMPLETED = False17#18# For the test docs, note the workspace must match the doc's idea of permissions.19# See data.py for the workspace definitions in which:20# 0 - public workspace, refdata21# 1 - public workspace, narrative22# 100 - private workspace, narrative23# 101 - private, inaccessible workspace, narrative24test_docs = [25    # Public doc, refdata26    {'name': 'public-doc1', 'access_group': '0', 'is_public': True, 'timestamp': 10},27    # Public doc, narrative28    {'name': 'public-doc2', 'access_group': '1', 'is_public': True, 'timestamp': 12},29    # Private but accessible doc30    {'name': 'private-doc1', 'is_public': False, 'access_group': '100', 'timestamp': 7},31    # Private but inaccessible doc32    {'name': 'private-doc2', 'is_public': False, 'access_group': '101', 'timestamp': 9},33]34narrative_docs = [35        {36            'name': 'narrative1',37            'narrative_title': 'narrative1',38            'is_public': True,39            'obj_id': 123,40            'access_group': '1',41            'timestamp': 1,42        },43]44def init_elasticsearch():45    """46    Initialize the indexes and documents on elasticsearch before running tests.47    """48    global _COMPLETED49    if _COMPLETED:50        return51    for index_name in index_names:52        create_index(index_name)53    create_index(narrative_index_name)54    for index_name in index_names:55        for doc in test_docs:56            create_doc(index_name, doc)57    for doc in narrative_docs:58        create_doc(narrative_index_name, doc)59    # create default_search alias for all fields.60    url = f"{_ES_URL}/_aliases"61    alias_name = config['index_prefix'] + config['prefix_delimiter'] + "default_search"62    body = {63        "actions": [64            {"add": {"indices": index_names, "alias": alias_name}}65        ]66    }67    resp = requests.post(url, data=json.dumps(body), headers={'Content-Type': 'application/json'})68    if not resp.ok:69        raise RuntimeError("Error creating aliases on ES:", resp.text)70    _COMPLETED = True71def create_index(index_name):72    # Check if exists73    resp = requests.head(_ES_URL + '/' + index_name)74    if resp.status_code == 200:75        return76    resp = requests.put(77        _ES_URL + '/' + index_name,78        data=json.dumps({79            'settings': {80                'index': {'number_of_shards': 2, 'number_of_replicas': 1}81            }82        }),83        headers={'Content-Type': 'application/json'},84    )85    if not resp.ok and resp.json()['error']['type'] != 'index_already_exists_exception':86        raise RuntimeError('Error creating index on ES:', resp.text)87def create_doc(index_name, data):88    # Wait for doc to sync89    url = '/'.join([  # type: ignore90        _ES_URL,91        index_name,92        '_doc',93        data['name'],94        '?refresh=wait_for'95    ])96    headers = {'Content-Type': 'application/json'}97    resp = requests.put(url, data=json.dumps(data), headers=headers)98    if not resp.ok:...

elasticsearch.py

Source:elasticsearch.py

1# -*- coding: utf-8 -*-2import json3class Elasticsearch(object):4    def __init__(self, es_ip, session=None):5        self._es_url = "http://%s:9200" % es_ip6        self._session = session7    @property8    def session(self):9        return self._session10    @session.setter11    def session(self, value):12        self._session = value13    def query_index_doc_count(self, index_name, condition_data):14        """15        Query the doc total count of the index16        :param index_name: the name of index17        :param condition_data: the querying condition18        :return: index doc count19        """20        uri = self._es_url + "/%s/_search?search_type=count" % index_name21        data = {"query":{"match":condition_data}}22        response = self.session.post(uri, json=data, verify=False)23        r = json.loads(response.content)24        return int(r["hits"]["total"])25    def query_cnt_ti_related_info(self, ioc_data=""):26        """27        Query index threat_intelligence_related_info doc count by condition28        :param ioc_data: the querying ioc value29        :return: index doc count30        """31        index = "threat_intelligence_related_info"32        query_data = {"ioc": ioc_data}33        return self.query_index_doc_count(index, query_data)34    def delete_index(self, index_name="*"):35        """36        Delete index, delete all by default 37        """38        uri = self._es_url + "/%s" % index_name39        response = self.session.delete(uri, verify=False)40        return json.loads(response.content)41    def query_cnt_ti_port(self, port=""):42        """43        Query index threat_intelligence_port doc count by condition44        :param port: the querying port value45        :return: index doc count46        """47        index = "threat_intelligence_port"48        query_data = {"portlist.port": port}49        return self.query_index_doc_count(index, query_data)50    def flush_es_event_index(self):51        uri = self._es_url + '/event*/_refresh'52        self.session.post(uri, verify=False)53    def get_valid_ioc(self, type):54        '''55        Get the valid ioc in 30 days.56        :return:57        '''58        url = self._es_url + '/threat_intelligence_credibility/_search?pretty'59        payload = {"query": {"match": {"type":type}}, "sort": {"timestamp": "desc"}, "size": 2}60        response = self._session.post(url, json=payload)61        response_content = json.loads(response.content)62        if response.status_code == 200:63            list_ioc = []64            for item in response_content['hits']['hits']:65                list_ioc.append(item['_source']['ioc'])66            return list_ioc67    def get_ioc_details_from_es(self, ioc):68        '''69        Get the detailed ioc info from elasticsearch.70        :param ioc: ioc71        :return: ioc info72        '''73        url = self._es_url + '/threat_intelligence/_search?pretty'74        payload = {"query": {"match": {"ioc": ioc}}}75        response = self._session.post(url, json=payload)76        response_content = json.loads(response.content)77        if response.status_code == 200 :78            print response_content['hits']['hits'][0]['_source']...

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.