234 lines
7.2 KiB
Python
234 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
|
#
|
|
# Copyright (C) 2017 Ken Fallon ken@fallon.ie
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
# UTF-8 Test >> ÇirçösÚáóíéőöÓÁśł <<
|
|
# 213.46.252.136 gateway
|
|
|
|
import sys, getopt, json, os, logging, sys, requests, datetime, re, dateutil.parser, base64, xmltodict, random, psycopg2, time, automationhat
|
|
|
|
from pprint import pprint
|
|
#import traceback
|
|
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
#logging.disable(logging.DEBUG)
|
|
settings = dict()
|
|
|
|
def usage(exitcode, message = None):
|
|
'''
|
|
prints usage and exits
|
|
'''
|
|
if message is not None:
|
|
logging.error( message )
|
|
print( os.path.basename(sys.argv[0]) + ' [--help -l <lab> -c <config> ]')
|
|
sys.exit(exitcode)
|
|
|
|
def get_lab_settings(settings):
|
|
'''
|
|
Get configuration for the lab, usually from http://172.30.218.244/autotest/settings.json
|
|
'''
|
|
if os.path.isfile( settings[ 'config_url' ] ):
|
|
with open( settings[ 'config_url' ] ) as json_file:
|
|
info_dict = json.load(json_file)
|
|
return info_dict
|
|
else:
|
|
session = requests.session()
|
|
resp = session.get(url=settings[ 'config_url' ], headers={'Accept-Encoding': 'gzip'})
|
|
if resp.status_code == 200:
|
|
info_dict = json.loads(resp.text)
|
|
return info_dict
|
|
return None
|
|
|
|
def argumentTest():
|
|
'''
|
|
Gets the required paramaters that the program needs in order to run. These can either be provided on the command line, or as environemental variables, or both.
|
|
|
|
export LAB="lab4b"
|
|
export config_url="http://172.30.218.244/autotest/settings.json"
|
|
export service="purcha#!/usr/bin/env python
|
|
|
|
import automationhat
|
|
import time
|
|
import os
|
|
import sys
|
|
|
|
counter = -1
|
|
lightPattern = [
|
|
[1,0,0],
|
|
[0,1,0],
|
|
[0,0,1],
|
|
[0,1,1],
|
|
[0,1,0],
|
|
[1,1,0],
|
|
[1,0,0],
|
|
[1,1,0],
|
|
[0,1,0],
|
|
[0,1,1],
|
|
[0,0,1],
|
|
[0,0,0],
|
|
[1,0,1],
|
|
[0,1,0],
|
|
[0,0,0],
|
|
]
|
|
|
|
def increment():
|
|
global counter
|
|
counter+=1
|
|
if counter==len(lightPattern):
|
|
counter = 0
|
|
|
|
def doCurrentCycle():
|
|
currentCycle = lightPattern[counter]
|
|
print currentCycle
|
|
if (currentCycle[0]==1) :
|
|
automationhat.output.one.on()
|
|
else:
|
|
automationhat.output.one.off()
|
|
|
|
if (currentCycle[1] == 1):
|
|
automationhat.output.two.on()
|
|
else:
|
|
automationhat.output.two.off()
|
|
|
|
if (currentCycle[2] == 1):
|
|
automationhat.output.three.on()
|
|
else:
|
|
automationhat.output.three.off()
|
|
|
|
time.sleep(2)
|
|
|
|
|
|
|
|
while True:
|
|
increment()
|
|
doCurrentCycle()se-service"
|
|
export customers="customers.json"
|
|
|
|
'''
|
|
lab = str()
|
|
config_url = str()
|
|
service = str()
|
|
settings = dict()
|
|
|
|
for evariable in [ 'LAB', 'config_url' ]:
|
|
if os.getenv( evariable ) is not None:
|
|
settings[ evariable ] = os.environ[ evariable ]
|
|
logging.debug( "Setting \"" + evariable + "\" set to \"" + settings[ evariable ] + "\" from an enviroment variable.")
|
|
|
|
try:
|
|
options, remainder = getopt.getopt(sys.argv[1:], 'hl:c:', [ 'help', 'lab=', 'config=' ])
|
|
|
|
except getopt.GetoptError:
|
|
usage(1, "Unrecognized option was found or missing argument." )
|
|
|
|
for opt, arg in options:
|
|
if opt in ('-h', '--help'):
|
|
usage(2)
|
|
elif opt in ('-l', '--lab'):
|
|
logging.debug( "Setting \"LAB\" set to \"" + arg + "\" from the \"" + opt + "\" command line argument.")
|
|
settings[ 'LAB' ] = arg
|
|
elif opt in ('-c', '--config'):
|
|
logging.debug( "Setting \"config_url\" set to \"" + arg + "\" from the \"" + opt + "\" command line argument.")
|
|
settings[ 'config_url' ] = arg
|
|
|
|
for param in [ 'LAB', 'config_url' ]:
|
|
if not param in settings:
|
|
usage( 3, "Cant find value for \"" + param +"\"")
|
|
|
|
try:
|
|
settings[ 'config' ] = get_lab_settings( settings )[ settings[ 'LAB' ] ]
|
|
if 'jenkins' in settings.keys():
|
|
logging.debug('Updating TraxIS Customer file to include the Purchase Service status.')
|
|
|
|
except Exception as e:
|
|
print('Error: Could not get Lab Configuration for "%s".' % settings[ 'LAB' ] )
|
|
traceback.print_exc()
|
|
exit(1)
|
|
|
|
return settings
|
|
|
|
|
|
#def get_server_instance():
|
|
#jenkins_url = 'http://172.22.137.160:8080'
|
|
#server = Jenkins(jenkins_url, username='autotest', password='autotest')
|
|
#return server
|
|
|
|
def get_server_instance(settings):
|
|
jenkins_url = 'http://%s:%s' % ( settings[ 'config' ][ 'jenkins' ][ 'host' ] , settings[ 'config' ][ 'jenkins' ][ 'port' ] )
|
|
server = Jenkins(jenkins_url, username = settings[ 'config' ][ 'jenkins' ][ 'user' ], password = settings[ 'config' ][ 'jenkins' ][ 'password' ])
|
|
return server
|
|
|
|
def alloff():
|
|
automationhat.relay.one.off()
|
|
automationhat.output.one.off()
|
|
automationhat.output.two.off()
|
|
automationhat.output.three.off()
|
|
|
|
# Set according to status
|
|
def setstatus(status):
|
|
if status != "":
|
|
alloff()
|
|
if status == "SUCCESS":
|
|
automationhat.output.three.on()
|
|
print( "SUCCESS" )
|
|
|
|
if status == "UNSTABLE":
|
|
automationhat.output.two.on()
|
|
print( "UNSTABLE" )
|
|
|
|
if status == "FAILURE":
|
|
automationhat.output.one.on()
|
|
print( "FAILURE" )
|
|
return
|
|
|
|
def get_job_details( settings ):
|
|
# Refer Example #1 for definition of function 'get_server_instance'
|
|
server = get_server_instance( settings )
|
|
settings[ 'teststatus' ] = str()
|
|
for job in settings[ 'config' ][ 'jenkins' ][ 'jobs' ]:
|
|
logging.debug( "Checking job %s" % job )
|
|
job_instance = server.get_job( job )
|
|
this_status = job_instance.get_last_build().get_status()
|
|
if not ( settings[ 'teststatus' ] == "FAILURE" or (settings[ 'teststatus' ] == "UNSTABLE" and this_status == "SUCCESS") ):
|
|
settings[ 'teststatus' ] = this_status
|
|
return settings
|
|
|
|
if __name__ == "__main__":
|
|
logging.debug( "start" )
|
|
try:
|
|
while True:
|
|
alloff()
|
|
automationhat.relay.one.on()
|
|
settings = argumentTest()
|
|
#pprint( settings[ 'config' ][ 'jenkins' ] )
|
|
#print( "Hello" )
|
|
#pprint ( get_server_instance( settings ).version )
|
|
settings = get_job_details( settings )
|
|
status = settings[ 'teststatus' ]
|
|
setstatus(status)
|
|
#print( get_server_instance().version )
|
|
time.sleep(300)
|
|
|
|
except Exception as e:
|
|
print("Error: " + str(e))
|
|
traceback.print_exc()
|
|
exit(1)
|
|
logging.debug('done')
|
|
|
|
'''
|
|
export LAB="lab4b"
|
|
export config_url="http://172.30.218.244/autotest/settings.json"
|
|
'''
|