#!/usr/bin/python # # This program and the accompanying materials # are made available under the terms of the Apache License, Version 2.0 # which accompanies this distribution, and is available at # # http://www.apache.org/licenses/LICENSE-2.0 # """ Module to generate Functest reporting for gitlab pages """ import argparse import datetime import logging import os import jinja2 import json import requests from prettytable import PrettyTable # Logger logging.basicConfig() LOGGER = logging.getLogger("Xtesting-ONAP-Status") LOGGER.setLevel("INFO") # LOGGER.setLevel("DEBUG") #PROXY = {} PROXY = {'http':'http://87.254.212.120:8080', 'https':'http://87.254.212.120:8080'} # PROXY = {'http': 'socks5h://127.0.0.1:8080', # 'https': 'socks5h://127.0.0.1:8080'} # Initialization URL_PRIVATE_BASE = "http://onap.api.testresults.opnfv.fr/api/v1/results" URL_BASE = "http://testresults.opnfv.org/onap/api/v1/results" URL_BASE_PODS = "http://testresults.opnfv.org/onap/api/v1/pods" REPORTINGDATE = datetime.datetime.now().strftime("%Y-%m-%d %H:%M") # init just connection_check to get the list of scenarios # as all the scenarios run connection_check # the following tests are the default daily tests INFRA_HEALTHCHECK = {'name': 'infrastructure-healthcheck', 'tests': {'onap-k8s', 'onap-helm', 'nodeport_check_certs', 'onap-k8s-teardown'}} # 'onap-k8s-teardown','internal_check_certs'}} HEALTHCHECK = {'name': 'healthcheck', 'tests': {'core', 'full', 'healthdist', 'postinstall', 'hv-ves', 'ves-collector', 'basic_cds', 'basic_onboard', 'cps-healthcheck'}} #SMOKE_USECASES = {'name': 'smoke usecases', # 'tests': {'basic_vm', 'basic_network', 'basic_cnf', 'cmpv2', # 'pnf-registrate', '5gbulkpm', 'basic_clamp', # 'basic_vm_macro', 'pnf_macro'}} SMOKE_USECASES = {'name': 'smoke usecases', 'tests': {'basic_vm', 'basic_network', 'basic_cnf', 'pnf-registrate', 'basic_clamp', 'basic_vm_macro', 'pnf_macro'}} SECURITY_USECASES = {'name': 'security', 'tests': {'root_pods', 'unlimitted_pods', 'nonssl_endpoints', 'jdpw_ports', 'kube_hunter'}} TIERS = [INFRA_HEALTHCHECK, HEALTHCHECK, SMOKE_USECASES, SECURITY_USECASES] TRENDS = [INFRA_HEALTHCHECK, HEALTHCHECK, SMOKE_USECASES, SECURITY_USECASES] # list of tests with dedicated reporting page to be referenced RESULT_URLS_LEGACY = { 'core': './xtesting-healthcheck/core/report.html', 'small': './xtesting-healthcheck/small/report.html', 'medium': './xtesting-healthcheck/medium/report.html', 'full': './xtesting-healthcheck/full/report.html', 'postinstall': './xtesting-healthcheck/postinstall/report.html', 'healthdist': './xtesting-healthcheck/healthdist/report.html', 'onap-k8s': './infrastructure-healthcheck/k8s/kubernetes-status/index.html', 'onap-k8s-teardown': './infrastructure-healthcheck/k8s-teardown/kubernetes-status/index.html', 'onap-helm': './infrastructure-healthcheck/k8s/onap-helm/helm.html', 'nodeport_check_certs': './infrastructure-healthcheck/k8s/nodeport_check_certs/certificates.html', 'internal_check_certs': './infrastructure-healthcheck/internal_check_certs/internal_check_certs/certificates.html', 'basic_vm': './smoke-usecases/basic_vm/reporting.html', 'basic_vm_macro': './smoke-usecases/basic_vm_macro/reporting.html', 'basic_network': './smoke-usecases/basic_network/reporting.html', 'basic_cnf': './smoke-usecases/basic_cnf/reporting.html', 'basic_cds': './smoke-usecases/basic_cds/reporting.html', 'basic_onboard': './smoke-usecases/basic_onboard/reporting.html', 'basic_clamp': './smoke-usecases/basic_clamp/reporting.html', 'pnf_macro': './smoke-usecases/pnf_macro/reporting.html', 'pnf-registrate': './xtesting-smoke-usecases-robot/pnf-registrate/report.html', '5gbulkpm': './xtesting-smoke-usecases-robot/5gbulkpm/report.html', 'hv-ves': './xtesting-smoke-usecases-robot/hv-ves/report.html', 'cmpv2': './xtesting-smoke-usecases-robot/cmpv2/report.html', 'dcaemod': './xtesting-smoke-usecases-robot/dcaemod/report.html', 'ves-collector': './xtesting-smoke-usecases-robot/ves-collector/report.html', 'root_pods': './security/root_pods/root_pods.log', 'unlimitted_pods': './security/unlimitted_pods/xtesting.log', 'cis_kubernetes': './security/cis_kubernetes/cis_kubernetes/cis_kubernetes.log', 'nonssl_endpoints': './security/nonssl_endpoints/xtesting.log', 'jdpw_ports': './security/jdpw_ports/xtesting.log', 'kube_hunter': './security/kube_hunter/xtesting.log', 'versions': './security/versions/versions.html', 'cps-healthcheck': './xtesting-healthcheck/cps-healthcheck/report.html', } # list of tests with dedicated reporting page to be referenced RESULT_URLS_S3 = { 'core': './core/report.html', 'full': './full/report.html', 'postinstall': './postinstall/report.html', 'healthdist': './healthdist/report.html', 'onap-k8s': './k8s/kubernetes-status/index.html', 'onap-k8s-teardown': './k8s-teardown/kubernetes-status/index.html', 'onap-helm': './k8s/onap-helm/helm.html', 'nodeport_check_certs': './k8s/nodeport_check_certs/certificates.html', 'internal_check_certs': './infrastructure-healthcheck/internal_check_certs/certificates.html', 'basic_vm': './basic_vm/reporting.html', 'basic_vm_macro': './basic_vm_macro/reporting.html', 'basic_network': './basic_network/reporting.html', 'basic_cnf': './basic_cnf/reporting.html', 'basic_cds': './basic_cds/reporting.html', 'basic_onboard': './basic_onboard/reporting.html', 'basic_clamp': './basic_clamp/reporting.html', 'pnf_macro': './pnf_macro/reporting.html', 'pnf-registrate': './pnf-registrate/report.html', '5gbulkpm': './5gbulkpm/report.html', 'hv-ves': './hv-ves/report.html', 'cmpv2': './cmpv2/cmpv2/report.html', 'dcaemod': './dcaemod/report.html', 'ves-collector': './ves-collector/report.html', 'root_pods': './root_pods/root_pods.log', 'unlimitted_pods': './unlimitted_pods/unlimitted_pods.log', 'cis_kubernetes': './cis_kubernetes/cis_kubernetes.log', 'nonssl_endpoints': './nonssl_endpoints/nonssl_endpoints.log', 'jdpw_ports': './jdpw_ports/jdpw_ports.log', 'kube_hunter': './kube_hunter/kube_hunter.log', 'versions': './security/versions.html', 'cps-healthcheck': './cps-healthcheck/report.html', } def get_lab_owner(pod_name): url = (URL_BASE_PODS + "?name=" + pod_name) response = requests.get(url, proxies=PROXY) response_json = response.json() try: lab_owner = response_json['pods'][0]['creator'] except KeyError: lab_owner = "unknown" except IndexError: lab_owner = "unknown" return lab_owner # Retrieve the Functest configuration to detect which tests are relevant # according to the pod, scenario PERIOD = 8 LOGGER.info("generate Xtesting reporting page") PARSER = argparse.ArgumentParser() PARSER.add_argument('-p', '--pod', help='Pod name') PARSER.add_argument('-d', '--db', help='Test DB URL') PARSER.add_argument('-t', '--build_tag', help='Build_tag') PARSER.add_argument('-m', '--mode', help='result retrieval mode', choices=['legacy', 's3'], default='legacy') ARGS = PARSER.parse_args() #PODS = ['onap_xtesting_openlab-OPNFV-oom', # 'onap_oom_gating_pod4_1-ONAP-oom', # 'onap_oom_gating_pod4_2-ONAP-oom', # 'onap_oom_gating_pod4_3-ONAP-oom', # 'onap_oom_gating_pod4_4-ONAP-oom', # 'onap_oom_gating_azure_1-OPNFV-oom', # 'onap_oom_gating_azure_2-OPNFV-oom', # 'onap_oom_gating_azure_3-OPNFV-oom', # 'onap_oom_gating_azure_4-OPNFV-oom', # 'onap_daily_pod4_master-ONAP-oom', # 'onap_daily_pod4_elalto-ONAP-oom'] PODS = ['ci-dualstack-master-daily'] if ARGS.pod is not None: PODS = [ARGS.pod] # adapt tests according to the typ of tests: daily/weekly/gating if "weekly" in ARGS.pod: # Complete the list with weekly tests SECURITY_USECASES['tests'].add('versions') INFRA_HEALTHCHECK['tests'].add('internal_check_certs') if "gating" in ARGS.pod: SECURITY_USECASES['tests'].remove('kube_hunter') # adapt test according to the version: guilin / honolulu / master if "guilin" in ARGS.pod: HEALTHCHECK['tests'].remove('dcaemod') SMOKE_USECASES['tests'].remove('basic_clamp') RESULT_URLS = RESULT_URLS_LEGACY LOGGER.info("init core result_url: %s", RESULT_URLS['core']) if ARGS.mode == "s3": LOGGER.info("use s3 mode for file retrieval") LOGGER.info("intended core result_url: %s", RESULT_URLS_S3['core']) RESULT_URLS = RESULT_URLS_S3 LOGGER.info("s3 core result_url: %s", RESULT_URLS['core']) LOGGER.info("final core result_url: %s", RESULT_URLS['core']) LOGGER.info("List of PODS: %s", PODS) for pod in PODS: LOGGER.info("POD: %s", pod) # Get the version lab_version = "master" lab_owner = get_lab_owner(pod) lab_owner = "NOKIA" LOGGER.info("Lab owner: %s", lab_owner) TREND_LINE = "" SCORE = 0 # Trend # ***** # calculation of the TREND SCORE_TREND = 0 if ARGS.db is not None: URL_BASE = str([ARGS.db][0]) LOGGER.info("Database: %s", URL_BASE) for tier_trend in TRENDS: tier_results = [] nb_tests = 0 nb_pass = 0 nb_fail = 0 score = 0 for test in tier_trend['tests']: project = 'integration' # Security tests affected to security project if tier_trend['name'] == 'security': project = 'security' url = (URL_BASE + "?project_name=" + project + "&case=" + test + "&pod_name=" + pod + "&last=5") response = requests.get(url, proxies=PROXY) response_json = response.json() # Note the 'u' must be used in python 2.7 # str(response_json).count("criteria': 'uFAIL") # it shall be removed if using python3 nb_fail = nb_fail + str(response_json).count("criteria': 'FAIL") nb_pass = nb_pass + str(response_json).count("criteria': 'PASS") try: score_trend = round(100 * nb_pass / (nb_pass + nb_fail)) except ZeroDivisionError: score_trend = 0 LOGGER.debug("Score Trend %s: %s", tier_trend, score_trend) tier_trend['score'] = score_trend # calculation of the overall SCORE for TREND NB_TIERS = 0 for tier_trend in TRENDS: NB_TIERS += 1 SCORE_TREND = SCORE_TREND + tier_trend['score'] SCORE_TREND = round(SCORE_TREND / NB_TIERS) LOGGER.info("Score Trend: %s", str(SCORE_TREND)) # calculation of the overall SCORE for tier in TIERS: tier_results = [] nb_tests = 0 nb_pass = 0 score = 0 for test in tier['tests']: # for Gating we consider the build_tag to retrieve the results # For daily runs, we do not. A build_tag is created based on # gitlab CI id and is different for each CI stage param_build_tag = "" if "gating" in pod and ARGS.build_tag is not None: param_build_tag = "&build_tag=" + str([ARGS.build_tag][0]) project = 'integration' # Security tests affected to security project if tier['name'] == 'security': project = 'security' # onap-k8s and onap-k8s-teardown are the same test # BUT # onap-k8s is executed after the installation (fresh installation) # onap-k8s-teardown after the tests # in case of tests executed in onap namespace, a test may trigger # an error status even it was OK at the end of the installation # a special uggly processing is then needed to avoid false negative search_test = test if test == "onap-k8s-teardown": search_test = "onap-k8s" nb_test_max = 5 url = (URL_BASE + "?project_name=" + project + "&case=" + search_test + "&period=" + str(PERIOD) + "&pod_name=" + pod + "&last=" + str(nb_test_max) + param_build_tag) LOGGER.debug("url: %s", url) response = requests.get(url, proxies=PROXY) response_json = response.json() response_url = "" if test in RESULT_URLS: response_url = RESULT_URLS[test] LOGGER.debug("response_json: %s", response_json) req_result = "" nb_results_found = len(response_json['results']) try: if test == "onap-k8s": req_result = response_json['results'][nb_results_found-1]['criteria'] else: req_result = response_json['results'][0]['criteria'] if lab_version == "unknown": lab_version = response_json['results'][0]['version'] except IndexError: req_result = None result = {'name': test, 'result': req_result, 'url': response_url} LOGGER.debug("result: %s", result) nb_tests += 1 if req_result == "PASS": nb_pass += 1 LOGGER.debug("nb_pass: %s", nb_pass) LOGGER.debug("nb_tests: %s", nb_tests) score = round(100 * nb_pass / nb_tests) LOGGER.debug("score: %s", score) tier_results.append(result) tier['score'] = score tier['results'] = tier_results # calculation of the overall SCORE NB_TIERS = 0 for tier in TIERS: NB_TIERS += 1 LOGGER.debug("Score %s", tier) SCORE = SCORE + tier['score'] SCORE = round(SCORE / NB_TIERS) LOGGER.info("Score: %s", str(SCORE)) # calculation of the evolution score versus trend if SCORE > 1.05*SCORE_TREND: # increasing TREND_LINE = "long arrow alternate up icon" LOGGER.info("Better status") elif SCORE < 0.95*SCORE_TREND: # decreasing TREND_LINE = "long arrow alternate down icon" LOGGER.info("Worst status") else: # stable TREND_LINE = "long arrow alternate right icon" LOGGER.info("stable status") TEMPLATELOADER = jinja2.FileSystemLoader(".") TEMPLATEENV = jinja2.Environment( loader=TEMPLATELOADER, autoescape=True) TEMPLATE_FILE = ("./index-tmpl-new.html") TEMPLATE = TEMPLATEENV.get_template(TEMPLATE_FILE) OUTPUT_TEXT = TEMPLATE.render( tiers=TIERS, pod=pod, period=PERIOD, date=REPORTINGDATE, score=SCORE, trend=TREND_LINE, lab_version=lab_version, lab_owner=lab_owner) FILENAME = "./index.html" with open(FILENAME, "w+") as fh: fh.write(OUTPUT_TEXT) # Generate txt reportingmy pretty Table vote=2 score_daily = [] dashboard_table = PrettyTable() dashboard_table.field_names = ["Test Name", "Category", "Status"] dashboard_table._max_width = {"Test Name" : 30, "Category": 40,"Status" : 10} #print(TIERS) for tier in TIERS: tier_score = {'tier': tier['name'], 'score': tier['score']} score_daily.append(tier_score) for test in tier['results']: if tier['name'] == "infrastructure-healthcheck": if test['name'] == "onap-k8s" or test['name'] == "onap-helm": if test['result'] == "FAIL" or test['result'] == None: vote-=2 if tier['name'] == "healthcheck" or tier['name'] == "smoke usecases": if test['result'] == "FAIL" or test['result'] == None: vote-=1 dashboard_table.add_row([test['name'],tier['name'], test['result']]) if vote < -2: vote = -2 LOGGER.info(dashboard_table) LOGGER.info("If I could, I would vote " + str(vote)) with open("./daily-status.txt", "w") as write_file: write_file.write(str(dashboard_table)) write_file.write("\n") write_file.write("**********************\n") write_file.write("* Automated vote: "+ str(vote) +"\n") write_file.write("**********************\n") write_file.close() # Gating vote # Infra HC onap-helm and onap-K8S MUST be OK # HC > 90 only 1 error OK in Full if not critical component # Smoke 1 # Generate heatlth json to build a time view odf the daily dashboard_table # create a json file for version tracking with open("./daily-scores.json", "w") as write_file: json.dump(score_daily, write_file)