Loading lib/tool_shed/util/hg_util.py +8 −4 Original line number Diff line number Diff line Loading @@ -5,6 +5,7 @@ import subprocess import tempfile from datetime import datetime from time import gmtime from typing import TYPE_CHECKING from galaxy.tool_shed.util import basic_util from galaxy.tool_shed.util.hg_util import ( Loading @@ -21,12 +22,15 @@ from galaxy.tool_shed.util.hg_util import ( ) from galaxy.util import unicodify if TYPE_CHECKING: from galaxy.util.path import StrPath log = logging.getLogger(__name__) INITIAL_CHANGELOG_HASH = "000000000000" def add_changeset(repo_path, path_to_filename_in_archive): def add_changeset(repo_path: "StrPath", path_to_filename_in_archive: "StrPath"): try: subprocess.check_output(["hg", "add", path_to_filename_in_archive], stderr=subprocess.STDOUT, cwd=repo_path) except Exception as e: Loading @@ -51,7 +55,7 @@ def archive_repository_revision(app, repository, archive_dir, changeset_revision raise Exception(error_message) def commit_changeset(repo_path: str, full_path_to_changeset: str, username: str, message: str) -> None: def commit_changeset(repo_path: "StrPath", full_path_to_changeset: "StrPath", username: str, message: str) -> None: try: subprocess.check_output( ["hg", "commit", "-u", username, "-m", message, full_path_to_changeset], Loading Loading @@ -215,7 +219,7 @@ def get_rev_label_from_changeset_revision(repo, changeset_revision, include_date return rev, label def remove_path(repo_path, selected_file): def remove_path(repo_path: "StrPath", selected_file: "StrPath"): cmd = ["hg", "remove", "--force", selected_file] try: subprocess.check_output(cmd, stderr=subprocess.STDOUT, cwd=repo_path) Loading @@ -236,7 +240,7 @@ def remove_path(repo_path, selected_file): raise Exception(error_message) def init_repository(repo_path): def init_repository(repo_path: "StrPath"): """ Create a new Mercurial repository in the given directory. """ Loading test/unit/tool_shed/test_hg_util.py +51 −37 Original line number Diff line number Diff line from typing import TYPE_CHECKING import pytest from tool_shed.util import hg_util if TYPE_CHECKING: from pathlib import Path def test_init_repository(tmpdir): hg_util.init_repository(str(tmpdir)) assert (tmpdir / ".hg").exists() def test_init_repository(tmp_path): hg_util.init_repository(tmp_path) assert (tmp_path / ".hg").exists() def test_init_repository_fails(tmpdir): test_init_repository(tmpdir) with pytest.raises(Exception) as exc_info: test_init_repository(tmpdir) assert "Error initializing repository" in str(exc_info.value) def test_init_repository_fails(tmp_path): hg_util.init_repository(tmp_path) with pytest.raises(Exception, match="Error initializing repository"): test_init_repository(tmp_path) def test_add_file_and_commmit_changeset(tmpdir): test_init_repository(tmpdir) path_to_add = tmpdir / "test.txt" path_to_add.write("test") hg_util.add_changeset(str(tmpdir), str(path_to_add)) hg_util.commit_changeset(str(tmpdir), str(path_to_add), "testuser", "testcommit") def _add_file_and_commmit_changeset(tmp_path: "Path"): path_to_add = tmp_path / "test.txt" path_to_add.write_text("test") hg_util.add_changeset(tmp_path, path_to_add) hg_util.commit_changeset(tmp_path, path_to_add, "testuser", "testcommit") return path_to_add def test_add_dir_and_commit_changeset(tmpdir): test_init_repository(tmpdir) path_to_add = tmpdir.mkdir("abc") (path_to_add / "test.txt").write("bla") hg_util.add_changeset(str(tmpdir), str(path_to_add)) hg_util.commit_changeset(str(tmpdir), str(path_to_add), "testuser", "testcommit") return path_to_add def test_add_file_and_commmit_changeset(tmp_path): hg_util.init_repository(tmp_path) _add_file_and_commmit_changeset(tmp_path) def _add_dir_and_commit_changeset(tmp_path: "Path"): dir_to_add = tmp_path / "abc" dir_to_add.mkdir() (dir_to_add / "test.txt").write_text("bla") hg_util.add_changeset(tmp_path, dir_to_add) hg_util.commit_changeset(tmp_path, dir_to_add, "testuser", "testcommit") return dir_to_add def test_add_dir_and_commit_changeset(tmp_path): hg_util.init_repository(tmp_path) _add_dir_and_commit_changeset(tmp_path) def test_remove_tracked_file(tmpdir): path_to_remove = test_add_file_and_commmit_changeset(tmpdir) hg_util.remove_path(str(tmpdir), path_to_remove) def test_remove_tracked_file(tmp_path): hg_util.init_repository(tmp_path) path_to_remove = _add_file_and_commmit_changeset(tmp_path) hg_util.remove_path(tmp_path, path_to_remove) def test_remove_untracked_file(tmpdir): test_init_repository(tmpdir) untracked_path = tmpdir / "untracked.txt" untracked_path.write("bla") hg_util.remove_path(str(tmpdir), str(untracked_path)) def test_remove_untracked_file(tmp_path): hg_util.init_repository(tmp_path) untracked_path = tmp_path / "untracked.txt" untracked_path.write_text("bla") hg_util.remove_path(tmp_path, untracked_path) def test_remove_tracked_dir(tmpdir): path_to_remove = test_add_dir_and_commit_changeset(tmpdir) hg_util.remove_path(str(tmpdir), path_to_remove) def test_remove_tracked_dir(tmp_path): hg_util.init_repository(tmp_path) dir_to_remove = _add_dir_and_commit_changeset(tmp_path) hg_util.remove_path(tmp_path, dir_to_remove) def test_remove_nonexistant_file_fails(tmpdir): test_init_repository(tmpdir) with pytest.raises(Exception) as exc_info: hg_util.remove_path(str(tmpdir), str(tmpdir / "some path")) assert "Error removing path" in str(exc_info.value) def test_remove_nonexistant_file_fails(tmp_path): hg_util.init_repository(tmp_path) with pytest.raises(Exception, match="Error removing path"): hg_util.remove_path(tmp_path, tmp_path / "some path") Loading
lib/tool_shed/util/hg_util.py +8 −4 Original line number Diff line number Diff line Loading @@ -5,6 +5,7 @@ import subprocess import tempfile from datetime import datetime from time import gmtime from typing import TYPE_CHECKING from galaxy.tool_shed.util import basic_util from galaxy.tool_shed.util.hg_util import ( Loading @@ -21,12 +22,15 @@ from galaxy.tool_shed.util.hg_util import ( ) from galaxy.util import unicodify if TYPE_CHECKING: from galaxy.util.path import StrPath log = logging.getLogger(__name__) INITIAL_CHANGELOG_HASH = "000000000000" def add_changeset(repo_path, path_to_filename_in_archive): def add_changeset(repo_path: "StrPath", path_to_filename_in_archive: "StrPath"): try: subprocess.check_output(["hg", "add", path_to_filename_in_archive], stderr=subprocess.STDOUT, cwd=repo_path) except Exception as e: Loading @@ -51,7 +55,7 @@ def archive_repository_revision(app, repository, archive_dir, changeset_revision raise Exception(error_message) def commit_changeset(repo_path: str, full_path_to_changeset: str, username: str, message: str) -> None: def commit_changeset(repo_path: "StrPath", full_path_to_changeset: "StrPath", username: str, message: str) -> None: try: subprocess.check_output( ["hg", "commit", "-u", username, "-m", message, full_path_to_changeset], Loading Loading @@ -215,7 +219,7 @@ def get_rev_label_from_changeset_revision(repo, changeset_revision, include_date return rev, label def remove_path(repo_path, selected_file): def remove_path(repo_path: "StrPath", selected_file: "StrPath"): cmd = ["hg", "remove", "--force", selected_file] try: subprocess.check_output(cmd, stderr=subprocess.STDOUT, cwd=repo_path) Loading @@ -236,7 +240,7 @@ def remove_path(repo_path, selected_file): raise Exception(error_message) def init_repository(repo_path): def init_repository(repo_path: "StrPath"): """ Create a new Mercurial repository in the given directory. """ Loading
test/unit/tool_shed/test_hg_util.py +51 −37 Original line number Diff line number Diff line from typing import TYPE_CHECKING import pytest from tool_shed.util import hg_util if TYPE_CHECKING: from pathlib import Path def test_init_repository(tmpdir): hg_util.init_repository(str(tmpdir)) assert (tmpdir / ".hg").exists() def test_init_repository(tmp_path): hg_util.init_repository(tmp_path) assert (tmp_path / ".hg").exists() def test_init_repository_fails(tmpdir): test_init_repository(tmpdir) with pytest.raises(Exception) as exc_info: test_init_repository(tmpdir) assert "Error initializing repository" in str(exc_info.value) def test_init_repository_fails(tmp_path): hg_util.init_repository(tmp_path) with pytest.raises(Exception, match="Error initializing repository"): test_init_repository(tmp_path) def test_add_file_and_commmit_changeset(tmpdir): test_init_repository(tmpdir) path_to_add = tmpdir / "test.txt" path_to_add.write("test") hg_util.add_changeset(str(tmpdir), str(path_to_add)) hg_util.commit_changeset(str(tmpdir), str(path_to_add), "testuser", "testcommit") def _add_file_and_commmit_changeset(tmp_path: "Path"): path_to_add = tmp_path / "test.txt" path_to_add.write_text("test") hg_util.add_changeset(tmp_path, path_to_add) hg_util.commit_changeset(tmp_path, path_to_add, "testuser", "testcommit") return path_to_add def test_add_dir_and_commit_changeset(tmpdir): test_init_repository(tmpdir) path_to_add = tmpdir.mkdir("abc") (path_to_add / "test.txt").write("bla") hg_util.add_changeset(str(tmpdir), str(path_to_add)) hg_util.commit_changeset(str(tmpdir), str(path_to_add), "testuser", "testcommit") return path_to_add def test_add_file_and_commmit_changeset(tmp_path): hg_util.init_repository(tmp_path) _add_file_and_commmit_changeset(tmp_path) def _add_dir_and_commit_changeset(tmp_path: "Path"): dir_to_add = tmp_path / "abc" dir_to_add.mkdir() (dir_to_add / "test.txt").write_text("bla") hg_util.add_changeset(tmp_path, dir_to_add) hg_util.commit_changeset(tmp_path, dir_to_add, "testuser", "testcommit") return dir_to_add def test_add_dir_and_commit_changeset(tmp_path): hg_util.init_repository(tmp_path) _add_dir_and_commit_changeset(tmp_path) def test_remove_tracked_file(tmpdir): path_to_remove = test_add_file_and_commmit_changeset(tmpdir) hg_util.remove_path(str(tmpdir), path_to_remove) def test_remove_tracked_file(tmp_path): hg_util.init_repository(tmp_path) path_to_remove = _add_file_and_commmit_changeset(tmp_path) hg_util.remove_path(tmp_path, path_to_remove) def test_remove_untracked_file(tmpdir): test_init_repository(tmpdir) untracked_path = tmpdir / "untracked.txt" untracked_path.write("bla") hg_util.remove_path(str(tmpdir), str(untracked_path)) def test_remove_untracked_file(tmp_path): hg_util.init_repository(tmp_path) untracked_path = tmp_path / "untracked.txt" untracked_path.write_text("bla") hg_util.remove_path(tmp_path, untracked_path) def test_remove_tracked_dir(tmpdir): path_to_remove = test_add_dir_and_commit_changeset(tmpdir) hg_util.remove_path(str(tmpdir), path_to_remove) def test_remove_tracked_dir(tmp_path): hg_util.init_repository(tmp_path) dir_to_remove = _add_dir_and_commit_changeset(tmp_path) hg_util.remove_path(tmp_path, dir_to_remove) def test_remove_nonexistant_file_fails(tmpdir): test_init_repository(tmpdir) with pytest.raises(Exception) as exc_info: hg_util.remove_path(str(tmpdir), str(tmpdir / "some path")) assert "Error removing path" in str(exc_info.value) def test_remove_nonexistant_file_fails(tmp_path): hg_util.init_repository(tmp_path) with pytest.raises(Exception, match="Error removing path"): hg_util.remove_path(tmp_path, tmp_path / "some path")