summaryrefslogtreecommitdiff
path: root/linaropy/git/workdir.py
blob: db125e42d954c8e7946e785f2ea78cb90926fbeb (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
import unittest
import logging
import os

import uuid

from ..proj import Proj

from ..cd import cd
from gitrepo import GitRepo

from clone import Clone

from sh import git
from sh import git_new_workdir
from sh import ErrorReturnCode

# TODO: Set the git repo pushurl.


class Workdir(GitRepo):
    # @ clone - clone this workdir from this Clone.
    # @ workdir - name of the proposed workdir.
    # @ track - derive the workdir from the branch designated by the string
    #           in 'track'.
    # @ branchname - the name of the local branch

    def __init__(self, proj, clone=None, workdir="", track=None, branchname=""):
        super(Workdir, self).__init__(proj)

        # TODO: The Clone can be in a different project dir than proj.  if
        # clone.clonedir is not in proj.projdir then create a symlink in the
        # current projdir.

        if not isinstance(clone, Clone):
            raise TypeError('Clone input parameter is not of type Clone')

        if not isinstance(workdir, str):
            raise TypeError('Workdir workdir parameter must be a string')

        # An empty workdir would result in an empty variable expansion
        # and a malformed git new-workdir expression.
        if workdir == "":
            raise EnvironmentError(
                'You must specify a workdir directory when creating a Workdir')

        if branchname == "":
            raise EnvironmentError(
                'You must specify a branchname when creating a Workdir')

        try:
            with cd(self.proj.projdir):
                logging.info(
                    "Workdir(): going to call git new-workdir tracking %s" % track)
                if track:
                    logging.info(
                        "Workdir(): Calling git new-workdir for workdir %s, tracking %s" % (workdir, track))
                    # TODO: Do we want to prevalidate that 'track' is a valid branch or tag?
                    # If we don't this will just throw an exception.
                    print git_new_workdir(clone.clonedir(), workdir, track, _err_to_out=True, _out="/dev/null")
                else:
                    logging.info(
                        "Workdir(): Calling git new-workdir for workdir %s, tracking %s" % (workdir, "origin/HEAD"))
                    # Always just checkout HEAD if the track is not specified.
                    print git_new_workdir(clone.clonedir(), workdir, "origin/HEAD", _err_to_out=True, _out="/dev/null")

        except ErrorReturnCode as exc:
            # if 'workdir' is None
            # If 'workdir' already exists.
            # If clone.clonedir() doesn't exist.
            # If track isn't a valid branch.
            raise EnvironmentError("Unable to create a git workdir ")

        self.repodir = self.proj.projdir + '/' + workdir

        try:
            with cd(self.repodir):
                git("checkout", "-b", branchname)
        except ErrorReturnCode as exc:
            raise EnvironmentError(
                "Unable to checkout branch %s in workdir %s" % (branchname, self.repodir))

        # TODO: Verify that self.repodir exists or 'raise' exception?

    # TODO: write this function or the workdir will remain in detached
    # head state if it's not checked out into a branch.
    # def checkout(self, ):

    def workdir(self):
        return self.repodir


class TestWorkdir(unittest.TestCase):
    testdirprefix = "WorkdirUT"

    @classmethod
    def setUpClass(cls):
        # We need a Proj dir to store the necessary Clone in.
        cls.proj = Proj(prefix=TestWorkdir.testdirprefix)

        # TODO: Perform a connectivity test to determine online or offline
        #      testing mode.

        # We'll create a Clone to be used by all the Workdir tests in order to
        # be nice to the git server.
        cls.rnclone = Clone(
            cls.proj, remote=u'http://git.linaro.org/toolchain/release-notes.git')
        # OFFLINE:
        # cls.rnclone=Clone(cls.proj,remote=u'/var/run/media/ryanarn/sidecar/reldir/release-notes')

        # All further tests will use this as the remote if they need to
        # create more clones.
        cls.rnremote = cls.rnclone.repodir

        # A common name to use for the workdir for all of the Workdir tests.
        cls.workdir = "testworkdir"

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

    # Every instance of TestWorkdir requires its own proj directory.
    def setUp(self):
        self.proj = Proj(prefix=TestWorkdir.testdirprefix)
        self.branchname = "testworkdir-" + str(uuid.uuid4())

    def tearDown(self):
        self.proj.cleanup()

    def test_workdir(self):
        self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                               workdir=TestWorkdir.workdir, track=None, branchname=self.branchname)
        self.assertTrue(os.path.isdir(self.workdir.workdir()))

    def test_workdir_with_default_track(self):
        self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                               workdir=TestWorkdir.workdir, branchname=self.branchname)
        self.assertTrue(os.path.isdir(self.workdir.workdir()))

    def test_workdir_with_track(self):
        self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                               workdir=TestWorkdir.workdir, track="origin/HEAD", branchname=self.branchname)
        self.assertTrue(os.path.isdir(self.workdir.workdir()))

    def test_workdir_with_checkoutbranch(self):
        self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                               workdir=TestWorkdir.workdir, branchname="startbranch")
        self.assertTrue(os.path.isdir(self.workdir.workdir()))

        # Test a new branch (derived from the existing branch)
        with self.workdir.checkoutbranch("TestWorkdir", self.workdir.getbranch()):
            self.assertEqual("TestWorkdir", self.workdir.getbranch())

        # Verify that leaving the context restored the original branch.
        self.assertEqual(self.workdir.getbranch(), "startbranch")

        # Test being able to switch to a branch that already exists.
        with self.workdir.checkoutbranch("TestWorkdir"):
            self.assertEqual("TestWorkdir", self.workdir.getbranch())

        self.assertEqual(self.workdir.getbranch(), "startbranch")

    def test_workdir_already_exists(self):
        self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                               workdir=TestWorkdir.workdir, branchname=self.branchname)

        # Should fail because TestWorkdir.workdir should already exist.
        with self.assertRaises(EnvironmentError):
            self.workdir2 = Workdir(self.proj, clone=TestWorkdir.rnclone,
                                    workdir=TestWorkdir.workdir, branchname=self.branchname + '_2')

    def test_workdir_exceptions(self):
        with self.assertRaises(TypeError):
            self.workdir = Workdir(
                self.proj, clone=TestWorkdir.rnclone, workdir=None, branchname=self.branchname)

        with self.assertRaises(EnvironmentError):
            self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone, workdir=TestWorkdir.workdir,
                                   track="someinvalidbranchname", branchname=self.branchname)

        # Test if clone.clonedir() doesn't exist by breaking repodir.
        brokenclone = Clone(TestWorkdir.proj, TestWorkdir.rnremote)
        brokenclone.repodir = "/breakthisrepodir"

        # Verify that a Clone with a broken repodir/clonedir results in an
        # exception.
        with self.assertRaises(EnvironmentError):
            self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                                   workdir=TestWorkdir.workdir, branchname=self.branchname)

        # 'clone' isn't a valid Clone type.
        with self.assertRaises(TypeError):
            self.workdir = Workdir(
                self.proj, clone="foobar", workdir=TestWorkdir.workdir, branchname=self.branchname)

    def test_workdir_exceptions(self):
        # TODO: branchname is empty.
        with self.assertRaises(EnvironmentError):
            self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                                   workdir=TestWorkdir.workdir, track="origin/HEAD", branchname="")

    def test_workdir_exceptions(self):
        # TODO: branchname already exists.
        self.workdir = Workdir(self.proj, clone=TestWorkdir.rnclone,
                               workdir=TestWorkdir.workdir, track="origin/HEAD", branchname=self.branchname)
        with self.assertRaises(EnvironmentError):
            self.workdir2 = Workdir(self.proj, clone=TestWorkdir.rnclone,
                                    workdir=TestWorkdir.workdir, track="origin/HEAD", branchname=self.branchname)

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