clone() - Code Metrics - Inspection of "Test valid_repository" - michaeljoseph/cookiecutter - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — ensure-repos-have-context ( bbbfa7...550348 )

by Michael

created 2016-07-28 17:36 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	61
rs	4.7368

How to fix Long Method Complexity

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

"""
cookiecutter.vcs
----------------

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


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


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

michaeljoseph / cookiecutter

Push — ensure-repos-have-context ( bbbfa7...550348 )

clone() D

Complexity

Size

Duplication

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like