osbuild: --checkpoint can now use globs

When developing or rebuilding manifests a lot it is common to want to
checkpoint everything to the store. It seems we all have small shell
scripts hanging around for this.

Let `--checkpoint` take a shell-like glob such as `--checkpoint="*"` to
checkpoint everything.

Note that there's a behavioral change here; previously `osbuild
--checkpoint=a` would error if that specific checkpoint wasn't found.
Now `osbuild` will only error if nothing was selected by the passed
globs.
This commit is contained in:
Simon de Vlieger 2023-08-04 12:31:23 +02:00 committed by Achilleas Koutsou
parent 14d31633a4
commit 427e82e0c0
2 changed files with 24 additions and 21 deletions

View file

@ -3,6 +3,7 @@ import contextlib
import hashlib
import json
import os
from fnmatch import fnmatch
from typing import Dict, Generator, Iterable, Iterator, List, Optional
from . import buildroot, host, objectstore, remoteloop
@ -472,27 +473,30 @@ class Manifest:
return results
def mark_checkpoints(self, checkpoints):
points = set(checkpoints)
def mark_checkpoints(self, patterns):
"""Match pipeline names, stage ids, and stage names against an iterable
of `fnmatch`-patterns."""
selected = []
def mark_stage(stage):
c = stage.id
if c in points:
stage.checkpoint = True
points.remove(c)
def matching(haystack):
return any(fnmatch(haystack, p) for p in patterns)
def mark_pipeline(pl):
if pl.name in points and pl.stages:
pl.stages[-1].checkpoint = True
points.remove(pl.name)
for pipeline in self.pipelines.values():
# checkpoints are marked on stages, if a pipeline has no stages we
# can't mark it
if not pipeline.stages:
continue
for stage in pl.stages:
mark_stage(stage)
if matching(pipeline.name):
selected.append(pipeline.name)
pipeline.stages[-1].checkpoint = True
for pl in self.pipelines.values():
mark_pipeline(pl)
for stage in pipeline.stages:
if matching(stage.id) or matching(stage.name):
selected.append(stage.id)
stage.checkpoint = True
return points
return selected
def get(self, name_or_id: str) -> Optional[Pipeline]:
pl = self.pipelines.get(name_or_id)