1# ex:ts=4:sw=4:sts=4:et 2# -*- tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- 3# 4# patchtestrepo: PatchTestRepo class used mainly to control a git repo from patchtest 5# 6# Copyright (C) 2016 Intel Corporation 7# 8# SPDX-License-Identifier: GPL-2.0-only 9# 10 11import os 12import utils 13import logging 14from patch import PatchTestPatch 15 16logger = logging.getLogger('patchtest') 17info=logger.info 18 19class PatchTestRepo(object): 20 21 # prefixes used for temporal branches/stashes 22 prefix = 'patchtest' 23 24 def __init__(self, patch, repodir, commit=None, branch=None): 25 self._repodir = repodir 26 self._patch = PatchTestPatch(patch) 27 self._current_branch = self._get_current_branch() 28 29 # targeted branch defined on the patch may be invalid, so make sure there 30 # is a corresponding remote branch 31 valid_patch_branch = None 32 if self._patch.branch in self.upstream_branches(): 33 valid_patch_branch = self._patch.branch 34 35 # Target Branch 36 # Priority (top has highest priority): 37 # 1. branch given at cmd line 38 # 2. branch given at the patch 39 # 3. current branch 40 self._branch = branch or valid_patch_branch or self._current_branch 41 42 # Target Commit 43 # Priority (top has highest priority): 44 # 1. commit given at cmd line 45 # 2. branch given at cmd line 46 # 3. branch given at the patch 47 # 3. current HEAD 48 self._commit = self._get_commitid(commit) or \ 49 self._get_commitid(branch) or \ 50 self._get_commitid(valid_patch_branch) or \ 51 self._get_commitid('HEAD') 52 53 self._workingbranch = "%s_%s" % (PatchTestRepo.prefix, os.getpid()) 54 55 # create working branch 56 self._exec({'cmd': ['git', 'checkout', '-b', self._workingbranch, self._commit]}) 57 58 self._patchmerged = False 59 60 # Check if patch can be merged using git-am 61 self._patchcanbemerged = True 62 try: 63 self._exec({'cmd': ['git', 'am', '--keep-cr'], 'input': self._patch.contents}) 64 except utils.CmdException as ce: 65 self._exec({'cmd': ['git', 'am', '--abort']}) 66 self._patchcanbemerged = False 67 finally: 68 # if patch was applied, remove it 69 if self._patchcanbemerged: 70 self._exec({'cmd':['git', 'reset', '--hard', self._commit]}) 71 72 # for debugging purposes, print all repo parameters 73 logger.debug("Parameters") 74 logger.debug("\tRepository : %s" % self._repodir) 75 logger.debug("\tTarget Commit : %s" % self._commit) 76 logger.debug("\tTarget Branch : %s" % self._branch) 77 logger.debug("\tWorking branch : %s" % self._workingbranch) 78 logger.debug("\tPatch : %s" % self._patch) 79 80 @property 81 def patch(self): 82 return self._patch.path 83 84 @property 85 def branch(self): 86 return self._branch 87 88 @property 89 def commit(self): 90 return self._commit 91 92 @property 93 def ismerged(self): 94 return self._patchmerged 95 96 @property 97 def canbemerged(self): 98 return self._patchcanbemerged 99 100 def _exec(self, cmds): 101 _cmds = [] 102 if isinstance(cmds, dict): 103 _cmds.append(cmds) 104 elif isinstance(cmds, list): 105 _cmds = cmds 106 else: 107 raise utils.CmdException({'cmd':str(cmds)}) 108 109 results = [] 110 cmdfailure = False 111 try: 112 results = utils.exec_cmds(_cmds, self._repodir) 113 except utils.CmdException as ce: 114 cmdfailure = True 115 raise ce 116 finally: 117 if cmdfailure: 118 for cmd in _cmds: 119 logger.debug("CMD: %s" % ' '.join(cmd['cmd'])) 120 else: 121 for result in results: 122 cmd, rc, stdout, stderr = ' '.join(result['cmd']), result['returncode'], result['stdout'], result['stderr'] 123 logger.debug("CMD: %s RCODE: %s STDOUT: %s STDERR: %s" % (cmd, rc, stdout, stderr)) 124 125 return results 126 127 def _get_current_branch(self, commit='HEAD'): 128 cmd = {'cmd':['git', 'rev-parse', '--abbrev-ref', commit]} 129 cb = self._exec(cmd)[0]['stdout'] 130 if cb == commit: 131 logger.warning('You may be detached so patchtest will checkout to master after execution') 132 cb = 'master' 133 return cb 134 135 def _get_commitid(self, commit): 136 137 if not commit: 138 return None 139 140 try: 141 cmd = {'cmd':['git', 'rev-parse', '--short', commit]} 142 return self._exec(cmd)[0]['stdout'] 143 except utils.CmdException as ce: 144 # try getting the commit under any remotes 145 cmd = {'cmd':['git', 'remote']} 146 remotes = self._exec(cmd)[0]['stdout'] 147 for remote in remotes.splitlines(): 148 cmd = {'cmd':['git', 'rev-parse', '--short', '%s/%s' % (remote, commit)]} 149 try: 150 return self._exec(cmd)[0]['stdout'] 151 except utils.CmdException: 152 pass 153 154 return None 155 156 def upstream_branches(self): 157 cmd = {'cmd':['git', 'branch', '--remotes']} 158 remote_branches = self._exec(cmd)[0]['stdout'] 159 160 # just get the names, without the remote name 161 branches = set(branch.split('/')[-1] for branch in remote_branches.splitlines()) 162 return branches 163 164 def merge(self): 165 if self._patchcanbemerged: 166 self._exec({'cmd': ['git', 'am', '--keep-cr'], 167 'input': self._patch.contents, 168 'updateenv': {'PTRESOURCE':self._patch.path}}) 169 self._patchmerged = True 170 171 def clean(self): 172 self._exec({'cmd':['git', 'checkout', '%s' % self._current_branch]}) 173 self._exec({'cmd':['git', 'branch', '-D', self._workingbranch]}) 174 self._patchmerged = False 175