Leonid PershinandCursor
0bad693b12
- Updated Selectel documentation to clarify error responses related to balance issues. - Introduced `peek_balance_rub` function for best-effort balance checks, handling API failures gracefully. - Improved error messages for insufficient funds and quota issues, specifying Selectel's 403 policy response. - Added balance checks in the doctor command to ensure users are informed about their balance status before attempting GPU creation. - Refactored exception handling in cloud operations to provide clearer feedback on balance-related errors. Co-authored-by: Cursor <cursoragent@cursor.com>
114 lines
3.6 KiB
Python
114 lines
3.6 KiB
Python
from types import SimpleNamespace
|
|
|
|
from gpu_rent.cli import _stop_after_failed_up
|
|
|
|
|
|
def _quiet(monkeypatch):
|
|
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)
|
|
|
|
|
|
def test_stop_after_failed_up_skips_when_no_server(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)
|
|
_quiet(monkeypatch)
|
|
monkeypatch.setattr(
|
|
"gpu_rent.state.load_state",
|
|
lambda: SimpleNamespace(server_id=None, floating_ip=None),
|
|
)
|
|
|
|
_stop_after_failed_up(SimpleNamespace(), RuntimeError("boom"))
|
|
assert calls == []
|
|
|
|
|
|
def test_stop_after_failed_up_runs_when_server_exists(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)
|
|
_quiet(monkeypatch)
|
|
monkeypatch.setattr(
|
|
"gpu_rent.state.load_state",
|
|
lambda: SimpleNamespace(server_id="s1", 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)
|
|
_quiet(monkeypatch)
|
|
monkeypatch.setattr(
|
|
"gpu_rent.state.load_state",
|
|
lambda: SimpleNamespace(server_id="s1", 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
|