92 lines
3.2 KiB
Python
92 lines
3.2 KiB
Python
from types import SimpleNamespace
|
|
|
|
from gpu_rent.cli import _stop_after_failed_up
|
|
|
|
|
|
def test_stop_after_failed_up_calls_cmd_stop(monkeypatch):
|
|
calls: list[dict] = []
|
|
|
|
def fake_stop(cfg, *, no_pull=False, log=None, destroy_disks=False):
|
|
calls.append({"no_pull": no_pull, "destroy_disks": destroy_disks})
|
|
return SimpleNamespace()
|
|
|
|
monkeypatch.setattr("gpu_rent.cli.cmd_stop", fake_stop)
|
|
monkeypatch.setattr("gpu_rent.cli.warn", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("gpu_rent.cli.ok", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("gpu_rent.cli.err", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("gpu_rent.cli.log", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(
|
|
"gpu_rent.state.load_state",
|
|
lambda: SimpleNamespace(floating_ip=None),
|
|
)
|
|
|
|
_stop_after_failed_up(SimpleNamespace(), RuntimeError("boom"))
|
|
assert calls == [{"no_pull": True, "destroy_disks": False}]
|
|
|
|
|
|
def test_stop_after_failed_up_prints_digest_when_fip(monkeypatch):
|
|
calls: list[dict] = []
|
|
digests: list[str] = []
|
|
|
|
def fake_stop(cfg, *, no_pull=False, log=None, destroy_disks=False):
|
|
calls.append({"no_pull": no_pull})
|
|
return SimpleNamespace()
|
|
|
|
monkeypatch.setattr("gpu_rent.cli.cmd_stop", fake_stop)
|
|
monkeypatch.setattr("gpu_rent.cli.warn", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("gpu_rent.cli.ok", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("gpu_rent.cli.err", lambda *_a, **_k: None)
|
|
monkeypatch.setattr("gpu_rent.cli.log", lambda *_a, **_k: None)
|
|
monkeypatch.setattr(
|
|
"gpu_rent.state.load_state",
|
|
lambda: SimpleNamespace(floating_ip="10.0.0.1"),
|
|
)
|
|
monkeypatch.setattr(
|
|
"gpu_rent.vm_logs.print_log_digest",
|
|
lambda cfg, host, **kw: digests.append(host),
|
|
)
|
|
|
|
_stop_after_failed_up(SimpleNamespace(enable_swarmui=True, llm_runtime="none"), RuntimeError("boom"))
|
|
assert digests == ["10.0.0.1"]
|
|
assert calls == [{"no_pull": True}]
|
|
|
|
|
|
def test_up_stop_on_fail_default_true(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("HOME", str(tmp_path))
|
|
monkeypatch.setenv("USERPROFILE", str(tmp_path))
|
|
for key in (
|
|
"OS_AUTH_URL",
|
|
"OS_USER_DOMAIN_NAME",
|
|
"OS_USERNAME",
|
|
"OS_PASSWORD",
|
|
"OS_PROJECT_ID",
|
|
"OS_REGION_NAME",
|
|
"GPU_RENT_AZ",
|
|
):
|
|
monkeypatch.setenv(key, "x")
|
|
monkeypatch.delenv("UP_STOP_ON_FAIL", raising=False)
|
|
from gpu_rent.config import load_config
|
|
|
|
cfg = load_config(require_auth=False)
|
|
assert cfg.up_stop_on_fail is True
|
|
|
|
|
|
def test_up_stop_on_fail_can_disable(monkeypatch, tmp_path):
|
|
monkeypatch.setenv("HOME", str(tmp_path))
|
|
monkeypatch.setenv("USERPROFILE", str(tmp_path))
|
|
for key in (
|
|
"OS_AUTH_URL",
|
|
"OS_USER_DOMAIN_NAME",
|
|
"OS_USERNAME",
|
|
"OS_PASSWORD",
|
|
"OS_PROJECT_ID",
|
|
"OS_REGION_NAME",
|
|
"GPU_RENT_AZ",
|
|
):
|
|
monkeypatch.setenv(key, "x")
|
|
monkeypatch.setenv("UP_STOP_ON_FAIL", "false")
|
|
from gpu_rent.config import load_config
|
|
|
|
cfg = load_config(require_auth=False)
|
|
assert cfg.up_stop_on_fail is False
|