ngn
ab580d6a18
Some checks failed
format / build (push) Failing after 6s
Signed-off-by: ngn <ngn@ngn.tf>
91 lines
2.5 KiB
Python
91 lines
2.5 KiB
Python
from .provider import provider
|
|
from .config import config
|
|
|
|
from .gitea import gitea
|
|
from .github import github
|
|
|
|
from os import remove, listdir, rmdir
|
|
from tempfile import mkdtemp
|
|
from os.path import join
|
|
import subprocess as sp
|
|
from typing import List
|
|
import atexit
|
|
|
|
PROVIDERS = {
|
|
"gitea": gitea,
|
|
"github": github,
|
|
}
|
|
|
|
|
|
class upstream:
|
|
def __init__(self, dir=".") -> None:
|
|
self.config: config = config(dir=dir)
|
|
self.provider: provider
|
|
self._tempdir: str = ""
|
|
|
|
url = self.config.url("upstream")
|
|
prov = self.config.str("provider")
|
|
|
|
for n, p in PROVIDERS.items():
|
|
if n == prov.lower():
|
|
self.provider = p(url)
|
|
break
|
|
|
|
if self.provider is None:
|
|
raise Exception("invalid provider %s" % provider)
|
|
|
|
if not self.provider.check():
|
|
raise Exception("upstream is not available")
|
|
|
|
# self.cleanup() will run when the program exits
|
|
atexit.register(self.cleanup)
|
|
|
|
def tempdir(self, path="") -> str:
|
|
if self._tempdir == "":
|
|
self._tempdir = mkdtemp(prefix="ups_")
|
|
return self._tempdir if path == "" else join(self._tempdir, path)
|
|
|
|
def commit(self, commit="") -> str:
|
|
if commit != "":
|
|
self.config.set_str("commit", commit)
|
|
return commit
|
|
return self.config.str("commit")
|
|
|
|
def until(self, until: str) -> List[str]:
|
|
return self.provider.until(until)
|
|
|
|
def last(self, count=1) -> List[str]:
|
|
return self.provider.last(count)
|
|
|
|
# download a patch for the commit and run all the scripts on it
|
|
def download(self, commit: str) -> str:
|
|
path = self.tempdir("%s.patch" % commit)
|
|
self.provider.download(commit, path)
|
|
|
|
for s in self.config.list("scripts"):
|
|
proc = sp.run(["sed", "-e", s, "-i", path], stderr=sp.PIPE)
|
|
if proc.returncode != 0:
|
|
raise Exception("script '%s': %s" % (s, proc.stderr))
|
|
|
|
return path
|
|
|
|
# apply the patch file using ups-apply
|
|
def apply(self, commit: str) -> None:
|
|
path = self.tempdir("%s.patch" % commit)
|
|
proc = sp.run(["ups-apply", path])
|
|
|
|
if proc.returncode != 0:
|
|
raise Exception(
|
|
"apply script returned non-zero exit code: %d" % proc.returncode
|
|
)
|
|
|
|
# cleanup the temp directory
|
|
def cleanup(self) -> None:
|
|
if self._tempdir == "":
|
|
return
|
|
|
|
for f in listdir(self._tempdir):
|
|
remove(join(self._tempdir, f))
|
|
|
|
rmdir(self._tempdir)
|