FreeCAD

1
# SPDX-License-Identifier: LGPL-2.1-or-later
2
# ***************************************************************************
3
# *                                                                         *
4
# *   Copyright (c) 2022 FreeCAD Project Association                        *
5
# *                                                                         *
6
# *   This file is part of FreeCAD.                                         *
7
# *                                                                         *
8
# *   FreeCAD is free software: you can redistribute it and/or modify it    *
9
# *   under the terms of the GNU Lesser General Public License as           *
10
# *   published by the Free Software Foundation, either version 2.1 of the  *
11
# *   License, or (at your option) any later version.                       *
12
# *                                                                         *
13
# *   FreeCAD is distributed in the hope that it will be useful, but        *
14
# *   WITHOUT ANY WARRANTY; without even the implied warranty of            *
15
# *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU      *
16
# *   Lesser General Public License for more details.                       *
17
# *                                                                         *
18
# *   You should have received a copy of the GNU Lesser General Public      *
19
# *   License along with FreeCAD. If not, see                               *
20
# *   <https://www.gnu.org/licenses/>.                                      *
21
# *                                                                         *
22
# ***************************************************************************
23

24
""" Wrapper around git executable to simplify calling git commands from Python. """
25

26
# pylint: disable=too-few-public-methods
27

28
import os
29
import platform
30
import shutil
31
import subprocess
32
from typing import List, Dict, Optional
33
import time
34

35
import addonmanager_utilities as utils
36
import addonmanager_freecad_interface as fci
37

38
translate = fci.translate
39

40

41
class NoGitFound(RuntimeError):
42
    """Could not locate the git executable on this system."""
43

44

45
class GitFailed(RuntimeError):
46
    """The call to git returned an error of some kind"""
47

48

49
def _ref_format_string() -> str:
50
    return (
51
        "--format=%(refname:lstrip=2)\t%(upstream:lstrip=2)\t%(authordate:rfc)\t%("
52
        "authorname)\t%(subject)"
53
    )
54

55

56
def _parse_ref_table(text: str):
57
    rows = text.splitlines()
58
    result = []
59
    for row in rows:
60
        columns = row.split("\t")
61
        result.append(
62
            {
63
                "ref_name": columns[0],
64
                "upstream": columns[1],
65
                "date": columns[2],
66
                "author": columns[3],
67
                "subject": columns[4],
68
            }
69
        )
70
    return result
71

72

73
class GitManager:
74
    """A class to manage access to git: mostly just provides a simple wrapper around
75
    the basic command-line calls. Provides optional asynchronous access to clone and
76
    update."""
77

78
    def __init__(self):
79
        self.git_exe = None
80
        self._find_git()
81
        if not self.git_exe:
82
            raise NoGitFound()
83

84
    def clone(self, remote, local_path, args: List[str] = None):
85
        """Clones the remote to the local path"""
86
        final_args = ["clone", "--recurse-submodules"]
87
        if args:
88
            final_args.extend(args)
89
        final_args.extend([remote, local_path])
90
        self._synchronous_call_git(final_args)
91

92
    def async_clone(self, remote, local_path, progress_monitor, args: List[str] = None):
93
        """Clones the remote to the local path, sending periodic progress updates
94
        to the passed progress_monitor. Returns a handle that can be used to
95
        cancel the job."""
96

97
    def checkout(self, local_path, spec, args: List[str] = None):
98
        """Checks out a specific git revision, tag, or branch. Any valid argument to
99
        git checkout can be submitted."""
100
        old_dir = os.getcwd()
101
        os.chdir(local_path)
102
        final_args = ["checkout"]
103
        if args:
104
            final_args.extend(args)
105
        final_args.append(spec)
106
        self._synchronous_call_git(final_args)
107
        os.chdir(old_dir)
108

109
    def dirty(self, local_path: str) -> bool:
110
        """Check for local changes"""
111
        old_dir = os.getcwd()
112
        os.chdir(local_path)
113
        result = False
114
        final_args = ["diff-index", "HEAD"]
115
        try:
116
            stdout = self._synchronous_call_git(final_args)
117
            if stdout:
118
                result = True
119
        except GitFailed:
120
            result = False
121
        os.chdir(old_dir)
122
        return result
123

124
    def detached_head(self, local_path: str) -> bool:
125
        """Check for detached head state"""
126
        old_dir = os.getcwd()
127
        os.chdir(local_path)
128
        result = False
129
        final_args = ["rev-parse", "--abbrev-ref", "--symbolic-full-name", "HEAD"]
130
        try:
131
            stdout = self._synchronous_call_git(final_args)
