<s> import json <EOL> import os <EOL> import unittest <EOL> import subprocess <EOL> import shlex <EOL> import shutil <EOL> import stat <EOL> import tempfile <EOL> import string <EOL> import random <EOL> import config <EOL> import requests <EOL> import time <EOL> import yaml <EOL> import logging <EOL> import pkg_resources <EOL> from pysflib . sfredmine import RedmineUtils <EOL> from subprocess import Popen , PIPE <EOL> logging . getLogger ( "<STR_LIT>" ) . setLevel ( logging . WARNING ) <EOL> logging . captureWarnings ( True ) <EOL> logger = logging . getLogger ( __name__ ) <EOL> logger . setLevel ( logging . DEBUG ) <EOL> EMPTY_JOB_XML = """<STR_LIT>""" <EOL> skipIf = unittest . skipIf <EOL> skip = unittest . skip <EOL> def _get_services ( ) : <EOL> cookie = get_cookie ( config . ADMIN_USER , <EOL> config . USERS [ config . ADMIN_USER ] [ '<STR_LIT:password>' ] ) <EOL> about = requests . get ( config . GATEWAY_URL + '<STR_LIT>' , <EOL> headers = { '<STR_LIT>' : '<STR_LIT:application/json>' } , <EOL> cookies = { '<STR_LIT>' : cookie } ) <EOL> if about . status_code > <NUM_LIT> : <EOL> return [ ] <EOL> try : <EOL> return about . json ( ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] <EOL> except ValueError : <EOL> raise Exception ( "<STR_LIT>" % ( config . GATEWAY_URL + <EOL> '<STR_LIT>' ) ) <EOL> def is_present ( service ) : <EOL> return service in _get_services ( ) <EOL> def has_issue_tracker ( ) : <EOL> return set ( config . ISSUE_TRACKERS ) & set ( _get_services ( ) ) <EOL> def skipIfServiceMissing ( service ) : <EOL> services = _get_services ( ) <EOL> if not services : <EOL> logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return skip <EOL> return skipIf ( service not in services , <EOL> '<STR_LIT>' % service ) <EOL> def skipIfServicePresent ( service ) : <EOL> services = _get_services ( ) <EOL> if not services : <EOL> logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return skip <EOL> return skipIf ( service in services , <EOL> '<STR_LIT>' % service ) <EOL> def skipIfIssueTrackerMissing ( ) : <EOL> services = _get_services ( ) <EOL> if not services : <EOL> logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return skip <EOL> return skipIf ( not set ( config . ISSUE_TRACKERS ) & set ( services ) , <EOL> '<STR_LIT>' ) <EOL> def skipIfIssueTrackerPresent ( ) : <EOL> services = _get_services ( ) <EOL> if not services : <EOL> logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return skip <EOL> tracker = list ( set ( config . ISSUE_TRACKERS ) & set ( services ) ) <EOL> if tracker : <EOL> return skipIf ( <EOL> '<STR_LIT>' % '<STR_LIT:U+002CU+0020>' . join ( tracker ) ) <EOL> def get_issue_tracker_utils ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> services = _get_services ( ) <EOL> if not services : <EOL> logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> return RedmineUtils ( config . GATEWAY_URL + "<STR_LIT>" , <EOL> * args , ** kwargs ) <EOL> if '<STR_LIT>' in services : <EOL> return RedmineUtils ( config . GATEWAY_URL + "<STR_LIT>" , <EOL> * args , ** kwargs ) <EOL> else : <EOL> from pysflib . interfaces . issuetracker import IssueTrackerUtils <EOL> return IssueTrackerUtils ( config . GATEWAY_URL , <EOL> * args , ** kwargs ) <EOL> def get_module_version ( module ) : <EOL> m = module <EOL> if not isinstance ( m , basestring ) : <EOL> m = module . __name__ <EOL> try : <EOL> return pkg_resources . get_distribution ( m ) . version <EOL> except pkg_resources . DistributionNotFound : <EOL> return "<STR_LIT:0>" <EOL> def create_random_str ( ) : <EOL> value = "<STR_LIT>" . join ( [ random . choice ( string . ascii_lowercase ) for _ in range ( <NUM_LIT:6> ) ] ) <EOL> return value <EOL> def set_private_key ( priv_key ) : <EOL> tempdir = tempfile . mkdtemp ( ) <EOL> priv_key_path = os . path . join ( tempdir , '<STR_LIT>' ) <EOL> file ( priv_key_path , '<STR_LIT:w>' ) . write ( priv_key ) <EOL> os . chmod ( priv_key_path , stat . S_IREAD | stat . S_IWRITE ) <EOL> return priv_key_path <EOL> def copytree ( src , dst , symlinks = False , ignore = None ) : <EOL> for item in os . listdir ( src ) : <EOL> s = os . path . join ( src , item ) <EOL> d = os . path . join ( dst , item ) <EOL> if os . path . isdir ( s ) : <EOL> shutil . copytree ( s , d , symlinks , ignore ) <EOL> else : <EOL> shutil . copy2 ( s , d ) <EOL> def get_cookie ( username , password ) : <EOL> url = "<STR_LIT>" % { '<STR_LIT>' : config . GATEWAY_URL } <EOL> resp = requests . post ( url , params = { '<STR_LIT:username>' : username , <EOL> '<STR_LIT:password>' : password , <EOL> '<STR_LIT>' : '<STR_LIT:/>' } , <EOL> allow_redirects = False ) <EOL> return resp . cookies . get ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def ssh_run_cmd ( sshkey_priv_path , user , host , subcmd ) : <EOL> host = '<STR_LIT>' % ( user , host ) <EOL> sshcmd = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> sshkey_priv_path , host ] <EOL> cmd = sshcmd + subcmd <EOL> p = Popen ( cmd , stdout = PIPE ) <EOL> return p . communicate ( ) <EOL> class Base ( unittest . TestCase ) : <EOL> pass <EOL> class Tool : <EOL> def __init__ ( self ) : <EOL> self . env = os . environ . copy ( ) <EOL> def exe ( self , cmd , cwd = None ) : <EOL> logger . debug ( '<STR_LIT>' % cmd ) <EOL> cmd = shlex . split ( cmd ) <EOL> ocwd = os . getcwd ( ) <EOL> output = '<STR_LIT>' <EOL> if cwd : <EOL> os . chdir ( cwd ) <EOL> try : <EOL> output = subprocess . check_output ( <EOL> cmd , stderr = subprocess . STDOUT , env = self . env ) <EOL> if output : <EOL> logger . debug ( '<STR_LIT>' % output . strip ( ) ) <EOL> except subprocess . CalledProcessError as err : <EOL> if err . output : <EOL> logger . exception ( "<STR_LIT>" % <EOL> ( err , err . output ) ) <EOL> else : <EOL> logger . exception ( err ) <EOL> finally : <EOL> os . chdir ( ocwd ) <EOL> return output <EOL> class ManageSfUtils ( Tool ) : <EOL> def __init__ ( self , url ) : <EOL> Tool . __init__ ( self ) <EOL> self . base_cmd = "<STR_LIT>" "<STR_LIT>" % ( url , config . GATEWAY_URL ) <EOL> def createProject ( self , name , user , options = None , cookie = None ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> base_cmd = self . base_cmd % ( user , passwd ) <EOL> if cookie : <EOL> base_cmd = base_cmd + "<STR_LIT>" % ( cookie ) <EOL> cmd = base_cmd + "<STR_LIT>" % name <EOL> if options : <EOL> for k , v in options . items ( ) : <EOL> cmd = cmd + "<STR_LIT>" + k + "<STR_LIT:U+0020>" + v <EOL> self . exe ( cmd ) <EOL> def deleteProject ( self , name , user ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd + "<STR_LIT>" <EOL> cmd = cmd % ( user , passwd , name ) <EOL> self . exe ( cmd ) <EOL> def replicationModifyConfig ( self , user , cmd , section , <EOL> setting = None , value = None ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( user , passwd ) + "<STR_LIT>" % ( cmd , section ) <EOL> if setting : <EOL> cmd = cmd + "<STR_LIT:U+0020>" + setting <EOL> if value : <EOL> cmd = cmd + "<STR_LIT:U+0020>" + value <EOL> self . exe ( cmd ) <EOL> def replicationTrigger ( self , user , project = None , url = None ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( user , passwd ) + "<STR_LIT>" <EOL> if project : <EOL> cmd = cmd + "<STR_LIT>" + project <EOL> if url : <EOL> cmd = cmd + "<STR_LIT>" + url <EOL> self . exe ( cmd ) <EOL> def addUsertoProjectGroups ( self , auth_user , project , new_user , groups ) : <EOL> passwd = config . USERS [ auth_user ] [ '<STR_LIT:password>' ] <EOL> umail = config . USERS [ new_user ] [ '<STR_LIT:email>' ] <EOL> cmd = self . base_cmd % ( auth_user , passwd ) <EOL> cmd = cmd + "<STR_LIT>" % project <EOL> cmd = cmd + "<STR_LIT>" % ( umail , groups ) <EOL> self . exe ( cmd ) <EOL> def deleteUserFromProjectGroups ( self , auth_user , <EOL> project , user , group = None ) : <EOL> passwd = config . USERS [ auth_user ] [ '<STR_LIT:password>' ] <EOL> umail = config . USERS [ user ] [ '<STR_LIT:email>' ] <EOL> cmd = self . base_cmd % ( auth_user , passwd ) + "<STR_LIT>" <EOL> cmd = cmd + "<STR_LIT>" % ( project , umail ) <EOL> if group : <EOL> cmd = cmd + "<STR_LIT>" % group <EOL> self . exe ( cmd ) <EOL> def list_active_members ( self , user ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( user , passwd ) + "<STR_LIT>" <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def create_gerrit_api_password ( self , user ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( user , passwd ) + "<STR_LIT>" <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def delete_gerrit_api_password ( self , user ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( user , passwd ) + "<STR_LIT>" <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def create_user ( self , user , password , email ) : <EOL> subcmd = ( "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( user , password , email , user ) ) <EOL> auth_user = config . ADMIN_USER <EOL> auth_password = config . USERS [ config . ADMIN_USER ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( auth_user , auth_password ) + subcmd <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def create_init_tests ( self , project , user ) : <EOL> subcmd = "<STR_LIT>" % project <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> cmd = self . base_cmd % ( user , passwd ) + subcmd <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def update_project_page ( self , user , project , url ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> subcmd = "<STR_LIT>" % ( project , url ) <EOL> cmd = self . base_cmd % ( user , passwd ) + subcmd <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def get_project_page ( self , user , project ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> subcmd = "<STR_LIT>" % project <EOL> cmd = self . base_cmd % ( user , passwd ) + subcmd <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> def delete_project_page ( self , user , project ) : <EOL> passwd = config . USERS [ user ] [ '<STR_LIT:password>' ] <EOL> subcmd = "<STR_LIT>" % project <EOL> cmd = self . base_cmd % ( user , passwd ) + subcmd <EOL> output = self . exe ( cmd ) <EOL> return output <EOL> class GerritGitUtils ( Tool ) : <EOL> def __init__ ( self , user , priv_key_path , email ) : <EOL> Tool . __init__ ( self ) <EOL> self . user = user <EOL> self . email = email <EOL> self . author = "<STR_LIT>" % ( self . user , email ) <EOL> self . priv_key_path = priv_key_path <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> ssh_wrapper = "<STR_LIT>" "<STR_LIT>" % os . path . abspath ( self . priv_key_path ) <EOL> wrapper_path = os . path . join ( self . tempdir , '<STR_LIT>' ) <EOL> file ( wrapper_path , '<STR_LIT:w>' ) . write ( ssh_wrapper ) <EOL> os . chmod ( wrapper_path , stat . S_IRWXU ) <EOL> self . env [ '<STR_LIT>' ] = wrapper_path <EOL> self . env [ '<STR_LIT>' ] = self . user <EOL> self . env [ '<STR_LIT>' ] = self . email <EOL> def config_review ( self , clone_dir ) : <EOL> self . exe ( "<STR_LIT>" % <EOL> config . GATEWAY_HOST , clone_dir ) <EOL> self . exe ( "<STR_LIT>" % <EOL> self . priv_key_path , clone_dir ) <EOL> self . exe ( "<STR_LIT>" , clone_dir ) <EOL> def list_open_reviews ( self , project , uri , port = <NUM_LIT> ) : <EOL> cmd = "<STR_LIT>" <EOL> cmd += "<STR_LIT>" <EOL> cmd += "<STR_LIT>" <EOL> reviews = self . exe ( cmd % ( os . path . abspath ( self . priv_key_path ) , <EOL> str ( port ) , <EOL> self . user , <EOL> uri , <EOL> project ) ) <EOL> array_json = "<STR_LIT:[>" + '<STR_LIT>' . join ( reviews . split ( '<STR_LIT:\n>' ) [ : - <NUM_LIT:1> ] ) + "<STR_LIT:]>" <EOL> j = json . loads ( array_json ) <EOL> return j [ : - <NUM_LIT:1> ] <EOL> def clone ( self , uri , target , config_review = True ) : <EOL> if not uri . startswith ( '<STR_LIT>' ) : <EOL> raise Exception ( "<STR_LIT>" % uri ) <EOL> cmd = "<STR_LIT>" % ( uri , target ) <EOL> self . exe ( cmd , self . tempdir ) <EOL> clone = os . path . join ( self . tempdir , target ) <EOL> if not os . path . isdir ( clone ) : <EOL> raise Exception ( "<STR_LIT>" % clone ) <EOL> self . exe ( '<STR_LIT>' % <EOL> self . user , clone ) <EOL> if config_review : <EOL> self . config_review ( clone ) <EOL> return clone <EOL> def fetch_meta_config ( self , clone_dir ) : <EOL> cmd = '<STR_LIT>' '<STR_LIT>' <EOL> self . exe ( cmd , clone_dir ) <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> def add_commit_in_branch ( self , clone_dir , branch , files = None , commit = None ) : <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> self . exe ( '<STR_LIT>' % branch , clone_dir ) <EOL> if not files : <EOL> file ( os . path . join ( clone_dir , '<STR_LIT>' ) , '<STR_LIT:w>' ) . write ( '<STR_LIT:data>' ) <EOL> files = [ '<STR_LIT>' ] <EOL> self . git_add ( clone_dir , files ) <EOL> if not commit : <EOL> commit = "<STR_LIT>" <EOL> self . exe ( "<STR_LIT>" % ( self . author , commit ) , <EOL> clone_dir ) <EOL> def add_commit_for_all_new_additions ( self , clone_dir , commit = None ) : <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> if not commit : <EOL> commit = "<STR_LIT>" <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> self . exe ( "<STR_LIT>" % ( self . author , commit ) , <EOL> clone_dir ) <EOL> def direct_push_branch ( self , clone_dir , branch ) : <EOL> self . exe ( '<STR_LIT>' % branch , clone_dir ) <EOL> self . exe ( '<STR_LIT>' % branch , clone_dir ) <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> def review_push_branch ( self , clone_dir , branch ) : <EOL> self . exe ( '<STR_LIT>' % branch , clone_dir ) <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> def git_add ( self , clone_dir , files = [ ] ) : <EOL> to_add = "<STR_LIT:U+0020>" . join ( files ) <EOL> self . exe ( '<STR_LIT>' % to_add , clone_dir ) <EOL> def add_commit_and_publish ( self , clone_dir , branch , <EOL> commit_msg , commit_author = None , <EOL> fnames = None ) : <EOL> self . exe ( '<STR_LIT>' % branch , clone_dir ) <EOL> if not fnames : <EOL> fname = create_random_str ( ) <EOL> data = '<STR_LIT:data>' <EOL> file ( os . path . join ( clone_dir , fname ) , '<STR_LIT:w>' ) . write ( data ) <EOL> fnames = [ fname ] <EOL> self . git_add ( clone_dir , fnames ) <EOL> if commit_msg : <EOL> author = '<STR_LIT>' % ( commit_author , <EOL> config . USERS [ commit_author ] [ '<STR_LIT:email>' ] ) if commit_author else self . author <EOL> self . exe ( "<STR_LIT>" % <EOL> ( author , commit_msg ) , clone_dir ) <EOL> else : <EOL> self . exe ( "<STR_LIT>" , clone_dir ) <EOL> self . exe ( "<STR_LIT>" , clone_dir ) <EOL> self . exe ( '<STR_LIT>' , clone_dir ) <EOL> def get_branches ( self , clone_dir , include_remotes = False ) : <EOL> cmd = '<STR_LIT>' <EOL> if include_remotes : <EOL> cmd += '<STR_LIT>' <EOL> out = self . exe ( cmd , clone_dir ) <EOL> return out . split ( ) <EOL> class JenkinsUtils : <EOL> def __init__ ( self ) : <EOL> with open ( '<STR_LIT>' % config . SF_BOOTSTRAP_DATA ) as fh : <EOL> yconfig = yaml . load ( fh ) <EOL> self . jenkins_user = '<STR_LIT>' <EOL> self . jenkins_password = yconfig . get ( '<STR_LIT>' ) <EOL> self . jenkins_url = config . GATEWAY_URL + "<STR_LIT>" <EOL> self . cookies = { '<STR_LIT>' : get_cookie ( config . USER_1 , <EOL> config . USER_1_PASSWORD ) } <EOL> def get ( self , url ) : <EOL> return requests . get ( url , <EOL> auth = ( self . jenkins_user , self . jenkins_password ) , <EOL> cookies = self . cookies ) <EOL> def post ( self , url , params , data , headers ) : <EOL> return requests . post ( url , <EOL> params = params , <EOL> data = data , <EOL> headers = headers , <EOL> auth = ( self . jenkins_user , <EOL> self . jenkins_password ) , <EOL> cookies = self . cookies ) <EOL> def create_job ( self , name ) : <EOL> url = "<STR_LIT>" % self . jenkins_url <EOL> headers = { '<STR_LIT>' : '<STR_LIT>' } <EOL> resp = self . post ( url , <EOL> params = { '<STR_LIT:name>' : name } , <EOL> data = EMPTY_JOB_XML , <EOL> headers = headers ) <EOL> return resp . status_code <EOL> def list_jobs ( self ) : <EOL> from xml . dom import minidom <EOL> url = "<STR_LIT>" % self . jenkins_url <EOL> resp = self . get ( url ) <EOL> if resp . status_code == <NUM_LIT:200> : <EOL> jobs = [ ] <EOL> for job in minidom . parseString ( resp . text ) . getElementsByTagName ( '<STR_LIT>' ) : <EOL> jobs . append ( job . firstChild . childNodes [ <NUM_LIT:0> ] . data ) <EOL> return jobs <EOL> return None <EOL> def get_last_build_number ( self , job_name , type ) : <EOL> url = "<STR_LIT>" % { <EOL> '<STR_LIT>' : self . jenkins_url , '<STR_LIT>' : job_name , '<STR_LIT:type>' : type } <EOL> try : <EOL> resp = self . get ( url ) <EOL> return int ( resp . text ) <EOL> except : <EOL> return <NUM_LIT:0> <EOL> def wait_till_job_completes ( self , job_name , last , type , max_retries = <NUM_LIT:30> ) : <EOL> retries = <NUM_LIT:0> <EOL> while True : <EOL> cur = self . get_last_build_number ( job_name , type ) <EOL> if cur > last : <EOL> break <EOL> elif retries > max_retries : <EOL> break <EOL> else : <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> retries += <NUM_LIT:1> </s>
