utils/test/scripts/shared_utils.py

   1 import urllib3
   2 import json
   3 http = urllib3.PoolManager()
   4
   5
   6 def delete_request(url, username, password, body=None):
   7     headers = urllib3.util.make_headers(basic_auth=':'.join([username, password]))
   8     http.request('DELETE', url, headers=headers, body=body)
   9
  10
  11 def publish_json(json_ojb, username, password, output_destination):
  12     json_dump = json.dumps(json_ojb)
  13     if output_destination == 'stdout':
  14         print json_dump
  15     else:
  16         headers = urllib3.util.make_headers(basic_auth=':'.join([username, password]))
  17         http.request('POST', output_destination, headers=headers, body=json_dump)
  18
  19
  20 def _get_nr_of_hits(elastic_json):
  21     return elastic_json['hits']['total']
  22
  23
  24 def get_elastic_data(elastic_url, username, password, body, field='_source'):
  25     # 1. get the number of results
  26     headers = urllib3.util.make_headers(basic_auth=':'.join([username, password]))
  27     elastic_json = json.loads(http.request('GET', elastic_url + '/_search?size=0', headers=headers, body=body).data)
  28     nr_of_hits = _get_nr_of_hits(elastic_json)
  29
  30     # 2. get all results
  31     elastic_json = json.loads(http.request('GET', elastic_url + '/_search?size={}'.format(nr_of_hits), headers=headers, body=body).data)
  32
  33     elastic_data = []
  34     for hit in elastic_json['hits']['hits']:
  35         elastic_data.append(hit[field])
  36     return elastic_data