132
            if stdout == "HEAD":
133
                result = True
134
        except GitFailed:
135
            result = False
136
        os.chdir(old_dir)
137
        return result
138

139
    def update(self, local_path):
140
        """Fetches and pulls the local_path from its remote"""
141
        old_dir = os.getcwd()
142
        os.chdir(local_path)
143
        try:
144
            self._synchronous_call_git(["fetch"])
145
            self._synchronous_call_git(["pull"])
146
            self._synchronous_call_git(["submodule", "update", "--init", "--recursive"])
147
        except GitFailed as e:
148
            fci.Console.PrintWarning(
149
                translate(
150
                    "AddonsInstaller",
151
                    "Basic git update failed with the following message:",
152
                )
153
                + str(e)
154
                + "\n"
155
            )
156
            fci.Console.PrintWarning(
157
                translate(
158
                    "AddonsInstaller",
159
                    "Backing up the original directory and re-cloning",
160
                )
161
                + "...\n"
162
            )
163
            remote = self.get_remote(local_path)
164
            with open(os.path.join(local_path, "ADDON_DISABLED"), "w", encoding="utf-8") as f:
165
                f.write(
166
                    "This is a backup of an addon that failed to update cleanly so "
167
                    "was re-cloned. It was disabled by the Addon Manager's git update "
168
                    "facility and can be safely deleted if the addon is working "
169
                    "properly."
170
                )
171
            os.chdir("..")
172
            os.rename(local_path, local_path + ".backup" + str(time.time()))
173
            self.clone(remote, local_path)
174
        os.chdir(old_dir)
175

176
    def status(self, local_path) -> str:
177
        """Gets the v1 porcelain status"""
178
        old_dir = os.getcwd()
179
        os.chdir(local_path)
180
        try:
181
            status = self._synchronous_call_git(["status", "-sb", "--porcelain"])
182
        except GitFailed as e:
183
            os.chdir(old_dir)
184
            raise e
185

186
        os.chdir(old_dir)
187
        return status
188

189
    def reset(self, local_path, args: List[str] = None):
190
        """Executes the git reset command"""
191
        old_dir = os.getcwd()
192
        os.chdir(local_path)
193
        final_args = ["reset"]
194
        if args:
195
            final_args.extend(args)
196
        try:
197
            self._synchronous_call_git(final_args)
198
        except GitFailed as e:
199
            os.chdir(old_dir)
200
            raise e
201
        os.chdir(old_dir)
202

203
    def async_fetch_and_update(self, local_path, progress_monitor, args=None):
204
        """Same as fetch_and_update, but asynchronous"""
205

206
    def update_available(self, local_path) -> bool:
207
        """Returns True if an update is available from the remote, or false if not"""
208
        old_dir = os.getcwd()
209
        os.chdir(local_path)
210
        try:
211
            self._synchronous_call_git(["fetch"])
212
            status = self._synchronous_call_git(["status", "-sb", "--porcelain"])
213
        except GitFailed as e:
214
            os.chdir(old_dir)
215
            raise e
216
        os.chdir(old_dir)
217
        return "behind" in status
218

219
    def current_tag(self, local_path) -> str:
220
        """Get the name of the currently checked-out tag if HEAD is detached"""
221
        old_dir = os.getcwd()
222
        os.chdir(local_path)
223
        try:
224
            tag = self._synchronous_call_git(["describe", "--tags"]).strip()
225
        except GitFailed as e:
226
            os.chdir(old_dir)
227
            raise e
228
        os.chdir(old_dir)
229
        return tag
230

231
    def current_branch(self, local_path) -> str:
232
        """Get the name of the current branch"""
233
        old_dir = os.getcwd()
234
        os.chdir(local_path)
235
        try:
236
            # This only works with git 2.22 and later (June 2019)
237
            # branch = self._synchronous_call_git(["branch", "--show-current"]).strip()
238

239
            # This is more universal (albeit more opaque to the reader):
240
            branch = self._synchronous_call_git(["rev-parse", "--abbrev-ref", "HEAD"]).strip()
241
        except GitFailed as e:
242
            os.chdir(old_dir)
243
            raise e
244
        os.chdir(old_dir)
245
        return branch
246

247
    def repair(self, remote, local_path):
248
        """Assumes that local_path is supposed to be a local clone of the given
249
        remote, and ensures that it is. Note that any local changes in local_path
250
        will be destroyed. This is achieved by archiving the old path, cloning an
251
        entirely new copy, and then deleting the old directory."""
252

253
        original_cwd = os.getcwd()
254

255
        # Make sure we are not currently in that directory, otherwise on Windows the
