Viewing files at commit 3cd2661fdbe9Commit detailsView latest

suites/merge/mergesuite/builder.py 62 lines · 2.5 KB · python
1
"""Deterministic Git history builder for suite-authored scenarios (no working tree)."""
2
from __future__ import annotations
3
4
import os
5
from pathlib import Path
6
import tempfile
7
from typing import Dict, List, Optional, Union
8
9
from .gitops import Git
10
11
# File spec: bytes (regular), ("exec", bytes), ("link", "target").
12
FileSpec = Union[bytes, tuple]
13
Files = Dict[str, FileSpec]
14
15
EPOCH = 1_750_000_000  # fixed, so imported Oak commit hashes are reproducible
16
17
18
class RepoBuilder(Git):
19
    def __init__(self, repo: Path, git_bin: str, home: Path):
20
        repo.mkdir(parents=True, exist_ok=True)
21
        super().__init__(repo, git_bin, home)
22
        self.run("init", "-q", "--object-format=sha1")
23
        self._tick = 0
24
25
    def commit(self, files: Files, parents: Optional[List[str]] = None, message: str = "c",
26
               when: Optional[int] = None, branch: Optional[str] = None) -> str:
27
        """Create a commit whose tree is exactly `files`. `when` is seconds since EPOCH
28
        (defaults to a monotone tick; may be negative to model clock skew)."""
29
        if when is None:
30
            self._tick += 60
31
            when = self._tick
32
        lines = []
33
        for path in sorted(files):
34
            spec = files[path]
35
            if isinstance(spec, bytes):
36
                mode, data = "100644", spec
37
            elif spec[0] == "exec":
38
                mode, data = "100755", spec[1]
39
            elif spec[0] == "link":
40
                mode, data = "120000", os.fsencode(spec[1])
41
            else:
42
                raise ValueError(f"bad file spec for {path}")
43
            oid = self.out("hash-object", "-w", "--stdin", input=data)
44
            lines.append(f"{mode} {oid}\t{path}")
45
        with tempfile.TemporaryDirectory(prefix="mergesuite-index-") as tmp:
46
            env = {"GIT_INDEX_FILE": str(Path(tmp) / "index")}
47
            self.run("update-index", "--add", "--index-info",
48
                     input=("\n".join(lines) + "\n").encode() if lines else b"", env_extra=env)
49
            tree = self.out("write-tree", env_extra=env)
50
        stamp = f"{EPOCH + when} +0000"
51
        args = ["commit-tree", tree]
52
        for p in parents or []:
53
            args += ["-p", p]  # order preserved: first parent first
54
        args += ["-m", message]
55
        commit = self.out(*args, env_extra={"GIT_AUTHOR_DATE": stamp, "GIT_COMMITTER_DATE": stamp})
56
        if branch:
57
            self.run("update-ref", f"refs/heads/{branch}", commit)
58
        return commit
59
60
61
def lines(*items: str) -> bytes:
62
    return "".join(f"{x}\n" for x in items).encode()