utils/test/scripts/shared_utils.py

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