256
        # "rename" will fail. To guarantee we aren't in it, change to it, then shift
257
        # up one.
258
        os.chdir(local_path)
259
        os.chdir("..")
260
        backup_path = local_path + ".backup" + str(time.time())
261
        os.rename(local_path, backup_path)
262
        try:
263
            self.clone(remote, local_path)
264
        except GitFailed as e:
265
            fci.Console.PrintError(
266
                translate("AddonsInstaller", "Failed to clone {} into {} using git").format(
267
                    remote, local_path
268
                )
269
            )
270
            os.chdir(original_cwd)
271
            raise e
272
        os.chdir(original_cwd)
273
        shutil.rmtree(backup_path, ignore_errors=True)
274

275
    def get_remote(self, local_path) -> str:
276
        """Get the repository that this local path is set to fetch from"""
277
        old_dir = os.getcwd()
278
        os.chdir(local_path)
279
        try:
280
            response = self._synchronous_call_git(["remote", "-v", "show"])
281
        except GitFailed as e:
282
            os.chdir(old_dir)
283
            raise e
284
        lines = response.split("\n")
285
        result = "(unknown remote)"
286
        for line in lines:
287
            if line.endswith("(fetch)"):
288
                # The line looks like:
289
                # origin  https://some/sort/of/path (fetch)
290

291
                segments = line.split()
292
                if len(segments) == 3:
293
                    result = segments[1]
294
                    break
295
                fci.Console.PrintWarning("Error parsing the results from git remote -v show:\n")
296
                fci.Console.PrintWarning(line + "\n")
297
        os.chdir(old_dir)
298
        return result
299

300
    def get_branches(self, local_path) -> List[str]:
301
        """Get a list of all available branches (local and remote)"""
302
        old_dir = os.getcwd()
303
        os.chdir(local_path)
304
        try:
305
            stdout = self._synchronous_call_git(["branch", "-a", "--format=%(refname:lstrip=2)"])
306
        except GitFailed as e:
307
            os.chdir(old_dir)
308
            raise e
309
        os.chdir(old_dir)
310
        branches = []
311
        for branch in stdout.split("\n"):
312
            branches.append(branch)
313
        return branches
314

315
    def get_branches_with_info(self, local_path) -> List[Dict[str, str]]:
316
        """Get a list of branches, where each entry is a dictionary with status information about
317
        the branch."""
318
        old_dir = os.getcwd()
319
        os.chdir(local_path)
320
        try:
321
            stdout = self._synchronous_call_git(["branch", "-a", _ref_format_string()])
322
            return _parse_ref_table(stdout)
323
        except GitFailed as e:
324
            os.chdir(old_dir)
325
            raise e
326

327
    def get_tags_with_info(self, local_path) -> List[Dict[str, str]]:
328
        """Get a list of branches, where each entry is a dictionary with status information about
329
        the branch."""
330
        old_dir = os.getcwd()
331
        os.chdir(local_path)
332
        try:
333
            stdout = self._synchronous_call_git(["tag", "-l", _ref_format_string()])
334
            return _parse_ref_table(stdout)
335
        except GitFailed as e:
336
            os.chdir(old_dir)
337
            raise e
338

339
    def get_last_committers(self, local_path, n=10):
340
        """Examine the last n entries of the commit history, and return a list of all
341
        the committers, their email addresses, and how many commits each one is
342
        responsible for.
343
        """
344
        old_dir = os.getcwd()
345
        os.chdir(local_path)
346
        authors = self._synchronous_call_git(["log", f"-{n}", "--format=%cN"]).split("\n")
347
        emails = self._synchronous_call_git(["log", f"-{n}", "--format=%cE"]).split("\n")
348
        os.chdir(old_dir)
349

350
        result_dict = {}
351
        for author, email in zip(authors, emails):
352
            if not author or not email:
353
                continue
354
            if author not in result_dict:
355
                result_dict[author] = {}
356
                result_dict[author]["email"] = [email]
357
                result_dict[author]["count"] = 1
358
            else:
359
                if email not in result_dict[author]["email"]:
360
                    # Same author name, new email address -- treat it as the same
361
                    # person with a second email, instead of as a whole new person
362
                    result_dict[author]["email"].append(email)
363
                result_dict[author]["count"] += 1
364
        return result_dict
365

366
    def get_last_authors(self, local_path, n=10):
367
        """Examine the last n entries of the commit history, and return a list of all
368
        the authors, their email addresses, and how many commits each one is
369
        responsible for.
370
        """
371
        old_dir = os.getcwd()
