feat: Remove dependency on git (#22)

* Linting

* Add dependency

* Get last version without git

* List commits without git

* Remove unused methods

* Minor fix

* Return tag name as bytes

* Update readme

* Update list_commits method

* Update formatting

* Typing

* Version 0.4.0

README.md

@@ -13,6 +13,12 @@ based on the conventional commits.
 Use the `changelog` command to generate a nicely formatted changelog
 (Github markdown compatible).
 
+## Requirements
+`convbump` does not have any external dependencies.
+
+`convbump` uses a pure Python library to access the Git repository and so does not
+require a `git` executable.
+
 ## Development
 The application is written in Python and uses
 [Poetry](https://python-poetry.org/docs/) to configure the package and manage

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "convbump"
-version = "0.3.1"
+version = "0.4.0"
 description = "Tool for Conventional Commits"
 authors = ["Max Kovykov <maxim.kovykov@avast.com>"]
 license = "BSD-3-Clause"
@@ -30,6 +30,7 @@ include = ["py.typed"]
 python = "^3.7"
 semver = "^2.13.0"
 click = "^8.0.3"
+dulwich = "^0.20.32"
 
 [tool.poetry.dev-dependencies]
 mypy = "^0.931"

src/convbump/conventional.py

@@ -84,7 +84,7 @@ def from_git_commit(cls, git_commit: Commit) -> ConventionalCommit:
             is_breaking=is_breaking,
             subject=subject,
             body=git_commit.body or None,
-            hash=git_commit.hash,
+            hash=git_commit.hash.decode()[:7],
         )
 
     def format(self) -> str:

src/convbump/git.py

@@ -1,16 +1,18 @@
 import re
-import subprocess
 from dataclasses import dataclass
+from operator import itemgetter
 from pathlib import Path
-from typing import Iterable, Iterator, List, Optional, Tuple
+from typing import List, Optional, Tuple
 
+from dulwich.objects import Commit as RawCommit
+from dulwich.repo import Repo
 from semver import VersionInfo as Version
 
 # Default version tag regex. It is used to find the last valid git tag.
 # This regex will match the following tags: "v1", "v1.0", "v1.0.0"
 TAG_REGEX = re.compile(
     r"""^
-        v                           # literal 'v'
+        refs/tags/v                 # literal 'refs/tags/v'
         (?P<major>\d+)              # Major version is required
         (?:\.                       # Optional non-capturing group with minor and patch versions
             (?P<minor>\d+)          # Optional minor version
@@ -27,72 +29,102 @@
 @dataclass(frozen=True)
 class Commit:
 
-    hash: str
+    hash: bytes
     subject: str
     body: Optional[str]
 
 
+def parse_message(message: str) -> Tuple[str, str]:  # Tuple[subject, message]
+    """Parse git message into subject and body."""
+
+    paragraphs = message.strip().split("\n\n")
+    maybe_subject = paragraphs[0]
+    if "\n" in maybe_subject:
+        subject = ""
+        body = message
+    else:
+        subject = maybe_subject
+        body = "\n\n".join(paragraphs[1:])
+
+    return subject, body
+
+
 class Git:
     def __init__(self, path: Path) -> None:
         self.path = path
+        self.repo = Repo(self.path)  # type: ignore
 
-    def list_commits(self, from_ref: Optional[str], to_ref: Optional[str] = None) -> List[Commit]:
-        def iter_commtis(commit_ids: Iterable[str]) -> Iterator[Commit]:
-            for commit_id in commit_ids:
-                yield self._parse_commits(["git", "log", "-1", r"--format=%h%n%s%n%b", commit_id])
+    def list_commits(
+        self, from_tag: Optional[bytes], to_tag: Optional[bytes] = None
+    ) -> List[Commit]:
+        """List commits from `from_tag` to `to_tag`. If `to_tag` is None,
+        list commits until the latest commits.
 
-        if from_ref is None:
-            command = ["git", "log", "--format=%H"]
-        elif from_ref and to_ref:
-            command = ["git", "log", "--format=%H", f"{from_ref}..{to_ref}"]
-        else:
-            command = ["git", "log", "--format=%H", f"{from_ref}..HEAD"]
+        If `from_tag` list commits from the first commits.
+
+        If `from_tag` and `to_tag` is None, list all commits.
+
+        `from_tag` and `to_tag` must be a full tag name:
+
+            refs/tags/tag_name
+        """
 
         try:
-            commit_ids = reversed(self._check_output(command).splitlines())
-        except subprocess.CalledProcessError:
+            walker = self.repo.get_walker(reverse=True)  # type: ignore
+        except KeyError:  # Repo is empty
             return []
 
-        return [commit for commit in iter_commtis(commit_ids)]
-
-    def retrieve_last_commit(self) -> str:
-        return self._check_output(["git", "log", "-1", "--format=%B"])
+        # Convert from_ref to SHA if it is a tag name
+        from_sha = self.repo.get_peeled(from_tag) if from_tag else None  # type: ignore
 
-    def retrieve_last_version(self) -> Tuple[Optional[str], Optional[Version]]:
-        """Retrieve last valid version from a tag. Any non-valid version tags are skipped.
-        Return a tuple with tag name and version or None."""
+        # Convert to_ref to SHA if it is a tag name
+        to_sha = self.repo.get_peeled(to_tag) if to_tag else None  # type: ignore
 
-        tags = reversed(self._check_output(["git", "tag", "--sort=v:refname"]).split("\n"))
-        for tag in tags:
-            match = TAG_REGEX.match(tag)
-            if match:
-                match_dict = match.groupdict()
-                return tag, Version(
-                    match_dict["major"], match_dict["minor"] or 0, match_dict["patch"] or 0
-                )
+        if from_sha is None:
+            add = True
         else:
-            return None, None
+            add = False
+        commits: List[Commit] = []
+        for entry in walker:
+            commit: RawCommit = entry.commit
+            hash = commit.id
 
-    def retrieve_tag_body(self, tag: str) -> str:
-        return self._check_output(["git", "tag", "-l", "--format=%(body)", tag])
+            if add:
+                message = commit.message.decode()
+                subject, body = parse_message(message)
+                commits.append(Commit(hash, subject, body or None))
 
-    def retrieve_tag_subject(self, tag: str) -> str:
-        return self._check_output(["git", "tag", "-l", "--format=%(subject)", tag])
+            if hash == from_sha:
+                add = True
+            if to_tag and hash == to_sha:
+                break
 
-    def _check_output(self, args: List[str]) -> str:
-        maybe_output = subprocess.check_output(args, cwd=self.path)
-        if maybe_output is not None:
-            return maybe_output.strip().decode("utf-8")
+        return commits
 
-        raise ValueError("git command return unexpected empty output")
+    def retrieve_last_version(self) -> Tuple[Optional[bytes], Optional[Version]]:
+        """Retrieve last valid version from a tag. Any non-valid version tags are skipped.
+        Return a tuple with tag name and version or None."""
 
-    def _parse_commits(self, args: List[str]) -> Commit:
-        maybe_output = subprocess.check_output(args, cwd=self.path)
-        if maybe_output is not None:
-            output = maybe_output.strip().decode("utf-8")
+        tag_refs = filter(lambda ref: ref.startswith(b"refs/tags"), self.repo.get_refs())
 
-            hash, subject, *body = output.split("\n")
+        tag_version_list = []
+        for tag in tag_refs:
+            match = TAG_REGEX.match(tag.decode())
+            if match:
+                match_dict = match.groupdict()
+
+                tag_version_list.append(
+                    (
+                        tag,
+                        Version(
+                            match_dict["major"], match_dict["minor"] or 0, match_dict["patch"] or 0
+                        ),
+                    )
+                )
 
-            return Commit(hash, subject, "\n".join(body) if body else None)
+        sorted_tags = sorted(tag_version_list, key=itemgetter(1))
 
-        raise ValueError("git command return unexpected empty output")
+        if sorted_tags:
+            return sorted_tags[-1]
+        else:
+            return None, None

tests/conftest.py

@@ -45,8 +45,8 @@ def git_config() -> None:
 
 @pytest.fixture()
 def create_git_repository(
-    git_config: None, tmp_path: Path
-) -> GitFactory:  # pylint: disable=unused-argument
+    git_config: None, tmp_path: Path  # pylint: disable=unused-argument
+) -> GitFactory:
     def _(commits: Collection[Union[CommitTuple, str]]) -> Git:
         subprocess.check_call(["git", "init"], cwd=tmp_path)
 
@@ -61,7 +61,7 @@ def _(commits: Collection[Union[CommitTuple, str]]) -> Git:
 
             subprocess.check_call(["git", "commit", "--allow-empty", "-m", message], cwd=tmp_path)
             if tag is not None:
-                subprocess.check_call(["git", "tag", tag], cwd=tmp_path)
+                subprocess.check_call(["git", "tag", "-a", "-m", "message", tag], cwd=tmp_path)
 
         return Git(tmp_path)
 

tests/test_conventional.py

@@ -78,7 +78,7 @@ def make_commit(commit_message: str) -> Commit:
     subject, *rest = commit_message.split("\n")
     body = "\n".join(rest) if rest else None
 
-    return Commit("f392ca5", subject, body)
+    return Commit(b"f392ca5", subject, body)
 
 
 COMMIT_PARAMS = [

tests/test_git.py

@@ -1,8 +1,24 @@
+from typing import Tuple
+
 import pytest
 from conftest import BREAKING_FEATURE, INITIAL_COMMIT, GitFactory
 from semver import VersionInfo as Version
 
-from convbump.git import TAG_REGEX
+from convbump.git import TAG_REGEX, parse_message
+
+MESSAGES = [
+    ("Subject\n\nBody", ("Subject", "Body")),
+    ("Subject\n\nParagraph\n\nParagraph", ("Subject", "Paragraph\n\nParagraph")),
+    ("Subject", ("Subject", "")),
+    ("First line\nSecond line", ("", "First line\nSecond line")),
+    ("", ("", "")),
+]
+
+
+@pytest.mark.parametrize("message, result", MESSAGES)
+def test_parse_message(message: str, result: Tuple[str, str]) -> None:
+
+    assert parse_message(message) == result
 
 
 def test_repository_empty(create_git_repository: GitFactory) -> None:
@@ -11,12 +27,23 @@ def test_repository_empty(create_git_repository: GitFactory) -> None:
     assert len(git.list_commits(None)) == 0
 
 
+def test_repository_list_commits(create_git_repository: GitFactory) -> None:
+    commits = ["First", "Second"]
+    git = create_git_repository(commits)
+
+    commit_list = git.list_commits(None)
+
+    assert len(commit_list) == 2
+
+    assert [c.subject for c in commit_list] == ["First", "Second"]
+
+
 def test_repository_list_commits_from_ref(create_git_repository: GitFactory) -> None:
 
     commits = ["First", ("Second", "v1"), "Third", "Fourth"]
     git = create_git_repository(commits)
 
-    commit_list = git.list_commits("v1")
+    commit_list = git.list_commits(b"refs/tags/v1")
 
     assert len(commit_list) == 2
 
@@ -28,18 +55,30 @@ def test_repository_list_commits_from_ref_to_ref(create_git_repository: GitFacto
     commits = [("First", "v1"), "Second", "Third", ("Fourth", "v2"), "Fifth"]
     git = create_git_repository(commits)
 
-    commit_list = git.list_commits("v1", "v2")
+    commit_list = git.list_commits(b"refs/tags/v1", b"refs/tags/v2")
 
     assert len(commit_list) == 3
 
     assert [c.subject for c in commit_list] == ["Second", "Third", "Fourth"]
 
 
+def test_repository_list_commits_to_ref(create_git_repository: GitFactory) -> None:
+
+    commits = [("First", "v1"), "Second", "Third", ("Fourth", "v2"), "Fifth"]
+    git = create_git_repository(commits)
+
+    commit_list = git.list_commits(None, b"refs/tags/v2")
+
+    assert len(commit_list) == 4
+
+    assert [c.subject for c in commit_list] == ["First", "Second", "Third", "Fourth"]
+
+
 PARAMS = [
-    ("v1", Version(1, 0, 0)),
-    ("v10", Version(10, 0, 0)),
-    ("v1.0", Version(1, 0, 0)),
-    ("v1.1.10", Version(1, 1, 10)),
+    ("refs/tags/v1", Version(1, 0, 0)),
+    ("refs/tags/v10", Version(10, 0, 0)),
+    ("refs/tags/v1.0", Version(1, 0, 0)),
+    ("refs/tags/v1.1.10", Version(1, 1, 10)),
 ]