taeglich_push.py ruft nach dem Vault-Push werkzeuge_spiegel.py auf: committeter Stand von .vault-tools/ und .claude/ ohne Trading-Werkzeuge und watermarks-remover, README aus werkzeuge-README.md. Fallkatalog 14 von 14, drei Mutanten erkannt. veroeffentlichung.md und Stamm nachgezogen; Gitea seit heute anonym lesbar. Co-Authored-By: Claude Opus 5.5 <[email protected]>
296 lines
10 KiB
Python
296 lines
10 KiB
Python
"""Fährt die Fälle gegen werkzeuge_spiegel.py, auf Wegwerf-Vaults mit echten bare-Remotes.
|
|
|
|
Ersteller: Pascal Michel
|
|
Erstellt: 2026-10-14
|
|
|
|
Jeder Wegwerf-Vault hat als origin ein bare-Repo claude-regelwerk.git, daneben
|
|
liegt claude-werkzeuge.git mit einem veralteten Stand. Der Spiegel leitet seine
|
|
Adresse aus der des Vaults ab, wie im Betrieb. Geprüft wird, was nachher
|
|
drüben liegt, nicht was das Skript meldet.
|
|
|
|
Aufruf:
|
|
python .vault-tools/auto-push/faelle_werkzeuge_spiegel.py
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import shutil
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
HIER = Path(__file__).resolve().parent
|
|
SPIEGEL = HIER / "werkzeuge_spiegel.py"
|
|
README = HIER / "werkzeuge-README.md"
|
|
|
|
CRLF_INHALT = b"zeile eins\r\nzeile zwei\r\n"
|
|
|
|
ERWARTET = {
|
|
"README.md",
|
|
".vault-tools/auto-push/werkzeuge_spiegel.py",
|
|
".vault-tools/auto-push/werkzeuge-README.md",
|
|
".vault-tools/linkcheck/a.py",
|
|
".vault-tools/linkcheck/crlf.txt",
|
|
".claude/agents/x.md",
|
|
}
|
|
|
|
|
|
def raeume(pfad: Path) -> None:
|
|
"""rmtree, das auch schreibgeschuetzte git-Objektdateien wegbekommt."""
|
|
|
|
def erzwinge(funktion, ziel, _fehler):
|
|
try:
|
|
os.chmod(ziel, stat.S_IWRITE)
|
|
funktion(ziel)
|
|
except OSError:
|
|
pass
|
|
|
|
shutil.rmtree(pfad, onexc=erzwinge)
|
|
|
|
|
|
def git(wo: Path, *argumente: str) -> subprocess.CompletedProcess[bytes]:
|
|
return subprocess.run(
|
|
["git", "-C", str(wo), *argumente],
|
|
capture_output=True,
|
|
stdin=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
|
|
|
|
def lade_spiegel():
|
|
spec = importlib.util.spec_from_file_location("werkzeuge_spiegel", SPIEGEL)
|
|
if spec is None or spec.loader is None:
|
|
raise RuntimeError(f"nicht ladbar: {SPIEGEL}")
|
|
modul = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(modul)
|
|
return modul
|
|
|
|
|
|
def schreibe(pfad: Path, inhalt: bytes) -> None:
|
|
pfad.parent.mkdir(parents=True, exist_ok=True)
|
|
pfad.write_bytes(inhalt)
|
|
|
|
|
|
def richte_repo_ein(wo: Path, remote: Path) -> None:
|
|
git(wo, "init", "-q")
|
|
git(wo, "symbolic-ref", "HEAD", "refs/heads/main")
|
|
git(wo, "config", "user.name", "Fall")
|
|
git(wo, "config", "user.email", "[email protected]")
|
|
git(wo, "config", "commit.gpgsign", "false")
|
|
git(wo, "config", "core.autocrlf", "false")
|
|
git(wo, "remote", "add", "origin", remote.as_uri())
|
|
|
|
|
|
def committe(wo: Path, text: str) -> None:
|
|
git(wo, "add", "-A")
|
|
git(wo, "commit", "-q", "-m", text)
|
|
|
|
|
|
def baue(basis: Path) -> tuple[Path, Path]:
|
|
"""Legt Vault, beide bare-Remotes und den veralteten Spiegelstand an."""
|
|
vault = basis / "vault"
|
|
remotes = basis / "remotes"
|
|
regelwerk = remotes / "claude-regelwerk.git"
|
|
werkzeuge = remotes / "claude-werkzeuge.git"
|
|
remotes.mkdir(parents=True)
|
|
subprocess.run(["git", "init", "-q", "--bare", str(regelwerk)], check=True)
|
|
subprocess.run(["git", "init", "-q", "--bare", "--initial-branch=main", str(werkzeuge)], check=True)
|
|
|
|
schreibe(vault / ".vault-tools" / "auto-push" / "werkzeuge_spiegel.py", SPIEGEL.read_bytes())
|
|
schreibe(vault / ".vault-tools" / "auto-push" / "werkzeuge-README.md", README.read_bytes())
|
|
schreibe(vault / ".vault-tools" / "linkcheck" / "a.py", b"print('a')\n")
|
|
schreibe(vault / ".vault-tools" / "linkcheck" / "crlf.txt", CRLF_INHALT)
|
|
schreibe(vault / ".vault-tools" / "trading-plattform-pruefung" / "t.py", b"trading\n")
|
|
schreibe(vault / ".vault-tools" / "watermarks-remover" / "w.sh", b"aus\n")
|
|
schreibe(vault / ".claude" / "agents" / "x.md", b"# x\n")
|
|
schreibe(vault / "Inbox" / "geheim.md", b"Kundendaten\n")
|
|
richte_repo_ein(vault, regelwerk)
|
|
committe(vault, "Grundstand")
|
|
# Ungesicherte Datei im Arbeitsbaum: darf nie hinueber.
|
|
schreibe(vault / ".vault-tools" / "linkcheck" / "lokal.tmp", b"nicht committet\n")
|
|
|
|
alt = basis / "alt"
|
|
schreibe(alt / "README.md", b"alt\n")
|
|
schreibe(alt / "veraltet.txt", b"weg damit\n")
|
|
richte_repo_ein(alt, werkzeuge)
|
|
committe(alt, "Alter Stand")
|
|
git(alt, "push", "-q", "origin", "main")
|
|
return vault, werkzeuge
|
|
|
|
|
|
def dateien(remote: Path) -> set[str]:
|
|
lauf = git(remote, "ls-tree", "-r", "--name-only", "main")
|
|
if lauf.returncode != 0:
|
|
return set()
|
|
return {z for z in lauf.stdout.decode("utf-8", errors="replace").splitlines() if z}
|
|
|
|
|
|
def inhalt(remote: Path, pfad: str) -> bytes:
|
|
return git(remote, "cat-file", "blob", f"main:{pfad}").stdout
|
|
|
|
|
|
def commits(remote: Path) -> int:
|
|
lauf = git(remote, "rev-list", "--count", "main")
|
|
if lauf.returncode != 0:
|
|
return -1
|
|
return int(lauf.stdout.decode().strip() or "0")
|
|
|
|
|
|
def main() -> int:
|
|
spiegel = lade_spiegel()
|
|
basis = Path(tempfile.mkdtemp(prefix="werkzeuge-spiegel-faelle-"))
|
|
gesamt = 0
|
|
bestanden = 0
|
|
|
|
def pruefe(name: str, ok: bool, text: str) -> None:
|
|
nonlocal gesamt, bestanden
|
|
gesamt += 1
|
|
bestanden += int(ok)
|
|
print(f"{'BESTANDEN' if ok else 'FEHLER':10} {name:42} {text}")
|
|
|
|
def wirft(wurzel: Path, klon: Path) -> str | None:
|
|
try:
|
|
spiegel.spiegele(wurzel, klon)
|
|
except spiegel.SpiegelFehler as fehler:
|
|
return str(fehler)
|
|
return None
|
|
|
|
try:
|
|
# --- Grundlinie -------------------------------------------------
|
|
vault, werkzeuge = baue(basis / "a")
|
|
klon = basis / "a" / "klon"
|
|
status, text = spiegel.spiegele(vault, klon)
|
|
drueben = dateien(werkzeuge)
|
|
pruefe(
|
|
"grundlinie",
|
|
status == "gespiegelt" and drueben == ERWARTET,
|
|
f"{status}: {text}; zu viel {sorted(drueben - ERWARTET)}, fehlt {sorted(ERWARTET - drueben)}",
|
|
)
|
|
pruefe(
|
|
"readme aus der quelle",
|
|
inhalt(werkzeuge, "README.md") == README.read_bytes(),
|
|
f"{len(inhalt(werkzeuge, 'README.md'))} Bytes drueben",
|
|
)
|
|
pruefe(
|
|
"bytes bleiben (crlf)",
|
|
inhalt(werkzeuge, ".vault-tools/linkcheck/crlf.txt") == CRLF_INHALT,
|
|
repr(inhalt(werkzeuge, ".vault-tools/linkcheck/crlf.txt")),
|
|
)
|
|
|
|
# --- Zweiter Lauf ohne Aenderung --------------------------------
|
|
vorher = commits(werkzeuge)
|
|
status, text = spiegel.spiegele(vault, klon)
|
|
pruefe(
|
|
"zweiter lauf -> unveraendert",
|
|
status == "unveraendert" and commits(werkzeuge) == vorher,
|
|
f"{status}: {text}; Commits {vorher}->{commits(werkzeuge)}",
|
|
)
|
|
|
|
# --- Committet geht mit, ungesichert nicht ----------------------
|
|
a_py = vault / ".vault-tools" / "linkcheck" / "a.py"
|
|
schreibe(a_py, b"print('neu')\n")
|
|
committe(vault, "a geaendert")
|
|
schreibe(a_py, b"print('ungesichert')\n")
|
|
status, text = spiegel.spiegele(vault, klon)
|
|
pruefe(
|
|
"committete aenderung geht mit",
|
|
status == "gespiegelt" and inhalt(werkzeuge, ".vault-tools/linkcheck/a.py") == b"print('neu')\n",
|
|
f"{status}: {text}; drueben {inhalt(werkzeuge, '.vault-tools/linkcheck/a.py')!r}",
|
|
)
|
|
|
|
# --- Loeschung geht mit -----------------------------------------
|
|
git(vault, "rm", "-q", ".claude/agents/x.md")
|
|
committe(vault, "x geloescht")
|
|
spiegel.spiegele(vault, klon)
|
|
pruefe(
|
|
"loeschung geht mit",
|
|
".claude/agents/x.md" not in dateien(werkzeuge),
|
|
f"drueben {sorted(dateien(werkzeuge))}",
|
|
)
|
|
|
|
# --- Schutz: nichts leeren, was nicht unser Klon ist ------------
|
|
fremd = basis / "fremd"
|
|
fremd.mkdir()
|
|
git(fremd, "init", "-q")
|
|
# Erreichbares Remote in anderer Schreibweise: ohne Schutz liefe der
|
|
# Abgleich durch und leerte den Ordner.
|
|
git(fremd, "remote", "add", "origin", str(werkzeuge))
|
|
schreibe(fremd / "bleibt.txt", b"x\n")
|
|
meldung = wirft(vault, fremd)
|
|
pruefe(
|
|
"fremder klon -> befund, nichts geleert",
|
|
meldung is not None and (fremd / "bleibt.txt").is_file(),
|
|
str(meldung)[:100],
|
|
)
|
|
|
|
innen = vault / "klon-im-vault"
|
|
schreibe(innen / "bleibt.txt", b"x\n")
|
|
meldung = wirft(vault, innen)
|
|
pruefe(
|
|
"klon im vault -> befund, nichts geleert",
|
|
meldung is not None and (innen / "bleibt.txt").is_file(),
|
|
str(meldung)[:100],
|
|
)
|
|
|
|
ohne_git = basis / "ohne-git"
|
|
schreibe(ohne_git / "bleibt.txt", b"x\n")
|
|
meldung = wirft(vault, ohne_git)
|
|
pruefe(
|
|
"ordner ohne git -> befund, nichts geleert",
|
|
meldung is not None and (ohne_git / "bleibt.txt").is_file(),
|
|
str(meldung)[:100],
|
|
)
|
|
|
|
# --- Fehlende Voraussetzungen sind Befunde ----------------------
|
|
vault2, werkzeuge2 = baue(basis / "b")
|
|
raeume(werkzeuge2)
|
|
meldung = wirft(vault2, basis / "b" / "klon")
|
|
pruefe("remote fehlt beim klonen -> befund", meldung is not None, str(meldung)[:100])
|
|
|
|
vault3, werkzeuge3 = baue(basis / "c")
|
|
git(vault3, "rm", "-q", ".vault-tools/auto-push/werkzeuge-README.md")
|
|
committe(vault3, "readme weg")
|
|
meldung = wirft(vault3, basis / "c" / "klon")
|
|
pruefe(
|
|
"readme-quelle fehlt -> befund, nichts gepusht",
|
|
meldung is not None and "README" in meldung and commits(werkzeuge3) == 1,
|
|
f"Commits {commits(werkzeuge3)}; {str(meldung)[:80]}",
|
|
)
|
|
pruefe(
|
|
"readme-quelle fehlt -> kein klon angelegt",
|
|
not (basis / "c" / "klon").exists(),
|
|
f"klon vorhanden: {(basis / 'c' / 'klon').exists()}",
|
|
)
|
|
|
|
vault4, werkzeuge4 = baue(basis / "d")
|
|
klon4 = basis / "d" / "klon"
|
|
spiegel.spiegele(vault4, klon4)
|
|
raeume(werkzeuge4)
|
|
schreibe(vault4 / ".vault-tools" / "linkcheck" / "a.py", b"print('d')\n")
|
|
committe(vault4, "a geaendert")
|
|
meldung = wirft(vault4, klon4)
|
|
pruefe("remote weg bei bestehendem klon -> befund", meldung is not None, str(meldung)[:100])
|
|
|
|
vault5, _ = baue(basis / "e")
|
|
git(vault5, "remote", "remove", "origin")
|
|
meldung = wirft(vault5, basis / "e" / "klon")
|
|
pruefe("vault ohne origin -> befund", meldung is not None, str(meldung)[:100])
|
|
|
|
print()
|
|
print(f"{bestanden} von {gesamt} Faellen bestanden.")
|
|
return 0 if bestanden == gesamt else 1
|
|
finally:
|
|
raeume(basis)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
except AttributeError:
|
|
pass
|
|
raise SystemExit(main())
|