#!/usr/bin/env python # # jose.lausuch@ericsson.com # valentin.boucher@orange.com # All rights reserved. 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 # import functools import json import logging import os import re import shutil import subprocess import sys import time from datetime import datetime as dt import dns.resolver import requests from six.moves import urllib import yaml from git import Repo from functest.utils import constants from functest.utils import decorators logger = logging.getLogger(__name__) # ---------------------------------------------------------- # # INTERNET UTILS # # ----------------------------------------------------------- def check_internet_connectivity(url='http://www.opnfv.org/'): """ Check if there is access to the internet """ try: urllib.request.urlopen(url, timeout=5) return True except urllib.error.URLError: return False def download_url(url, dest_path): """ Download a file to a destination path given a URL """ name = url.rsplit('/')[-1] dest = dest_path + "/" + name try: response = urllib.request.urlopen(url) except (urllib.error.HTTPError, urllib.error.URLError): return False with open(dest, 'wb') as f: shutil.copyfileobj(response, f) return True # ---------------------------------------------------------- # # CI UTILS # # ----------------------------------------------------------- def get_git_branch(repo_path): """ Get git branch name """ repo = Repo(repo_path) branch = repo.active_branch return branch.name def get_installer_type(): """ Get installer type (fuel, apex, joid, compass) """ try: installer = os.environ['INSTALLER_TYPE'] except KeyError: logger.error("Impossible to retrieve the installer type") installer = "Unknown_installer" return installer def get_scenario(): """ Get scenario """ try: scenario = os.environ['DEPLOY_SCENARIO'] except KeyError: logger.info("Impossible to retrieve the scenario." "Use default os-nosdn-nofeature-noha") scenario = "os-nosdn-nofeature-noha" return scenario def get_version(): """ Get version """ # Use the build tag to retrieve the version # By default version is unknown # if launched through CI the build tag has the following format # jenkins------ # e.g. jenkins-functest-fuel-opnfv-jump-2-daily-master-190 # jenkins-functest-fuel-baremetal-weekly-master-8 # use regex to match branch info rule = "(dai|week)ly-(.+?)-[0-9]*" build_tag = get_build_tag() m = re.search(rule, build_tag) if m: return m.group(2) else: return "unknown" def get_pod_name(): """ Get PoD Name from env variable NODE_NAME """ try: return os.environ['NODE_NAME'] except KeyError: logger.info( "Unable to retrieve the POD name from environment. " + "Using pod name 'unknown-pod'") return "unknown-pod" def get_build_tag(): """ Get build tag of jenkins jobs """ try: build_tag = os.environ['BUILD_TAG'] except KeyError: logger.info("Impossible to retrieve the build tag") build_tag = "none" return build_tag def get_db_url(): """ Returns DB URL """ # TODO use CONST mechanism try: # if TEST_DB_URL declared in env variable, use it! db_url = os.environ['TEST_DB_URL'] except KeyError: db_url = get_functest_config('results.test_db_url') return db_url def logger_test_results(project, case_name, status, details): pod_name = get_pod_name() scenario = get_scenario() version = get_version() build_tag = get_build_tag() logger.info( "\n" "****************************************\n" "\t %(p)s/%(n)s results \n\n" "****************************************\n" "DB:\t%(db)s\n" "pod:\t%(pod)s\n" "version:\t%(v)s\n" "scenario:\t%(s)s\n" "status:\t%(c)s\n" "build tag:\t%(b)s\n" "details:\t%(d)s\n" % {'p': project, 'n': c