1
0
Fork 0

Adding upstream version 0.16.4.

Signed-off-by: Daniel Baumann <daniel@debian.org>
This commit is contained in:
Daniel Baumann 2025-02-11 18:49:40 +01:00
parent 7cdc86fc2c
commit 6e615d8555
Signed by: daniel
GPG key ID: FBB4F0E80A80222F
11 changed files with 143 additions and 124 deletions

View file

@ -61,7 +61,7 @@ symbol | meaning
---|--- ---|---
`+`| staged changes `+`| staged changes
`*`| unstaged changes `*`| unstaged changes
`_`| untracked files/folders `?`| untracked files/folders
The bookkeeping sub-commands are The bookkeeping sub-commands are
@ -333,6 +333,9 @@ For example, the default setting corresponds to
branch,commit_msg,commit_time branch,commit_msg,commit_time
``` ```
Here `branch` includes both branch name and status. To get the branch name alone, use `branch_name`.
### customize git command flags ### customize git command flags
One can set custom flags to run `git` commands. For example, with One can set custom flags to run `git` commands. For example, with

View file

@ -47,7 +47,7 @@
- `+`: 暂存(staged) - `+`: 暂存(staged)
- `*`: 未暂存unstaged - `*`: 未暂存unstaged
- `_`: 未追踪untracked - `?`: 未追踪untracked
基础指令: 基础指令:

View file

@ -1,3 +1,3 @@
import pkg_resources import pkg_resources
__version__ = pkg_resources.get_distribution('gita').version __version__ = pkg_resources.get_distribution("gita").version

View file

@ -2,8 +2,11 @@ import os
def get_config_dir() -> str: def get_config_dir() -> str:
root = os.environ.get('GITA_PROJECT_HOME') or os.environ.get('XDG_CONFIG_HOME') or os.path.join( root = (
os.path.expanduser('~'), '.config') os.environ.get("GITA_PROJECT_HOME")
or os.environ.get("XDG_CONFIG_HOME")
or os.path.join(os.path.expanduser("~"), ".config")
)
return os.path.join(root, "gita") return os.path.join(root, "gita")

View file

