Initial commit

First version of commit iterators & office_hours tester

Makefile

@@ -0,0 +1,12 @@
+all: unittest lint mypy
+
+unittest:
+	python3 -m unittest -v
+
+lint:
+	pylint git_oh.py test.py
+
+mypy:
+	mypy git_oh.py
+
+.PHONY: lint mypy

NOTES

@@ -0,0 +1,3 @@
+- creating commit at arbitrary dates :
+ - git commit --date option change the authored date
+ - GIT_COMMITTER_DATE env variable change the committed datetime

git_oh.py

@@ -0,0 +1,86 @@
+#!/usr/bin/python3
+""" Entrouvert test """
+import datetime
+import shutil
+import tempfile
+import warnings
+from collections.abc import Sequence
+from typing import Generator
+
+
+import git
+
+from git.repo.base import Repo
+from git.objects.commit import Commit
+
+
+def in_office_hours(moment:datetime.datetime=datetime.datetime.now(),
+        starthour:datetime.time=datetime.time(8,0,0),
+        stophour:datetime.time=datetime.time(20,0,0),
+        weekend:Sequence[int]=(5,6))->bool:
+    """ Indicates if a moment is in office hours.
+
+    Office hours are localized, comparisons are done without
+    taking care of tzoffset : in fact, if a working day starts at
+    08:00, 07:59:59+0200 is off as 07:59:59+0000 is.
+
+    Arguments :
+        - moment : the moment to compare with office hours
+        - starthour : standard day start hour
+        - stophour : standard day stop hour
+        - weekend : list of days off (0,1,....,6)
+    """
+    for dow in weekend:
+        if dow < 0 or dow > 6:
+            raise ValueError("Weekend days are integer in [0..6]")
+
+    if starthour.tzinfo is not None or stophour.tzinfo is not None:
+        warnings.warn("Start & stop hours should not indicate any \
+tzinfo : comparisons are done without taking tzoffset in considaration")
+
+    if moment.weekday() in weekend:
+        return False
+    localtime = moment.time()
+    return starthour <= localtime <= stophour
+
+
+def iter_commits(repo:Repo)->Generator[Commit, None, None]:
+    """ Generator on all git commits in given repository.
+
+    Recursively iterate on each commit in each branch/ref not yielding
+    duplicates commits (based on commit hashes)
+
+    Arguments :
+        - repo : The repository instance to fetch commits from
+    """
+    encountered = set()
+    for ref in repo.refs:
+        for commit in repo.iter_commits(ref.name):
+            if commit.binsha not in encountered:
+                encountered.add(commit.binsha)
+                yield commit
+
+
+class TempRemoteRepo(Repo):
+    """ A temporary repository referencing a remote repository
+
+    Allows to iterate on all commits without cloning the remote
+    """
+
+    def __init__(self, remote_url:str):
+        """ Initialize a new empty repository referencing a remote repo
+
+        Arguments :
+            - remote_url : The url of the remote repo to reference
+        """
+        self.temppath = tempfile.mkdtemp(prefix="git_oh_")
+        git.Repo.init(self.temppath)
+        super().__init__(self.temppath)
+        self.create_remote("origin", remote_url).fetch()
+
+    def __del__(self):
+        shutil.rmtree(self.temppath)
+        super().__del__()
+
+    def __iter__(self)->Generator[Commit, None, None]:
+        return iter_commits(self)

