Leonid Pershin
82e36129cd
- Integrated `print_access_card` functionality into the `up` command for both tunneled and non-tunneled scenarios. - Removed the deprecated `print_mcp_snippet` function from the session management flow. - Updated the `tunnel_forwards` function to streamline port handling for LLM runtimes. - Enhanced test cases to reflect changes in access card printing and MCP snippet logging.
84 lines
2.4 KiB
Python
84 lines
2.4 KiB
Python
from gpu_rent.notify import print_mcp_snippet
|
|
from gpu_rent.snapshot import ensure_boot_snapshot
|
|
|
|
|
|
class _Cfg:
|
|
boot_snapshot_name = "gpu-rent-boot-ok"
|
|
swarmui_local_port = 17801
|
|
notify_ready = True
|
|
llm_runtime = "none"
|
|
ollama_local_port = 17811
|
|
llamacpp_local_port = 17812
|
|
|
|
|
|
def test_ensure_boot_snapshot_skips_existing():
|
|
class Snap:
|
|
id = "snap1"
|
|
name = "gpu-rent-boot-ok"
|
|
|
|
class Conn:
|
|
class block_storage:
|
|
@staticmethod
|
|
def create_snapshot(**kwargs):
|
|
raise AssertionError("should not create")
|
|
|
|
logs = []
|
|
import gpu_rent.snapshot as snap_mod
|
|
|
|
# monkey via attribute
|
|
orig = snap_mod.find_snapshot_by_name
|
|
snap_mod.find_snapshot_by_name = lambda conn, name: Snap()
|
|
try:
|
|
got = ensure_boot_snapshot(Conn(), boot_volume_id="v1", cfg=_Cfg(), log=logs.append)
|
|
assert got is not None
|
|
assert any("уже есть" in m for m in logs)
|
|
finally:
|
|
snap_mod.find_snapshot_by_name = orig
|
|
|
|
|
|
def test_ensure_boot_snapshot_creates(monkeypatch):
|
|
created = {}
|
|
|
|
class Snap:
|
|
def __init__(self):
|
|
self.id = "new"
|
|
self.status = "creating"
|
|
|
|
class Conn:
|
|
class block_storage:
|
|
@staticmethod
|
|
def create_snapshot(**kwargs):
|
|
created.update(kwargs)
|
|
return Snap()
|
|
|
|
@staticmethod
|
|
def get_snapshot(sid):
|
|
s = Snap()
|
|
s.status = "available"
|
|
return s
|
|
|
|
import gpu_rent.snapshot as snap_mod
|
|
|
|
monkeypatch.setattr(snap_mod, "find_snapshot_by_name", lambda conn, name: None)
|
|
monkeypatch.setattr(snap_mod.time, "sleep", lambda s: None)
|
|
logs = []
|
|
got = ensure_boot_snapshot(Conn(), boot_volume_id="vol-1", cfg=_Cfg(), log=logs.append)
|
|
assert created["volume_id"] == "vol-1"
|
|
assert created["force"] is True
|
|
assert created["name"] == "gpu-rent-boot-ok"
|
|
assert got.status == "available"
|
|
|
|
|
|
def test_mcp_snippet(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"gpu_rent.access_card.load_state",
|
|
lambda: type("S", (), {"notes": {}})(),
|
|
)
|
|
logs = []
|
|
print_mcp_snippet(_Cfg(), logs.append)
|
|
text = "\n".join(logs)
|
|
assert "17801/mcp" in text
|
|
assert "mcp.json" in text
|
|
assert "SwarmUI UI" in text
|
|
assert "gpu-rent stop" in text
|