@ -3,7 +3,7 @@ import csv
import subprocess import subprocess
from enum import Enum from enum import Enum
from pathlib import Path from pathlib import Path
from functools import lru_cache from functools import lru_cache, partial
from typing import Tuple, List, Callable, Dict from typing import Tuple, List, Callable, Dict
from . import common from . import common
@ -13,24 +13,25 @@ class Color(Enum):
""" """
Terminal color Terminal color
""" """
black = '\x1b[30m'
red = '\x1b[31m' # local diverges from remote black = "\x1b[30m"
green = '\x1b[32m' # local == remote red = "\x1b[31m" # local diverges from remote
yellow = '\x1b[33m' # local is behind green = "\x1b[32m" # local == remote
blue = '\x1b[34m' yellow = "\x1b[33m" # local is behind
purple = '\x1b[35m' # local is ahead blue = "\x1b[34m"
cyan = '\x1b[36m' purple = "\x1b[35m" # local is ahead
white = '\x1b[37m' # no remote branch cyan = "\x1b[36m"
end = '\x1b[0m' white = "\x1b[37m" # no remote branch
b_black = '\x1b[30;1m' end = "\x1b[0m"
b_red = '\x1b[31;1m' b_black = "\x1b[30;1m"
b_green = '\x1b[32;1m' b_red = "\x1b[31;1m"
b_yellow = '\x1b[33;1m' b_green = "\x1b[32;1m"
b_blue = '\x1b[34;1m' b_yellow = "\x1b[33;1m"
b_purple = '\x1b[35;1m' b_blue = "\x1b[34;1m"
b_cyan = '\x1b[36;1m' b_purple = "\x1b[35;1m"
b_white = '\x1b[37;1m' b_cyan = "\x1b[36;1m"
underline = '\x1B[4m' b_white = "\x1b[37;1m"
underline = "\x1B[4m"
# Make f"{Color.foo}" expand to Color.foo.value . # Make f"{Color.foo}" expand to Color.foo.value .
# #
@ -40,26 +41,24 @@ class Color(Enum):
default_colors = { default_colors = {
'no-remote': Color.white.name, "no-remote": Color.white.name,
'in-sync': Color.green.name, "in-sync": Color.green.name,
'diverged': Color.red.name, "diverged": Color.red.name,
'local-ahead': Color.purple.name, "local-ahead": Color.purple.name,
'remote-ahead': Color.yellow.name, "remote-ahead": Color.yellow.name,
} }
def show_colors(): # pragma: no cover def show_colors(): # pragma: no cover
""" """ """
"""
for i, c in enumerate(Color, start=1): for i, c in enumerate(Color, start=1):
if c != Color.end and c != Color.underline: if c != Color.end and c != Color.underline:
print(f'{c.value}{c.name:<8} ', end='') print(f"{c.value}{c.name:<8} ", end="")
if i % 9 == 0: if i % 9 == 0:
print() print()
print(f'{Color.end}') print(f"{Color.end}")
for situation, c in sorted(get_color_encoding().items()): for situation, c in sorted(get_color_encoding().items()):
print(f'{situation:<12}: {Color[c].value}{c:<8}{Color.end} ') print(f"{situation:<12}: {Color[c].value}{c:<8}{Color.end} ")
@lru_cache() @lru_cache()
@ -69,9 +68,9 @@ def get_color_encoding() -> Dict[str, str]:
In the format of {situation: color name} In the format of {situation: color name}
""" """
# custom settings # custom settings
csv_config = Path(common.get_config_fname('color.csv')) csv_config = Path(common.get_config_fname("color.csv"))
if csv_config.is_file(): if csv_config.is_file():
with open(csv_config, 'r') as f: with open(csv_config, "r") as f:
reader = csv.DictReader(f) reader = csv.DictReader(f)
colors = next(reader) colors = next(reader)
else: else:
@ -79,7 +78,7 @@ def get_color_encoding() -> Dict[str, str]:
return colors return colors
def get_info_funcs() -> List[Callable[[str], str]]: def get_info_funcs(no_colors=False) -> List[Callable[[str], str]]:
""" """
Return the functions to generate `gita ll` information. All these functions Return the functions to generate `gita ll` information. All these functions
take the repo path as input and return the corresponding information as str. take the repo path as input and return the corresponding information as str.
@ -88,10 +87,11 @@ def get_info_funcs() -> List[Callable[[str], str]]:
to_display = get_info_items() to_display = get_info_items()
# This re-definition is to make unit test mocking to work # This re-definition is to make unit test mocking to work
all_info_items = { all_info_items = {
'branch': get_repo_status, "branch": partial(get_repo_status, no_colors=no_colors),
'commit_msg': get_commit_msg, "branch_name": get_repo_branch,
'commit_time': get_commit_time, "commit_msg": get_commit_msg,
'path': get_path, "commit_time": get_commit_time,
"path": get_path,
} }
return [all_info_items[k] for k in to_display] return [all_info_items[k] for k in to_display]
@ -101,15 +101,15 @@ def get_info_items() -> List[str]:
Return the information items to be displayed in the `gita ll` command. Return the information items to be displayed in the `gita ll` command.
""" """
# custom settings # custom settings
csv_config = Path(common.get_config_fname('info.csv')) csv_config = Path(common.get_config_fname("info.csv"))
if csv_config.is_file(): if csv_config.is_file():
with open(csv_config, 'r') as f: with open(csv_config, "r") as f:
reader = csv.reader(f) reader = csv.reader(f)
display_items = next(reader) display_items = next(reader)
display_items = [x for x in display_items if x in ALL_INFO_ITEMS] display_items = [x for x in display_items if x in ALL_INFO_ITEMS]
else: else:
# default settings # default settings
display_items = ['branch', 'commit_msg', 'commit_time'] display_items = ["branch", "commit_msg", "commit_time"]
return display_items return display_items
@ -119,43 +119,48 @@ def get_path(prop: Dict[str, str]) -> str:
# TODO: do we need to add the flags here too? # TODO: do we need to add the flags here too?
def get_head(path: str) -> str: def get_head(path: str) -> str:
result = subprocess.run('git symbolic-ref -q --short HEAD || git describe --tags --exact-match', result = subprocess.run(
"git symbolic-ref -q --short HEAD || git describe --tags --exact-match",
shell=True, shell=True,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
universal_newlines=True, universal_newlines=True,
cwd=path) cwd=path,
)
return result.stdout.strip() return result.stdout.strip()
def run_quiet_diff(flags: List[str], args: List[str]) -> int: def run_quiet_diff(flags: List[str], args: List[str], path) -> int:
""" """
Return the return code of git diff `args` in quiet mode Return the return code of git diff `args` in quiet mode
""" """
result = subprocess.run( result = subprocess.run(
['git'] + flags + ['diff', '--quiet'] + args, ["git"] + flags + ["diff", "--quiet"] + args,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
cwd=path,
) )
return result.returncode return result.returncode
def get_common_commit() -> str: def get_common_commit(path) -> str:
""" """
Return the hash of the common commit of the local and upstream branches. Return the hash of the common commit of the local and upstream branches.
""" """
result = subprocess.run('git merge-base @{0} @{u}'.split(), result = subprocess.run(
"git merge-base @{0} @{u}".split(),
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True) universal_newlines=True,
cwd=path,
)
return result.stdout.strip() return result.stdout.strip()
def has_untracked(flags: List[str]) -> bool: def has_untracked(flags: List[str], path) -> bool:
""" """
Return True if untracked file/folder exists Return True if untracked file/folder exists
""" """
cmd = ['git'] + flags + 'ls-files -zo --exclude-standard'.split() cmd = ["git"] + flags + "ls-files -zo --exclude-standard".split()
result = subprocess.run(cmd, result = subprocess.run(cmd, stdout=subprocess.PIPE, cwd=path)
stdout=subprocess.PIPE)
return bool(result.stdout) return bool(result.stdout)
@ -164,12 +169,14 @@ def get_commit_msg(prop: Dict[str, str]) -> str:
Return the last commit message. Return the last commit message.
""" """
# `git show-branch --no-name HEAD` is faster than `git show -s --format=%s` # `git show-branch --no-name HEAD` is faster than `git show -s --format=%s`
cmd = ['git'] + prop['flags'] + 'show-branch --no-name HEAD'.split() cmd = ["git"] + prop["flags"] + "show-branch --no-name HEAD".split()
result = subprocess.run(cmd, result = subprocess.run(
cmd,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
universal_newlines=True, universal_newlines=True,
cwd=prop['path']) cwd=prop["path"],
)
return result.stdout.strip() return result.stdout.strip()
@ -177,60 +184,65 @@ def get_commit_time(prop: Dict[str, str]) -> str:
""" """
Return the last commit time in parenthesis. Return the last commit time in parenthesis.
""" """
cmd = ['git'] + prop['flags'] + 'log -1 --format=%cd --date=relative'.split() cmd = ["git"] + prop["flags"] + "log -1 --format=%cd --date=relative".split()
result = subprocess.run(cmd, result = subprocess.run(
cmd,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
universal_newlines=True, universal_newlines=True,
cwd=prop['path']) cwd=prop["path"],
)
return f"({result.stdout.strip()})" return f"({result.stdout.strip()})"
def get_repo_status(prop: Dict[str, str], no_colors=False) -> str: def get_repo_status(prop: Dict[str, str], no_colors=False) -> str:
head = get_head(prop['path']) head = get_head(prop["path"])
dirty, staged, untracked, color = _get_repo_status(prop, no_colors) dirty, staged, untracked, color = _get_repo_status(prop, no_colors)
if color: if color:
return f'{color}{head+" "+dirty+staged+untracked:<10}{Color.end}' return f"{color}{head+' ['+dirty+staged+untracked+']':<13}{Color.end}"
return f'{head+" "+dirty+staged+untracked:<10}' return f"{head+' ['+dirty+staged+untracked+']':<13}"
def get_repo_branch(prop: Dict[str, str]) -> str:
return get_head(prop["path"])
def _get_repo_status(prop: Dict[str, str], no_colors: bool) -> Tuple[str]: def _get_repo_status(prop: Dict[str, str], no_colors: bool) -> Tuple[str]:
""" """
Return the status of one repo Return the status of one repo
""" """
path = prop['path'] path = prop["path"]
flags = prop['flags'] flags = prop["flags"]
os.chdir(path) dirty = "*" if run_quiet_diff(flags, [], path) else ""
dirty = '*' if run_quiet_diff(flags, []) else '' staged = "+" if run_quiet_diff(flags, ["--cached"], path) else ""
staged = '+' if run_quiet_diff(flags, ['--cached']) else '' untracked = "?" if has_untracked(flags, path) else ""
untracked = '_' if has_untracked(flags) else ''
if no_colors: if no_colors:
return dirty, staged, untracked, '' return dirty, staged, untracked, ""
colors = {situ: Color[name].value colors = {situ: Color[name].value for situ, name in get_color_encoding().items()}
for situ, name in get_color_encoding().items()} diff_returncode = run_quiet_diff(flags, ["@{u}", "@{0}"], path)
diff_returncode = run_quiet_diff(flags, ['@{u}', '@{0}'])
has_no_remote = diff_returncode == 128 has_no_remote = diff_returncode == 128
has_no_diff = diff_returncode == 0 has_no_diff = diff_returncode == 0
if has_no_remote: if has_no_remote:
color = colors['no-remote'] color = colors["no-remote"]
elif has_no_diff: elif has_no_diff:
color = colors['in-sync'] color = colors["in-sync"]
else: else:
common_commit = get_common_commit() common_commit = get_common_commit(path)
outdated = run_quiet_diff(flags, ['@{u}', common_commit]) outdated = run_quiet_diff(flags, ["@{u}", common_commit], path)
if outdated: if outdated:
diverged = run_quiet_diff(flags, ['@{0}', common_commit]) diverged = run_quiet_diff(flags, ["@{0}", common_commit], path)
color = colors['diverged'] if diverged else colors['remote-ahead'] color = colors["diverged"] if diverged else colors["remote-ahead"]
else: # local is ahead of remote else: # local is ahead of remote
color = colors['local-ahead'] color = colors["local-ahead"]
return dirty, staged, untracked, color return dirty, staged, untracked, color
ALL_INFO_ITEMS = { ALL_INFO_ITEMS = {
'branch': get_repo_status, "branch",
'commit_msg': get_commit_msg, "branch_name",
'commit_time': get_commit_time, "commit_msg",
'path': get_path, "commit_time",
} "path",
}

