gitman.models.source.Source.update_files() - Code Metrics - jacebrowning/gitman - Measure and Improve Code Quality continuously with Scrutinizer

gitman.models.source.Source.update_files() F
last analyzed 2020-08-20 15:07 UTC

↳ Parent: gitman.models.source

Complexity

Conditions

Size

Total Lines	79
Code Lines	54

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	44
CRAP Score	16.2716

Importance

Changes

Metric	Value
cc	16
eloc	54
nop	6
dl	0
loc	79
ccs	44
cts	49
cp	0.898
crap	16.2716
rs	2.4
c	0
b	0
f	0

How to fix Long Method Complexity

import os
from dataclasses import dataclass, field
from typing import List, Optional

import log

from .. import common, exceptions, git, shell


@dataclass
class Source:
    """A dictionary of `git` and `ln` arguments."""

    name: Optional[str]
    type: str
    repo: str
    sparse_paths: List[str] = field(default_factory=list)
    rev: str = 'master'
    link: Optional[str] = None
    scripts: List[str] = field(default_factory=list)

    DIRTY = '<dirty>'
    UNKNOWN = '<unknown>'

    def __post_init__(self):
        if self.name is None:
            self.name = self._infer_name(self.repo)
        self.type = self.type or 'git'

    def __repr__(self):
        return "<source {}>".format(self)

    def __str__(self):
        pattern = "['{t}'] '{r}' @ '{v}' in '{d}'"
        if self.link:
            pattern += " <- '{s}'"
        return pattern.format(
            t=self.type, r=self.repo, v=self.rev, d=self.name, s=self.link
        )

    def __eq__(self, other):
        return self.name == other.name

    def __ne__(self, other):
        return self.name != other.name

    def __lt__(self, other):
        return self.name < other.name

    def update_files(
        self,
        force=False,
        force_interactive=False,
        fetch=False,
        clean=True,
        skip_changes=False,
    ):
        """Ensure the source matches the specified revision."""
        log.info("Updating source files...")

        # Clone the repository if needed
        assert self.name
        if not os.path.exists(self.name):
            git.clone(
                self.type,
                self.repo,
                self.name,
                sparse_paths=self.sparse_paths,
                rev=self.rev,
            )

        # Enter the working tree
        shell.cd(self.name)
        if not git.valid():
            if force:
                git.rebuild(self.type, self.repo)
                fetch = True
            else:
                raise self._invalid_repository

        # Check for uncommitted changes
        if not force:
            log.debug("Confirming there are no uncommitted changes...")
            if skip_changes:
                if git.changes(
                    self.type, include_untracked=clean, display_status=False
                ):
                    common.show(
                        f'Skipped update due to uncommitted changes in {os.getcwd()}',
                        color='git_changes',
                    )
                    return
            elif force_interactive:
                if git.changes(
                    self.type, include_untracked=clean, display_status=False
                ):
                    common.show(
                        f'Uncommitted changes found in {os.getcwd()}',
                        color='git_changes',
                    )

                    while True:
                        yn_input = str(
                            input("Do you want to overwrite? (Y/N)[Y]: ")
                        ).rstrip('\r\n')

                        if yn_input.lower() == "y" or not yn_input:
                            break

                        if yn_input.lower() == "n":
                            common.show(
                                f'Skipped update in {os.getcwd()}', color='git_changes'
                            )
                            return

            else:
                if git.changes(self.type, include_untracked=clean):
                    raise exceptions.UncommittedChanges(
                        f'Uncommitted changes in {os.getcwd()}'
                    )

        # Fetch the desired revision
        if fetch or git.is_fetch_required(self.type, self.rev):
            git.fetch(self.type, self.repo, self.name, rev=self.rev)

        # Update the working tree to the desired revision
        git.update(
            self.type, self.repo, self.name, fetch=fetch, clean=clean, rev=self.rev
        )

    def create_link(self, root, force=False):
        """Create a link from the target name to the current directory."""
        if not self.link:
            return

        log.info("Creating a symbolic link...")

        target = os.path.join(root, self.link)
        source = os.path.relpath(os.getcwd(), os.path.dirname(target))

        if os.path.islink(target):
            os.remove(target)
        elif os.path.exists(target):
            if force:
                shell.rm(target)
            else:
                msg = "Preexisting link location at {}".format(target)
                raise exceptions.UncommittedChanges(msg)

        shell.ln(source, target)

    def run_scripts(self, force=False, show_shell_stdout=False):
        log.info("Running install scripts...")

        # Enter the working tree
        shell.cd(self.name)
        if not git.valid():
            raise self._invalid_repository

        # Check for scripts
        if not self.scripts or not self.scripts[0]:
            common.show("(no scripts to run)", color='shell_info')
            common.newline()
            return

        # Run all scripts
        for script in self.scripts:
            try:
                shell.call(script, _shell=True, _show_stdout=show_shell_stdout)
            except exceptions.ShellError as exc:
                common.show(*exc.output, color='shell_error')
                cmd = exc.program
                if force:
                    log.debug("Ignored error from call to '%s'", cmd)
                else:
                    msg = "Command '{}' failed in {}".format(cmd, os.getcwd())
                    raise exceptions.ScriptFailure(msg)
        common.newline()

    def identify(self, allow_dirty=True, allow_missing=True, skip_changes=False):
        """Get the path and current repository URL and hash."""
        assert self.name
        if os.path.isdir(self.name):

            shell.cd(self.name)
            if not git.valid():
                raise self._invalid_repository

            path = os.getcwd()
            url = git.get_url(self.type)
            if git.changes(
                self.type,
                display_status=not allow_dirty and not skip_changes,
                _show=not skip_changes,
            ):

                if allow_dirty:
                    common.show(self.DIRTY, color='git_dirty', log=False)
                    common.newline()
                    return path, url, self.DIRTY

                if skip_changes:
                    msg = ("Skipped lock due to uncommitted changes " "in {}").format(
                        os.getcwd()
                    )
                    common.show(msg, color='git_changes')
                    common.newline()
                    return path, url, self.DIRTY

                msg = "Uncommitted changes in {}".format(os.getcwd())
                raise exceptions.UncommittedChanges(msg)

            rev = git.get_hash(self.type, _show=True)
            common.show(rev, color='git_rev', log=False)
            common.newline()
            return path, url, rev

        if allow_missing:
            return os.getcwd(), '<missing>', self.UNKNOWN

        raise self._invalid_repository

    def lock(self, rev=None, allow_dirty=False, skip_changes=False):
        """Create a locked source object.

        Return a locked version of the current source if not dirty
        otherwise None.
        """

        if rev is None:
            _, _, rev = self.identify(
                allow_dirty=allow_dirty, allow_missing=False, skip_changes=skip_changes
            )

        if rev == self.DIRTY:
            return None

        source = self.__class__(
            type=self.type,
            repo=self.repo,
            name=self.name,
            rev=rev,
            link=self.link,
            scripts=self.scripts,
            sparse_paths=self.sparse_paths,
        )
        return source

    @property
    def _invalid_repository(self):
        assert self.name
        path = os.path.join(os.getcwd(), self.name)
        msg = """

            Not a valid repository: {}
            During install you can rebuild a repo with a missing .git directory using the --force option
            """.format(
            path
        )
        return exceptions.InvalidRepository(msg)

    @staticmethod
    def _infer_name(repo):
        filename = repo.split('/')[-1]
        name = filename.split('.')[0]
        return name


