File: submit.py

package info (click to toggle)
git-ubuntu 1.1-2
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 1,688 kB
  • sloc: python: 13,378; sh: 480; makefile: 2
file content (249 lines) | stat: -rw-r--r-- 7,899 bytes parent folder | download
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
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
import argparse
import logging
import os
from pygit2 import Commit
import shutil
from subprocess import CalledProcessError
import sys
import tempfile
import textwrap
from gitubuntu.__main__ import top_level_defaults
from gitubuntu.git_repository import (
    derive_target_branch,
    GitUbuntuRepository,
    git_dep14_tag,
)
from gitubuntu.run import decode_binary, run
from gitubuntu.source_information import GitUbuntuSourceInformation, launchpad_login_auth

from lazr.restfulclient.errors import BadRequest


def parse_args(subparsers=None, base_subparsers=None):
    kwargs = dict(description='Submit changes as a Launchpad Merge Proposal (EXPERIMENTAL)',
                  formatter_class=argparse.RawTextHelpFormatter,
                 )
    if base_subparsers:
        kwargs['parents'] = base_subparsers
    if subparsers:
        parser = subparsers.add_parser('submit', **kwargs)
        parser.set_defaults(func=cli_main)
    else:
        parser = argparse.ArgumentParser(**kwargs)

    parser.add_argument('--target-branch', type=str,
        help='Branch to target with merge proposal. If not specified, '
             'the nearest remote branch to the current branch, '
             'if unique, is used.'
    )
    parser.add_argument('--reviewer', action='append',
        help='Specify reviewers for the proposed merge. This '
             'will default to the canonical-server-reporter team. This option can be '
             'specified multiple times.',
        default=['canonical-server-reporter']
    )
    parser.add_argument('-d', '--directory', type=str,
        help='Use git repository at specified location.',
        default=os.path.abspath(os.getcwd())
    )
    parser.add_argument('-f', '--force',
        help='Overwrite existing tags and branches, where applicable.',
        action='store_true'
    )
    parser.add_argument('--target-user', type=str,
        help='Target user\'s repository to merge into',
        default='git-ubuntu-import'
    )
    parser.add_argument('--branch', type=str,
        help='The branch to merge. Defaults to branch pointed to by HEAD.'
    )
    parser.add_argument('--no-push', action='store_true',
                        help='Do not push the specified branch before '
                             'submitting the merge proposal'
    )
    parser.add_argument('-l', '--lp-user', type=str, help=argparse.SUPPRESS)
    if not subparsers:
        return parser.parse_args()
    return 'submit - %s' % kwargs['description']

def cli_main(args):
    try:
        user = args.lp_user
    except AttributeError:
        user = None

    return main(
        directory=args.directory,
        force=args.force,
        target_user=args.target_user,
        target_branch=args.target_branch,
        no_push=args.no_push,
        reviewers=args.reviewer,
        user=user,
        branch=args.branch,
        proto=args.proto,
    )

def main(
    directory,
    force,
    target_user,
    target_branch,
    no_push,
    reviewers,
    user=None,
    branch=None,
    proto=top_level_defaults.proto,
):
    """Entry point to submit

    Arguments:
    @directory: string path to directory containing local repository
    @force: if True, force the merge instead of erroring
    @target_user: string Launchpad user whose repository is the target
    @target_branch: string branch name in the target repository
    @no_push: if True, do not push the source branch before proposing
    the MP
    @reviewers: list of string Launchpad usernames to use as reviewers
    of the MP
    @user: string user to authenticate to Launchpad as
    @branch: string branch to propose for merging
    @proto: string protocol to use (one of 'http', 'https', 'git')

    If user is None, value of `git config gitubuntu.lpuser` will be
    used.

    If branch is None, HEAD is used.

    Returns 0 if the MP was created successfully; 1 otherwise.
    """
    repo = GitUbuntuRepository(directory, user, proto)

    if branch:
        source_branch = 'refs/heads/%s' % branch
        if not repo.get_head_by_name(branch):
            logging.warning("Specified branch (%s) does not exist locally.",
                branch,
            )
        commitish_string = branch
    else:
        if repo.raw_repo.head_is_detached:
            logging.error("Please create a local branch before submitting.")
            return 1
        source_branch = repo.raw_repo.head.name
        commitish_string = 'HEAD'

    logging.debug("source branch: %s", source_branch)

    pkgname = repo.get_changelog_srcpkg_from_treeish(commitish_string)

    logging.debug("source package: %s", pkgname)

    if target_branch and target_user:
        target_head_string = target_branch
    else:
        if target_user == 'git-ubuntu-import':
            namespace = 'pkg'
        else:
            namespace = target_user
        target_head_string = derive_target_branch(
            repo,
            commitish_string,
            namespace,
        )
        if len(target_head_string) == 0:
            return 1

    logging.debug("target branch: %s", target_head_string)

    if not no_push:
        try:
            # the refspec is specific because we may not be on the source
            # branch
            repo.git_run(
                [
                    'push',
                    repo.lp_user,
                    '%s:%s' % (source_branch, source_branch),
                ]
            )
        except CalledProcessError:
            logging.error(
                "Unable to push %s to remote %s",
                source_branch,
                repo.lp_user,
            )
            return 1

    lp = launchpad_login_auth()

    # get target git_ref object
    path='~%s/ubuntu/+source/%s/+git/%s' % (
        target_user,
        pkgname,
        pkgname,
    )
    target_git_repo = lp.git_repositories.getByPath(path=path)
    if target_git_repo is None:
        logging.error(
            "Unable to find target repository (path=%s)",
            path,
        )
        return 1

    target_git_ref = target_git_repo.getRefByPath(
        path='refs/heads/%s' % target_head_string
    )
    logging.debug("target git ref: %s", target_git_ref)

    logging.debug("Deduced user: %s", repo.lp_user)
    # get source git_ref object
    path='~%s/ubuntu/+source/%s/+git/%s' % (
        repo.lp_user,
        pkgname,
        pkgname,
    )
    source_git_repo = lp.git_repositories.getByPath(path=path)
    if source_git_repo is None:
        logging.error(
            "Unable to find source repository (path=%s)",
            path,
        )
        return 1

    source_git_ref = source_git_repo.getRefByPath(path=source_branch)
    if source_git_ref == None:
        logging.error(
            "Unable to find branch %s in %s's repository. Is a `git "
            "push` needed or does the branch name need to be "
            "specified with --branch?",
            source_branch,
            repo.lp_user,
        )
        return 1
    logging.debug("Source git ref: %s", source_git_ref)

    # create MP
    try:
        mp = source_git_ref.createMergeProposal(merge_target=target_git_ref)
    except BadRequest as e:
        logging.error("Unable to create merge proposal: %s", e)
        return 1
    # only take unique reviewers
    for potential_reviewer in set(reviewers):
        try:
            mp.nominateReviewer(reviewer=lp.people[potential_reviewer])
        except KeyError:
            logging.error(
                "Unable to nominate reviewer %s: not found",
                potential_reviewer,
            )
        except:
            logging.error(
                "Unable to nominate reviewer %s: unknown error",
                potential_reviewer,
            )
    print("Your merge proposal is now available at: %s" % mp.web_link)
    print("If it looks ok, please move it to the 'Needs Review' state.")

    return 0