summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--.github/workflows/nos.yml20
-rw-r--r--README.md4
-rw-r--r--gita/__main__.py4
-rw-r--r--gita/common.py2
-rw-r--r--gita/info.py8
-rw-r--r--gita/utils.py262
-rw-r--r--setup.py2
-rw-r--r--tests/test_main.py13
-rw-r--r--tests/test_utils.py289
9 files changed, 371 insertions, 233 deletions
diff --git a/.github/workflows/nos.yml b/.github/workflows/nos.yml
index 9b81b05..52cf572 100644
--- a/.github/workflows/nos.yml
+++ b/.github/workflows/nos.yml
@@ -4,24 +4,26 @@ on: [push, pull_request]
jobs:
build:
- runs-on: ubuntu-latest
+ runs-on: ubuntu-20.04
strategy:
matrix:
- os: [ubuntu-latest, macos-latest, windows-latest]
- python-version: [3.6, 3.7, 3.8, 3.9]
+ os: [ubuntu-20.04, macos-latest, windows-latest]
+ python-version: [3.6, 3.7, 3.8, 3.9, "3.10", "3.11"]
steps:
- - uses: actions/checkout@v2
+ - uses: actions/checkout@v3
- name: Set up Python ${{ matrix.python-version }}
- uses: actions/setup-python@v2
+ uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependences
run: |
- python -m pip install --upgrade pip
+ python -m pip install --upgrade pip wheel
pip install -r requirements.txt
- pip install .
+ pip install -e .
- name: Pytest
run: |
- pytest tests --cov=./gita
+ pytest tests --cov=./gita --cov-report=xml
- name: Upload coverage to Codecov
- uses: codecov/codecov-action@v2
+ uses: codecov/codecov-action@v3
+ with:
+ token: ${{ secrets.CODECOV_TOKEN }}
diff --git a/README.md b/README.md
index d1cd42e..630d41a 100644
--- a/README.md
+++ b/README.md
@@ -128,8 +128,8 @@ If more than one repos are specified, the `git` command runs asynchronously,
with the exception of `log`, `difftool` and `mergetool`,
which require non-trivial user input.
-Repo configuration is saved in `$XDG_CONFIG_HOME/gita/repos.csv`
-(most likely `~/.config/gita/repos.csv`).
+Repo configuration global is saved in `$XDG_CONFIG_HOME/gita/repos.csv`
+(most likely `~/.config/gita/repos.csv`) or if you prefered at project configuration add environment variable `GITA_PROJECT_HOME`.
## Installation
diff --git a/gita/__main__.py b/gita/__main__.py
index 2517958..4cc0e1e 100644
--- a/gita/__main__.py
+++ b/gita/__main__.py
@@ -48,10 +48,10 @@ def _group_name(name: str, exclude_old_names=True) -> str:
def _path_name(name: str) -> str:
"""
- Return absolute path without trailing /
+ Return absolute path
"""
if name:
- return os.path.abspath(name).rstrip(os.path.sep)
+ return os.path.abspath(name)
return ""
diff --git a/gita/common.py b/gita/common.py
index e1a3dde..994e5e0 100644
--- a/gita/common.py
+++ b/gita/common.py
@@ -2,7 +2,7 @@ import os
def get_config_dir() -> str:
- root = os.environ.get('XDG_CONFIG_HOME') or os.path.join(
+ root = 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")
diff --git a/gita/info.py b/gita/info.py
index ee302d0..57bb1a8 100644
--- a/gita/info.py
+++ b/gita/info.py
@@ -9,7 +9,7 @@ from typing import Tuple, List, Callable, Dict
from . import common
-class Color(str, Enum):
+class Color(Enum):
"""
Terminal color
"""
@@ -32,6 +32,12 @@ class Color(str, Enum):
b_white = '\x1b[37;1m'
underline = '\x1B[4m'
+ # Make f"{Color.foo}" expand to Color.foo.value .
+ #
+ # See https://stackoverflow.com/a/24487545
+ def __str__(self):
+ return f"{self.value}"
+
default_colors = {
'no-remote': Color.white.name,
diff --git a/gita/utils.py b/gita/utils.py
index 5332255..76aebce 100644
--- a/gita/utils.py
+++ b/gita/utils.py
@@ -9,6 +9,8 @@ from functools import lru_cache, partial
from pathlib import Path
from typing import List, Dict, Coroutine, Union, Iterator, Tuple
from collections import Counter, defaultdict
+from concurrent.futures import ThreadPoolExecutor
+import multiprocessing
from . import info
from . import common
@@ -17,24 +19,25 @@ from . import common
MAX_INT = sys.maxsize
-def get_relative_path(kid: str, parent: str) -> Union[List[str], None]:
+def get_relative_path(kid: os.PathLike, parent: str) -> Union[List[str], None]:
"""
- Return the relative path depth if relative, otherwise MAX_INT.
+ Return the relative path depth if relative, otherwise None.
- Both the `kid` and `parent` should be absolute paths without trailing /
+ Both the `kid` and `parent` should be absolute paths
"""
- # Note that os.path.commonpath has no trailing /
- # TODO: python3.9 pathlib has is_relative_to() function
- # TODO: Maybe use os.path.commonprefix? since it's faster?
- if parent == '':
+ if parent == "":
return None
- if parent == os.path.commonpath((kid, parent)):
- rel = os.path.normpath(os.path.relpath(kid, parent)).split(os.sep)
- if rel == ['.']:
- rel = []
- return rel
- else:
+
+ p_kid = Path(kid)
+ # p_kid = Path(kid).resolve()
+ try:
+ p_rel = p_kid.relative_to(parent)
+ except ValueError:
return None
+ rel = str(p_rel).split(os.sep)
+ if rel == ["."]:
+ rel = []
+ return rel
@lru_cache()
@@ -43,16 +46,22 @@ def get_repos() -> Dict[str, Dict[str, str]]:
Return a `dict` of repo name to repo absolute path and repo type
"""
- path_file = common.get_config_fname('repos.csv')
+ path_file = common.get_config_fname("repos.csv")
repos = {}
if os.path.isfile(path_file) and os.stat(path_file).st_size > 0:
with open(path_file) as f:
- rows = csv.DictReader(f, ['path', 'name', 'type', 'flags'],
- restval='') # it's actually a reader
- repos = {r['name']:
- {'path': r['path'], 'type': r['type'],
- 'flags': r['flags'].split()}
- for r in rows if is_git(r['path'], include_bare=True)}
+ rows = csv.DictReader(
+ f, ["path", "name", "type", "flags"], restval=""
+ ) # it's actually a reader
+ repos = {
+ r["name"]: {
+ "path": r["path"],
+ "type": r["type"],
+ "flags": r["flags"].split(),
+ }
+ for r in rows
+ if is_git(r["path"], include_bare=True)
+ }
return repos
@@ -65,20 +74,19 @@ def get_context() -> Union[Path, None]:
"""
config_dir = Path(common.get_config_dir())
- matches = list(config_dir.glob('*.context'))
+ matches = list(config_dir.glob("*.context"))
if len(matches) > 1:
print("Cannot have multiple .context file")
sys.exit(1)
if not matches:
return None
ctx = matches[0]
- if ctx.stem == 'auto':
- cwd = str(Path.cwd())
+ if ctx.stem == "auto":
# The context is set to be the group with minimal distance to cwd
candidate = None
min_dist = MAX_INT
for gname, prop in get_groups().items():
- rel = get_relative_path(cwd, prop['path'])
+ rel = get_relative_path(Path.cwd(), prop["path"])
if rel is None:
continue
d = len(rel)
@@ -88,7 +96,7 @@ def get_context() -> Union[Path, None]:
if not candidate:
ctx = None
else:
- ctx = ctx.with_name(f'{candidate}.context')
+ ctx = ctx.with_name(f"{candidate}.context")
return ctx
@@ -98,19 +106,23 @@ def get_groups() -> Dict[str, Dict[str, Union[str, List]]]:
Return a `dict` of group name to group properties such as repo names and
group path.
"""
- fname = common.get_config_fname('groups.csv')
+ fname = common.get_config_fname("groups.csv")
groups = {}
+ repos = get_repos()
# Each line is: group-name:repo1 repo2 repo3:group-path
if os.path.isfile(fname) and os.stat(fname).st_size > 0:
- with open(fname, 'r') as f:
- rows = csv.DictReader(f, ['name', 'repos', 'path'],
- restval='', delimiter=':')
+ with open(fname, "r") as f:
+ rows = csv.DictReader(
+ f, ["name", "repos", "path"], restval="", delimiter=":"
+ )
+ # filter out invalid repos
groups = {
- r['name']: {
- 'repos': r['repos'].split(),
- 'path': r['path']
- }
- for r in rows}
+ r["name"]: {
+ "repos": [repo for repo in r["repos"].split() if repo in repos],
+ "path": r["path"],
+ }
+ for r in rows
+ }
return groups
@@ -121,7 +133,7 @@ def delete_repo_from_groups(repo: str, groups: Dict[str, Dict]) -> bool:
deleted = False
for name in groups:
try:
- groups[name]['repos'].remove(repo)
+ groups[name]["repos"].remove(repo)
except ValueError as e:
pass
else:
@@ -130,20 +142,18 @@ def delete_repo_from_groups(repo: str, groups: Dict[str, Dict]) -> bool:
def replace_context(old: Union[Path, None], new: str):
- """
-
- """
- auto = Path(common.get_config_dir()) / 'auto.context'
+ """ """
+ auto = Path(common.get_config_dir()) / "auto.context"
if auto.exists():
old = auto
- if new == 'none': # delete
+ if new == "none": # delete
old and old.unlink()
elif old:
# ctx.rename(ctx.with_stem(new_name)) # only works in py3.9
- old.rename(old.with_name(f'{new}.context'))
+ old.rename(old.with_name(f"{new}.context"))
else:
- Path(auto.with_name(f'{new}.context')).write_text('')
+ Path(auto.with_name(f"{new}.context")).write_text("")
def get_choices() -> List[Union[str, None]]:
@@ -162,10 +172,8 @@ def get_choices() -> List[Union[str, None]]:
def is_submodule_repo(p: Path) -> bool:
- """
-
- """
- if p.is_file() and '.git/modules' in p.read_text():
+ """ """
+ if p.is_file() and ".git/modules" in p.read_text():
return True
return False
@@ -183,7 +191,7 @@ def is_git(path: str, include_bare=False, exclude_submodule=False) -> bool:
# A more reliable way to differentiable regular and worktree repos is to
# compare the result of `git rev-parse --git-dir` and
# `git rev-parse --git-common-dir`
- loc = os.path.join(path, '.git')
+ loc = os.path.join(path, ".git")
# TODO: we can display the worktree repos in a different font.
if os.path.exists(loc):
if exclude_submodule and is_submodule_repo(Path(loc)):
@@ -192,11 +200,13 @@ def is_git(path: str, include_bare=False, exclude_submodule=False) -> bool:
if not include_bare:
return False
# detect bare repo
- got = subprocess.run('git rev-parse --is-bare-repository'.split(),
- stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
- cwd=path
- )
- if got.returncode == 0 and got.stdout == b'true\n':
+ got = subprocess.run(
+ "git rev-parse --is-bare-repository".split(),
+ stdout=subprocess.PIPE,
+ stderr=subprocess.DEVNULL,
+ cwd=path,
+ )
+ if got.returncode == 0 and got.stdout == b"true\n":
return True
return False
@@ -211,16 +221,16 @@ def rename_repo(repos: Dict[str, Dict[str, str]], repo: str, new_name: str):
prop = repos[repo]
del repos[repo]
repos[new_name] = prop
- write_to_repo_file(repos, 'w')
+ write_to_repo_file(repos, "w")
groups = get_groups()
for g, values in groups.items():
- members = values['repos']
+ members = values["repos"]
if repo in members:
members.remove(repo)
members.append(new_name)
- groups[g]['repos'] = sorted(members)
- write_to_groups_file(groups, 'w')
+ groups[g]["repos"] = sorted(members)
+ write_to_groups_file(groups, "w")
def write_to_repo_file(repos: Dict[str, Dict[str, str]], mode: str):
@@ -228,40 +238,43 @@ def write_to_repo_file(repos: Dict[str, Dict[str, str]], mode: str):
@param repos: each repo is {name: {properties}}
"""
# The 3rd column is repo type; unused field
- data = [(prop['path'], name, '', ' '.join(prop['flags']))
- for name, prop in repos.items()]
- fname = common.get_config_fname('repos.csv')
+ data = [
+ (prop["path"], name, "", " ".join(prop["flags"]))
+ for name, prop in repos.items()
+ ]
+ fname = common.get_config_fname("repos.csv")
os.makedirs(os.path.dirname(fname), exist_ok=True)
- with open(fname, mode, newline='') as f:
- writer = csv.writer(f, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
+ with open(fname, mode, newline="") as f:
+ writer = csv.writer(f, delimiter=",", quotechar='"', quoting=csv.QUOTE_MINIMAL)
writer.writerows(data)
# TODO: combine with the repo writer
def write_to_groups_file(groups: Dict[str, Dict], mode: str):
- """
-
- """
- fname = common.get_config_fname('groups.csv')
+ """ """
+ fname = common.get_config_fname("groups.csv")
os.makedirs(os.path.dirname(fname), exist_ok=True)
if not groups: # all groups are deleted
- Path(fname).write_text('')
+ Path(fname).write_text("")
else:
# delete the group if there are no repos
for name in list(groups):
- if not groups[name]['repos']:
+ if not groups[name]["repos"]:
del groups[name]
- with open(fname, mode, newline='') as f:
+ with open(fname, mode, newline="") as f:
data = [
- (group, ' '.join(prop['repos']), prop['path'])
- for group, prop in groups.items()
- ]
- writer = csv.writer(f, delimiter=':', quotechar='"', quoting=csv.QUOTE_MINIMAL)
+ (group, " ".join(prop["repos"]), prop["path"])
+ for group, prop in groups.items()
+ ]
+ writer = csv.writer(
+ f, delimiter=":", quotechar='"', quoting=csv.QUOTE_MINIMAL
+ )
writer.writerows(data)
-def _make_name(path: str, repos: Dict[str, Dict[str, str]],
- name_counts: Counter) -> str:
+def _make_name(
+ path: str, repos: Dict[str, Dict[str, str]], name_counts: Counter
+) -> str:
"""
Given a new repo `path`, create a repo name. By default, basename is used.
If name collision exists, further include parent path name.
@@ -276,17 +289,19 @@ def _make_name(path: str, repos: Dict[str, Dict[str, str]],
return name
-def add_repos(repos: Dict[str, Dict[str, str]], new_paths: List[str],
- include_bare=False,
- exclude_submodule=False,
- dry_run=False,
- ) -> Dict[str, Dict[str, str]]:
+def add_repos(
+ repos: Dict[str, Dict[str, str]],
+ new_paths: List[str],
+ include_bare=False,
+ exclude_submodule=False,
+ dry_run=False,
+) -> Dict[str, Dict[str, str]]:
"""
Write new repo paths to file; return the added repos.
@param repos: name -> path
"""
- existing_paths = {prop['path'] for prop in repos.values()}
+ existing_paths = {prop["path"] for prop in repos.values()}
new_paths = {p for p in new_paths if is_git(p, include_bare, exclude_submodule)}
new_paths = new_paths - existing_paths
new_repos = {}
@@ -296,21 +311,21 @@ def add_repos(repos: Dict[str, Dict[str, str]], new_paths: List[str],
for p in new_paths:
print(p)
return {}
- name_counts = Counter(
- os.path.basename(os.path.normpath(p)) for p in new_paths
- )
- new_repos = {_make_name(path, repos, name_counts): {
- 'path': path,
- 'flags': '',
- } for path in new_paths}
- write_to_repo_file(new_repos, 'a+')
+ name_counts = Counter(os.path.basename(os.path.normpath(p)) for p in new_paths)
+ new_repos = {
+ _make_name(path, repos, name_counts): {
+ "path": path,
+ "flags": "",
+ }
+ for path in new_paths
+ }
+ write_to_repo_file(new_repos, "a+")
else:
- print('No new repos found!')
+ print("No new repos found!")
return new_repos
-def _generate_dir_hash(repo_path: str, paths: List[str]) -> Tuple[
- Tuple[str, ...], str]:
+def _generate_dir_hash(repo_path: str, paths: List[str]) -> Tuple[Tuple[str, ...], str]:
"""
Return relative parent strings, and the parent head string
@@ -322,13 +337,12 @@ def _generate_dir_hash(repo_path: str, paths: List[str]) -> Tuple[
if rel is not None:
break
else:
- return (), ''
+ return (), ""
head, tail = os.path.split(p)
return (tail, *rel), head
-def auto_group(repos: Dict[str, Dict[str, str]], paths: List[str]
- ) -> Dict[str, Dict]:
+def auto_group(repos: Dict[str, Dict[str, str]], paths: List[str]) -> Dict[str, Dict]:
"""
@params repos: repos to be grouped
@@ -337,17 +351,17 @@ def auto_group(repos: Dict[str, Dict[str, str]], paths: List[str]
# i.e., each repo should be contained in one and only one path
new_groups = defaultdict(dict)
for repo_name, prop in repos.items():
- hash, head = _generate_dir_hash(prop['path'], paths)
+ hash, head = _generate_dir_hash(prop["path"], paths)
if not hash:
continue
- for i in range(1, len(hash)+1):
- group_name = '-'.join(hash[:i])
+ for i in range(1, len(hash) + 1):
+ group_name = "-".join(hash[:i])
prop = new_groups[group_name]
- prop['path'] = os.path.join(head, *hash[:i])
- if 'repos' not in prop:
- prop['repos'] = [repo_name]
+ prop["path"] = os.path.join(head, *hash[:i])
+ if "repos" not in prop:
+ prop["repos"] = [repo_name]
else:
- prop['repos'].append(repo_name)
+ prop["repos"].append(repo_name)
# FIXME: need to make sure the new group names don't clash with old ones
# or repo names
return new_groups
@@ -359,7 +373,7 @@ def parse_clone_config(fname: str) -> Iterator[List[str]]:
"""
with open(fname) as f:
for line in f:
- yield line.strip().split(',')
+ yield line.strip().split(",")
async def run_async(repo_name: str, path: str, cmds: List[str]) -> Union[None, str]:
@@ -374,7 +388,8 @@ async def run_async(repo_name: str, path: str, cmds: List[str]) -> Union[None, s
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE,
start_new_session=True,
- cwd=path)
+ cwd=path,
+ )
stdout, stderr = await process.communicate()
for pipe in (stdout, stderr):
if pipe:
@@ -389,7 +404,7 @@ def format_output(s: str, prefix: str):
"""
Prepends every line in given string with the given prefix.
"""
- return ''.join([f'{prefix}: {line}' for line in s.splitlines(keepends=True)])
+ return "".join([f"{prefix}: {line}" for line in s.splitlines(keepends=True)])
def exec_async_tasks(tasks: List[Coroutine]) -> List[Union[None, str]]:
@@ -397,7 +412,7 @@ def exec_async_tasks(tasks: List[Coroutine]) -> List[Union[None, str]]:
Execute tasks asynchronously
"""
# TODO: asyncio API is nicer in python 3.7
- if platform.system() == 'Windows':
+ if platform.system() == "Windows":
loop = asyncio.ProactorEventLoop()
asyncio.set_event_loop(loop)
else:
@@ -415,17 +430,20 @@ def describe(repos: Dict[str, Dict[str, str]], no_colors: bool = False) -> str:
Return the status of all repos
"""
if repos:
- name_width = max(len(n) for n in repos) + 1
- funcs = info.get_info_funcs()
+ name_width = len(max(repos, key=len)) + 1
+ funcs = info.get_info_funcs()
- 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)
+ 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)
- for name in sorted(repos):
- info_items = ' '.join(f(repos[name]) for f in funcs)
- yield f'{name:<{name_width}}{info_items}'
+ num_threads = min(multiprocessing.cpu_count(), len(repos))
+ with ThreadPoolExecutor(max_workers=num_threads) as executor:
+ for line in executor.map(
+ lambda repo: f'{repo:<{name_width}}{" ".join(f(repos[repo]) for f in funcs)}',
+ sorted(repos)):
+ yield line
def get_cmds_from_files() -> Dict[str, Dict[str, str]]:
@@ -442,15 +460,15 @@ def get_cmds_from_files() -> Dict[str, Dict[str, str]]:
"""
# default config file
fname = os.path.join(os.path.dirname(__file__), "cmds.json")
- with open(fname, 'r') as f:
+ with open(fname, "r") as f:
cmds = json.load(f)
# custom config file
root = common.get_config_dir()
- fname = os.path.join(root, 'cmds.json')
+ fname = os.path.join(root, "cmds.json")
custom_cmds = {}
if os.path.isfile(fname) and os.path.getsize(fname):
- with open(fname, 'r') as f:
+ with open(fname, "r") as f:
custom_cmds = json.load(f)
# custom commands shadow default ones
@@ -458,8 +476,10 @@ def get_cmds_from_files() -> Dict[str, Dict[str, str]]:
return cmds
-def parse_repos_and_rest(input: List[str], quote_mode=False,
- ) -> Tuple[Dict[str, Dict[str, str]], List[str]]:
+def parse_repos_and_rest(
+ input: List[str],
+ quote_mode=False,
+) -> Tuple[Dict[str, Dict[str, str]], List[str]]:
"""
Parse gita input arguments
@@ -481,7 +501,7 @@ def parse_repos_and_rest(input: List[str], quote_mode=False,
if not names and ctx:
names = [ctx.stem]
if quote_mode and i + 1 != len(input):
- print(input[i], 'is not a repo or group' )
+ print(input[i], "is not a repo or group")
sys.exit(2)
if names:
@@ -490,7 +510,7 @@ def parse_repos_and_rest(input: List[str], quote_mode=False,
if k in repos:
chosen[k] = repos[k]
if k in groups:
- for r in groups[k]['repos']:
+ for r in groups[k]["repos"]:
chosen[r] = repos[r]
# if not set here, all repos are chosen
repos = chosen
diff --git a/setup.py b/setup.py
index 97f3223..71b23ec 100644
--- a/setup.py
+++ b/setup.py
@@ -7,7 +7,7 @@ with open("README.md", encoding="utf-8") as f:
setup(
name="gita",
packages=["gita"],
- version="0.16.2",
+ version="0.16.3",
license="MIT",
description="Manage multiple git repos with sanity",
long_description=long_description,
diff --git a/tests/test_main.py b/tests/test_main.py
index e36c2d8..0f2eeb5 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -94,14 +94,14 @@ class TestLsLl:
out, err = capfd.readouterr()
assert err == ""
assert "gita" in out
- assert info.Color.end in out
+ assert info.Color.end.value in out
# no color on branch name
__main__.main(["ll", "-C"])
out, err = capfd.readouterr()
assert err == ""
assert "gita" in out
- assert info.Color.end not in out
+ assert info.Color.end.value not in out
__main__.main(["ls", "gita"])
out, err = capfd.readouterr()
@@ -367,8 +367,9 @@ class TestGroupCmd:
assert err == ""
assert "xx yy\n" == out
+ @patch("gita.utils.get_repos", return_value={"a": "", "b": "", "c": "", "d": ""})
@patch("gita.common.get_config_fname", return_value=GROUP_FNAME)
- def test_ll(self, _, capfd):
+ def test_ll(self, _, __, capfd):
args = argparse.Namespace()
args.to_group = None
args.group_cmd = None
@@ -382,8 +383,9 @@ class TestGroupCmd:
== "\x1b[4mxx\x1b[0m: \n - a\n - b\n\x1b[4myy\x1b[0m: \n - a\n - c\n - d\n"
)
+ @patch("gita.utils.get_repos", return_value={"a": "", "b": "", "c": "", "d": ""})
@patch("gita.common.get_config_fname", return_value=GROUP_FNAME)
- def test_ll_with_group(self, _, capfd):
+ def test_ll_with_group(self, _, __, capfd):
args = argparse.Namespace()
args.to_group = None
args.group_cmd = None
@@ -394,9 +396,10 @@ class TestGroupCmd:
assert err == ""
assert "a c d\n" == out
+ @patch("gita.utils.get_repos", return_value={"a": "", "b": "", "c": "", "d": ""})
@patch("gita.common.get_config_fname", return_value=GROUP_FNAME)
@patch("gita.utils.write_to_groups_file")
- def test_rename(self, mock_write, _):
+ def test_rename(self, mock_write, *_):
args = argparse.Namespace()
args.gname = "xx"
args.new_name = "zz"
diff --git a/tests/test_utils.py b/tests/test_utils.py
index 7433532..9679116 100644
--- a/tests/test_utils.py
+++ b/tests/test_utils.py
@@ -6,134 +6,237 @@ from unittest.mock import patch, mock_open
from gita import utils, info
from conftest import (
- PATH_FNAME, PATH_FNAME_EMPTY, PATH_FNAME_CLASH, GROUP_FNAME, TEST_DIR,
+ PATH_FNAME,
+ PATH_FNAME_EMPTY,
+ PATH_FNAME_CLASH,
+ GROUP_FNAME,
+ TEST_DIR,
)
-@pytest.mark.parametrize('input, expected', [
- ([], ({'repo1': {'path': '/a/bcd/repo1', 'type': '', 'flags': []}, 'xxx': {'path': '/a/b/c/repo3', 'type': '', 'flags': []}, 'repo2': {'path': '/e/fgh/repo2', 'type': '', 'flags': []}}, [])),
- (['st'], ({'repo1': {'path': '/a/bcd/repo1', 'type': '', 'flags': []}, 'xxx': {'path': '/a/b/c/repo3', 'type': '', 'flags': []}, 'repo2': {'path': '/e/fgh/repo2', 'type': '', 'flags': []}}, ['st'])),
- (['repo1', 'st'], ({'repo1': {'flags': [], 'path': '/a/bcd/repo1', 'type': ''}}, ['st'])),
- (['repo1'], ({'repo1': {'flags': [], 'path': '/a/bcd/repo1', 'type': ''}}, [])),
- ])
-@patch('gita.utils.is_git', return_value=True)
-@patch('gita.common.get_config_fname', return_value=PATH_FNAME)
+@pytest.mark.parametrize(
+ "kid, parent, expected",
+ [
+ ("/a/b/repo", "/a/b", ["repo"]),
+ ("/a/b/repo", "/a", ["b", "repo"]),
+ ("/a/b/repo", "/a/", ["b", "repo"]),
+ ("/a/b/repo", "", None),
+ ("/a/b/repo", "/a/b/repo", []),
+ ],
+)
+def test_get_relative_path(kid, parent, expected):
+ assert expected == utils.get_relative_path(kid, parent)
+
+
+@pytest.mark.parametrize(
+ "input, expected",
+ [
+ (
+ [],
+ (
+ {
+ "repo1": {"path": "/a/bcd/repo1", "type": "", "flags": []},
+ "xxx": {"path": "/a/b/c/repo3", "type": "", "flags": []},
+ "repo2": {"path": "/e/fgh/repo2", "type": "", "flags": []},
+ },
+ [],
+ ),
+ ),
+ (
+ ["st"],
+ (
+ {
+ "repo1": {"path": "/a/bcd/repo1", "type": "", "flags": []},
+ "xxx": {"path": "/a/b/c/repo3", "type": "", "flags": []},
+ "repo2": {"path": "/e/fgh/repo2", "type": "", "flags": []},
+ },
+ ["st"],
+ ),
+ ),
+ (
+ ["repo1", "st"],
+ ({"repo1": {"flags": [], "path": "/a/bcd/repo1", "type": ""}}, ["st"]),
+ ),
+ (["repo1"], ({"repo1": {"flags": [], "path": "/a/bcd/repo1", "type": ""}}, [])),
+ ],
+)
+@patch("gita.utils.is_git", return_value=True)
+@patch("gita.common.get_config_fname", return_value=PATH_FNAME)
def test_parse_repos_and_rest(mock_path_fname, _, input, expected):
got = utils.parse_repos_and_rest(input)
assert got == expected
-@pytest.mark.parametrize('repo_path, paths, expected', [
- ('/a/b/c/repo', ['/a/b'], (('b', 'c'), '/a')),
- ])
+@pytest.mark.parametrize(
+ "repo_path, paths, expected",
+ [
+ ("/a/b/c/repo", ["/a/b"], (("b", "c"), "/a")),
+ ],
+)
def test_generate_dir_hash(repo_path, paths, expected):
got = utils._generate_dir_hash(repo_path, paths)
assert got == expected
-@pytest.mark.parametrize('repos, paths, expected', [
- ({'r1': {'path': '/a/b//repo1'}, 'r2': {'path': '/a/b/repo2'}},
- ['/a/b'], {'b': {'repos': ['r1', 'r2'], 'path': '/a/b'}}),
- ({'r1': {'path': '/a/b//repo1'}, 'r2': {'path': '/a/b/c/repo2'}},
- ['/a/b'], {'b': {'repos': ['r1', 'r2'], 'path': '/a/b'},
- 'b-c': {'repos': ['r2'], 'path': "/a/b/c"}}),
- ({'r1': {'path': '/a/b/c/repo1'}, 'r2': {'path': '/a/b/c/repo2'}},
- ['/a/b'], {'b-c': {'repos': ['r1', 'r2'], 'path': '/a/b/c'},
- 'b': {'path': '/a/b', 'repos': ['r1', 'r2']}}),
- ])
+@pytest.mark.parametrize(
+ "repos, paths, expected",
+ [
+ (
+ {"r1": {"path": "/a/b//repo1"}, "r2": {"path": "/a/b/repo2"}},
+ ["/a/b"],
+ {"b": {"repos": ["r1", "r2"], "path": "/a/b"}},
+ ),
+ (
+ {"r1": {"path": "/a/b//repo1"}, "r2": {"path": "/a/b/c/repo2"}},
+ ["/a/b"],
+ {
+ "b": {"repos": ["r1", "r2"], "path": "/a/b"},
+ "b-c": {"repos": ["r2"], "path": "/a/b/c"},
+ },
+ ),
+ (
+ {"r1": {"path": "/a/b/c/repo1"}, "r2": {"path": "/a/b/c/repo2"}},
+ ["/a/b"],
+ {
+ "b-c": {"repos": ["r1", "r2"], "path": "/a/b/c"},
+ "b": {"path": "/a/b", "repos": ["r1", "r2"]},
+ },
+ ),
+ ],
+)
def test_auto_group(repos, paths, expected):
got = utils.auto_group(repos, paths)
assert got == expected
-@pytest.mark.parametrize('test_input, diff_return, expected', [
- ([{'abc': {'path': '/root/repo/', 'type': '', 'flags': []}}, False],
- True, 'abc \x1b[31mrepo *+_ \x1b[0m msg xx'),
- ([{'abc': {'path': '/root/repo/', 'type': '', 'flags': []}}, True],
- True, 'abc repo *+_ msg xx'),
- ([{'repo': {'path': '/root/repo2/', 'type': '', 'flags': []}}, False],
- False, 'repo \x1b[32mrepo _ \x1b[0m msg xx'),
-])
+@pytest.mark.parametrize(
+ "test_input, diff_return, expected",
+ [
+ (
+ [{"abc": {"path": "/root/repo/", "type": "", "flags": []}}, False],
+ True,
+ "abc \x1b[31mrepo *+_ \x1b[0m msg xx",
+ ),
+ (
+ [{"abc": {"path": "/root/repo/", "type": "", "flags": []}}, True],
+ True,
+ "abc repo *+_ msg xx",
+ ),
+ (
+ [{"repo": {"path": "/root/repo2/", "type": "", "flags": []}}, False],
+ False,
+ "repo \x1b[32mrepo _ \x1b[0m msg xx",
+ ),
+ ],
+)
def test_describe(test_input, diff_return, expected, monkeypatch):
- monkeypatch.setattr(info, 'get_head', lambda x: 'repo')
- monkeypatch.setattr(info, 'run_quiet_diff', lambda *_: diff_return)
- monkeypatch.setattr(info, 'get_commit_msg', lambda *_: "msg")
- monkeypatch.setattr(info, 'get_commit_time', lambda *_: "xx")
- monkeypatch.setattr(info, 'has_untracked', lambda *_: True)
- monkeypatch.setattr('os.chdir', lambda x: None)
+ monkeypatch.setattr(info, "get_head", lambda x: "repo")
+ monkeypatch.setattr(info, "run_quiet_diff", lambda *_: diff_return)
+ monkeypatch.setattr(info, "get_commit_msg", lambda *_: "msg")
+ monkeypatch.setattr(info, "get_commit_time", lambda *_: "xx")
+ monkeypatch.setattr(info, "has_untracked", lambda *_: True)
+ monkeypatch.setattr("os.chdir", lambda x: None)
info.get_color_encoding.cache_clear() # avoid side effect
assert expected == next(utils.describe(*test_input))
-@pytest.mark.parametrize('path_fname, expected', [
- (PATH_FNAME, {
- 'repo1': {'path': '/a/bcd/repo1', 'type': '', 'flags': []},
- 'repo2': {'path': '/e/fgh/repo2', 'type': '', 'flags': []},
- 'xxx': {'path': '/a/b/c/repo3', 'type': '', 'flags': []},
- }),
- (PATH_FNAME_EMPTY, {}),
- (PATH_FNAME_CLASH, {
- 'repo2': {'path': '/e/fgh/repo2', 'type': '', 'flags': ['--haha', '--pp']},
- 'repo1': {'path': '/root/x/repo1', 'type': '', 'flags': []}
- }),
-])
-@patch('gita.utils.is_git', return_value=True)
-@patch('gita.common.get_config_fname')
+@pytest.mark.parametrize(
+ "path_fname, expected",
+ [
+ (
+ PATH_FNAME,
+ {
+ "repo1": {"path": "/a/bcd/repo1", "type": "", "flags": []},
+ "repo2": {"path": "/e/fgh/repo2", "type": "", "flags": []},
+ "xxx": {"path": "/a/b/c/repo3", "type": "", "flags": []},
+ },
+ ),
+ (PATH_FNAME_EMPTY, {}),
+ (
+ PATH_FNAME_CLASH,
+ {
+ "repo2": {
+ "path": "/e/fgh/repo2",
+ "type": "",
+ "flags": ["--haha", "--pp"],
+ },
+ "repo1": {"path": "/root/x/repo1", "type": "", "flags": []},
+ },
+ ),
+ ],
+)
+@patch("gita.utils.is_git", return_value=True)
+@patch("gita.common.get_config_fname")
def test_get_repos(mock_path_fname, _, path_fname, expected):
mock_path_fname.return_value = path_fname
utils.get_repos.cache_clear()
assert utils.get_repos() == expected
-@patch('gita.common.get_config_dir')
+@patch("gita.common.get_config_dir")
def test_get_context(mock_config_dir):
mock_config_dir.return_value = TEST_DIR
utils.get_context.cache_clear()
- assert utils.get_context() == TEST_DIR / 'xx.context'
+ assert utils.get_context() == TEST_DIR / "xx.context"
- mock_config_dir.return_value = '/'
+ mock_config_dir.return_value = "/"
utils.get_context.cache_clear()
assert utils.get_context() == None
-@pytest.mark.parametrize('group_fname, expected', [
- (GROUP_FNAME, {'xx': {'repos': ['a', 'b'], 'path': ''},
- 'yy': {'repos': ['a', 'c', 'd'], 'path': ''}}),
-])
-@patch('gita.common.get_config_fname')
-def test_get_groups(mock_group_fname, group_fname, expected):
+@pytest.mark.parametrize(
+ "group_fname, expected",
+ [
+ (
+ GROUP_FNAME,
+ {
+ "xx": {"repos": ["a", "b"], "path": ""},
+ "yy": {"repos": ["a", "c", "d"], "path": ""},
+ },
+ ),
+ ],
+)
+@patch("gita.common.get_config_fname")
+@patch("gita.utils.get_repos", return_value={"a": "", "b": "", "c": "", "d": ""})
+def test_get_groups(_, mock_group_fname, group_fname, expected):
mock_group_fname.return_value = group_fname
utils.get_groups.cache_clear()
assert utils.get_groups() == expected
-@patch('os.path.isfile', return_value=True)
-@patch('os.path.getsize', return_value=True)
+@patch("os.path.isfile", return_value=True)
+@patch("os.path.getsize", return_value=True)
def test_custom_push_cmd(*_):
- with patch('builtins.open',
- mock_open(read_data='{"push":{"cmd":"hand","help":"me","allow_all":true}}')):
+ with patch(
+ "builtins.open",
+ mock_open(read_data='{"push":{"cmd":"hand","help":"me","allow_all":true}}'),
+ ):
cmds = utils.get_cmds_from_files()
- assert cmds['push'] == {'cmd': 'hand', 'help': 'me', 'allow_all': True}
+ assert cmds["push"] == {"cmd": "hand", "help": "me", "allow_all": True}
@pytest.mark.parametrize(
- 'path_input, expected',
+ "path_input, expected",
[
- (['/home/some/repo'], '/home/some/repo,some/repo,,\r\n'), # add one new
- (['/home/some/repo1', '/repo2'],
- {'/repo2,repo2,,\r\n', # add two new
- '/home/some/repo1,repo1,,\r\n'}), # add two new
- (['/home/some/repo1', '/nos/repo'],
- '/home/some/repo1,repo1,,\r\n'), # add one old one new
- ])
-@patch('os.makedirs')
-@patch('gita.utils.is_git', return_value=True)
+ (["/home/some/repo"], "/home/some/repo,some/repo,,\r\n"), # add one new
+ (
+ ["/home/some/repo1", "/repo2"],
+ {"/repo2,repo2,,\r\n", "/home/some/repo1,repo1,,\r\n"}, # add two new
+ ), # add two new
+ (
+ ["/home/some/repo1", "/nos/repo"],
+ "/home/some/repo1,repo1,,\r\n",
+ ), # add one old one new
+ ],
+)
+@patch("os.makedirs")
+@patch("gita.utils.is_git", return_value=True)
def test_add_repos(_0, _1, path_input, expected, monkeypatch):
- monkeypatch.setenv('XDG_CONFIG_HOME', '/config')
- with patch('builtins.open', mock_open()) as mock_file:
- utils.add_repos({'repo': {'path': '/nos/repo'}}, path_input)
- mock_file.assert_called_with('/config/gita/repos.csv', 'a+', newline='')
+ monkeypatch.setenv("XDG_CONFIG_HOME", "/config")
+ with patch("builtins.open", mock_open()) as mock_file:
+ utils.add_repos({"repo": {"path": "/nos/repo"}}, path_input)
+ mock_file.assert_called_with("/config/gita/repos.csv", "a+", newline="")
handle = mock_file()
if type(expected) == str:
handle.write.assert_called_once_with(expected)
@@ -145,21 +248,22 @@ def test_add_repos(_0, _1, path_input, expected, monkeypatch):
assert not kwargs
-@patch('gita.utils.write_to_groups_file')
-@patch('gita.utils.write_to_repo_file')
+@patch("gita.utils.write_to_groups_file")
+@patch("gita.utils.write_to_repo_file")
def test_rename_repo(mock_write, _):
- repos = {'r1': {'path': '/a/b', 'type': None},
- 'r2': {'path': '/c/c', 'type': None}}
- utils.rename_repo(repos, 'r2', 'xxx')
- mock_write.assert_called_once_with(repos, 'w')
+ repos = {"r1": {"path": "/a/b", "type": None}, "r2": {"path": "/c/c", "type": None}}
+ utils.rename_repo(repos, "r2", "xxx")
+ mock_write.assert_called_once_with(repos, "w")
def test_async_output(capfd):
tasks = [
- utils.run_async('myrepo', '.', [
- 'python3', '-c',
- f"print({i});import time; time.sleep({i});print({i})"
- ]) for i in range(4)
+ utils.run_async(
+ "myrepo",
+ ".",
+ ["python3", "-c", f"print({i});import time; time.sleep({i});print({i})"],
+ )
+ for i in range(4)
]
# I don't fully understand why a new loop is needed here. Without a new
# loop, "pytest" fails but "pytest tests/test_utils.py" works. Maybe pytest
@@ -168,12 +272,15 @@ def test_async_output(capfd):
utils.exec_async_tasks(tasks)
out, err = capfd.readouterr()
- assert err == ''
- assert out == 'myrepo: 0\nmyrepo: 0\n\nmyrepo: 1\nmyrepo: 1\n\nmyrepo: 2\nmyrepo: 2\n\nmyrepo: 3\nmyrepo: 3\n\n'
+ assert err == ""
+ assert (
+ out
+ == "myrepo: 0\nmyrepo: 0\n\nmyrepo: 1\nmyrepo: 1\n\nmyrepo: 2\nmyrepo: 2\n\nmyrepo: 3\nmyrepo: 3\n\n"
+ )
def test_is_git(tmpdir):
with tmpdir.as_cwd():
- subprocess.run('git init --bare .'.split())
+ subprocess.run("git init --bare .".split())
assert utils.is_git(Path.cwd()) is False
assert utils.is_git(Path.cwd(), include_bare=True) is True