Completed
Push — master ( 56245b...a9d00e )
by
unknown
38s
created

clone()   F

Complexity

Conditions 11

Size

Total Lines 63

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 11
c 0
b 0
f 0
dl 0
loc 63
rs 3.9705

How to fix   Long Method    Complexity   

Long Method

Small methods make your code easier to understand, in particular if combined with a good name. Besides, if your method is small, finding a good name is usually much easier.

For example, if you find yourself adding comments to a method's body, this is usually a good sign to extract the commented part to a new method, and use the comment as a starting point when coming up with a good name for this new method.

Commonly applied refactorings include:

Complexity

Complex classes like clone() often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes.

Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.

1
# -*- coding: utf-8 -*-
2
3
"""Helper functions for working with version control systems."""
4
5
from __future__ import unicode_literals
6
import logging
7
import os
8
import subprocess
9
10
from whichcraft import which
11
12
from .exceptions import (
13
    RepositoryNotFound, RepositoryCloneFailed, UnknownRepoType, VCSNotInstalled
14
)
15
from .utils import make_sure_path_exists, prompt_and_delete
16
17
logger = logging.getLogger(__name__)
18
19
20
BRANCH_ERRORS = [
21
    'error: pathspec',
22
    'unknown revision',
23
]
24
25
26
def identify_repo(repo_url):
27
    """Determine if `repo_url` should be treated as a URL to a git or hg repo.
28
29
    Repos can be identified by prepending "hg+" or "git+" to the repo URL.
30
31
    :param repo_url: Repo URL of unknown type.
32
    :returns: ('git', repo_url), ('hg', repo_url), or None.
33
    """
34
    repo_url_values = repo_url.split('+')
35
    if len(repo_url_values) == 2:
36
        repo_type = repo_url_values[0]
37
        if repo_type in ["git", "hg"]:
38
            return repo_type, repo_url_values[1]
39
        else:
40
            raise UnknownRepoType
41
    else:
42
        if 'git' in repo_url:
43
            return 'git', repo_url
44
        elif 'bitbucket' in repo_url:
45
            return 'hg', repo_url
46
        else:
47
            raise UnknownRepoType
48
49
50
def is_vcs_installed(repo_type):
51
    """
52
    Check if the version control system for a repo type is installed.
53
54
    :param repo_type:
55
    """
56
    return bool(which(repo_type))
57
58
59
def clone(repo_url, checkout=None, clone_to_dir='.', no_input=False):
60
    """Clone a repo to the current directory.
61
62
    :param repo_url: Repo URL of unknown type.
63
    :param checkout: The branch, tag or commit ID to checkout after clone.
64
    :param clone_to_dir: The directory to clone to.
65
                         Defaults to the current directory.
66
    :param no_input: Suppress all user prompts when calling via API.
67
    """
68
    # Ensure that clone_to_dir exists
69
    clone_to_dir = os.path.expanduser(clone_to_dir)
70
    make_sure_path_exists(clone_to_dir)
71
72
    # identify the repo_type
73
    repo_type, repo_url = identify_repo(repo_url)
74
75
    # check that the appropriate VCS for the repo_type is installed
76
    if not is_vcs_installed(repo_type):
77
        msg = "'{0}' is not installed.".format(repo_type)
78
        raise VCSNotInstalled(msg)
79
80
    repo_url = repo_url.rstrip('/')
81
    tail = os.path.split(repo_url)[1]
82
    if repo_type == 'git':
83
        repo_dir = os.path.normpath(os.path.join(clone_to_dir,
84
                                                 tail.rsplit('.git')[0]))
85
    elif repo_type == 'hg':
86
        repo_dir = os.path.normpath(os.path.join(clone_to_dir, tail))
87
    logger.debug('repo_dir is {0}'.format(repo_dir))
88
89
    if os.path.isdir(repo_dir):
90
        clone = prompt_and_delete(repo_dir, no_input=no_input)
91
    else:
92
        clone = True
93
94
    if clone:
95
        try:
96
            subprocess.check_output(
97
                [repo_type, 'clone', repo_url],
98
                cwd=clone_to_dir,
99
                stderr=subprocess.STDOUT,
100
            )
101
            if checkout is not None:
102
                subprocess.check_output(
103
                    [repo_type, 'checkout', checkout],
104
                    cwd=repo_dir,
105
                    stderr=subprocess.STDOUT,
106
                )
107
        except subprocess.CalledProcessError as clone_error:
108
            output = clone_error.output.decode('utf-8')
109
            if 'not found' in output.lower():
110
                raise RepositoryNotFound(
111
                    'The repository {} could not be found, '
112
                    'have you made a typo?'.format(repo_url)
113
                )
114
            if any(error in output for error in BRANCH_ERRORS):
115
                raise RepositoryCloneFailed(
116
                    'The {} branch of repository {} could not found, '
117
                    'have you made a typo?'.format(checkout, repo_url)
118
                )
119
            raise
120
121
    return repo_dir
122