clone() - Code Metrics - Inspection of "Update HISTORY.rst for #781" - michaeljoseph/cookiecutter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 4d3455...bf0e43 )

by Michael

created 2016-12-31 18:58 UTC

clone() D

↳ Parent: Project

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Importance

Changes	3
Bugs	0	Features	0

Metric	Value
cc	10
c	3
b	0
f	0
dl	0
loc	59
rs	4.8648

How to fix Long Method Complexity

# -*- coding: utf-8 -*-

"""Helper functions for working with version control systems."""

from __future__ import unicode_literals
import logging
import os
import subprocess
import sys

from whichcraft import which

from .exceptions import (
    RepositoryNotFound, RepositoryCloneFailed, UnknownRepoType, VCSNotInstalled
)
from .prompt import read_user_yes_no
from .utils import make_sure_path_exists, rmtree

logger = logging.getLogger(__name__)


BRANCH_ERRORS = [
    'error: pathspec',
    'unknown revision',
]


def prompt_and_delete_repo(repo_dir, no_input=False):
    """Ask the user whether it's okay to delete the previously-cloned repo.

    If yes, deletes it. Otherwise, Cookiecutter exits.

    :param repo_dir: Directory of previously-cloned repo.
    :param no_input: Suppress prompt to delete repo and just delete it.
    """
    # Suppress prompt if called via API
    if no_input:
        ok_to_delete = True
    else:
        question = (
            "You've cloned {} before. "
            "Is it okay to delete and re-clone it?"
        ).format(repo_dir)

        ok_to_delete = read_user_yes_no(question, 'yes')

    if ok_to_delete:
        rmtree(repo_dir)
    else:
        sys.exit()


def identify_repo(repo_url):
    """Determine if `repo_url` should be treated as a URL to a git or hg repo.

    Repos can be identified by prepending "hg+" or "git+" to the repo URL.

    :param repo_url: Repo URL of unknown type.
    :returns: ('git', repo_url), ('hg', repo_url), or None.
    """
    repo_url_values = repo_url.split('+')
    if len(repo_url_values) == 2:
        repo_type = repo_url_values[0]
        if repo_type in ["git", "hg"]:
            return repo_type, repo_url_values[1]
        else:
            raise UnknownRepoType
    else:
        if 'git' in repo_url:
            return 'git', repo_url
        elif 'bitbucket' in repo_url:
            return 'hg', repo_url
        else:
            raise UnknownRepoType


def is_vcs_installed(repo_type):
    """
    Check if the version control system for a repo type is installed.

    :param repo_type:
    """
    return bool(which(repo_type))


