|
|
|
@ -2,10 +2,10 @@ import json
|
|
|
|
|
import re
|
|
|
|
|
import subprocess as sub
|
|
|
|
|
from abc import ABC, abstractmethod
|
|
|
|
|
from typing import Optional, Union
|
|
|
|
|
from typing import Optional
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
from os import environ
|
|
|
|
|
import logging
|
|
|
|
|
from ..infrastructure import ExecutionApi
|
|
|
|
|
|
|
|
|
|
# TODO: jem, zam - 2023_04_18: Discuss if we can move more functionality to domain?
|
|
|
|
|
class FileHandler(ABC):
|
|
|
|
@ -188,7 +188,7 @@ class SystemApi():
|
|
|
|
|
self.stderr = [""]
|
|
|
|
|
self.exitcode = 0
|
|
|
|
|
|
|
|
|
|
def run(self, args: list[str]):
|
|
|
|
|
def run(self, args: list[str]): #TODO Make stateless
|
|
|
|
|
sanitized_args = []
|
|
|
|
|
for arg in args:
|
|
|
|
|
str_arg = str(arg)
|
|
|
|
@ -208,14 +208,9 @@ class SystemApi():
|
|
|
|
|
else:
|
|
|
|
|
self.stderr = None
|
|
|
|
|
|
|
|
|
|
self.exitcode = stream.returncode
|
|
|
|
|
|
|
|
|
|
def run_checked(self, *args):
|
|
|
|
|
self.run(args)
|
|
|
|
|
|
|
|
|
|
logging.warning(f"err: {self.stderr}")
|
|
|
|
|
logging.warning(f"exit: {self.exitcode}")
|
|
|
|
|
|
|
|
|
|
if len(self.stderr) > 0:
|
|
|
|
|
raise Exception(f"Command failed with: {self.stderr}")
|
|
|
|
|
|
|
|
|
@ -223,61 +218,61 @@ class SystemApi():
|
|
|
|
|
class GitApi():
|
|
|
|
|
|
|
|
|
|
def __init__(self):
|
|
|
|
|
self.system_api = SystemApi()
|
|
|
|
|
self.execution_api = ExecutionApi()
|
|
|
|
|
|
|
|
|
|
def get_latest_n_commits(self, n: int):
|
|
|
|
|
self.system_api.run_checked(
|
|
|
|
|
'git', 'log', '--oneline', '--format="%s %b"', f'-n {n}')
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
self.execution_api.execute(
|
|
|
|
|
f'git log --oneline --format="%s %b" -n {n}')
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def get_latest_commit(self):
|
|
|
|
|
output = self.get_latest_n_commits(1)
|
|
|
|
|
return " ".join(output)
|
|
|
|
|
|
|
|
|
|
def tag_annotated(self, annotation: str, message: str, count: int):
|
|
|
|
|
self.system_api.run_checked(
|
|
|
|
|
self.execution_api.execute(
|
|
|
|
|
'git', 'tag', '-a', annotation, '-m', message, f'HEAD~{count}')
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def tag_annotated_second_last(self, annotation: str, message:str):
|
|
|
|
|
self.tag_annotated(annotation, message, 1)
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def get_latest_tag(self):
|
|
|
|
|
self.system_api.run_checked('git', 'describe', '--tags', '--abbrev=0')
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
self.execution_api.execute('git', 'describe', '--tags', '--abbrev=0')
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def get_current_branch(self):
|
|
|
|
|
self.system_api.run_checked('git', 'branch', '--show-current')
|
|
|
|
|
return ''.join(self.system_api.stdout).rstrip()
|
|
|
|
|
self.execution_api.execute('git', 'branch', '--show-current')
|
|
|
|
|
return ''.join(self.execution_api.stdout).rstrip()
|
|
|
|
|
|
|
|
|
|
def init(self, default_branch: str = "main"):
|
|
|
|
|
self.system_api.run_checked('git', 'init', '-b', default_branch)
|
|
|
|
|
self.execution_api.execute('git', 'init', '-b', default_branch)
|
|
|
|
|
|
|
|
|
|
def set_user_config(self, email: str, name: str):
|
|
|
|
|
self.system_api.run_checked('git', 'config', 'user.email', email)
|
|
|
|
|
self.system_api.run_checked('git', 'config', 'user.name', name)
|
|
|
|
|
self.execution_api.execute('git', 'config', 'user.email', email)
|
|
|
|
|
self.execution_api.execute('git', 'config', 'user.name', name)
|
|
|
|
|
|
|
|
|
|
def add_file(self, file_path: Path):
|
|
|
|
|
self.system_api.run_checked('git', 'add', file_path)
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
self.execution_api.execute('git', 'add', file_path)
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def add_remote(self, origin: str, url: str):
|
|
|
|
|
self.system_api.run_checked('git', 'remote', 'add', origin, url)
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
self.execution_api.execute('git', 'remote', 'add', origin, url)
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def commit(self, commit_message: str):
|
|
|
|
|
self.system_api.run_checked(
|
|
|
|
|
self.execution_api.execute(
|
|
|
|
|
'git', 'commit', '-m', commit_message)
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def push(self):
|
|
|
|
|
self.system_api.run_checked('git', 'push')
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
self.execution_api.execute('git', 'push')
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
def checkout(self, branch: str):
|
|
|
|
|
self.system_api.run_checked('git', 'checkout', branch)
|
|
|
|
|
return self.system_api.stdout
|
|
|
|
|
self.execution_api.execute('git', 'checkout', branch)
|
|
|
|
|
return self.execution_api.stdout
|
|
|
|
|
|
|
|
|
|
class EnvironmentApi():
|
|
|
|
|
|
|
|
|
|