utils/test/dashboard/dashboard/common/elastic_access.py

   1 import json
   2 import urlparse
   3
   4 import urllib3
   5
   6 http = urllib3.PoolManager()
   7
   8
   9 def _request(method, url, creds=None, body=None):
  10     headers = urllib3.make_headers(basic_auth=creds)
  11     return http.request(method, url, headers=headers, body=body)
  12
  13
  14 def _post(url, creds=None, body=None):
  15     return _request('POST', url, creds=creds, body=body)
  16
  17
  18 def _get(url, creds=None, body=None):
  19     return json.loads(_request('GET', url, creds=creds, body=body).data)
  20
  21
  22 def delete_docs(url, creds=None, body=None):
  23     return _request('DELETE', url, creds=creds, body=body)
  24
  25
  26 def publish_docs(url, creds=None, body=None):
  27     result = _post(url, creds=creds, body=(json.dumps(body)))
  28     return result.status, result.data
  29
  30
  31 def _get_docs_nr(url, creds=None, body=None):
  32     res_data = _get('{}/_search?size=0'.format(url), creds=creds, body=body)
  33     print(type(res_data), res_data)
  34     return res_data['hits']['total']
  35
  36
  37 def get_docs(url, creds=None, body=None, field='_source'):
  38
  39     docs_nr = _get_docs_nr(url, creds=creds, body=body)
  40     res_data = _get('{}/_search?size={}'.format(url, docs_nr),
  41                     creds=creds, body=body)
  42
  43     docs = []
  44     for hit in res_data['hits']['hits']:
  45         docs.append(hit[field])
  46     return docs
  47
  48
  49 def publish_kibana(url, creds, type, id, body):
  50     url = urlparse.urljoin(url, '/.kibana/{}/{}'.format(type, id))
  51     publish_docs(url, creds, body)