Switch to sha-256 git repo and use git tools again

This commit is contained in:
Adam Majer 2024-04-08 14:55:37 +02:00
parent c9e07e536f
commit 4a37489132

View File

@ -4,7 +4,6 @@ import os
import pathlib import pathlib
import subprocess import subprocess
import pygit2
import requests import requests
from lib.binary import BINARY from lib.binary import BINARY
@ -20,11 +19,6 @@ class Git:
self.committer = committer self.committer = committer
self.committer_email = committer_email self.committer_email = committer_email
self.repo = None
def is_open(self):
return self.repo is not None
def exists(self): def exists(self):
"""Check if the path is a valid git repository""" """Check if the path is a valid git repository"""
return (self.path / ".git").exists() return (self.path / ".git").exists()
@ -35,35 +29,60 @@ class Git:
self.open() self.open()
def open(self): def open(self):
# Convert the path to string, to avoid some limitations in subprocess.run(
# older pygit2 ['git', 'init', '--object-format=sha256', '-b', 'factory'],
self.repo = pygit2.init_repository(str(self.path)) cwd=self.path,
check=True,
)
def is_dirty(self): def is_dirty(self):
"""Check if there is something to commit""" """Check if there is something to commit"""
assert self.is_open() status_str = subprocess.run(
['git', 'status', '--porcelain=2'],
return self.repo.status() cwd=self.path,
stdout=subprocess.PIPE,
check=True
).stdout.decode('utf-8')
return len(list(filter(None, status_str.split('\n')))) > 0
def branches(self): def branches(self):
return list(self.repo.branches) br=subprocess.run(
['git', 'for-each-ref', '--format=%(refname:short)', 'refs/heads/'],
cwd=self.path,
check=True,
stdout=subprocess.PIPE
).stdout.decode('utf-8').split()
if len(br) == 0:
br.append('factory') # unborn branch?
return br
def branch(self, branch, commit=None): def branch(self, branch, commit='HEAD'):
if not commit: commit = subprocess.run(
commit = self.repo.head ['git', 'rev-parse', '--verify', '--end-of-options', commit + '^{commit}'],
else: cwd=self.path,
commit = self.repo.get(commit) check=True,
self.repo.branches.local.create(branch, commit) stdout=subprocess.PIPE
).stdout.decode('utf-8').strip()
return subprocess.run(['git', 'branch', branch, commit], check=True)
def checkout(self, branch): def checkout(self, branch):
"""Checkout into the branch HEAD""" """Checkout into the branch HEAD"""
new_branch = False new_branch = False
ref = f"refs/heads/{branch}"
if branch not in self.branches(): if branch not in self.branches():
self.repo.references["HEAD"].set_target(ref) subprocess.run(
['git', 'branch', '-q', branch, 'HEAD'],
cwd=self.path,
check=True
)
new_branch = True new_branch = True
else: else:
self.repo.checkout(ref) ref = f"refs/heads/{branch}"
if (self.path/'.git'/ref).exists():
subprocess.run(
['git', 'checkout', '-q', branch],
cwd=self.path,
check=True
)
return new_branch return new_branch
def commit( def commit(
@ -87,30 +106,62 @@ class Git:
committer_time = committer_time if committer_time else user_time committer_time = committer_time if committer_time else user_time
if self.is_dirty(): if self.is_dirty():
self.repo.index.add_all() subprocess.run(
["git", "add", "--all", "."],
cwd=self.path,
check=True,
)
self.repo.index.write() tree_id = subprocess.run(
author = pygit2.Signature(user, user_email, int(user_time.timestamp())) ['git', 'write-tree'],
committer = pygit2.Signature( cwd=self.path,
committer, committer_email, int(committer_time.timestamp()) check=True,
stdout=subprocess.PIPE
).stdout.decode('utf-8').strip()
parent_array = []
if isinstance(parents, list):
for parent in filter(None, parents):
parent_array = parent_array + ['-p', parent]
elif isinstance(parents, str):
parents_array = ['-p', parents]
commit_id = subprocess.run(
['git', 'commit-tree'] + parent_array + [tree_id],
cwd=self.path,
env={
"GIT_AUTHOR_NAME": user,
"GIT_AUTHOR_EMAIL": user_email,
"GIT_AUTHOR_DATE": str(int(committer_time.timestamp())),
"GIT_COMMITTER_NAME": committer,
"GIT_COMMITTER_EMAIL": committer_email,
"GIT_COMMITTER_DATE": str(int(committer_time.timestamp())),
},
input=message.encode('utf-8'),
check=True,
stdout=subprocess.PIPE
).stdout.decode('utf-8').rstrip()
subprocess.run(
['git', 'reset', '--soft', commit_id],
cwd=self.path,
check=True,
) )
return commit_id
tree = self.repo.index.write_tree() def branch_head(self, branch='HEAD'):
return self.repo.create_commit( return subprocess.run(
"HEAD", author, committer, message, tree, parents ['git', 'rev-parse', '--verify', '--end-of-options', branch],
) cwd=self.path,
check=True,
def last_commit(self): stdout=subprocess.PIPE
try: ).stdout.decode('utf-8').strip()
return self.repo.head.target
except:
return None
def branch_head(self, branch):
return self.repo.references["refs/heads/" + branch].target
def set_branch_head(self, branch, commit): def set_branch_head(self, branch, commit):
self.repo.references["refs/heads/" + branch].set_target(commit) return subprocess.run(
['git', 'branch', '-f', branch, commit],
cwd=self.path,
check=True,
)
def gc(self): def gc(self):
logging.debug(f"Garbage recollect and repackage {self.path}") logging.debug(f"Garbage recollect and repackage {self.path}")
@ -121,17 +172,21 @@ class Git:
stderr=subprocess.STDOUT, stderr=subprocess.STDOUT,
) )
def clean(self): # def clean(self):
for path, _ in self.repo.status().items(): # for path, _ in self.repo.status().items():
logging.debug(f"Cleaning {path}") # logging.debug(f"Cleaning {path}")
try: # try:
(self.path / path).unlink() # (self.path / path).unlink()
self.repo.index.remove(path) # self.repo.index.remove(path)
except Exception as e: # except Exception as e:
logging.warning(f"Error removing file {path}: {e}") # logging.warning(f"Error removing file {path}: {e}")
def add(self, filename): def add(self, filename):
self.repo.index.add(filename) subprocess.run(
['git', 'add', filename],
cwd=self.path,
check=True,
)
def add_default_lfs_gitattributes(self, force=False): def add_default_lfs_gitattributes(self, force=False):
if not (self.path / ".gitattributes").exists() or force: if not (self.path / ".gitattributes").exists() or force:
@ -185,9 +240,11 @@ class Git:
return any(fnmatch.fnmatch(filename, line) for line in patterns) return any(fnmatch.fnmatch(filename, line) for line in patterns)
def remove(self, file: pathlib.Path): def remove(self, file: pathlib.Path):
self.repo.index.remove(file.name) subprocess.run(
(self.path / file).unlink() ['git', 'rm', '-q', '--ignore-unmatch', file.name],
cwd=self.path,
check=True,
)
patterns = self.get_specific_lfs_gitattributes() patterns = self.get_specific_lfs_gitattributes()
if file.name in patterns: if file.name in patterns:
patterns.remove(file.name) patterns.remove(file.name)
@ -213,19 +270,22 @@ class Git:
if response.status_code not in (201, 409): if response.status_code not in (201, 409):
print(response.data) print(response.data)
url = f"gitea@src.opensuse.org:{org_name}/{repo_name}.git" url = f"gitea@src.opensuse.org:{org_name}/{repo_name}.git"
self.repo.remotes.create("origin", url) subprocess.run(
['git', 'remote', 'add', 'origin', url],
cwd=self.path,
check=True,
)
def push(self, force=False): def push(self, force=False):
remo = self.repo.remotes["origin"] cmd = ['git', 'push'];
if force:
cmd.append('-f')
cmd.append('origin')
cmd.append('refs/heads/factory');
cmd.append('refs/heads/devel');
subprocess.run(
cmd,
cwd=self.path,
check=True,
)
keypair = pygit2.KeypairFromAgent("gitea")
callbacks = pygit2.RemoteCallbacks(credentials=keypair)
refspecs = ["refs/heads/factory"]
develspec = "refs/heads/devel"
if develspec in self.repo.references:
if force:
refspecs.append(f"+{develspec}:{develspec}")
else:
refspecs.append("{develspec}:{develspec}")
remo.push(refspecs, callbacks=callbacks)