103 lines
3.5 KiB
Python
103 lines
3.5 KiB
Python
import json
|
|
from unittest.mock import patch
|
|
|
|
import pytest
|
|
|
|
from cloud import VM, VMCreateRequest, VMManager, VMStateRequest
|
|
|
|
|
|
class Response:
|
|
def __init__(self, payload: dict | None = None) -> None:
|
|
self.payload = payload
|
|
|
|
def read(self) -> bytes:
|
|
return b"" if self.payload is None else json.dumps(self.payload).encode()
|
|
|
|
def __enter__(self) -> "Response":
|
|
return self
|
|
|
|
def __exit__(self, *args: object) -> None:
|
|
return None
|
|
|
|
|
|
@patch("cloud.vm_manager.urlopen")
|
|
def test_list_vms_encodes_repeated_filters(urlopen) -> None:
|
|
urlopen.return_value = Response({"items": []})
|
|
manager = VMManager("token", base_url="https://api.example")
|
|
|
|
response = manager.list_vms("project-id", vm_ids=["vm-1", "vm-2"], statuses=["running"])
|
|
assert response.items == []
|
|
|
|
request = urlopen.call_args.args[0]
|
|
assert request.full_url == (
|
|
"https://api.example/api/v1/vms?project_id=project-id&vm_ids=vm-1&vm_ids=vm-2&statuses=running"
|
|
)
|
|
assert request.get_header("Authorization") == "Bearer token"
|
|
|
|
|
|
@patch("cloud.vm_manager.urlopen")
|
|
def test_create_vm_sends_documented_payload(urlopen) -> None:
|
|
urlopen.return_value = Response([{"id": "vm-id", "name": "game-vm"}])
|
|
manager = VMManager("token", base_url="https://api.example")
|
|
payload = VMCreateRequest(
|
|
project_id="project-id",
|
|
name="game-vm",
|
|
disks=[{"disk_id": "disk-id"}],
|
|
)
|
|
|
|
assert manager.create_vm(payload)[0].id == "vm-id"
|
|
|
|
request = urlopen.call_args.args[0]
|
|
assert request.method == "POST"
|
|
assert request.full_url == "https://api.example/api/v1/vms"
|
|
assert json.loads(request.data) == [payload.model_dump(exclude_none=True)]
|
|
|
|
|
|
@patch("cloud.vm_manager.urlopen")
|
|
def test_create_vms_uses_v1_batch_endpoint(urlopen) -> None:
|
|
urlopen.return_value = Response([{"id": "vm-id", "name": "game-vm"}])
|
|
manager = VMManager("token", base_url="https://api.example")
|
|
payload = VMCreateRequest(
|
|
project_id="project-id",
|
|
name="game-vm",
|
|
disks=[{"disk_id": "disk-id"}],
|
|
subnets=[{"subnet_id": "subnet-id", "new_floating_ip": True}],
|
|
)
|
|
|
|
assert manager.create_vms([payload])[0].id == "vm-id"
|
|
|
|
request = urlopen.call_args.args[0]
|
|
assert request.full_url == "https://api.example/api/v1/vms"
|
|
assert json.loads(request.data) == [payload.model_dump(exclude_none=True)]
|
|
|
|
|
|
@patch("cloud.vm_manager.urlopen")
|
|
def test_start_vm_uses_bulk_state_endpoint(urlopen) -> None:
|
|
urlopen.return_value = Response()
|
|
manager = VMManager("token", base_url="https://api.example")
|
|
|
|
manager.start_vm("vm-id")
|
|
|
|
request = urlopen.call_args.args[0]
|
|
assert request.method == "PUT"
|
|
assert request.full_url == "https://api.example/api/v1/vms"
|
|
assert json.loads(request.data) == [{"id": "vm-id", "state": "running"}]
|
|
|
|
|
|
@patch("cloud.vm_manager.time.sleep")
|
|
def test_wait_for_vms_ready_polls_until_running(sleep) -> None:
|
|
manager = VMManager("token")
|
|
initial = VM(id="vm-id", name="game-vm", state="creating")
|
|
states = iter(["creating", "running"])
|
|
|
|
with patch.object(manager, "get_vm", side_effect=lambda _: initial.model_copy(update={"state": next(states)})):
|
|
ready = manager.wait_for_vms_ready([initial], timeout=1, poll_interval=0.01)
|
|
|
|
assert ready[0].state == "running"
|
|
sleep.assert_called_once_with(0.01)
|
|
|
|
|
|
def test_set_state_rejects_unknown_state() -> None:
|
|
with pytest.raises(ValueError, match="Input should be 'running' or 'stopped'"):
|
|
VMStateRequest(id="vm-id", state="rebooting")
|