| 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()
|