test.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+""" Tests for git_oh script """
+
+import datetime
+import gc
+import os
+import random
+import shutil
+import tempfile
+import unittest
+
+import git
+
+import git_oh
+
+
+class TestGitIterations(unittest.TestCase):
+    """ Testing iterator on git commits """
+
+    def setUp(self):
+        self.repo_path = tempfile.mkdtemp(prefix="git_oh_test_")
+        self.repo = git.Repo.init(self.repo_path)
+        self.actors = [git.Actor(f"User{i:d}", f"user{i:d}@localhost")
+                       for i in range(8)]
+
+    def tearDown(self):
+        shutil.rmtree(self.repo_path)
+
+    def test_metadata(self):
+        """ Fetch commits and their metadata in a simple git repository """
+        fromisoformat = datetime.datetime.fromisoformat
+        commit_date_fmt = "2022-12-13T23:32:2%d+0200"
+
+        commits = [
+                self.git_commit_mod(author=self.actors[0],
+                    commit_date=fromisoformat(commit_date_fmt % i))
+                for i in range(10)]
+        found_commits = list(git_oh.iter_commits(self.repo))
+
+        set_a = {c.hexsha for c in commits}
+        set_b = {c.hexsha for c in found_commits}
+
+        self.assertEqual(set_a, set_b)
+
+        for commit in found_commits:
+            self.assertEqual(commit.author, self.actors[0])
+            self.assertLessEqual(commit.committed_datetime,
+                    fromisoformat(commit_date_fmt % 9))
+            self.assertGreaterEqual(commit.committed_datetime,
+                    fromisoformat(commit_date_fmt % 0))
+
+    def test_branches(self):
+        """ Fetch commits from a repo with mutliple branches """
+
+        commits = [self.git_commit_mod(author=self.actors[0])
+                for _ in range(10)]
+
+        self.repo.git.checkout("HEAD", b="new_branch")
+
+        commits += [self.git_commit_mod(author=self.actors[0])
+                for _ in range(10)]
+
+        self.repo.git.checkout("HEAD", b="another_branch")
+
+        commits += [self.git_commit_mod(author=self.actors[0])
+                for _ in range(10)]
+
+        self.repo.git.checkout("new_branch")
+
+        found_commits = {commit.hexsha
+            for commit in git_oh.iter_commits(self.repo)}
+        commits = {commit.hexsha for commit in commits}
+
+        self.assertEqual(commits, found_commits)
+
+
+    def test_remote(self):
+        """ Testing TempRemoteRepo class commit fetch __iter__ method """
+        commits = {self.git_commit_mod(author=self.actors[0]).hexsha
+                for _ in range(10)}
+        repo = git_oh.TempRemoteRepo(f"file://{self.repo_path:s}")
+
+        found_commits = {commit.hexsha for commit in repo}
+
+        self.assertEqual(commits, found_commits)
+
+
+    def test_remote_cleanup(self):
+        """ Testing TempRemoteRepo class cleanup """
+        repo = git_oh.TempRemoteRepo(f"file://{self.repo_path:s}")
+        tmppath = repo.temppath
+
+        _ = list(repo)
+
+        self.assertTrue(os.path.isdir(tmppath))
+        del repo
+        gc.collect() # asking gc to call repo.__del__()
+        self.assertFalse(os.path.isdir(tmppath))
+
+
+    def test_remote_branches(self):
+        """ Testing TempRemoteRepo branch commit fetch """
+
+        commits = []
+        for i in range(5):
+            commits += [self.git_commit_mod(author=self.actors[0])
+                    for _ in range(10)]
+            self.repo.git.checkout("HEAD", b=f"branch-{i:d}")
+
+        repo = git_oh.TempRemoteRepo(f"file://{self.repo_path:s}")
+
+        found_commits = {commit.hexsha for commit in repo}
+        commits = {commit.hexsha for commit in commits}
+
+        self.assertEqual(commits, found_commits)
+
+
+    def git_commit_mod(self, msg="new commit", filename="foo.txt",
+            **commit_kwargs):
+        """ Add a random modification to repository and commit them
+            Arguments :
+                - msg : commit message
+                - filename : the filename to modify/add/commit
+                - **commit_kwargs : see help(git.index.base.IndexFile.commit)
+            Returns :
+                - the commit instance
+        """
+        with open(os.path.join(self.repo_path, filename), "w+",
+                encoding="utf-8") as repo_fp:
+            repo_fp.write(random.choices("abcdef")[0])
+        self.repo.index.add([filename])
+        return self.repo.index.commit(msg, **commit_kwargs)
+
+
+if __name__ == "__main__":
+    unittest.main()