Browse Source

add token checking and creating into result update script

Change-Id: I5dd8798bc778e9777658d5d0cff7bc156f9570f0
Megan Guiney 4 years ago
committed by Megan
  1. 191
  2. 4


@ -19,26 +19,59 @@
Test result update & verify script for the local refstack database.
import os
import argparse
import datetime
import json
import jwt
import os
import requests
import sys
from collections import namedtuple
import json
def generate_token(keyfile, _id):
# get private key in string format
with open(keyfile) as pemfile:
secret =
exp = datetime.timedelta(seconds=100500)
payload =\
"user_openid": _id,
"exp": + exp
token = jwt.encode(payload, secret, algorithm="RS256")
with open("./token", "w+") as tokenfile:
print("token stored in the current working directory.")
def testAuth(link, tokenfile):
auth_test_url = link.split("/v1")[0] + "/#/profile"
with open(tokenfile) as tokenfile:
token =
headers = {"Authorization": "Bearer " + token}
response = requests.get(auth_test_url, headers)
if response.status_code == 200:
return True, token
print("token auth failed. status code = %d" % (response.status_code))
print("auth failure detail: %s" % (response.text))
return False, None
def getData(entry):
"""Extract and reformat data from a product data csv"""
guidelines = ['2015.03', '2015.04', '2015.05', '2015.07', '2016.01',
'2016.08', '2017.01', '2017.09']
components = ['platform', 'compute', 'storage', 'object']
guidelines = ["2015.03", "2015.04", "2015.05", "2015.07", "2016.01",
"2016.08", "2017.01", "2017.09"]
# NOTE: Storage is an alias of Object
components = ["platform", "compute", "storage", "object"]
if len(entry) < 10:
return None, None, None
refstackLink = entry[9].strip()
guideline = entry[4].strip()
target = entry[5].lower().strip()
if refstackLink:
testId = refstackLink.split('/')[-1]
testId = refstackLink.split("/")[-1]
refstackLink = None
testId = None
@ -53,27 +86,28 @@ def getData(entry):
def linkChk(link, token):
"""Check existence of and access to api result link"""
print("now checking result: " + link)
print("checking result with a test ID of: %s" % (link.split("/")[-1]))
if not link:
return False
if " " in link:
return False
headers = {'Authorization': 'Bearer ' + token}
headers = {"Authorization": "Bearer " + token}
response = requests.get(link, headers)
if response.status_code == 200:
return json.loads(response.text)
elif response.status_code == 401 or response.status_code == 403:
print("Authentication Failed. link check response code: " +
print("Authentication Failed. link check response code: %d" %
return False
elif response.status_code == 400:
print("Malformed Request. link response code: " +
print("Malformed Request. link response code: %d" %
return False
print("Link check response_status_code=" +
print("Link check response_status_code = %d" %
print("Link check response detail: %s" % (response.text))
return False
except requests.exceptions as err:
@ -82,95 +116,96 @@ def linkChk(link, token):
def updateField(header, apiLink, raw_data):
"""Update a given metadata field"""
valid_keytype = ['shared', 'guideline', 'target']
valid_keytype = ["shared", "guideline", "target"]
keytype = raw_data.type
keyval = raw_data.value
if keytype not in valid_keytype or not keyval:
updresult = "%s keypair does not exist" % (keytype)
return updresult, False
link = apiLink.strip() + '/meta/' + keytype
link = apiLink.strip() + "/meta/" + keytype
response =, data=keyval, headers=header)
if response.status_code != 201:
print('update response status code=%d' %
print('update response text=' + response.text)
print("update response status code=%d" %
print("update response text=%s" % (response.text))
updresult = ("%s field update failed. reason: %s" %
(keytype, response.text.replace(',', ' ')))
(keytype, response.text.replace(",", " ")))
return updresult, False
updresult = "%s field update successful," % (keytype)
updresult = ("%s field update successful," % (keytype))
return updresult, True
def updateResult(apiLink, target, guideline, token, record):
"""Update metadata for result and verify if all updates are a success"""
MetadataField = namedtuple('MetadataField', ['type', 'value'])
MetadataField = namedtuple("MetadataField", ["type", "value"])
success = []
header = {'Authorization': 'Bearer ' + token}
with open(record, 'a') as r:
header = {"Authorization": "Bearer " + token}
with open(record, "a") as r:
r.write(str( + "," + apiLink + ",")
# update the shared field
data = MetadataField('shared', 'true')
data = MetadataField("shared", "true")
shared_result, shared_status = updateField(header, apiLink, data)
# update the target field
data = MetadataField('target', target)
data = MetadataField("target", target)
target_result, target_status = updateField(header, apiLink, data)
# update the guideline field
data = MetadataField('guideline', guideline + '.json')
data = MetadataField("guideline", guideline + ".json")
gl_result, gl_status = updateField(header, apiLink, data)
if not all(success):
r.write('unable to verify.\n')
r.write("unable to verify.\n")
return False
# if there were no update failures, we can verify the result
# this is the operation most likely to fail, so extra checks are
# in order
print('Test Result updated successfully. Attempting verification.')
print("Test Result updated successfully. Attempting verification.")
response = requests.put(apiLink,
json={'verification_status': 1},
json={"verification_status": 1},
except Exception as ex:
print('Exception raised while verifying test result: %s' %
print("Exception raised while verifying test result: %s" %
r.write('verification failed: %s\n' % (str(ex)))
r.write("verification failed: %s\n" % (str(ex)))
return False
updated = verification_chk(apiLink, header)
if response.status_code not in (200, 201):
print('verification failure status code=%d' %
print('verification failure detail=%s' %
r.write('verification unsuccessful: detail: %s\n' %
print("verification failure status code=%d" %
print("verification failure detail=%s" %
r.write("verification unsuccessful: detail: %s\n" %
return False
elif not updated:
print("verification_status field failed to update")
r.write('verification status update failed. detail: %s\n' %
r.write("verification status update failed. detail: %s\n" %
return False
print('Test result verified!\n')
r.write('Test result successfully verified\n')
print("Test result verified!\n")
r.write("Test result successfully verified\n")
return True
def verification_chk(link, header):
response = requests.get(link, header)
status = int(response.json()['verification_status'])
status = int(response.json()["verification_status"])
if status == 1:
return True
return False
except Exception as ex:
print('Exception raised while ensuring update of ' +
'verification status: ' + str(ex))
"Exception raised while ensuring verification status update: %s" %
return False
@ -178,51 +213,75 @@ def main():
linect = 0
parser = argparse.ArgumentParser(
"Update the internal RefStack db using a csv file")
parser.add_argument("--file", "-f", metavar='f', type=str, action="store",
# token handling options- we either need the path of a working token,
# or the data to generate a new token
token_flags = parser.add_mutually_exclusive_group(required=True)
token_flags.add_argument("--tokenfile", type=str, action="store",
help=("Absolute path to a json web token to "
"use to auth to the RefStack API"))
token_flags.add_argument("--generate", nargs=2,
metavar=("ssh-key", "openstack-id"),
help=("data needed to create a new auth token "
"ssh - key should be an absolute path to "
"a rsa ssh key. openstack - id indicates "
"an openstackid url to use for auth. "
"example: "
"<your id >"))
# non token-related flags
parser.add_argument("--file", "-f", metavar="f", type=str, action="store",
help="csv source for the data to use in updates")
"--endpoint", "-e", metavar='e',
"--endpoint", "-e", metavar="e",
type=str, action="store", required=True,
help="the base URL of the endpoint. ex:")
parser.add_argument("--token", "-t", metavar="t", type=str,
action="store", required=True, help="API auth token")
parser.add_argument("--record", "-r", metavar="r", type=str,
action="store", default="verification_results.csv",
help="name of file to output update & verification " +
" run record data into")
result = parser.parse_args()
infile = result.file
record = result.record
endpoint = result.endpoint
token = result.token
help=("name of file to output update & verification "
"run record data into"))
args = parser.parse_args()
infile = args.file
record = args.record
endpoint = args.endpoint
if args.generate:
keypath = args.generate[0]
_id = args.generate[1]
generate_token(keypath, _id)
tokenfile = "./token"
tokenfile = args.tokenfile
auth_success, token = testAuth(endpoint, tokenfile)
if not auth_success:
print(("Please enter either a valid token or an openstackid and the "
"absolute path to an rsa ssh key."))
with open(infile) as f:
for line in f:
linect = linect + 1
entry = line.split(",")
testId, guideline, target = getData(entry)
if None in (testId, guideline, target):
"entry found at line " + str(linect) +
" cannot be updated and verified: entry incomplete.\n")
print(("entry found at line %d cannot be updated and "
"verified: entry incomplete.\n") % (linect))
apiLink = os.path.join(endpoint, 'results', testId)
apiLink = os.path.join(endpoint, "results", testId)
testResult = linkChk(apiLink, token)
if testResult:
if testResult.get('verification_status'):
if testResult.get("verification_status"):
print("Result has been verified.\n")
"Result link is valid. Updating...")
"Result link is valid. Updating result with ID %s"
% (testId))
success = updateResult(apiLink, target, guideline,
token, record)
if not success:
print("update of the results with the ID " +
testId + " failed. please recheck your " +
"spreadsheet and try again\n")
print(("update of the results with the ID %s "
"failed. please recheck your spreadsheet "
"and try again" % (testId)))
print("the test result " + testId + " cannot be " +
"verified due to a link verification failure\n")
print(("the test result: % s cannot be updated or "
"verified due to a broken result link." % (testId)))


@ -7,7 +7,7 @@ successful in the usage of the script
The script can be run using the following formatting:
"./ --file /tmp/datasource.csv --endpoint --token <my-token>". In order to --tokenfile <token file path>". In order to
successfully update and verify results, you will need admin rights
for the refstack server in question. Instructions on how to get
these for your local install can be found at
@ -64,7 +64,7 @@ Because editing arbitrary test results requires administrative privileges,
an auth token must be used with the RefStack API. This token can be
generated by entering the command "jwt --key="$( cat <path to private key>
)" --alg=RS256 user_openid=<openstackid> exp=+100500". This generates a
json web token, which we must link using the "-t" or "--token" flag. Because
json web token, which we must link using the "--tokenfile" flag. Because
we cannot auth without this token, the token is a required flag.
The script will go through each line of the CSV, grabbing the refstack link,