#!/usr/bin/python
# -*- coding: utf-8 -*-
#
# Copyright (C) 2013-2019:
# This file is part of Shinken Enterprise, all rights reserved.
import time
import sys
from optparse import OptionParser
import pycurl
from StringIO import StringIO
from urllib import urlencode
import base64
def call_url(url, debug):
buffer = StringIO()
my_curl = pycurl.Curl()
my_curl.exception = None
my_curl.setopt(my_curl.URL, url)
my_curl.setopt(my_curl.WRITEFUNCTION, buffer.write)
try:
my_curl.perform()
except pycurl.error as pycurl_error:
if pycurl_error.args[0] == pycurl.E_COULDNT_CONNECT and my_curl.exception:
print 'The curl failed on url "%s "with the error : %s' % (url, my_curl.exception)
sys.exit(2)
else:
print 'The curl failed on url "%s "with the error : %s' % (url, pycurl_error)
sys.exit(2)
http_code = my_curl.getinfo(my_curl.RESPONSE_CODE)
my_curl.close()
data = buffer.getvalue().strip()
if http_code != 200:
print 'The call to "%s" return an http error %s' % (url, http_code)
if debug:
print 'The full result is :\n%s' % data
sys.exit(2)
return data
def call_api(opts, route, data={}):
host = opts.host
port = opts.port
password = opts.password
if password:
url = "http://%s:%s/%s?password=%s&%s" % (host, port, route, base64.b64encode(password), urlencode(data))
else:
url = "http://%s:%s/%s?%s" % (host, port, route, urlencode(data))
if opts.debug:
print '[DEBUG] Will call url [%s]' % url
return call_url(url, opts.debug)
def get_parsed_options():
parser = OptionParser(description='trusted source')
parser.add_option('-u', '--user', dest="user", default='admin', help='Set the user')
parser.add_option('-p', '--password', dest="password", default='admin', help='Set the password')
parser.add_option('-H', '--host', dest="host", default='localhost', help='Set the host')
parser.add_option('-P', '--port', dest="port", default='7765', help='Set the port')
parser.add_option('-s', '--source', dest="source", help='Set the source')
parser.add_option('-t', '--type', dest="item_type", default='hosts', help='Set the item_type')
parser.add_option('-d', '--debug', action='store_true', dest='debug', help='Set debug mode')
opts, args = parser.parse_args()
if not opts.source:
parser.error('Please set the source.')
return opts
def main():
opts = get_parsed_options()
data = {
'login' : opts.user,
'source_name': opts.source,
}
print "\nCall the import for source [%s]" % opts.source
ret = call_api(opts, 'trusted-source/v1/force-source-import', data)
if not ret == '"OK"':
sys.exit(-2)
print "OK !\n"
if opts.debug:
print "[DEBUG] Result is : %s" % ret
print "\nMerge in progress "
while call_api(opts, 'trusted-source/v1/get-api-source-controller-importing') == 'true':
time.sleep(1)
print '.'
print "OK !\n"
print"\nApply diff and import to staging"
del data['source_name']
data['filter'] = "sources:%s" % opts.source
ret = call_api(opts, 'trusted-source/v1/force-trusted-source-behaviour', data)
print "OK !\n"
if opts.debug:
print "[DEBUG] Result is : %s" % ret
print"\nPut in production"
data['item_type'] = opts.item_type
ret = call_api(opts, 'trusted-source/v1/put-in-production', data)
print "OK !\n"
if opts.debug:
print "[DEBUG] Result is : %s" % ret
if __name__ == '__main__':
main()
|