292 lines
9.3 KiB
Python
292 lines
9.3 KiB
Python
import fnmatch
|
|
import logging
|
|
import os
|
|
import pathlib
|
|
import subprocess
|
|
|
|
import requests
|
|
|
|
from lib.binary import BINARY
|
|
|
|
LFS_SUFFIX = "filter=lfs diff=lfs merge=lfs -text"
|
|
|
|
|
|
class Git:
|
|
"""Local git repository"""
|
|
|
|
def __init__(self, path, committer=None, committer_email=None):
|
|
self.path = pathlib.Path(path)
|
|
self.committer = committer
|
|
self.committer_email = committer_email
|
|
|
|
def exists(self):
|
|
"""Check if the path is a valid git repository"""
|
|
return (self.path / ".git").exists()
|
|
|
|
def create(self):
|
|
"""Create a local git repository"""
|
|
self.path.mkdir(parents=True, exist_ok=True)
|
|
self.open()
|
|
|
|
def open(self):
|
|
subprocess.run(
|
|
['git', 'init', '--object-format=sha256', '-b', 'factory'],
|
|
cwd=self.path,
|
|
check=True,
|
|
)
|
|
|
|
def is_dirty(self):
|
|
"""Check if there is something to commit"""
|
|
status_str = subprocess.run(
|
|
['git', 'status', '--porcelain=2'],
|
|
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):
|
|
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='HEAD'):
|
|
commit = subprocess.run(
|
|
['git', 'rev-parse', '--verify', '--end-of-options', commit + '^{commit}'],
|
|
cwd=self.path,
|
|
check=True,
|
|
stdout=subprocess.PIPE
|
|
).stdout.decode('utf-8').strip()
|
|
return subprocess.run(['git', 'branch', branch, commit], check=True)
|
|
|
|
def checkout(self, branch):
|
|
"""Checkout into the branch HEAD"""
|
|
new_branch = False
|
|
if branch not in self.branches():
|
|
subprocess.run(
|
|
['git', 'branch', '-q', branch, 'HEAD'],
|
|
cwd=self.path,
|
|
check=True
|
|
)
|
|
new_branch = True
|
|
else:
|
|
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
|
|
|
|
def commit(
|
|
self,
|
|
user,
|
|
user_email,
|
|
user_time,
|
|
message,
|
|
parents=None,
|
|
committer=None,
|
|
committer_email=None,
|
|
committer_time=None,
|
|
):
|
|
"""Add all the files and create a new commit in the current HEAD"""
|
|
|
|
if not committer:
|
|
committer = self.committer if self.committer else self.user
|
|
committer_email = (
|
|
self.committer_email if self.committer_email else self.user_email
|
|
)
|
|
committer_time = committer_time if committer_time else user_time
|
|
|
|
if self.is_dirty():
|
|
subprocess.run(
|
|
["git", "add", "--all", "."],
|
|
cwd=self.path,
|
|
check=True,
|
|
)
|
|
|
|
tree_id = subprocess.run(
|
|
['git', 'write-tree'],
|
|
cwd=self.path,
|
|
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": f"{int(user_time.timestamp())} +0000",
|
|
"GIT_COMMITTER_NAME": committer,
|
|
"GIT_COMMITTER_EMAIL": committer_email,
|
|
"GIT_COMMITTER_DATE": f"{int(committer_time.timestamp())} +0000",
|
|
},
|
|
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
|
|
|
|
def branch_head(self, branch='HEAD'):
|
|
return subprocess.run(
|
|
['git', 'rev-parse', '--verify', '--end-of-options', branch],
|
|
cwd=self.path,
|
|
check=True,
|
|
stdout=subprocess.PIPE
|
|
).stdout.decode('utf-8').strip()
|
|
|
|
def set_branch_head(self, branch, commit):
|
|
return subprocess.run(
|
|
['git', 'branch', '-f', branch, commit],
|
|
cwd=self.path,
|
|
check=True,
|
|
)
|
|
|
|
def gc(self):
|
|
logging.debug(f"Garbage recollect and repackage {self.path}")
|
|
subprocess.run(
|
|
["git", "gc", "--auto"],
|
|
cwd=self.path,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
)
|
|
|
|
# def clean(self):
|
|
# for path, _ in self.repo.status().items():
|
|
# logging.debug(f"Cleaning {path}")
|
|
# try:
|
|
# (self.path / path).unlink()
|
|
# self.repo.index.remove(path)
|
|
# except Exception as e:
|
|
# logging.warning(f"Error removing file {path}: {e}")
|
|
|
|
def add(self, filename):
|
|
subprocess.run(
|
|
['git', 'add', filename],
|
|
cwd=self.path,
|
|
check=True,
|
|
)
|
|
|
|
def add_default_lfs_gitattributes(self, force=False):
|
|
if not (self.path / ".gitattributes").exists() or force:
|
|
with (self.path / ".gitattributes").open("w") as f:
|
|
content = ["## Default LFS"]
|
|
content += [f"*{b} {LFS_SUFFIX}" for b in sorted(BINARY)]
|
|
f.write("\n".join(content))
|
|
f.write("\n")
|
|
self.add(".gitattributes")
|
|
|
|
def add_specific_lfs_gitattributes(self, binaries):
|
|
self.add_default_lfs_gitattributes(force=True)
|
|
if binaries:
|
|
with (self.path / ".gitattributes").open("a") as f:
|
|
content = ["## Specific LFS patterns"]
|
|
content += [f"{b} {LFS_SUFFIX}" for b in sorted(binaries)]
|
|
f.write("\n".join(content))
|
|
f.write("\n")
|
|
self.add(".gitattributes")
|
|
|
|
def get_specific_lfs_gitattributes(self):
|
|
with (self.path / ".gitattributes").open() as f:
|
|
patterns = [
|
|
line.split()[0]
|
|
for line in f
|
|
if line.strip() and not line.startswith("#")
|
|
]
|
|
binary = {f"*{b}" for b in BINARY}
|
|
return [p for p in patterns if p not in binary]
|
|
|
|
def add_lfs(self, filename, sha256, size):
|
|
with (self.path / filename).open("w") as f:
|
|
f.write("version https://git-lfs.github.com/spec/v1\n")
|
|
f.write(f"oid sha256:{sha256}\n")
|
|
f.write(f"size {size}\n")
|
|
self.add(filename)
|
|
|
|
if not self.is_lfs_tracked(filename):
|
|
logging.debug(f"Add specific LFS file {filename}")
|
|
specific_patterns = self.get_specific_lfs_gitattributes()
|
|
specific_patterns.append(filename)
|
|
self.add_specific_lfs_gitattributes(specific_patterns)
|
|
|
|
def is_lfs_tracked(self, filename):
|
|
with (self.path / ".gitattributes").open() as f:
|
|
patterns = (
|
|
line.split()[0]
|
|
for line in f
|
|
if line.strip() and not line.startswith("#")
|
|
)
|
|
return any(fnmatch.fnmatch(filename, line) for line in patterns)
|
|
|
|
def remove(self, file: pathlib.Path):
|
|
subprocess.run(
|
|
['git', 'rm', '-q', '--ignore-unmatch', file.name],
|
|
cwd=self.path,
|
|
check=True,
|
|
)
|
|
patterns = self.get_specific_lfs_gitattributes()
|
|
if file.name in patterns:
|
|
patterns.remove(file.name)
|
|
self.add_specific_lfs_gitattributes(patterns)
|
|
|
|
def add_gitea_remote(self, package):
|
|
repo_name = package.replace("+", "_")
|
|
org_name = "rpm"
|
|
|
|
if not os.getenv("GITEA_TOKEN"):
|
|
logging.warning("Not adding a remote due to missing $GITEA_TOKEN")
|
|
return
|
|
|
|
url = f"https://src.opensuse.org/api/v1/org/{org_name}/repos"
|
|
response = requests.post(
|
|
url,
|
|
data={"name": repo_name},
|
|
headers={"Authorization": f"token {os.getenv('GITEA_TOKEN')}"},
|
|
timeout=10,
|
|
)
|
|
# 409 Conflict (Already existing)
|
|
# 201 Created
|
|
if response.status_code not in (201, 409):
|
|
print(response.data)
|
|
url = f"gitea@src.opensuse.org:{org_name}/{repo_name}.git"
|
|
subprocess.run(
|
|
['git', 'remote', 'add', 'origin', url],
|
|
cwd=self.path,
|
|
check=True,
|
|
)
|
|
|
|
def push(self, force=False):
|
|
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,
|
|
)
|
|
|