372
        os.chdir(local_path)
373
        authors = self._synchronous_call_git(["log", f"-{n}", "--format=%aN"])
374
        emails = self._synchronous_call_git(["log", f"-{n}", "--format=%aE"])
375
        os.chdir(old_dir)
376

377
        result_dict = {}
378
        for author, email in zip(authors, emails):
379
            if author not in result_dict:
380
                result_dict[author]["email"] = [email]
381
                result_dict[author]["count"] = 1
382
            else:
383
                if email not in result_dict[author]["email"]:
384
                    # Same author name, new email address -- treat it as the same
385
                    # person with a second email, instead of as a whole new person
386
                    result_dict[author]["email"].append(email)
387
                result_dict[author]["count"] += 1
388
        return result_dict
389

390
    def migrate_branch(self, local_path: str, old_branch: str, new_branch: str) -> None:
391
        """Rename a branch (used when the remote branch name changed). Assumes that "origin"
392
        exists."""
393
        old_dir = os.getcwd()
394
        os.chdir(local_path)
395
        try:
396
            self._synchronous_call_git(["branch", "-m", old_branch, new_branch])
397
            self._synchronous_call_git(["fetch", "origin"])
398
            self._synchronous_call_git(["branch", "--unset-upstream"])
399
            self._synchronous_call_git(["branch", f"--set-upstream-to=origin/{new_branch}"])
400
            self._synchronous_call_git(["pull"])
401
        except GitFailed as e:
402
            fci.Console.PrintWarning(
403
                translate(
404
                    "AddonsInstaller",
405
                    "Git branch rename failed with the following message:",
406
                )
407
                + str(e)
408
                + "\n"
409
            )
410
            os.chdir(old_dir)
411
            raise e
412
        os.chdir(old_dir)
413

414
    def _find_git(self):
415
        # Find git. In preference order
416
        #   A) The value of the GitExecutable user preference
417
        #   B) The executable located in the same directory as FreeCAD and called "git"
418
        #   C) The result of a shutil search for your system's "git" executable
419
        prefs = fci.ParamGet("User parameter:BaseApp/Preferences/Addons")
420
        git_exe = prefs.GetString("GitExecutable", "Not set")
421
        if not git_exe or git_exe == "Not set" or not os.path.exists(git_exe):
422
            fc_dir = fci.DataPaths().home_dir
423
            git_exe = os.path.join(fc_dir, "bin", "git")
424
            if "Windows" in platform.system():
425
                git_exe += ".exe"
426

427
        if platform.system() == "Darwin" and not self._xcode_command_line_tools_are_installed():
428
            return
429

430
        if not git_exe or not os.path.exists(git_exe):
431
            git_exe = shutil.which("git")
432

433
        if not git_exe or not os.path.exists(git_exe):
434
            return
435

436
        prefs.SetString("GitExecutable", git_exe)
437
        self.git_exe = git_exe
438

439
    def _xcode_command_line_tools_are_installed(self) -> bool:
440
        """On Macs, there is *always* an executable called "git", but sometimes it's just a
441
        script that tells the user to install XCode's Command Line tools. So the existence of git
442
        on the Mac actually requires us to check for that installation."""
443
        try:
444
            subprocess.check_output(["xcode-select", "-p"])
445
            return True
446
        except subprocess.CalledProcessError:
447
            return False
448

449
    def _synchronous_call_git(self, args: List[str]) -> str:
450
        """Calls git and returns its output."""
451
        final_args = [self.git_exe]
452
        final_args.extend(args)
453

454
        try:
455
            proc = utils.run_interruptable_subprocess(final_args)
456
        except subprocess.CalledProcessError as e:
457
            raise GitFailed(
458
                f"Git returned a non-zero exit status: {e.returncode}\n"
459
                + f"Called with: {' '.join(final_args)}\n\n"
460
                + f"Returned stderr:\n{e.stderr}"
461
            ) from e
462

463
        return proc.stdout
464

465

466
def initialize_git() -> Optional[GitManager]:
467
    """If git is enabled, locate the git executable if necessary and return a new
468
    GitManager object. The executable location is saved in user preferences for reuse,
469
    and git can be disabled by setting the disableGit parameter in the Addons
470
    preference group. Returns None if for any of those reasons we aren't using git."""
471

472
    git_manager = None
473
    pref = fci.ParamGet("User parameter:BaseApp/Preferences/Addons")
474
    disable_git = pref.GetBool("disableGit", False)
475
    if not disable_git:
476
        try:
477
            git_manager = GitManager()
478
        except NoGitFound:
479
            pass
480
    return git_manager
481