Initial idea

As long as you fill everything out correctly, this'll work.

On running the script, the read config is printed.  Then, on a 1s
timer, updates are polled from the remote.  On a change, you should
see output for the setup and build functions, with a final message
telling you if it passed or failed.

Works quite well

1 files changed, 125 insertions, 0 deletions

diff --git a/main.py b/main.py
new file mode 100644
index 0000000..b4e4737
--- /dev/null
+++ b/main.py
@@ -0,0 +1,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()