def clone(repo_url, checkout=None, clone_to_dir='.', no_input=False):
    """Clone a repo to the current directory.

    :param repo_url: Repo URL of unknown type.
    :param checkout: The branch, tag or commit ID to checkout after clone.
    :param clone_to_dir: The directory to clone to.
                         Defaults to the current directory.
    :param no_input: Suppress all user prompts when calling via API.
    """
    # Ensure that clone_to_dir exists
    clone_to_dir = os.path.expanduser(clone_to_dir)
    make_sure_path_exists(clone_to_dir)

    # identify the repo_type
    repo_type, repo_url = identify_repo(repo_url)

    # check that the appropriate VCS for the repo_type is installed
    if not is_vcs_installed(repo_type):
        msg = "'{0}' is not installed.".format(repo_type)
        raise VCSNotInstalled(msg)

    repo_url = repo_url.rstrip('/')
    tail = os.path.split(repo_url)[1]
    if repo_type == 'git':
        repo_dir = os.path.normpath(os.path.join(clone_to_dir,
                                                 tail.rsplit('.git')[0]))
    elif repo_type == 'hg':
        repo_dir = os.path.normpath(os.path.join(clone_to_dir, tail))
    logger.debug('repo_dir is {0}'.format(repo_dir))

    if os.path.isdir(repo_dir):
        prompt_and_delete_repo(repo_dir, no_input=no_input)

    try:
        subprocess.check_output(
            [repo_type, 'clone', repo_url],
            cwd=clone_to_dir,
            stderr=subprocess.STDOUT,
        )
        if checkout is not None:
            subprocess.check_output(
                [repo_type, 'checkout', checkout],
                cwd=repo_dir,
                stderr=subprocess.STDOUT,
            )
    except subprocess.CalledProcessError as clone_error:
        if 'not found' in clone_error.output.lower():
            raise RepositoryNotFound(
                'The repository {} could not be found, '
                'have you made a typo?'.format(repo_url)
            )
        if any(error in clone_error.output for error in BRANCH_ERRORS):
            raise RepositoryCloneFailed(
                'The {} branch of repository {} could not found, '
                'have you made a typo?'.format(checkout, repo_url)
            )
        raise

    return repo_dir


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			import sys
10
11			from whichcraft import which
12
13			from .exceptions import (
14			RepositoryNotFound, RepositoryCloneFailed, UnknownRepoType, VCSNotInstalled
15			)
16			from .prompt import read_user_yes_no
17			from .utils import make_sure_path_exists, rmtree
18
19			logger = logging.getLogger(__name__)
20
21
22			BRANCH_ERRORS = [
23			'error: pathspec',
24			'unknown revision',
25			]
26
27
28			def prompt_and_delete_repo(repo_dir, no_input=False):
29			"""Ask the user whether it's okay to delete the previously-cloned repo.
30
31			If yes, deletes it. Otherwise, Cookiecutter exits.
32
33			:param repo_dir: Directory of previously-cloned repo.
34			:param no_input: Suppress prompt to delete repo and just delete it.
35			"""
36			# Suppress prompt if called via API
37			if no_input:
38			ok_to_delete = True
39			else:
40			question = (
41			"You've cloned {} before. "
42			"Is it okay to delete and re-clone it?"
43			).format(repo_dir)
44
45			ok_to_delete = read_user_yes_no(question, 'yes')
46
47			if ok_to_delete:
48			rmtree(repo_dir)
49			else:
50			sys.exit()
51
52
53			def identify_repo(repo_url):
54			"""Determine if `repo_url` should be treated as a URL to a git or hg repo.
55
56			Repos can be identified by prepending "hg+" or "git+" to the repo URL.
57
58			:param repo_url: Repo URL of unknown type.
59			:returns: ('git', repo_url), ('hg', repo_url), or None.
60			"""
61			repo_url_values = repo_url.split('+')
62			if len(repo_url_values) == 2:
63			repo_type = repo_url_values[0]
64			if repo_type in ["git", "hg"]:
65			return repo_type, repo_url_values[1]
66			else:
67			raise UnknownRepoType
68			else:
69			if 'git' in repo_url:
70			return 'git', repo_url
71			elif 'bitbucket' in repo_url:
72			return 'hg', repo_url
73			else:
74			raise UnknownRepoType
75
76
77			def is_vcs_installed(repo_type):
78			"""
79			Check if the version control system for a repo type is installed.
80
81			:param repo_type:
82			"""
83			return bool(which(repo_type))
84
85
86			def clone(repo_url, checkout=None, clone_to_dir='.', no_input=False):
87			"""Clone a repo to the current directory.
88
89			:param repo_url: Repo URL of unknown type.
90			:param checkout: The branch, tag or commit ID to checkout after clone.
91			:param clone_to_dir: The directory to clone to.
92			Defaults to the current directory.
93			:param no_input: Suppress all user prompts when calling via API.
94			"""
95			# Ensure that clone_to_dir exists
96			clone_to_dir = os.path.expanduser(clone_to_dir)
97			make_sure_path_exists(clone_to_dir)
98
99			# identify the repo_type
100			repo_type, repo_url = identify_repo(repo_url)
101
102			# check that the appropriate VCS for the repo_type is installed
103			if not is_vcs_installed(repo_type):
104			msg = "'{0}' is not installed.".format(repo_type)
105			raise VCSNotInstalled(msg)
106
107			repo_url = repo_url.rstrip('/')
108			tail = os.path.split(repo_url)[1]
109			if repo_type == 'git':
110			repo_dir = os.path.normpath(os.path.join(clone_to_dir,
111			tail.rsplit('.git')[0]))
112			elif repo_type == 'hg':
113			repo_dir = os.path.normpath(os.path.join(clone_to_dir, tail))
114			logger.debug('repo_dir is {0}'.format(repo_dir))
115
116			if os.path.isdir(repo_dir):
117			prompt_and_delete_repo(repo_dir, no_input=no_input)
118
119			try:
120			subprocess.check_output(
121			[repo_type, 'clone', repo_url],
122			cwd=clone_to_dir,
123			stderr=subprocess.STDOUT,
124			)
125			if checkout is not None:
126			subprocess.check_output(
127			[repo_type, 'checkout', checkout],
128			cwd=repo_dir,
129			stderr=subprocess.STDOUT,
130			)
131			except subprocess.CalledProcessError as clone_error:
132			if 'not found' in clone_error.output.lower():
133			raise RepositoryNotFound(
134			'The repository {} could not be found, '
135			'have you made a typo?'.format(repo_url)
136			)
137			if any(error in clone_error.output for error in BRANCH_ERRORS):
138			raise RepositoryCloneFailed(
139			'The {} branch of repository {} could not found, '
140			'have you made a typo?'.format(checkout, repo_url)
141			)
142			raise
143
144			return repo_dir
145

michaeljoseph / cookiecutter

Push — master ( 4d3455...bf0e43 )

clone() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like