Leonid Pershin
a563ae06c4
- Updated version number in pyproject.toml and __init__.py to 0.2.0. - Revised README.md to reflect the current state of the project, including usage instructions and setup steps. - Improved CLI documentation in cli.md, adding details about new commands and their functionalities. - Enhanced the quick start section in README.md for better clarity on initial setup. - Updated local folder documentation to clarify file handling and commands. - Added a new command for listing GPU flavors and improved error handling in the CLI. - Implemented a watchdog feature in the tunnel to manage server states effectively.
76 lines
2.2 KiB
Python
76 lines
2.2 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
|
|
|
|
|
|
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(capsys):
|
|
logs = []
|
|
print_mcp_snippet(_Cfg(), logs.append)
|
|
text = "\n".join(logs)
|
|
assert "17801/mcp" in text
|
|
assert "gpu-rent tunnel" in text
|
|
assert "mcp.json" in text
|