diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/clash_path_file | 4 | ||||
| -rw-r--r-- | tests/conftest.py | 9 | ||||
| -rw-r--r-- | tests/mock_group_file | 4 | ||||
| -rw-r--r-- | tests/mock_path_file | 2 | ||||
| -rw-r--r-- | tests/test_info.py | 7 | ||||
| -rw-r--r-- | tests/test_main.py | 663 | ||||
| -rw-r--r-- | tests/test_utils.py | 306 | ||||
| -rw-r--r-- | tests/xx.context | 0 |
8 files changed, 801 insertions, 194 deletions
diff --git a/tests/clash_path_file b/tests/clash_path_file index 4abbfca..33eeae2 100644 --- a/tests/clash_path_file +++ b/tests/clash_path_file @@ -1,3 +1,3 @@ -/a/bcd/repo1,repo1 -/e/fgh/repo2,repo2 +/a/bcd/repo1,repo1, +/e/fgh/repo2,repo2,,--haha --pp /root/x/repo1,repo1 diff --git a/tests/conftest.py b/tests/conftest.py index b3e59ed..5236a90 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -8,10 +8,11 @@ def fullpath(fname: str): return str(TEST_DIR / fname) -PATH_FNAME = fullpath('mock_path_file') -PATH_FNAME_EMPTY = fullpath('empty_path_file') -PATH_FNAME_CLASH = fullpath('clash_path_file') -GROUP_FNAME = fullpath('mock_group_file') +PATH_FNAME = fullpath("mock_path_file") +PATH_FNAME_EMPTY = fullpath("empty_path_file") +PATH_FNAME_CLASH = fullpath("clash_path_file") +GROUP_FNAME = fullpath("mock_group_file") + def async_mock(): """ diff --git a/tests/mock_group_file b/tests/mock_group_file index 32f0a64..d0d950c 100644 --- a/tests/mock_group_file +++ b/tests/mock_group_file @@ -1,2 +1,2 @@ -xx: [a, b] -yy: [a, c, d] +xx:a b
+yy:a c d
diff --git a/tests/mock_path_file b/tests/mock_path_file index 2a5f9f9..81dc9ef 100644 --- a/tests/mock_path_file +++ b/tests/mock_path_file @@ -1,4 +1,4 @@ /a/bcd/repo1,repo1 -/a/b/c/repo3,xxx +/a/b/c/repo3,xxx,, /e/fgh/repo2,repo2 diff --git a/tests/test_info.py b/tests/test_info.py index 025aedc..0af8a47 100644 --- a/tests/test_info.py +++ b/tests/test_info.py @@ -4,13 +4,14 @@ from unittest.mock import patch, MagicMock from gita import info -@patch('subprocess.run') +@patch("subprocess.run") def test_run_quiet_diff(mock_run): mock_return = MagicMock() mock_run.return_value = mock_return - got = info.run_quiet_diff(['my', 'args']) + got = info.run_quiet_diff(["--flags"], ["my", "args"], "/a/b/c") mock_run.assert_called_once_with( - ['git', 'diff', '--quiet', 'my', 'args'], + ["git", "--flags", "diff", "--quiet", "my", "args"], stderr=subprocess.DEVNULL, + cwd="/a/b/c", ) assert got == mock_return.returncode diff --git a/tests/test_main.py b/tests/test_main.py index ff28111..a877160 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -1,178 +1,615 @@ import pytest from unittest.mock import patch +from pathlib import Path import argparse +import asyncio import shlex from gita import __main__ -from gita import utils +from gita import utils, info from conftest import ( - PATH_FNAME, PATH_FNAME_EMPTY, PATH_FNAME_CLASH, GROUP_FNAME, - async_mock + PATH_FNAME, + PATH_FNAME_EMPTY, + PATH_FNAME_CLASH, + GROUP_FNAME, + async_mock, + TEST_DIR, ) +@patch("gita.utils.get_repos", return_value={"aa"}) +def test_group_name(_): + got = __main__._group_name("xx") + assert got == "xx" + with pytest.raises(SystemExit): + __main__._group_name("aa") + + +class TestAdd: + @pytest.mark.parametrize( + "input, expected", + [ + (["add", "."], ""), + ], + ) + @patch("gita.common.get_config_fname") + def test_add(self, mock_path_fname, tmp_path, input, expected): + def side_effect(input, _=None): + return tmp_path / f"{input}.txt" + + mock_path_fname.side_effect = side_effect + utils.get_repos.cache_clear() + __main__.main(input) + utils.get_repos.cache_clear() + got = utils.get_repos() + assert len(got) == 1 + assert got["gita"]["type"] == expected + + +@pytest.mark.parametrize( + "path_fname, expected", + [ + (PATH_FNAME, ""), + (PATH_FNAME_CLASH, "repo2: ['--haha', '--pp']\n"), + ], +) +@patch("gita.utils.is_git", return_value=True) +@patch("gita.utils.get_groups", return_value={}) +@patch("gita.common.get_config_fname") +def test_flags(mock_path_fname, _, __, path_fname, expected, capfd): + mock_path_fname.return_value = path_fname + utils.get_repos.cache_clear() + __main__.main(["flags"]) + out, err = capfd.readouterr() + assert err == "" + assert out == expected + + class TestLsLl: - @patch('gita.utils.get_config_fname') - def testLl(self, mock_path_fname, capfd, tmp_path): - """ functional test """ + @patch("gita.common.get_config_fname") + def test_ll(self, mock_path_fname, capfd, tmp_path): + """ + functional test + """ + # avoid modifying the local configuration - mock_path_fname.return_value = tmp_path / 'path_config.txt' - __main__.main(['add', '.']) + def side_effect(input, _=None): + return tmp_path / f"{input}.txt" + + mock_path_fname.side_effect = side_effect + utils.get_repos.cache_clear() + __main__.main(["add", "."]) out, err = capfd.readouterr() - assert err == '' - assert 'Found 1 new repo(s).\n' == out + assert err == "" + assert "Found 1 new repo(s).\n" == out # in production this is not needed utils.get_repos.cache_clear() - __main__.main(['ls']) + __main__.main(["ls"]) + out, err = capfd.readouterr() + assert err == "" + assert "gita\n" == out + + __main__.main(["ll"]) out, err = capfd.readouterr() - assert err == '' - assert 'gita\n' == out + assert err == "" + assert "gita" in out + assert info.Color.end.value in out - __main__.main(['ll']) + # no color on branch name + __main__.main(["ll", "-C"]) out, err = capfd.readouterr() - assert err == '' - assert 'gita' in out + assert err == "" + assert "gita" in out + assert info.Color.end.value not in out - __main__.main(['ls', 'gita']) + __main__.main(["ls", "gita"]) out, err = capfd.readouterr() - assert err == '' - assert out.strip() == utils.get_repos()['gita'] - - def testLs(self, monkeypatch, capfd): - monkeypatch.setattr(utils, 'get_repos', - lambda: {'repo1': '/a/', 'repo2': '/b/'}) - monkeypatch.setattr(utils, 'describe', lambda x: x) - __main__.main(['ls']) + assert err == "" + assert out.strip() == utils.get_repos()["gita"]["path"] + + def test_ls(self, monkeypatch, capfd): + monkeypatch.setattr( + utils, + "get_repos", + lambda: {"repo1": {"path": "/a/"}, "repo2": {"path": "/b/"}}, + ) + monkeypatch.setattr(utils, "describe", lambda x: x) + __main__.main(["ls"]) out, err = capfd.readouterr() - assert err == '' + assert err == "" assert out == "repo1 repo2\n" - __main__.main(['ls', 'repo1']) + __main__.main(["ls", "repo1"]) out, err = capfd.readouterr() - assert err == '' - assert out == '/a/\n' - - @pytest.mark.parametrize('path_fname, expected', [ - (PATH_FNAME, - "repo1 cmaster dsu\x1b[0m msg\nrepo2 cmaster dsu\x1b[0m msg\nxxx cmaster dsu\x1b[0m msg\n"), - (PATH_FNAME_EMPTY, ""), - (PATH_FNAME_CLASH, - "repo1 cmaster dsu\x1b[0m msg\nrepo2 cmaster dsu\x1b[0m msg\nx/repo1 cmaster dsu\x1b[0m msg\n" - ), - ]) - @patch('gita.utils.is_git', return_value=True) - @patch('gita.info.get_head', return_value="master") - @patch('gita.info._get_repo_status', return_value=("d", "s", "u", "c")) - @patch('gita.info.get_commit_msg', return_value="msg") - @patch('gita.utils.get_config_fname') - def testWithPathFiles(self, mock_path_fname, _0, _1, _2, _3, path_fname, - expected, capfd): - mock_path_fname.return_value = path_fname + assert err == "" + assert out == "/a/\n" + + @pytest.mark.parametrize( + "path_fname, expected", + [ + ( + PATH_FNAME, + "repo1 \x1b[31mmaster [*+?⇕] \x1b[0m msg \nrepo2 \x1b[31mmaster [*+?⇕] \x1b[0m msg \nxxx \x1b[31mmaster [*+?⇕] \x1b[0m msg \n", + ), + (PATH_FNAME_EMPTY, ""), + ( + PATH_FNAME_CLASH, + "repo1 \x1b[31mmaster [*+?⇕] \x1b[0m msg \nrepo2 \x1b[31mmaster [*+?⇕] \x1b[0m msg \n", + ), + ], + ) + @patch("gita.utils.is_git", return_value=True) + @patch("gita.info.get_head", return_value="master") + @patch( + "gita.info._get_repo_status", + return_value=("dirty", "staged", "untracked", "diverged"), + ) + @patch("gita.info.get_commit_msg", return_value="msg") + @patch("gita.info.get_commit_time", return_value="") + @patch("gita.common.get_config_fname") + def test_with_path_files( + self, mock_path_fname, _0, _1, _2, _3, _4, path_fname, expected, capfd + ): + def side_effect(input, _=None): + if input == "repos.csv": + return path_fname + return f"/{input}" + + mock_path_fname.side_effect = side_effect utils.get_repos.cache_clear() - __main__.main(['ll']) + __main__.main(["ll"]) out, err = capfd.readouterr() print(out) - assert err == '' + assert err == "" assert out == expected -@patch('os.path.isfile', return_value=True) -@patch('gita.utils.get_config_fname', return_value='some path') -@patch('gita.utils.get_repos', return_value={'repo1': '/a/', 'repo2': '/b/'}) -@patch('gita.utils.write_to_repo_file') +@pytest.mark.parametrize( + "input, expected", + [ + ({"repo1": {"path": "/a/"}, "repo2": {"path": "/b/"}}, ""), + ], +) +@patch("subprocess.run") +@patch("gita.utils.get_repos") +def test_freeze(mock_repos, mock_run, input, expected, capfd): + mock_repos.return_value = input + __main__.main(["freeze"]) + assert mock_run.call_count == 2 + out, err = capfd.readouterr() + assert err == "" + assert out == expected + + +@patch("subprocess.run") +def test_clone_with_url(mock_run): + args = argparse.Namespace() + args.clonee = "http://abc.com/repo1" + args.preserve_path = None + args.directory = "/home/xxx" + args.from_file = False + args.dry_run = False + __main__.f_clone(args) + cmds = ["git", "clone", args.clonee] + mock_run.assert_called_once_with(cmds, cwd=args.directory) + + +@patch( + "gita.utils.parse_clone_config", + return_value=[["git@github.com:user/repo.git", "repo", "/a/repo"]], +) +@patch("gita.utils.run_async", new=async_mock()) +@patch("subprocess.run") +def test_clone_with_config_file(*_): + asyncio.set_event_loop(asyncio.new_event_loop()) + args = argparse.Namespace() + args.clonee = "freeze_filename" + args.preserve_path = False + args.directory = None + args.from_file = True + args.dry_run = False + __main__.f_clone(args) + mock_run = utils.run_async.mock + assert mock_run.call_count == 1 + cmds = ["git", "clone", "git@github.com:user/repo.git"] + mock_run.assert_called_once_with("repo", Path.cwd(), cmds) + + +@patch( + "gita.utils.parse_clone_config", + return_value=[["git@github.com:user/repo.git", "repo", "/a/repo"]], +) +@patch("gita.utils.run_async", new=async_mock()) +@patch("subprocess.run") +def test_clone_with_preserve_path(*_): + asyncio.set_event_loop(asyncio.new_event_loop()) + args = argparse.Namespace() + args.clonee = "freeze_filename" + args.directory = None + args.from_file = True + args.preserve_path = True + args.dry_run = False + __main__.f_clone(args) + mock_run = utils.run_async.mock + assert mock_run.call_count == 1 + cmds = ["git", "clone", "git@github.com:user/repo.git", "/a/repo"] + mock_run.assert_called_once_with("repo", Path.cwd(), cmds) + + +@patch("os.makedirs") +@patch("os.path.isfile", return_value=True) +@patch("gita.common.get_config_fname", return_value="some path") +@patch( + "gita.utils.get_repos", + return_value={ + "repo1": {"path": "/a/", "type": ""}, + "repo2": {"path": "/b/", "type": ""}, + }, +) +@patch("gita.utils.write_to_repo_file") def test_rm(mock_write, *_): args = argparse.Namespace() - args.repo = ['repo1'] + args.repo = ["repo1"] __main__.f_rm(args) - mock_write.assert_called_once_with({'repo2': '/b/'}, 'w') + mock_write.assert_called_once_with({"repo2": {"path": "/b/", "type": ""}}, "w") def test_not_add(): # this won't write to disk because the repo is not valid - __main__.main(['add', '/home/some/repo/']) + __main__.main(["add", "/home/some/repo/"]) -@patch('gita.utils.get_repos', return_value={'repo2': '/d/efg'}) -@patch('subprocess.run') +@patch("gita.utils.get_repos", return_value={"repo2": {"path": "/d/efg", "flags": []}}) +@patch("subprocess.run") def test_fetch(mock_run, *_): - __main__.main(['fetch']) - mock_run.assert_called_once_with(['git', 'fetch'], cwd='/d/efg') + asyncio.set_event_loop(asyncio.new_event_loop()) + __main__.main(["fetch"]) + mock_run.assert_called_once_with(["git", "fetch"], cwd="/d/efg", shell=False) @patch( - 'gita.utils.get_repos', return_value={ - 'repo1': '/a/bc', - 'repo2': '/d/efg' - }) -@patch('gita.utils.run_async', new=async_mock()) -@patch('subprocess.run') + "gita.utils.get_repos", + return_value={ + "repo1": {"path": "/a/bc", "flags": []}, + "repo2": {"path": "/d/efg", "flags": []}, + }, +) +@patch("gita.utils.run_async", new=async_mock()) +@patch("subprocess.run") def test_async_fetch(*_): - __main__.main(['fetch']) + __main__.main(["fetch"]) mock_run = utils.run_async.mock assert mock_run.call_count == 2 - cmds = ['git', 'fetch'] + cmds = ["git", "fetch"] # print(mock_run.call_args_list) - mock_run.assert_any_call('repo1', '/a/bc', cmds) - mock_run.assert_any_call('repo2', '/d/efg', cmds) + mock_run.assert_any_call("repo1", "/a/bc", cmds) + mock_run.assert_any_call("repo2", "/d/efg", cmds) -@pytest.mark.parametrize('input', [ - 'diff --name-only --staged', - "commit -am 'lala kaka'", -]) -@patch('gita.utils.get_repos', return_value={'repo7': 'path7'}) -@patch('subprocess.run') +@pytest.mark.parametrize( + "input", + [ + "diff --name-only --staged", + "commit -am 'lala kaka'", + ], +) +@patch("gita.utils.get_repos", return_value={"repo7": {"path": "path7", "flags": []}}) +@patch("subprocess.run") def test_superman(mock_run, _, input): mock_run.reset_mock() - args = ['super', 'repo7'] + shlex.split(input) + args = ["super", "repo7"] + shlex.split(input) __main__.main(args) - expected_cmds = ['git'] + shlex.split(input) - mock_run.assert_called_once_with(expected_cmds, cwd='path7') - - -@pytest.mark.parametrize('input, expected', [ - ('a', {'xx': ['b'], 'yy': ['c', 'd']}), - ("c", {'xx': ['a', 'b'], 'yy': ['a', 'd']}), - ("a b", {'yy': ['c', 'd']}), -]) -@patch('gita.utils.get_repos', return_value={'a': '', 'b': '', 'c': '', 'd': ''}) -@patch('gita.utils.get_config_fname', return_value=GROUP_FNAME) -@patch('gita.utils.write_to_groups_file') -def test_ungroup(mock_write, _, __, input, expected): - utils.get_groups.cache_clear() - args = ['ungroup'] + shlex.split(input) + expected_cmds = ["git"] + shlex.split(input) + mock_run.assert_called_once_with(expected_cmds, cwd="path7", shell=False) + + +@pytest.mark.parametrize( + "input", + [ + "diff --name-only --staged", + "commit -am 'lala kaka'", + ], +) +@patch("gita.utils.get_repos", return_value={"repo7": {"path": "path7", "flags": []}}) +@patch("subprocess.run") +def test_shell(mock_run, _, input): + mock_run.reset_mock() + args = ["shell", "repo7", input] __main__.main(args) - mock_write.assert_called_once_with(expected, 'w') + expected_cmds = input + mock_run.assert_called_once_with( + expected_cmds, cwd="path7", shell=True, stderr=-2, stdout=-1 + ) -@patch('gita.utils.get_config_fname', return_value=GROUP_FNAME) -def test_group_display(_, capfd): - args = argparse.Namespace() - args.to_group = None - utils.get_groups.cache_clear() - __main__.f_group(args) - out, err = capfd.readouterr() - assert err == '' - assert 'xx: a b\nyy: a c d\n' == out +class TestContext: + @patch("gita.utils.get_context", return_value=None) + def test_display_no_context(self, _, capfd): + __main__.main(["context"]) + out, err = capfd.readouterr() + assert err == "" + assert "Context is not set\n" == out + + @patch("gita.utils.get_context", return_value=Path("gname.context")) + @patch("gita.utils.get_groups", return_value={"gname": {"repos": ["a", "b"]}}) + def test_display_context(self, _, __, capfd): + __main__.main(["context"]) + out, err = capfd.readouterr() + assert err == "" + assert "gname: a b\n" == out + + @patch("gita.utils.get_context") + def test_reset(self, mock_ctx): + __main__.main(["context", "none"]) + mock_ctx.return_value.unlink.assert_called() + + @patch("gita.utils.get_context", return_value=None) + @patch("gita.common.get_config_dir", return_value=TEST_DIR) + @patch("gita.utils.get_groups", return_value={"lala": ["b"], "kaka": []}) + def test_set_first_time(self, *_): + ctx = TEST_DIR / "lala.context" + assert not ctx.is_file() + __main__.main(["context", "lala"]) + assert ctx.is_file() + ctx.unlink() + + @patch("gita.common.get_config_dir", return_value=TEST_DIR) + @patch("gita.utils.get_groups", return_value={"lala": ["b"], "kaka": []}) + @patch("gita.utils.get_context") + def test_set_second_time(self, mock_ctx, *_): + __main__.main(["context", "kaka"]) + mock_ctx.return_value.rename.assert_called() + + +class TestGroupCmd: + @patch("gita.common.get_config_fname", return_value=GROUP_FNAME) + def test_ls(self, _, capfd): + args = argparse.Namespace() + args.to_group = None + args.group_cmd = "ls" + utils.get_groups.cache_clear() + __main__.f_group(args) + out, err = capfd.readouterr() + 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): + args = argparse.Namespace() + args.to_group = None + args.group_cmd = None + args.to_show = None + utils.get_groups.cache_clear() + __main__.f_group(args) + out, err = capfd.readouterr() + assert err == "" + assert ( + out + == "\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): + args = argparse.Namespace() + args.to_group = None + args.group_cmd = None + args.to_show = "yy" + utils.get_groups.cache_clear() + __main__.f_group(args) + out, err = capfd.readouterr() + 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, *_): + args = argparse.Namespace() + args.gname = "xx" + args.new_name = "zz" + args.group_cmd = "rename" + utils.get_groups.cache_clear() + __main__.f_group(args) + expected = { + "yy": {"repos": ["a", "c", "d"], "path": ""}, + "zz": {"repos": ["a", "b"], "path": ""}, + } + mock_write.assert_called_once_with(expected, "w") -@patch('gita.utils.is_git', return_value=True) -@patch('gita.utils.get_config_fname', return_value=PATH_FNAME) -@patch('gita.utils.rename_repo') + @patch("gita.info.get_color_encoding", return_value=info.default_colors) + @patch("gita.common.get_config_fname", return_value=GROUP_FNAME) + def test_rename_error(self, *_): + utils.get_groups.cache_clear() + with pytest.raises(SystemExit, match="1"): + __main__.main("group rename xx yy".split()) + + @pytest.mark.parametrize( + "input, expected", + [ + ("xx", {"yy": {"repos": ["a", "c", "d"], "path": ""}}), + ("xx yy", {}), + ], + ) + @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_rm(self, mock_write, _, __, input, expected): + utils.get_groups.cache_clear() + args = ["group", "rm"] + shlex.split(input) + __main__.main(args) + mock_write.assert_called_once_with(expected, "w") + + @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_add(self, mock_write, *_): + args = argparse.Namespace() + args.to_group = ["a", "c"] + args.group_cmd = "add" + args.gname = "zz" + utils.get_groups.cache_clear() + __main__.f_group(args) + mock_write.assert_called_once_with( + {"zz": {"repos": ["a", "c"], "path": ""}}, "a+" + ) + + @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_add_to_existing(self, mock_write, *_): + args = argparse.Namespace() + args.to_group = ["a", "c"] + args.group_cmd = "add" + args.gname = "xx" + utils.get_groups.cache_clear() + __main__.f_group(args) + mock_write.assert_called_once_with( + { + "xx": {"repos": ["a", "b", "c"], "path": ""}, + "yy": {"repos": ["a", "c", "d"], "path": ""}, + }, + "w", + ) + + @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_rm_repo(self, mock_write, *_): + args = argparse.Namespace() + args.to_rm = ["a", "c"] + args.group_cmd = "rmrepo" + args.gname = "xx" + utils.get_groups.cache_clear() + __main__.f_group(args) + mock_write.assert_called_once_with( + { + "xx": {"repos": ["b"], "path": ""}, + "yy": {"repos": ["a", "c", "d"], "path": ""}, + }, + "w", + ) + + @patch("gita.common.get_config_fname") + def test_integration(self, mock_path_fname, tmp_path, capfd): + def side_effect(input, _=None): + return tmp_path / f"{input}.csv" + + mock_path_fname.side_effect = side_effect + + __main__.main("add .".split()) + utils.get_repos.cache_clear() + __main__.main("group add gita -n test".split()) + utils.get_groups.cache_clear() + __main__.main("ll test".split()) + out, err = capfd.readouterr() + assert err == "" + assert "gita" in out + + +@patch("gita.utils.is_git", return_value=True) +@patch("gita.common.get_config_fname", return_value=PATH_FNAME) +@patch("gita.utils.rename_repo") def test_rename(mock_rename, _, __): utils.get_repos.cache_clear() - args = ['rename', 'repo1', 'abc'] + args = ["rename", "repo1", "abc"] __main__.main(args) mock_rename.assert_called_once_with( - {'repo1': '/a/bcd/repo1', 'repo2': '/e/fgh/repo2', - 'xxx': '/a/b/c/repo3'}, - 'repo1', 'abc') + { + "repo1": {"path": "/a/bcd/repo1", "type": "", "flags": []}, + "xxx": {"path": "/a/b/c/repo3", "type": "", "flags": []}, + "repo2": {"path": "/e/fgh/repo2", "type": "", "flags": []}, + }, + "repo1", + "abc", + ) -@patch('os.path.isfile', return_value=False) -def test_info(mock_isfile, capfd): - __main__.f_info(None) +class TestInfo: + @patch("gita.common.get_config_fname", return_value="") + def test_ll(self, _, capfd): + args = argparse.Namespace() + args.info_cmd = None + __main__.f_info(args) + out, err = capfd.readouterr() + assert ( + "In use: branch,commit_msg,commit_time\nUnused: branch_name,path\n" == out + ) + assert err == "" + + @patch("gita.common.get_config_fname") + def test_add(self, mock_get_fname, tmpdir): + args = argparse.Namespace() + args.info_cmd = "add" + args.info_item = "path" + with tmpdir.as_cwd(): + csv_config = Path.cwd() / "info.csv" + mock_get_fname.return_value = csv_config + __main__.f_info(args) + items = info.get_info_items() + assert items == ["branch", "commit_msg", "commit_time", "path"] + + @patch("gita.common.get_config_fname") + def test_rm(self, mock_get_fname, tmpdir): + args = argparse.Namespace() + args.info_cmd = "rm" + args.info_item = "commit_msg" + with tmpdir.as_cwd(): + csv_config = Path.cwd() / "info.csv" + mock_get_fname.return_value = csv_config + __main__.f_info(args) + items = info.get_info_items() + assert items == ["branch", "commit_time"] + + +@patch("gita.common.get_config_fname") +def test_set_color(mock_get_fname, tmpdir): + args = argparse.Namespace() + args.color_cmd = "set" + args.color = "b_white" + args.situation = "no_remote" + with tmpdir.as_cwd(): + csv_config = Path.cwd() / "colors.csv" + mock_get_fname.return_value = csv_config + __main__.f_color(args) + + info.get_color_encoding.cache_clear() # avoid side effect + items = info.get_color_encoding() + info.get_color_encoding.cache_clear() # avoid side effect + assert items == { + "no_remote": "b_white", + "in_sync": "green", + "diverged": "red", + "local_ahead": "purple", + "remote_ahead": "yellow", + } + + +@pytest.mark.parametrize( + "input, expected", + [ + ({"repo1": {"path": "/a/"}, "repo2": {"path": "/b/"}}, ""), + ], +) +@patch("gita.utils.write_to_groups_file") +@patch("gita.utils.write_to_repo_file") +@patch("gita.utils.get_repos") +def test_clear( + mock_repos, + mock_write_to_repo_file, + mock_write_to_groups_file, + input, + expected, + capfd, +): + mock_repos.return_value = input + __main__.main(["clear"]) + assert mock_write_to_repo_file.call_count == 1 + mock_write_to_repo_file.assert_called_once_with({}, "w") + assert mock_write_to_groups_file.call_count == 1 + mock_write_to_groups_file.assert_called_once_with({}, "w") out, err = capfd.readouterr() - assert 'In use: branch,commit_msg\nUnused: path\n' == out - assert err == '' + assert err == "" + assert out == expected diff --git a/tests/test_utils.py b/tests/test_utils.py index 3128041..2936f0e 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1,111 +1,269 @@ import pytest import asyncio +import subprocess +from pathlib import Path 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, + PATH_FNAME, + PATH_FNAME_EMPTY, + PATH_FNAME_CLASH, + GROUP_FNAME, + TEST_DIR, ) -@pytest.mark.parametrize('test_input, diff_return, expected', [ - ({ - 'abc': '/root/repo/' - }, True, 'abc \x1b[31mrepo *+_ \x1b[0m msg'), - ({ - 'repo': '/root/repo2/' - }, False, 'repo \x1b[32mrepo _ \x1b[0m msg'), -]) +@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")), + ], +) +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"]}, + }, + ), + ], +) +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", + ), + ], +) 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, 'has_untracked', lambda: True) - monkeypatch.setattr('os.chdir', lambda x: None) - print('expected: ', repr(expected)) - print('got: ', repr(next(utils.describe(test_input)))) - assert expected == next(utils.describe(test_input)) - - -@pytest.mark.parametrize('path_fname, expected', [ - (PATH_FNAME, { - 'repo1': '/a/bcd/repo1', - 'repo2': '/e/fgh/repo2', - 'xxx': '/a/b/c/repo3', - }), - (PATH_FNAME_EMPTY, {}), - (PATH_FNAME_CLASH, { - 'repo1': '/a/bcd/repo1', - 'repo2': '/e/fgh/repo2', - 'x/repo1': '/root/x/repo1' - }), -]) -@patch('gita.utils.is_git', return_value=True) -@patch('gita.utils.get_config_fname') + 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(info, "get_common_commit", lambda x: "") + + 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") 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 -@pytest.mark.parametrize('group_fname, expected', [ - (GROUP_FNAME, {'xx': ['a', 'b'], 'yy': ['a', 'c', 'd']}), -]) -@patch('gita.utils.get_config_fname') -def test_get_groups(mock_group_fname, group_fname, expected): +@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" + + 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") +@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:\n cmd: hand\n help: me')): + 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'} + 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,repo\n'), # add one new - (['/home/some/repo1', '/repo2'], - {'/repo2,repo2\n/home/some/repo1,repo1\n', # add two new - '/home/some/repo1,repo1\n/repo2,repo2\n'}), # add two new - (['/home/some/repo1', '/nos/repo'], - '/home/some/repo1,repo1\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': '/nos/repo'}, path_input) - mock_file.assert_called_with('/config/gita/repo_path', 'a+') + 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) else: - handle.write.assert_called_once() + # the write order is random + assert handle.write.call_count == 2 args, kwargs = handle.write.call_args assert args[0] in expected assert not kwargs -@patch('gita.utils.write_to_repo_file') -def test_rename_repo(mock_write): - utils.rename_repo({'r1': '/a/b', 'r2': '/c/c'}, 'r2', 'xxx') - mock_write.assert_called_once_with({'r1': '/a/b', 'xxx': '/c/c'}, 'w') +@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") 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 @@ -114,5 +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()) + assert utils.is_git(Path.cwd()) is False + assert utils.is_git(Path.cwd(), include_bare=True) is True diff --git a/tests/xx.context b/tests/xx.context new file mode 100644 index 0000000..e69de29 --- /dev/null +++ b/tests/xx.context |