Scan ru-6 by Nova availability zones (a/b/c). First boot uses FLAVOR_SIZE_PRESET=cheap; reruns stay on the disk segment. Co-authored-by: Cursor <cursoragent@cursor.com>
59 lines
2.2 KiB
Python
59 lines
2.2 KiB
Python
from gpu_rent.config import load_config
|
|
from gpu_rent.flavor_presets import format_char_impact, normalize_size_preset
|
|
from gpu_rent.placement import apply_az_policy, existing_disk_az
|
|
from gpu_rent.pools import PoolGpuOffer
|
|
from gpu_rent.state import SessionState
|
|
|
|
|
|
def test_normalize_size_preset():
|
|
assert normalize_size_preset("RAM") == "ram"
|
|
assert normalize_size_preset("nope") == "cheap"
|
|
assert "GPU" in "\n".join(format_char_impact())
|
|
|
|
|
|
def test_apply_az_policy_disk_wins(monkeypatch):
|
|
monkeypatch.setenv("OS_AUTH_URL", "https://example.invalid/identity/v3")
|
|
monkeypatch.setenv("OS_USER_DOMAIN_NAME", "1")
|
|
monkeypatch.setenv("OS_USERNAME", "u")
|
|
monkeypatch.setenv("OS_PASSWORD", "p")
|
|
monkeypatch.setenv("OS_PROJECT_ID", "proj")
|
|
monkeypatch.setenv("OS_REGION_NAME", "ru-6")
|
|
monkeypatch.setenv("GPU_RENT_AZ", "ru-6b")
|
|
cfg = load_config(require_auth=True)
|
|
best = PoolGpuOffer(region="ru-6", multizone=True, recommended_az="ru-6a")
|
|
placed = apply_az_policy(cfg, disk_az="ru-6a", best=best)
|
|
assert placed.reason == "disk"
|
|
assert placed.cfg.gpu_rent_az == "ru-6a"
|
|
|
|
|
|
def test_apply_az_policy_first_boot_cheap(monkeypatch):
|
|
monkeypatch.setenv("OS_AUTH_URL", "https://example.invalid/identity/v3")
|
|
monkeypatch.setenv("OS_USER_DOMAIN_NAME", "1")
|
|
monkeypatch.setenv("OS_USERNAME", "u")
|
|
monkeypatch.setenv("OS_PASSWORD", "p")
|
|
monkeypatch.setenv("OS_PROJECT_ID", "proj")
|
|
monkeypatch.setenv("OS_REGION_NAME", "ru-6")
|
|
monkeypatch.setenv("GPU_RENT_AZ", "ru-6b")
|
|
cfg = load_config(require_auth=True)
|
|
best = PoolGpuOffer(region="ru-6", multizone=True, recommended_az="ru-6a")
|
|
placed = apply_az_policy(cfg, disk_az=None, best=best)
|
|
assert placed.reason == "cheap"
|
|
assert placed.cfg.gpu_rent_az == "ru-6a"
|
|
|
|
|
|
def test_existing_disk_az_from_volume():
|
|
class Vol:
|
|
availability_zone = "ru-6a"
|
|
|
|
class Storage:
|
|
def get_volume(self, vid):
|
|
assert vid == "boot-1"
|
|
return Vol()
|
|
|
|
class Conn:
|
|
block_storage = Storage()
|
|
|
|
cfg = load_config(require_auth=False)
|
|
state = SessionState(boot_volume_id="boot-1")
|
|
assert existing_disk_az(Conn(), cfg, state) == "ru-6a"
|