View file

@ -5,7 +5,7 @@ import csv
import asyncio import asyncio
import platform import platform
import subprocess import subprocess
from functools import lru_cache, partial from functools import lru_cache
from pathlib import Path from pathlib import Path
from typing import List, Dict, Coroutine, Union, Iterator, Tuple from typing import List, Dict, Coroutine, Union, Iterator, Tuple
from collections import Counter, defaultdict from collections import Counter, defaultdict
@ -431,18 +431,14 @@ def describe(repos: Dict[str, Dict[str, str]], no_colors: bool = False) -> str:
""" """
if repos: if repos:
name_width = len(max(repos, key=len)) + 1 name_width = len(max(repos, key=len)) + 1
funcs = info.get_info_funcs() funcs = info.get_info_funcs(no_colors=no_colors)
get_repo_status = info.get_repo_status
if get_repo_status in funcs and no_colors:
idx = funcs.index(get_repo_status)
funcs[idx] = partial(get_repo_status, no_colors=True)
num_threads = min(multiprocessing.cpu_count(), len(repos)) num_threads = min(multiprocessing.cpu_count(), len(repos))
with ThreadPoolExecutor(max_workers=num_threads) as executor: with ThreadPoolExecutor(max_workers=num_threads) as executor:
for line in executor.map( for line in executor.map(
lambda repo: f'{repo:<{name_width}}{" ".join(f(repos[repo]) for f in funcs)}', lambda name: f'{name:<{name_width}}{" ".join(f(repos[name]) for f in funcs)}',
sorted(repos)): sorted(repos),
):
yield line yield line