1	1	import os
2	1	from dataclasses import dataclass, field
3	1	from typing import List, Optional
4
5	1	import log
6	1
7		from .. import common, exceptions, git, shell
8	1
9
10		@dataclass
11	1	class Source:
12		"""A dictionary of `git` and `ln` arguments."""
13
14	1	name: Optional[str]
15	1	type: str
16	1	repo: str
17	1	sparse_paths: List[str] = field(default_factory=list)
18	1	rev: str = 'master'
19	1	link: Optional[str] = None
20		scripts: List[str] = field(default_factory=list)
21
22	1	DIRTY = '<dirty>'
23	1	UNKNOWN = '<unknown>'
24
25	1	def __post_init__(self):
26	1	if self.name is None:
27	1	self.name = self._infer_name(self.repo)
28	1	self.type = self.type or 'git'
29	1
30	1	def __repr__(self):
31	1	return "<source {}>".format(self)
32
33	1	def __str__(self):
34	1	pattern = "['{t}'] '{r}' @ '{v}' in '{d}'"
35	1	if self.link:
36	1	pattern += " <- '{s}'"
37		return pattern.format(
38	1	t=self.type, r=self.repo, v=self.rev, d=self.name, s=self.link
39	1	)
40
41	1	def __eq__(self, other):
42	1	return self.name == other.name
43	1
44	1	def __ne__(self, other):
45	1	return self.name != other.name
46
47	1	def __lt__(self, other):
48	1	return self.name < other.name
49
50	1	def update_files(
51	1	self,
52		force=False,
53	1	force_interactive=False,
54	1	fetch=False,
55		clean=True,
56	1	skip_changes=False,
57		):
58	1	"""Ensure the source matches the specified revision."""
59		log.info("Updating source files...")
60
61	1	# Clone the repository if needed
62	1	assert self.name
63		if not os.path.exists(self.name):
64		git.clone(
65	1	self.type,
66	1	self.repo,
67	1	self.name,
68		sparse_paths=self.sparse_paths,
69		rev=self.rev,
70	1	)
71	1
72	1	# Enter the working tree
73		shell.cd(self.name)
74		if not git.valid():
75		if force:
76		git.rebuild(self.type, self.repo)
77	1	fetch = True
78		else:
79		raise self._invalid_repository
80	1
81		# Check for uncommitted changes
82		if not force:
83	1	log.debug("Confirming there are no uncommitted changes...")
84		if skip_changes:
85	1	if git.changes(
86		self.type, include_untracked=clean, display_status=False
87	1	):
88	1	common.show(
89		f'Skipped update due to uncommitted changes in {os.getcwd()}',
90	1	color='git_changes',
91		)
92	1	return
93		elif force_interactive:
94		if git.changes(
95		self.type, include_untracked=clean, display_status=False
96	1	):
97	1	common.show(
98		f'Uncommitted changes found in {os.getcwd()}',
99	1	color='git_changes',
100	1	)
101	1
102	1	while True:
103	1	yn_input = str(
104		input("Do you want to overwrite? (Y/N)[Y]: ")
105	1	).rstrip('\r\n')
106	1
107		if yn_input.lower() == "y" or not yn_input:
108	1	break
109
110	1	if yn_input.lower() == "n":
111	1	common.show(
112		f'Skipped update in {os.getcwd()}', color='git_changes'
113		)
114	1	return
115	1
116		else:
117		if git.changes(self.type, include_untracked=clean):
118		raise exceptions.UncommittedChanges(
119	1	f'Uncommitted changes in {os.getcwd()}'
120	1	)
121	1
122	1	# Fetch the desired revision
123		if fetch or git.is_fetch_required(self.type, self.rev):
124		git.fetch(self.type, self.repo, self.name, rev=self.rev)
125	1
126	1	# Update the working tree to the desired revision
127	1	git.update(
128	1	self.type, self.repo, self.name, fetch=fetch, clean=clean, rev=self.rev
129	1	)
130	1
131	1	def create_link(self, root, force=False):
132	1	"""Create a link from the target name to the current directory."""
133		if not self.link:
134	1	return
135	1
136		log.info("Creating a symbolic link...")
137
138	1	target = os.path.join(root, self.link)
139		source = os.path.relpath(os.getcwd(), os.path.dirname(target))
140	1
141		if os.path.islink(target):
142	1	os.remove(target)
143		elif os.path.exists(target):
144	1	if force:
145	1	shell.rm(target)
146	1	else:
147		msg = "Preexisting link location at {}".format(target)
148	1	raise exceptions.UncommittedChanges(msg)
149	1
150	1	shell.ln(source, target)
151	1
152	1	def run_scripts(self, force=False, show_shell_stdout=False):
153	1	log.info("Running install scripts...")
154
155		# Enter the working tree
156		shell.cd(self.name)
157		if not git.valid():
158		raise self._invalid_repository
159	1
160	1	# Check for scripts
161	1	if not self.scripts or not self.scripts[0]:
162	1	common.show("(no scripts to run)", color='shell_info')
163		common.newline()
164	1	return
165
166	1	# Run all scripts
167		for script in self.scripts:
168		try:
169		shell.call(script, _shell=True, _show_stdout=show_shell_stdout)
170	1	except exceptions.ShellError as exc:
171		common.show(*exc.output, color='shell_error')
172	1	cmd = exc.program
173		if force:
174	1	log.debug("Ignored error from call to '%s'", cmd)
175	1	else:
176	1	msg = "Command '{}' failed in {}".format(cmd, os.getcwd())
177		raise exceptions.ScriptFailure(msg)
178	1	common.newline()
179
180	1	def identify(self, allow_dirty=True, allow_missing=True, skip_changes=False):
181		"""Get the path and current repository URL and hash."""
182	1	assert self.name
183	1	if os.path.isdir(self.name):
184	1
185		shell.cd(self.name)
186	1	if not git.valid():
187		raise self._invalid_repository
188	1
189	1	path = os.getcwd()
190	1	url = git.get_url(self.type)
191		if git.changes(
192		self.type,
193		display_status=not allow_dirty and not skip_changes,
194		_show=not skip_changes,
195		):
196
197		if allow_dirty:
198		common.show(self.DIRTY, color='git_dirty', log=False)
199		common.newline()
200		return path, url, self.DIRTY
201
202		if skip_changes:
203		msg = ("Skipped lock due to uncommitted changes " "in {}").format(
204		os.getcwd()
205		)
206		common.show(msg, color='git_changes')
207		common.newline()
208		return path, url, self.DIRTY
209
210		msg = "Uncommitted changes in {}".format(os.getcwd())
211		raise exceptions.UncommittedChanges(msg)
212
213		rev = git.get_hash(self.type, _show=True)
214		common.show(rev, color='git_rev', log=False)
215		common.newline()
216		return path, url, rev
217
218		if allow_missing:
219		return os.getcwd(), '<missing>', self.UNKNOWN
220
221		raise self._invalid_repository
222
223		def lock(self, rev=None, allow_dirty=False, skip_changes=False):
224		"""Create a locked source object.
225
226		Return a locked version of the current source if not dirty
227		otherwise None.
228		"""
229
230		if rev is None:
231		_, _, rev = self.identify(
232		allow_dirty=allow_dirty, allow_missing=False, skip_changes=skip_changes
233		)
234
235		if rev == self.DIRTY:
236		return None
237
238		source = self.__class__(
239		type=self.type,
240		repo=self.repo,
241		name=self.name,
242		rev=rev,
243		link=self.link,
244		scripts=self.scripts,
245		sparse_paths=self.sparse_paths,
246		)
247		return source
248
249		@property
250		def _invalid_repository(self):
251		assert self.name
252		path = os.path.join(os.getcwd(), self.name)
253		msg = """
254
255		Not a valid repository: {}
256		During install you can rebuild a repo with a missing .git directory using the --force option
257		""".format(
258		path
259		)
260		return exceptions.InvalidRepository(msg)
261
262		@staticmethod
263		def _infer_name(repo):
264		filename = repo.split('/')[-1]
265		name = filename.split('.')[0]
266		return name
267

jacebrowning / gitman

gitman.models.source.Source.update_files() F last analyzed 2020-08-20 15:07 UTC

Complexity

Size

Duplication

Code Coverage

Importance

How to fix Long Method Complexity

Long Method

Complexity

Duplication Side-by-Side

Filter issues like

gitman.models.source.Source.update_files() F
last analyzed 2020-08-20 15:07 UTC