1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
import tempfile
import shutil
import subprocess
import os
from time import sleep
from typing import Callable, Union
from json import dump, load
from dataclasses import dataclass
def clearscreen():
if os.name == 'nt':
os.system('cls')
else:
os.system('clear')
def cmd(command, cwd = "./"):
result = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, cwd=cwd)
return (result.returncode, result.stdout)
def pull_latest_git_hash(url: str, branch: str):
return cmd(["git", "ls-remote", url, f"refs/heads/{branch}"])[1].strip().split()[0]
@dataclass
class GitState:
url: str
branch: str
git_hash: str
def __init__(self, url: str, branch: str):
self.url = url
self.branch = branch
self.git_hash = pull_latest_git_hash(url, branch)
def new_update(self):
return self.git_hash != pull_latest_git_hash(self.url, self.branch)
def update(self):
self.git_hash = pull_latest_git_hash(self.url, self.branch)
@dataclass
class Task:
name: str
git: GitState
build_command: [str]
@staticmethod
def from_json(json: dict[str, Union[str, [str]]]):
return Task(json["name"], GitState(json["url"], json["branch"]), json["build"]);
def to_json(self) -> dict[str, str]:
obj = dict()
obj["name"] = self.name
obj["url"] = self.git.url
obj["branch"] = self.git.branch
obj["build"] = self.build_command
return obj
def setup(self, directory):
# Pull URL into temp directory
return cmd(["git", "clone", "--progress", self.git.url, directory])
def build(self, directory):
# Run self.build_command, returning exit code and any output
return cmd(self.build_command, directory)
def teardown(self, directory):
# delete /tmp/{self.name} if it exists
shutil.rmtree(directory)
def on_update(self):
# Presuming a new update, perform a complete sanitised build.
temp_directory = tempfile.mkdtemp()
self.git.update()
print(f"[{self.name}:setup]: Starting setup")
code, _ = self.setup(temp_directory)
print(f"[{self.name}:setup]: Exited with {code}")
print(f"[{self.name}:build]: Starting build")
code, output = self.build(temp_directory)
print(f"[{self.name}:build]: Exited with {code}")
self.teardown(temp_directory)
return (code, output.split("\n"))
@dataclass
class Config:
config_path: str
tasks: [Task]
def __init__(self, path: str):
self.config_path = path
data = None
with open(path, "r") as fp:
data = load(fp)
self.tasks = [Task.from_json(json) for json in data]
def write(self):
config_json = [task.to_json() for task in tasks]
with open(config_path, "w") as fp:
dump(config_json, fp)
def poll_task_updates(self):
results = dict()
for i, task in enumerate(self.tasks):
if task.git.new_update():
results[task.name] = task.on_update()
self.tasks[i] = task
return results
def poll_task_updates_on_timer(self):
while True:
sleep(1)
updates = self.poll_task_updates()
if len(updates) != 0:
for name in updates:
code, output = updates[name]
status = "PASSED" if code == 0 else "FAILED"
print(f"[{name}]: {status}")
if __name__ == "__main__":
x = Config("./dbcd.json")
print(x)
x.poll_task_updates_on_timer()
|
