[𝘀𝗽𝗿] initial version

Created using spr 1.3.7

Author: boomanaiden154

premerge/advisor/git_utils.py

@@ -0,0 +1,78 @@
+import sqlite3
+import os
+import subprocess
+import logging
+
+REPOSITORY_URL = "https://github.com/llvm/llvm-project"
+FIRST_COMMIT_SHA = "f8f7f1b67c8ee5d81847955dc36fab86a6d129ad"
+
+
+def _clone_repository_if_not_present(
+    repository_path: str, repository_url=REPOSITORY_URL
+):
+    if not os.path.exists(os.path.join(repository_path, ".git")):
+        logging.info("Cloning git repository.")
+        subprocess.run(
+            ["git", "clone", repository_url, os.path.basename(repository_path)],
+            cwd=os.path.dirname(repository_path),
+            check=True,
+        )
+        logging.info("Finished cloning git repository.")
+
+
+def _get_and_add_commit_index(
+    commit_sha: str,
+    repository_path: str,
+    db_connection: sqlite3.Connection,
+    first_commit_sha,
+) -> int:
+    # Ensure the repository is up to date.
+    subprocess.run(["git", "fetch"], cwd=repository_path, check=True)
+    # Get the highest indexed commit so we can ensure we only add new
+    # commits.
+    latest_commit_info = db_connection.execute(
+        "SELECT * FROM commits ORDER BY commit_index DESC"
+    ).fetchone()
+    commits_to_add = []
+    if latest_commit_info:
+        latest_sha, latest_index = latest_commit_info
+    else:
+        latest_sha = first_commit_sha
+        latest_index = 1
+    log_output = subprocess.run(
+        ["git", "log", "--oneline", "--no-abbrev", f"{latest_sha}..{commit_sha}"],
+        cwd=repository_path,
+        stdout=subprocess.PIPE,
+        check=True,
+    )
+    log_lines = log_output.stdout.decode("utf-8").split("\n")[:-1]
+    for line_index, log_line in enumerate(log_lines):
+        line_commit_sha = log_line.split(" ")[0]
+        commit_index = latest_index + len(log_lines) - line_index
+        commits_to_add.append((line_commit_sha, commit_index))
+    db_connection.executemany("INSERT INTO commits VALUES(?, ?)", commits_to_add)
+    if not latest_commit_info:
+        commits_to_add.append((first_commit_sha, 1))
+    return commits_to_add[0][1]
+
+
+def get_commit_index(
+    commit_sha: str,
+    repository_path: str,
+    db_connection: sqlite3.Connection,
+    first_commit_sha=FIRST_COMMIT_SHA,
+) -> int:
+    _clone_repository_if_not_present(repository_path)
+    # Check to see if we already have the commit in the DB.
+    commit_matches = db_connection.execute(
+        "SELECT * FROM commits WHERE commit_sha=?", (commit_sha,)
+    ).fetchall()
+    if len(commit_matches) > 1:
+        raise ValueError("Expected only one entry per commit SHA")
+    elif len(commit_matches) == 1:
+        return commit_matches[0][1]
+    # We have not seen this commit before. Count the index and then add it to
+    # the DB.
+    return _get_and_add_commit_index(
+        commit_sha, repository_path, db_connection, first_commit_sha
+    )

premerge/advisor/git_utils_test.py

@@ -0,0 +1,124 @@
+import unittest
+import tempfile
+import sqlite3
+import subprocess
+import os
+
+import git_utils
+
+
+class GitUtilsTest(unittest.TestCase):
+    def setUp(self):
+        self.db_file = tempfile.NamedTemporaryFile()
+        self.db_connection = sqlite3.connect(self.db_file.name)
+        self.db_connection.execute("CREATE TABLE commits(commit_sha, commit_index)")
+        self.repository_path = tempfile.TemporaryDirectory(delete=False)
+
+    def tearDown(self):
+        self.db_file.close()
+        # self.repository_path.cleanup()
+
+    def setup_repository(self, commit_count: int) -> list[str]:
+        subprocess.run(["git", "init"], cwd=self.repository_path.name, check=True)
+        for commit_index in range(commit_count):
+            with open(
+                os.path.join(self.repository_path.name, str(commit_index)), "w"
+            ) as commit_file:
+                commit_file.write("test")
+            subprocess.run(
+                ["git", "add", "--all"], cwd=self.repository_path.name, check=True
+            )
+            subprocess.run(
+                [
+                    "git",
+                    "commit",
+                    "--author='test <test@example.com>'",
+                    "-m",
+                    "message",
+                ],
+                cwd=self.repository_path.name,
+                check=True,
+            )
+        log_process = subprocess.run(
+            ["git", "log", "--oneline", "--no-abbrev"],
+            cwd=self.repository_path.name,
+            stdout=subprocess.PIPE,
+            check=True,
+        )
+        commit_shas = []
+        for log_line in log_process.stdout.decode("utf-8").split("\n")[:-1]:
+            commit_shas.append(log_line.split(" ")[0])
+        commit_shas.reverse()
+        return commit_shas
+
+    def test_clone_repository(self):
+        self.setup_repository(5)
+        utils_repo_folder = tempfile.TemporaryDirectory()
+        utils_repo_path = os.path.join(utils_repo_folder.name, "repo")
+        git_utils._clone_repository_if_not_present(
+            utils_repo_path, self.repository_path.name
+        )
+        log_process = subprocess.run(
+            ["git", "log", "--oneline", "--no-abbrev", "--max-count=5"],
+            cwd=utils_repo_path,
+            stdout=subprocess.PIPE,
+            check=True,
+        )
+        self.assertEqual(len(log_process.stdout.decode("utf-8").split("\n")) - 1, 5)
+
+    def test_get_index_from_db(self):
+        self.setup_repository(1)
+        self.db_connection.execute(
+            "INSERT INTO commits VALUES(?, ?)",
+            ("f3939dc5093826c05f2a78ce1b0af769cd48fdab", 5),
+        )
+        self.assertEqual(
+            git_utils.get_commit_index(
+                "f3939dc5093826c05f2a78ce1b0af769cd48fdab",
+                self.repository_path.name,
+                self.db_connection,
+            ),
+            5,
+        )
+
+    def test_get_first_commit_from_git(self):
+        commit_shas = self.setup_repository(2)
+        self.assertEqual(
+            git_utils.get_commit_index(
+                commit_shas[1],
+                self.repository_path.name,
+                self.db_connection,
+                commit_shas[0],
+            ),
+            2,
+        )
+
+    def test_get_index_from_git(self):
+        commit_shas = self.setup_repository(3)
+        self.db_connection.execute(
+            "INSERT INTO commits VALUES(?, ?)", (commit_shas[1], 3)
+        )
+        self.assertEqual(
+            git_utils.get_commit_index(
+                commit_shas[2], self.repository_path.name, self.db_connection
+            ),
+            4,
+        )
+
+    def test_get_index_from_git_multiple_commits(self):
+        commit_shas = self.setup_repository(4)
+        self.db_connection.execute(
+            "INSERT INTO commits VALUES(?, ?)", (commit_shas[1], 3)
+        )
+        self.assertEqual(
+            git_utils.get_commit_index(
+                commit_shas[3], self.repository_path.name, self.db_connection
+            ),
+            5,
+        )
+        self.assertEqual(
+            git_utils.get_commit_index(
+                commit_shas[2], self.repository_path.name, self.db_connection
+            ),
+            4,
+        )