summaryrefslogtreecommitdiff
path: root/gitrepo.py
blob: 06a79deff484a5c794cba0a9c7921df9e35b367d (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
import unittest
import logging
import os

from sh import git
from sh import ErrorReturnCode
from sh import ErrorReturnCode_128

import subprocess

# Use this to manage 'cd' context
from contextlib import contextmanager

from proj import Proj
# Return to prevdir when context is exited.
# Use in the following manner:
#
#    with cd("foodir"):
#       operation1_in_foodir
#       operation2_in_foodir
#
#    operation3_in_prevdir
#
@contextmanager
def cd(newdir):
    prevdir = os.getcwd()
    os.chdir(os.path.expanduser(newdir))
    try:
        yield
    finally:
        os.chdir(prevdir)

# This is called 'GitRepo' because a repository for a project doesn't only have
# to be a git repository.  Abstract base class used to define the operations
# that both Clone and Workdir share.  Notice that there is no actual repodir
# specified here.  You must instantiate a subclass in order to use/test these
# operations.  This is because 'clones' and 'workdirs' are created differently
# but you operate on them in the exact same way.
class GitRepo(object):
    def __init__(self, proj):
	self.repodir=u""
	if isinstance(proj, Proj):
            self.proj=proj
        else:
	    raise TypeError('proj input parameter is not of type Proj')

    def branchexists(self, branch):
	with cd(self.repodir):
	    try:
                # Quote the branchname because it might have strange
                # characters in it.
		br="%s" % branch
                git("rev-parse", "--verify", br)
	    except ErrorReturnCode_128:
		return False
	    return True

    def getbranch(self):
	try:
	    with cd(self.repodir):
		branch=git("rev-parse", "--abbrev-ref", "HEAD").stdout.rstrip()
        except ErrorReturnCode:
	    raise EnvironmentError("Unable to get the current branch")
	return branch

    # TODO: Fully test this.
    # Stash current changes and change to new branch.
    # Return to previous branch when context is exited.
    @contextmanager
    def checkoutbranch(self, branch, track=""):
	with cd(self.repodir):
	    # The stashid will remain empty if there were no changes in the
	    # directory to stash.
	    stashid=u""

	    try:
                self.prevbranch=git("rev-parse", "--abbrev-ref", "HEAD").stdout.rstrip()
            except ErrorReturnCode:
	         raise EnvironmentError("Unable to get the current branch")

	    try:
		# TODO: do we want to do a blanket git add *?
                stashid=git("stash", "create").stdout.rstrip()
		# If there's no stash then there's no reason to save.
		if stashid!="":
		    git("stash", "save")
            except ErrorReturnCode:
	        raise EnvironmentError("Unable to get the current branch")

	    try:
		# TODO This is a nop as it is.
	        git("rev-parse", "--verify",  branch)
		# if the branch exists the previous statement won't cause an
		# exception so we know we can just check it out.
                git("checkout", branch)
            except ErrorReturnCode:
	        try:
		    # if it doesn't exist we need to checkout a new branch
                    git("checkout", "-b", branch, track).stdout.rstrip()
                except ErrorReturnCode as exc:
	            raise EnvironmentError("Unable to checkout -b %s %s" % (branch, track))
        try:
            yield
        finally:
	    try:
	        with cd(self.repodir):
                    git("checkout", self.prevbranch)
		    # If there's no stashid then there were no files stashed,
		    # so don't stash apply.
		    if stashid!="":
                        git("stash", "apply", stashid)
            except ErrorReturnCode, exc:
		    raise EnvironmentError("Unable to return to the previous branch: %s" % exc.stderr)

    # TODO: Write a unit test for this.
    def add(self, filetogitadd):
	# TODO: Determine if filetogitadd is relative or absolute.
	# TODO: verify that filetogitadd is in self.repodir
	try:
	    with cd(self.repodir):
		git("add", filetogitadd)
        except ErrorReturnCode:
	    raise EnvironmentError("Unable to git add " + filetogitadd)

    # TODO: Write a unit test for this.
    # TODO: fix the default
    def commit(self, message="default"):
	print "Committing changes."
	try:
	    with cd(self.repodir):
		# Git commit first with a boiler plate message and then allow the user
		# to amend.
		if git("status", "--porcelain"):
	#	print git("commit", "-m", '%s' % message, _err_to_out=True)
		    subprocess.call(["git", "commit", "-m", message])
		# using python sh will suppress the git editor
		    subprocess.call(["git", "commit", "--amend"])
        except ErrorReturnCode:
	    raise EnvironmentError("Unable to git commit ")

    def log(self, number):
	try:
	    with cd(self.repodir):
	        print git("log", "-n %d" % number)
        except ErrorReturnCode:
	    raise EnvironmentError("Unable to git add " + filetogitadd)

    # TODO: Does this need to 'cd' first?
    def editFile(self, toedit):
	editor = os.getenv('EDITOR')
	if not editor:
	    editor='/usr/bin/vim'
	os.system(editor + ' ' + self.repodir + '/' + toedit)

	self.add(toedit)

	return

    # TODO: Test this function with both dryrun=True and dryrun=False
    def pushToBranch(self, tobranch, dryrun=True):
	try:
	    # TODO: Check for un-merged commits.
	    with cd(self.repodir):
		branch=self.getbranch()
	        if dryrun:
	            git("push", "--dry-run", branch, tobranch)
	        else:
	            git("push", branch, tobranch)
        except ErrorReturnCode:
	    raise EnvironmentError("Unable to push branch %s to %s" % (branch, tobranch))

    def __str__(self):
	return self.repodir

# TestGitRepo is an abstract baseclass. Verify that the constructor
# will throw an exception if you try to run operations without
# defining a subclass.
class TestGitRepo(unittest.TestCase):
    repoprefix="GitRepoUT"

    # This class is only used to test the GitRepo functions since, as an,
    # abstract-baseclass, GitRepo doesn't define repodir, and we need an
    # actual repository to test git based functions.
    class DerivedGitRepo(GitRepo):
        def __init__(self, proj):
	    super(TestGitRepo.DerivedGitRepo,self).__init__(proj)
	    self.remote=u'http://git.linaro.org/toolchain/release-notes.git'
	    dgr_dirname="release-notes"
	    self.repodir=self.proj.projdir + "/" + dgr_dirname
	    try:
	        with cd(self.proj.projdir):
	            git("clone", self.remote, dgr_dirname, _err_to_out=True, _out="/dev/null")
            except ErrorReturnCode:
                raise EnvironmentError("Git unable to clone into " + self.repodir)

    @classmethod
    def setUpClass(cls):
	cls.proj=Proj(prefix=TestGitRepo.repoprefix, persist=False)
	cls.dgr=TestGitRepo.DerivedGitRepo(cls.proj)

    @classmethod
    def tearDownClass(cls):
        cls.proj.cleanup()

    # Test that the abstract baseclass throws an exception if one of the
    # functions are called from the baseclass without a derived class.
    def test_repo(self):
	self.repo=GitRepo(self.proj)
	with self.assertRaises(OSError):
            # We haven't set a repodir so this should throw an exception.
	    branch=self.repo.getbranch()

    # TODO: Create a test-branch in the repo so it's always there.

    def test_branchexists(self):
       self.assertTrue(self.dgr.branchexists("remotes/origin/releases/linaro-4.9-2015.05")) 
       
    def test_not_branchexists(self):
       self.assertFalse(self.dgr.branchexists("foobar"))

    def test_getbranch(self):
        with cd(self.dgr.repodir):
	    try:
                git("checkout", "-b", "master_test", "origin/master").stdout.rstrip()
            except ErrorReturnCode as exc:
	        raise EnvironmentError("Unable to checkout -b %s %s" % (branch, track))
	    self.assertEquals(self.dgr.getbranch(), "master_test")

    # TODO: Test checkoutbranch with various combinations of polluted
    # directories.

if __name__ == '__main__':
    #logging.basicConfig(level="INFO")
    unittest.main()