View file

@ -7,7 +7,7 @@ with open("README.md", encoding="utf-8") as f:
setup( setup(
name="gita", name="gita",
packages=["gita"], packages=["gita"],
version="0.16.3", version="0.16.4",
license="MIT", license="MIT",
description="Manage multiple git repos with sanity", description="Manage multiple git repos with sanity",
long_description=long_description, long_description=long_description,

View file

@ -8,10 +8,11 @@ def fullpath(fname: str):
return str(TEST_DIR / fname) return str(TEST_DIR / fname)
PATH_FNAME = fullpath('mock_path_file') PATH_FNAME = fullpath("mock_path_file")
PATH_FNAME_EMPTY = fullpath('empty_path_file') PATH_FNAME_EMPTY = fullpath("empty_path_file")
PATH_FNAME_CLASH = fullpath('clash_path_file') PATH_FNAME_CLASH = fullpath("clash_path_file")
GROUP_FNAME = fullpath('mock_group_file') GROUP_FNAME = fullpath("mock_group_file")
def async_mock(): def async_mock():
""" """

View file

@ -4,13 +4,14 @@ from unittest.mock import patch, MagicMock
from gita import info from gita import info
@patch('subprocess.run') @patch("subprocess.run")
def test_run_quiet_diff(mock_run): def test_run_quiet_diff(mock_run):
mock_return = MagicMock() mock_return = MagicMock()
mock_run.return_value = mock_return mock_run.return_value = mock_return
got = info.run_quiet_diff(['--flags'], ['my', 'args']) got = info.run_quiet_diff(["--flags"], ["my", "args"], "/a/b/c")
mock_run.assert_called_once_with( mock_run.assert_called_once_with(
['git', '--flags', 'diff', '--quiet', 'my', 'args'], ["git", "--flags", "diff", "--quiet", "my", "args"],
stderr=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
cwd="/a/b/c",
) )
assert got == mock_return.returncode assert got == mock_return.returncode

View file

@ -71,6 +71,7 @@ class TestLsLl:
""" """
functional test functional test
""" """
# avoid modifying the local configuration # avoid modifying the local configuration
def side_effect(input, _=None): def side_effect(input, _=None):
return tmp_path / f"{input}.txt" return tmp_path / f"{input}.txt"
@ -129,12 +130,12 @@ class TestLsLl:
[ [
( (
PATH_FNAME, PATH_FNAME,
"repo1 cmaster dsu\x1b[0m msg \nrepo2 cmaster dsu\x1b[0m msg \nxxx cmaster dsu\x1b[0m msg \n", "repo1 cmaster [dsu] \x1b[0m msg \nrepo2 cmaster [dsu] \x1b[0m msg \nxxx cmaster [dsu] \x1b[0m msg \n",
), ),
(PATH_FNAME_EMPTY, ""), (PATH_FNAME_EMPTY, ""),
( (
PATH_FNAME_CLASH, PATH_FNAME_CLASH,
"repo1 cmaster dsu\x1b[0m msg \nrepo2 cmaster dsu\x1b[0m msg \n", "repo1 cmaster [dsu] \x1b[0m msg \nrepo2 cmaster [dsu] \x1b[0m msg \n",
), ),
], ],
) )
@ -527,7 +528,9 @@ class TestInfo:
args.info_cmd = None args.info_cmd = None
__main__.f_info(args) __main__.f_info(args)
out, err = capfd.readouterr() out, err = capfd.readouterr()
assert "In use: branch,commit_msg,commit_time\nUnused: path\n" == out assert (
"In use: branch,commit_msg,commit_time\nUnused: branch_name,path\n" == out
)
assert err == "" assert err == ""
@patch("gita.common.get_config_fname") @patch("gita.common.get_config_fname")

