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
126
127
128
129
130
131
132
133
134
135
|
# Copyright (C) 2025 Aryadev Chavali
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the Unlicense for details.
# You may distribute and modify this code under the terms of the Unlicense,
# which you should have received a copy of along with this program. If not,
# please go to <https://unlicense.org/>.
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()
|
