Something went wrong. Try again.
bayes for days
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672"""A corpus has to outlive the worktree it was recorded in, and say what it is.
Both halves were paid for on the same afternoon: one `git worktree remove--force` destroyed 360 matches and 21937 decisions, and a fit against a corpusrecorded with an older feature set produced a weight vector nobody could tellfrom a good one. The tests below are those two failures, made cheap."""
import contextlibimport gzipimport ioimport jsonimport osimport shutilimport tempfileimport unittestfrom dataclasses import asdictfrom pathlib import Pathfrom unittest import mock
from sds.corpus import ( DATA, MANIFEST, CorpusError, Manifest, Review, Stale, basis_fingerprint, check, check_basis, check_data, corpora_root, listing, recorded_decisions, resolve, save, save_run, stamp,)from sds.epoch import EPOCHfrom sds.train import read_corpusfrom tests.synthetic import LOCAL_FEATURE, write_run
PLANTED = {"expected_damage": 1.0, "overkill": -0.5, "p_kill": 0.25}
class CorpusTest(unittest.TestCase): """A store in a temporary directory, so nothing here touches the real one."""
def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.store = self.root / "corpora" self.previous = os.environ.get("SDS_CORPORA") os.environ["SDS_CORPORA"] = str(self.store) self.addCleanup(self.tmp.cleanup) self.addCleanup(self._restore)
def _restore(self): if self.previous is None: os.environ.pop("SDS_CORPORA", None) else: os.environ["SDS_CORPORA"] = self.previous
def run_dir(self, name="run", **kwargs): directory = write_run(self.root / name, PLANTED, matches=3, decisions=6, **kwargs) (directory / "baseline.json").write_text( json.dumps( { "bot": "sds-bot --weights weights/fitted.json --explore 0.1", "scenario": "suite:mirror:abcdef123456", } ) ) return directory
class TestSavedCorpusOutlivesItsRun(CorpusTest): """The failure this module exists because of.
The 360-match corpus was a directory of symlinks into a worktree. Removing the worktree removed the corpus, and `runs/` being gitignored meant there was nothing left that even named it. """
def test_data_survives_the_source_directory_being_deleted(self): source = self.run_dir() manifest = save(source, "overnight") shutil.rmtree(source) self.assertFalse(source.exists())
self.assertTrue(manifest.data.is_dir()) corpus = read_corpus(manifest.data) self.assertEqual(len(corpus.matches), manifest.matches) self.assertEqual(len(corpus.decisions), manifest.decisions)
def test_the_per_match_log_and_report_are_kept(self): """The only written record of why MegaMek refused something.
These were dropped on the grounds that the `.log` was the bulk of a run directory. On a real 120-match run it is 4.0M against 7.3G of decision logs, and the victory condition, the seat map and every refusal reason are in it and nowhere else. A corpus exists so the run can be deleted. """ source = self.run_dir() (source / "mirror-lance-seed1-abcdef12.log").write_text( " victory : a side loses at 70% of its BV destroyed\n" ) (source / "mirror-lance-seed1-abcdef12.report.html").write_text("<html></html>") manifest = save(source, "overnight") shutil.rmtree(source)
kept = {path.name for path in manifest.data.iterdir()} self.assertIn("mirror-lance-seed1-abcdef12.log", kept) self.assertIn("mirror-lance-seed1-abcdef12.report.html", kept) self.assertIn( "victory", (manifest.data / "mirror-lance-seed1-abcdef12.log").read_text(), )
def test_a_plain_log_is_compressed_on_the_way_into_the_store(self): """A rescue must not put the uncompressed form on the disk twice.
A recording run compresses its own logs as each match ends, so this only fires for an older run directory. That is the case that hurts: the run which took this machine to 98% left 17G in `runs/`, and copying it would have been the second 17G against 8.4G free. """ source = self.run_dir() wanted = {path.name: path.read_bytes() for path in source.glob("*.decisions.jsonl")} self.assertTrue(wanted) manifest = save(source, "overnight")
stored = sorted(p.name for p in manifest.data.glob("*.decisions.jsonl*")) self.assertEqual(stored, sorted(name + ".gz" for name in wanted)) for name, body in wanted.items(): self.assertEqual(gzip.decompress((manifest.data / (name + ".gz")).read_bytes()), body)
def test_the_arrival_check_still_names_a_log_that_did_not_get_there(self): # Comparing names with `.gz` stripped must not turn the check that # caught `honest-movement-r8` into one that passes on anything. source = self.run_dir() from sds.match import compress_log as original
def only_the_first(path, *args, **kwargs): if getattr(only_the_first, "done", False): path.unlink() return path only_the_first.done = True return original(path, *args, **kwargs)
with ( mock.patch("sds.match.compress_log", only_the_first), self.assertRaises(CorpusError) as caught, ): save(source, "half-arrived") self.assertIn("did not reach the store", str(caught.exception))
def test_nothing_in_the_store_is_a_symlink(self): manifest = save(self.run_dir(), "overnight") for path in manifest.directory.rglob("*"): self.assertFalse(path.is_symlink(), f"{path} is a symlink")
def test_the_store_is_outside_the_run(self): source = self.run_dir() manifest = save(source, "overnight") self.assertNotIn(source.resolve(), manifest.data.resolve().parents) self.assertEqual(manifest.directory.parent, corpora_root())
def test_the_manifest_sits_beside_the_ignored_data(self): # The layout the .gitignore rule depends on: `corpora/*/data/` ignored, # the manifest one level up and committed. manifest = save(self.run_dir(), "overnight") self.assertTrue((manifest.directory / MANIFEST).is_file()) self.assertEqual(manifest.data.name, DATA)
def test_saving_over_a_corpus_needs_asking(self): self.run_dir() save(self.root / "run", "overnight") with self.assertRaises(CorpusError) as caught: save(self.root / "run", "overnight") self.assertIn("--replace", str(caught.exception)) save(self.root / "run", "overnight", replace=True)
class TestARecordingRunIsSavedWithoutBeingAsked(CorpusTest): """Whether the work survives must not be a flag.
Two corpora have been destroyed by the same shape of accident: the run had decision logs, nobody passed `--corpus`, nothing was saved, and nothing said so. The second was the asymmetric corpus every other figure in `docs/TACTICS.md` was measured against, and it is not recoverable. """
def test_a_run_with_decisions_is_saved_under_its_own_name(self): run = self.run_dir("20260829T041807Z-bench") manifest, invented = save_run(run, None) self.assertTrue(invented) self.assertEqual(manifest.name, "20260829T041807Z-bench") self.assertTrue((corpora_root() / manifest.name / DATA).is_dir())
def test_the_saved_data_outlives_the_run_directory(self): run = self.run_dir("20260829T041807Z-bench") manifest, _ = save_run(run, None) shutil.rmtree(run) self.assertTrue((corpora_root() / manifest.name / DATA).is_dir()) self.assertTrue(read_corpus(corpora_root() / manifest.name / DATA))
def test_a_given_name_wins_and_is_not_reported_as_invented(self): run = self.run_dir("20260829T041807Z-bench") manifest, invented = save_run(run, "tactics-hub-mirrored") self.assertFalse(invented) self.assertEqual(manifest.name, "tactics-hub-mirrored")
def test_a_run_that_recorded_nothing_is_refused_rather_than_stored(self): bare = self.root / "no-decisions" bare.mkdir() (bare / "a.json").write_text("{}") self.assertFalse(recorded_decisions(bare)) with self.assertRaises(CorpusError): save_run(bare, None)
def test_a_named_save_of_an_empty_run_still_refuses(self): bare = self.root / "no-decisions-named" bare.mkdir() with self.assertRaises(CorpusError): save_run(bare, "wishful")
class TestTheManifestCannotDescribeDataThatIsNotThere(CorpusTest): """`honest-movement-r8`, made cheap.
It sits in the store with a manifest asserting 5944 decisions over 242 results, 120 host logs, 120 HTML reports and zero decision logs - saved before `CORPUS_GLOBS` learned the `.gz` suffix, so `scan` read the compressed logs and the copy loop never saw them. A corpus that lies about holding training data is worse than one that is missing, because it satisfies every check anybody thinks to run. """
def test_a_log_that_does_not_reach_the_store_is_refused(self): run = self.run_dir("dropped") real_copy = shutil.copy2
def skip_the_decision_logs(src, dst, *args, **kwargs): if ".decisions.jsonl" in Path(src).name: return dst return real_copy(src, dst, *args, **kwargs)
with ( mock.patch("sds.corpus.shutil.copy2", skip_the_decision_logs), self.assertRaises(CorpusError) as caught, ): save(run, "silently-empty") self.assertIn("did not reach the store", str(caught.exception))
def test_the_half_written_corpus_is_not_left_behind(self): run = self.run_dir("dropped") real_copy = shutil.copy2
def skip_the_decision_logs(src, dst, *args, **kwargs): if ".decisions.jsonl" in Path(src).name: return dst return real_copy(src, dst, *args, **kwargs)
with ( mock.patch("sds.corpus.shutil.copy2", skip_the_decision_logs), contextlib.suppress(CorpusError), ): save(run, "silently-empty") self.assertFalse((corpora_root() / "silently-empty" / DATA).exists())
def test_a_healthy_save_keeps_every_decision_log(self): run = self.run_dir("intact") manifest = save(run, "intact") data = corpora_root() / manifest.name / DATA source = {p.name for p in run.glob("*.decisions.jsonl")} self.assertTrue(source) # Compressed on the way in, so the names carry a `.gz`. What is checked # is that every log got there, which is what the store's own arrival # check compares too. self.assertEqual( {name + ".gz" for name in source}, {p.name for p in data.glob("*.decisions.jsonl.gz")}, )
class TestProvenanceComesFromTheRunNotTheSaver(CorpusTest): """The rescue case, which is the one that defeats the staleness guard.
`stamp` writes the commit, epoch and basis into the run directory as the run starts. `save` used to overwrite all three with the *saving* checkout's, which is right only when the run and the save are the same checkout. Rescuing an old run from a newer worktree therefore stamped it with a basis it was never recorded under - and since `sds train` refuses a corpus whose basis differs from the current build, the mis-stamp does not merely mislabel it, it makes the guard pass. """
def stamped_run(self, name="rescued"): run = self.run_dir(name) # What `stamp` writes as the run starts: the provenance of the machine # and checkout that actually played the matches. stamped = Manifest( name=name, recorded="2026-08-26T02:18:18Z", commit="1d6bb72-dirty", epoch=EPOCH, basis="4a963bc90ee9", image="lance-blue/sds-runner:dfold", ) (run / MANIFEST).write_text(json.dumps(asdict(stamped), indent=2, sort_keys=True)) return run
def test_the_runs_own_basis_and_commit_survive_the_save(self): run = self.stamped_run() manifest = save(run, "rescued-corpus") self.assertEqual(manifest.basis, "4a963bc90ee9") self.assertEqual(manifest.commit, "1d6bb72-dirty") self.assertEqual(manifest.recorded, "2026-08-26T02:18:18Z")
def test_the_counts_are_still_this_saves_own_work(self): run = self.stamped_run() manifest = save(run, "rescued-corpus") self.assertEqual(manifest.matches, 3) self.assertTrue(manifest.decisions)
def test_a_run_with_no_stamp_says_so_rather_than_pretending(self): run = self.run_dir("unstamped") (run / MANIFEST).unlink(missing_ok=True) errors = io.StringIO() with contextlib.redirect_stderr(errors): save(run, "unstamped-corpus") self.assertIn("carries no manifest of its own", errors.getvalue())
class TestManifestRecordsWhatProducedIt(CorpusTest): def test_it_carries_the_provenance_a_fit_needs(self): manifest = save(self.run_dir(), "overnight", notes="the overnight one") self.assertEqual(manifest.epoch, EPOCH) self.assertEqual(manifest.scenario, "suite:mirror:abcdef123456") self.assertEqual(manifest.weights, "weights/fitted.json") self.assertTrue(manifest.explore) self.assertAlmostEqual(manifest.explore_rate, 0.1) self.assertEqual(manifest.matches, 3) self.assertEqual(manifest.notes, "the overnight one") self.assertEqual(manifest.source, "run")
def test_the_feature_list_is_sorted_and_complete(self): manifest = save(self.run_dir(), "overnight") expected = sorted([*PLANTED, LOCAL_FEATURE]) self.assertEqual(manifest.features, expected) self.assertEqual(manifest.learnable, sorted(PLANTED)) self.assertEqual(manifest.local, [LOCAL_FEATURE])
def test_it_round_trips_through_the_file(self): saved = save(self.run_dir(), "overnight") loaded = Manifest.load("overnight") self.assertEqual(loaded, saved)
def test_a_manifest_from_a_newer_sds_is_named_rather_than_crashed_on(self): manifest = save(self.run_dir(), "overnight") path = manifest.directory / MANIFEST document = json.loads(path.read_text()) document["provenance_v2"] = "something this version has never heard of" path.write_text(json.dumps(document)) with self.assertRaises(CorpusError) as caught: Manifest.load("overnight") self.assertIn("provenance_v2", str(caught.exception))
class TestStaleCorpusIsRefused(CorpusTest): """The second failure: a fit against a corpus that is not about this bot.
Same rule as `baseline.Incomparable`. The numbers are the same shape and mean different things, so a mismatch is an error rather than a footnote in a report nobody reads to the end. """
def test_a_different_epoch_names_what_changed(self): manifest = save(self.run_dir(), "overnight") manifest.epoch = EPOCH - 1 with self.assertRaises(Stale) as caught: check(manifest, reference=[]) message = str(caught.exception) self.assertIn(f"epoch {EPOCH - 1}", message) self.assertIn(f"epoch {EPOCH}", message) # The epoch's own note, not just its number: which change made the two # runs different games. from sds.epoch import describe
self.assertIn(describe(EPOCH), message)
def test_the_same_epoch_is_fine(self): manifest = save(self.run_dir(), "overnight") manifest.explore = False self.assertEqual(check(manifest, reference=[]), Review())
def test_requested_features_the_corpus_lacks_are_named(self): manifest = save(self.run_dir(), "overnight") asked = ["p_kill", "expected_damage", "level_tmm", "cover_quality", "los_out"] with self.assertRaises(Stale) as caught: check(manifest, requested=asked, reference=[]) message = str(caught.exception) self.assertIn("3 of the 5", message) for absent in ("level_tmm", "cover_quality", "los_out"): self.assertIn(absent, message) # And not the two it does have. self.assertNotIn("p_kill,", message)
def test_a_feature_the_baseline_m_uses_and_the_corpus_lacks_is_warned_about(self): # The expensive one: 21 features recorded, 26 in the checkout, and a fit # that silently produces a vector about the 21. A warning rather than a # refusal only because every corpus recorded today is missing the # positional features, so refusing would fire on every fit and teach # everybody to pass --stale-ok without reading it. manifest = save(self.run_dir(), "overnight") reference = sorted([*PLANTED, "exposure", "concentration"]) review = check(manifest, reference=reference) self.assertEqual(len(review.warnings), 1) message = review.warnings[0] self.assertIn("2 of them are absent", message) self.assertIn("exposure", message) self.assertIn("concentration", message) self.assertIn(str(len(manifest.features)), message)
def test_a_feature_the_corpus_has_and_the_baseline_does_not_is_a_note(self): manifest = save(self.run_dir(), "overnight") review = check(manifest, reference=["p_kill"]) self.assertTrue(any("expected_damage" in note for note in review.notes))
def test_exploration_is_said_out_loud(self): manifest = save(self.run_dir(), "overnight") review = check(manifest, reference=[]) self.assertTrue(any("off-policy" in note for note in review.notes))
def test_data_the_manifest_does_not_describe_is_refused(self): # The manifest is what gets committed and read years later, so it has to # be the truth about the directory beside it. manifest = save(self.run_dir(), "overnight") check_data(manifest, manifest.features) with self.assertRaises(Stale) as caught: check_data(manifest, [*manifest.features, "smokestack_index"]) self.assertIn("smokestack_index", str(caught.exception))
class TestResolvingATarget(CorpusTest): def test_a_name_resolves_to_the_data_and_its_manifest(self): saved = save(self.run_dir(), "overnight") target, manifest = resolve("overnight") self.assertEqual(target, saved.data) self.assertEqual(manifest, saved)
def test_a_corpus_directory_resolves_the_same_way(self): saved = save(self.run_dir(), "overnight") target, manifest = resolve(saved.directory) self.assertEqual(target, saved.data) self.assertIsNotNone(manifest)
def test_a_plain_run_directory_has_no_manifest(self): source = self.run_dir() target, manifest = resolve(source) self.assertEqual(target, source) self.assertIsNone(manifest)
class TestListing(CorpusTest): def test_it_lists_by_name(self): self.run_dir() save(self.root / "run", "beta") save(self.root / "run", "alpha") self.assertEqual([m.name for m in listing()], ["alpha", "beta"])
def test_an_empty_store_lists_nothing(self): self.assertEqual(listing(), [])
if __name__ == "__main__": unittest.main()
class TestBasisFingerprint(unittest.TestCase): """A name list cannot see a feature that keeps its name and changes meaning.
`p_kill` asked `P(total damage >= centre torso health)` and was overconfident by between 3.5x and 1144x depending on the volley's shape. Its replacement asks `P(a lethal location is destroyed)`. Same name, same normalisation, a different quantity, and every weight fitted against the old one is void. """
def entry(self, name="p_kill", norm="bounded", description="The chance of a kill."): return {"name": name, "normalisation": norm, "description": description}
def test_the_same_basis_hashes_the_same(self): a = [self.entry(), self.entry("overkill")] self.assertEqual(basis_fingerprint(a), basis_fingerprint(list(reversed(a))))
def test_a_changed_description_changes_the_basis(self): before = basis_fingerprint([self.entry(description="P(total >= CT health).")]) after = basis_fingerprint([self.entry(description="P(a lethal location dies).")]) self.assertNotEqual(before, after)
def test_a_changed_normalisation_changes_the_basis(self): self.assertNotEqual( basis_fingerprint([self.entry(norm="bounded")]), basis_fingerprint([self.entry(norm="rank")]), )
def test_an_added_feature_changes_the_basis(self): self.assertNotEqual( basis_fingerprint([self.entry()]), basis_fingerprint([self.entry(), self.entry("p_breach")]), )
def test_a_mismatched_basis_is_refused_and_says_why(self): manifest = Manifest(name="old", basis="aaaaaaaaaaaa") with self.assertRaises(Stale) as caught: check_basis(manifest, "bbbbbbbbbbbb") message = str(caught.exception) self.assertIn("aaaaaaaaaaaa", message) self.assertIn("bbbbbbbbbbbb", message)
def test_a_corpus_with_no_basis_is_not_refused(self): # Recorded before the fingerprint existed. Not worth refusing over. check_basis(Manifest(name="old", basis=""), "bbbbbbbbbbbb")
# What the bot would answer. The stamp reads its basis out of the built# binary, and whether a worktree has one built is not what these tests are# about - a fresh worktree has no `target/release/sds-bot` and the whole class# would fail there for a reason that has nothing to do with stamping.CATALOGUE = [ {"name": "expected_damage", "normalisation": "fraction", "description": "One."}, {"name": "cover_quality", "normalisation": "fraction", "description": "Two."}, {"name": "enemy_nearest", "normalisation": "local", "description": "Three."},]
class AStampedRunIsCheckable(unittest.TestCase): """A run directory says what recorded it, so `fit` cannot skip the check.
The hole this closes: `fit` runs `check_basis` only when a manifest is found, and a bench run never wrote one. It printed a note instead, and a note is a thing you read past at four in the morning. A corpus recorded against a retired basis was fitted that way. """
def setUp(self): self.root = Path(tempfile.mkdtemp()) self.run = self.root / "20260824T000000Z-bench" self.run.mkdir(parents=True) patch = mock.patch("sds.catalogue.from_bot", return_value={"features": CATALOGUE}) patch.start() self.addCleanup(patch.stop)
def tearDown(self): shutil.rmtree(self.root, ignore_errors=True)
def test_the_stamp_makes_the_directory_it_writes_into(self): """The run directory does not exist yet when the stamp is written.
`_run_dir` names it and the first match creates it, so a stamp that assumed the directory was there wrote nothing and printed a note - the whole fix, silently absent, found by running it rather than by this suite, which had been handing `stamp` a directory that already existed. """ fresh = self.root / "20260825T000000Z-bench" self.assertFalse(fresh.exists()) stamp(fresh) self.assertTrue((fresh / "manifest.json").is_file())
def test_a_stamped_run_resolves_to_a_manifest(self): stamp(self.run) _, manifest = resolve(self.run) self.assertIsNotNone(manifest, "a stamped run must bring its manifest to `fit`")
def test_the_stamp_carries_what_the_guards_read(self): manifest = stamp(self.run, bot="sds-bot --explore 0.3", explore_rate=0.3) # The two the guards actually use, plus the provenance a reader wants. self.assertTrue(manifest.basis, "no basis means check_basis cannot fire") self.assertTrue(manifest.features, "no features means check_data cannot fire") self.assertEqual(manifest.epoch, EPOCH) self.assertTrue(manifest.explore) self.assertEqual(manifest.explore_rate, 0.3)
def test_a_stamped_run_refuses_a_fit_across_a_basis_change(self): stamp(self.run) _, manifest = resolve(self.run) with self.assertRaises(Stale): check_basis(manifest, "0000deadbeef")
def test_a_basis_the_stamp_could_not_read_is_said_out_loud(self): """An empty basis is the hole reopening, so it must not be quiet.
`check_basis` returns early on an empty basis, so a stamp that failed to read the binary leaves a manifest that looks checked and is not. """ with ( mock.patch("sds.catalogue.from_bot", side_effect=RuntimeError("not built")), contextlib.redirect_stderr(io.StringIO()) as caught, ): manifest = stamp(self.run) self.assertEqual(manifest.basis, "") self.assertIn("will not be checked", caught.getvalue())
def test_counts_are_zero_until_the_corpus_is_saved(self): # Written before the first match, so there is nothing to count yet. # `check` and `check_basis` read neither, and `corpus save` fills them. manifest = stamp(self.run) self.assertEqual((manifest.matches, manifest.decisions), (0, 0))
class TestRunnerImageIsPinned(unittest.TestCase): """A run records the image it executed in, and the name is not `:latest`.
The default used to be a bare `lance-blue/sds-runner`, which docker reads as `:latest`: one mutable name on a daemon several agents share. A rebuild by any of them replaced the image a running benchmark was using, between one match and the next, and nothing failed - the matches carried on in a different environment. On this machine `:latest` and a fresh build of the Dockerfile were already two different image IDs. """
def test_the_tag_comes_from_the_dockerfile(self): import hashlib
from sds.match import REPO, default_image
digest = hashlib.sha256((REPO / "bridge" / "Dockerfile").read_bytes()).hexdigest() self.assertEqual(default_image(), f"lance-blue/sds-runner:df{digest[:12]}")
def test_build_and_run_derive_the_same_name(self): # `scripts/build.sh` computes it in bash and `match.py` in python. If # they ever disagree the harness runs an image nothing built. import subprocess
from sds.match import REPO, default_image
out = subprocess.run( ["sha256sum", str(REPO / "bridge" / "Dockerfile")], capture_output=True, text=True, check=True, ).stdout self.assertEqual(default_image(), f"lance-blue/sds-runner:df{out[:12]}")
def test_an_explicit_override_still_wins(self): import os from unittest import mock
from sds.match import default_image
with mock.patch.dict(os.environ, {"SDS_IMAGE": "someone/else:1"}): self.assertEqual(default_image(), "someone/else:1")
def test_the_stamp_records_it(self): directory = Path(tempfile.mkdtemp()) / "20260101T000000Z-bench" from sds import corpus
manifest = corpus.stamp(directory, bot="/work/bot", weights="w.json", image="pinned:1") self.assertEqual(manifest.image, "pinned:1") self.assertEqual(json.loads((directory / "manifest.json").read_text())["image"], "pinned:1")
def test_a_manifest_written_before_this_still_reads(self): # Every stored corpus predates the field. directory = Path(tempfile.mkdtemp()) directory.mkdir(parents=True, exist_ok=True) (directory / "manifest.json").write_text(json.dumps({"name": "old", "commit": "abc"})) from sds import corpus
self.assertEqual(corpus.Manifest.read(directory / "manifest.json").image, "")