Ulrich
2 years ago
5 changed files with 188 additions and 8 deletions
@ -1,6 +0,0 @@ |
|||
<?xml version="1.0" encoding="UTF-8"?> |
|||
<project version="4"> |
|||
<component name="VcsDirectoryMappings"> |
|||
<mapping directory="$PROJECT_DIR$" vcs="Git" /> |
|||
</component> |
|||
</project> |
@ -0,0 +1,60 @@ |
|||
import unittest |
|||
import os |
|||
import inspect |
|||
import utils.path_tool |
|||
import basic.message |
|||
import basic.program |
|||
import basic.constants as B |
|||
import test.constants |
|||
import test.testtools |
|||
import utils.path_const as P |
|||
import utils.git_tool |
|||
|
|||
HOME_PATH = test.constants.HOME_PATH |
|||
OS_SYSTEM = test.constants.OS_SYSTEM |
|||
|
|||
# here you can select single testfunction for developping the tests |
|||
TEST_FUNCTIONS = ["test_01run", "test_02status", "test_03log"] |
|||
#TEST_FUNCTIONS = [ "test_01status"] |
|||
verbose = False |
|||
|
|||
class MyTestCase(unittest.TestCase): |
|||
mymsg = "--------------------------------------------------------------" |
|||
|
|||
def test_01run(self): |
|||
global mymsg |
|||
actfunction = str(inspect.currentframe().f_code.co_name) |
|||
cnttest = 0 |
|||
if actfunction not in TEST_FUNCTIONS: |
|||
return |
|||
job = test.testtools.getJob() |
|||
utils.git_tool.runGit(job, B.ATTR_PATH_PROGRAM, "git status") |
|||
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|||
|
|||
def test_02status(self): |
|||
global mymsg |
|||
actfunction = str(inspect.currentframe().f_code.co_name) |
|||
cnttest = 0 |
|||
if actfunction not in TEST_FUNCTIONS: |
|||
return |
|||
job = test.testtools.getJob() |
|||
utils.git_tool.gitStatus(job, B.ATTR_PATH_PROGRAM) |
|||
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|||
|
|||
def test_03log(self): |
|||
global mymsg |
|||
actfunction = str(inspect.currentframe().f_code.co_name) |
|||
cnttest = 0 |
|||
if actfunction not in TEST_FUNCTIONS: |
|||
return |
|||
job = test.testtools.getJob() |
|||
utils.git_tool.gitLog(job, B.ATTR_PATH_COMPS) |
|||
MyTestCase.mymsg += "\n----- "+actfunction+" : "+str(cnttest) |
|||
|
|||
def test_zzz(self): |
|||
if verbose: print(MyTestCase.mymsg) |
|||
|
|||
|
|||
if __name__ == '__main__': |
|||
verbose = True |
|||
unittest.main() |
@ -0,0 +1,89 @@ |
|||
#!/usr/bin/python |
|||
# -*- coding: utf-8 -*- |
|||
# --------------------------------------------------------------------------------------------------------- |
|||
# Author : Ulrich Carmesin |
|||
# Source : gitea.ucarmesin.de |
|||
# --------------------------------------------------------------------------------------------------------- |
|||
import os |
|||
import re |
|||
import subprocess |
|||
import sys |
|||
import basic.toolHandling |
|||
import utils.data_const as D |
|||
import basic.constants as B |
|||
import basic.text_const as T |
|||
import utils.date_tool |
|||
|
|||
DEFAULT_CNT_COMMITS = 10 |
|||
COMMIT_ID = "commit" |
|||
COMMIT_AUTHOR = "author" |
|||
COMMIT_DATE = "date" |
|||
COMMIT_COMMENT = "comment" |
|||
|
|||
def runGit(job, repo, cmd): |
|||
cdpath = "" |
|||
if os.path.isdir(repo): |
|||
cdpath = repo |
|||
elif repo in job.conf.confs[B.SUBJECT_PATH]: |
|||
cdpath = job.conf.confs[B.SUBJECT_PATH][repo] |
|||
else: |
|||
raise Exception(T.EXP_PATH_MISSING, repo) |
|||
os.chdir(cdpath) |
|||
text = "" |
|||
process = subprocess.Popen(cmd, stdout=subprocess.PIPE, shell=True) |
|||
btext = process.communicate()[0] |
|||
text = btext.decode('utf-8') |
|||
return text |
|||
|
|||
def gitStatus(job, repo): |
|||
text = runGit(job, repo, "git status") |
|||
|
|||
def gitLog(job, repo, arg="", cnt=DEFAULT_CNT_COMMITS): |
|||
""" |
|||
extracts the last cnt commits into th structure |
|||
:param job: |
|||
:param repo: |
|||
:param cnt: |
|||
:return: [ {commit: "", author: ] |
|||
""" |
|||
text = runGit(job, repo, "git log") |
|||
if len(arg) > 1: |
|||
arg = " -- "+arg |
|||
else: |
|||
arg = "" |
|||
text = runGit(job, repo, "git log --pretty=format:\"%H | %cn | %cd | %s\""+arg) |
|||
print(text) |
|||
logs = [] |
|||
for l in text.split("\n"): |
|||
res = {} |
|||
a = l.split("|") |
|||
res[COMMIT_ID] = a[0].strip() |
|||
res[COMMIT_AUTHOR] = a[1].strip() |
|||
cdate = utils.date_tool.parseDate(a[2].strip()) |
|||
res[COMMIT_DATE] = utils.date_tool.getFormatDatetupel(cdate, utils.date_tool.F_DIR) |
|||
res[COMMIT_COMMENT] = a[3].strip() |
|||
logs.append(res) |
|||
print(str(logs)) |
|||
|
|||
|
|||
def gitCommits(job, repo, arg=""): |
|||
if len(arg) > 1: |
|||
arg = " -- "+arg |
|||
else: |
|||
arg = "" |
|||
text = runGit(job, repo, "git log --pretty=format:\"%H | %cn | %cd | %s\""+arg) |
|||
print(text) |
|||
|
|||
def gitPull(job, repo): |
|||
if "git" not in job.conf.confs[B.SUBJECT_TOOL] or repo not in job.conf.confs[B.SUBJECT_TOOL]["git"]: |
|||
raise Exception(T.EXP_CONFIG_MISSING, "tool.git."+repo) |
|||
#print(str(job.conf.confs[B.SUBJECT_TOOL]["git"][repo])) |
|||
master = job.conf.confs[B.SUBJECT_TOOL]["git"][repo]["master"] |
|||
remote = job.conf.confs[B.SUBJECT_TOOL]["git"][repo]["remote"] |
|||
text = runGit(job, repo, "git status") |
|||
if "Commit vor" in text: |
|||
print("Das Repository "+repo+" bitte zuerst pushen \n") |
|||
#print(text+"\"-------------------------------------\n") |
|||
else: |
|||
print("Repository "+repo+" kann gepullt werden.") |
|||
#runGit(job, repo, "git pull "+remote+" "+master) |
Loading…
Reference in new issue