File size: 1,489 Bytes
dfb775d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 | """Deploy registry + hot-swap atomicity."""
from __future__ import annotations
import pytest
from mindxtrain.deploy.hot_swap import promote_canary_to_live, rollback_live
from mindxtrain.deploy.registry import DeployRegistry, Slot
def test_registry_round_trip(tmp_path):
reg = DeployRegistry(tmp_path / "reg.json")
assert reg.get("live") is None
reg.set(Slot(name="live", run_id="r1", blake3="a" * 64))
got = reg.get("live")
assert got is not None
assert got.run_id == "r1"
def test_promote_canary_to_live_retains_previous(tmp_path):
reg = DeployRegistry(tmp_path / "reg.json")
reg.set(Slot(name="live", run_id="v1", blake3="0" * 64))
reg.set(Slot(name="canary", run_id="v2", blake3="1" * 64))
new_live = promote_canary_to_live(tmp_path / "reg.json")
assert new_live.run_id == "v2"
assert reg.get("previous_live") is not None
assert reg.get("previous_live").run_id == "v1"
assert reg.get("canary") is None
def test_rollback_live(tmp_path):
reg = DeployRegistry(tmp_path / "reg.json")
reg.set(Slot(name="live", run_id="v1", blake3="0" * 64))
reg.set(Slot(name="canary", run_id="v2", blake3="1" * 64))
promote_canary_to_live(tmp_path / "reg.json")
rolled = rollback_live(tmp_path / "reg.json")
assert rolled.run_id == "v1"
def test_rollback_with_no_previous_raises(tmp_path):
DeployRegistry(tmp_path / "reg.json")
with pytest.raises(RuntimeError):
rollback_live(tmp_path / "reg.json")
|