View file

@ -115,17 +115,17 @@ def test_auto_group(repos, paths, expected):
( (
[{"abc": {"path": "/root/repo/", "type": "", "flags": []}}, False], [{"abc": {"path": "/root/repo/", "type": "", "flags": []}}, False],
True, True,
"abc \x1b[31mrepo *+_ \x1b[0m msg xx", "abc \x1b[31mrepo [*+?] \x1b[0m msg xx",
), ),
( (
[{"abc": {"path": "/root/repo/", "type": "", "flags": []}}, True], [{"abc": {"path": "/root/repo/", "type": "", "flags": []}}, True],
True, True,
"abc repo *+_ msg xx", "abc repo [*+?] msg xx",
), ),
( (
[{"repo": {"path": "/root/repo2/", "type": "", "flags": []}}, False], [{"repo": {"path": "/root/repo2/", "type": "", "flags": []}}, False],
False, False,
"repo \x1b[32mrepo _ \x1b[0m msg xx", "repo \x1b[32mrepo [?] \x1b[0m msg xx",
), ),
], ],
) )
@ -135,7 +135,7 @@ def test_describe(test_input, diff_return, expected, monkeypatch):
monkeypatch.setattr(info, "get_commit_msg", lambda *_: "msg") monkeypatch.setattr(info, "get_commit_msg", lambda *_: "msg")
monkeypatch.setattr(info, "get_commit_time", lambda *_: "xx") monkeypatch.setattr(info, "get_commit_time", lambda *_: "xx")
monkeypatch.setattr(info, "has_untracked", lambda *_: True) monkeypatch.setattr(info, "has_untracked", lambda *_: True)
monkeypatch.setattr("os.chdir", lambda x: None) monkeypatch.setattr(info, "get_common_commit", lambda x: "")
info.get_color_encoding.cache_clear() # avoid side effect info.get_color_encoding.cache_clear() # avoid side effect
assert expected == next(utils.describe(*test_input)) assert expected == next(utils.describe(*test_input))