little update

tests/test_environment.py

@@ -0,0 +1,23 @@
+import os
+from pathlib import Path
+
+from dotenv import load_dotenv
+
+
+def test_dotenv_loads_missing_environment_value(tmp_path: Path, monkeypatch) -> None:
+    env_file = tmp_path / ".env"
+    env_file.write_text("CLOUD_PROJECT_ID=project-from-file\n")
+    monkeypatch.delenv("CLOUD_PROJECT_ID", raising=False)
+
+    assert load_dotenv(env_file) is True
+    assert os.environ["CLOUD_PROJECT_ID"] == "project-from-file"
+
+
+def test_dotenv_does_not_override_shell_environment(tmp_path: Path, monkeypatch) -> None:
+    env_file = tmp_path / ".env"
+    env_file.write_text("CLOUD_PROJECT_ID=project-from-file\n")
+    monkeypatch.setenv("CLOUD_PROJECT_ID", "project-from-shell")
+
+    load_dotenv(env_file)
+
+    assert os.environ["CLOUD_PROJECT_ID"] == "project-from-shell"

tests/test_iam.py

@@ -0,0 +1,26 @@
+import json
+from unittest.mock import patch
+
+from cloud.iam import IAMClient
+
+
+class Response:
+    def read(self) -> bytes:
+        return json.dumps({"access_token": "fresh-token"}).encode()
+
+    def __enter__(self) -> "Response":
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        return None
+
+
+@patch("cloud.iam.urlopen")
+def test_fetches_iam_token(urlopen) -> None:
+    urlopen.return_value = Response()
+
+    assert IAMClient("key-id", "secret").get_access_token() == "fresh-token"
+
+    request = urlopen.call_args.args[0]
+    assert request.full_url == "https://iam.api.cloud.ru/api/v1/auth/token"
+    assert json.loads(request.data) == {"keyId": "key-id", "secret": "secret"}

tests/test_kernel_consumer.py

@@ -0,0 +1,46 @@
+from uuid import uuid4
+
+from kernel_consumer.main import Settings, _image_metadata, create_game_vms
+
+
+class Cloud:
+    def __init__(self) -> None:
+        self.requests = []
+
+    def create_vms(self, requests):
+        self.requests.extend(requests)
+        return []
+
+    def wait_for_vms_ready(self, vms):
+        return vms
+
+
+def test_creates_one_vm_per_team_plus_public_extra() -> None:
+    cloud = Cloud()
+    settings = Settings(
+        cloud_project_id="project-id",
+        cloud_availability_zone_id="zone-id",
+        cloud_availability_zone_name=None,
+        cloud_vm_password="Password123",
+    )
+    network = type("Network", (), {"cloud_subnet_id": "subnet-id"})()
+    request = {"team_count": 2}
+
+    create_game_vms(request, uuid4(), network, cloud, settings)
+
+    assert len(cloud.requests) == 3
+    assert [item.subnets[0]["new_floating_ip"] for item in cloud.requests] == [False, False, True]
+    assert all(item.subnets[0]["subnet_id"] == "subnet-id" for item in cloud.requests)
+    assert all(item.availability_zone_id == "zone-id" for item in cloud.requests)
+    assert all(item.availability_zone_name is None for item in cloud.requests)
+
+
+def test_rejects_password_without_uppercase_latin_letter() -> None:
+    settings = Settings(cloud_vm_password="lowercase123")
+
+    try:
+        _image_metadata(settings, "game-host")
+    except ValueError as error:
+        assert str(error) == "CLOUD_VM_PASSWORD must contain at least one uppercase Latin letter"
+    else:
+        raise AssertionError("expected invalid password to be rejected")

tests/test_networking_manager.py

@@ -0,0 +1,151 @@
+from uuid import uuid4
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cloud.models import AvailabilityZone, Subnet
+from cloud.vm_manager import VMAPIError
+from database.models import Base, NetworkPoolEntry
+from networking import NetworkingManager, NetworkPoolExhaustedError
+
+
+class Cloud:
+    def __init__(self) -> None:
+        self.subnets: dict[str, Subnet] = {}
+        self.created: list[str] = []
+        self.requests = []
+        self.zones: list[AvailabilityZone] = []
+        self.subnet_states: list[str] = []
+
+    def get_subnet(self, subnet_id: str) -> Subnet:
+        if subnet_id not in self.subnets:
+            raise VMAPIError(404, "not found")
+        subnet = self.subnets[subnet_id]
+        if self.subnet_states:
+            subnet.state = self.subnet_states.pop(0)
+        return subnet
+
+    def list_availability_zones(self) -> list[AvailabilityZone]:
+        return self.zones
+
+    def create_subnet(self, request):
+        self.requests.append(request)
+        subnet = Subnet(
+            id=f"subnet-{len(self.subnets) + 1}",
+            name=request.name,
+            subnet_address=request.subnet_address,
+        )
+        self.subnets[subnet.id] = subnet
+        self.created.append(subnet.subnet_address)
+        return subnet
+
+
+@pytest.fixture
+def session_factory():
+    engine = create_engine("sqlite://")
+    Base.metadata.create_all(engine)
+    return sessionmaker(engine, expire_on_commit=False)
+
+
+def test_allocates_new_network_and_reuses_after_release(session_factory) -> None:
+    cloud = Cloud()
+    manager = NetworkingManager(session_factory, cloud, "project-id", availability_zone_id="zone-id")
+    first_game = uuid4()
+
+    first = manager.getNetworkFromPool(first_game)
+    assert first.cidr == "10.0.0.0/20"
+    assert first.game_id == first_game
+    assert cloud.created == ["10.0.0.0/20"]
+
+    manager.releaseNetworkToPool(first_game)
+    reused = manager.getNetworkFromPool(uuid4())
+    assert reused.cidr == first.cidr
+    assert len(cloud.created) == 1
+
+
+def test_waits_for_new_subnet_to_be_ready(session_factory) -> None:
+    cloud = Cloud()
+    cloud.subnet_states = ["creating", "created"]
+
+    entry = NetworkingManager(
+        session_factory,
+        cloud,
+        "project-id",
+        availability_zone_id="zone-id",
+        poll_interval=0.001,
+    ).getNetworkFromPool(uuid4())
+
+    assert entry.cidr == "10.0.0.0/20"
+
+
+def test_discards_missing_cloud_network_and_creates_replacement(session_factory) -> None:
+    cloud = Cloud()
+    with session_factory.begin() as session:
+        session.add(NetworkPoolEntry(cidr="10.0.0.0/20", cloud_subnet_id="gone"))
+
+    entry = NetworkingManager(session_factory, cloud, "project-id", availability_zone_id="zone-id").getNetworkFromPool(uuid4())
+
+    assert entry.cidr == "10.0.0.0/20"
+    assert entry.cloud_subnet_id == "subnet-1"
+
+
+def test_passes_vpc_and_availability_zone_to_subnet_create(session_factory) -> None:
+    cloud = Cloud()
+
+    NetworkingManager(
+        session_factory,
+        cloud,
+        "project-id",
+        vpc_id="vpc-id",
+        availability_zone_id="zone-id",
+    ).getNetworkFromPool(uuid4())
+
+    request = cloud.requests[0]
+    assert request.subnet_address == "10.0.0.0/20"
+    assert request.vpc_id == "vpc-id"
+    assert request.availability_zone_id == "zone-id"
+    assert request.model_dump(exclude_none=True)["availability_zone_id"] == "zone-id"
+
+
+def test_resolves_availability_zone_name_to_id(session_factory) -> None:
+    cloud = Cloud()
+    cloud.zones.append(AvailabilityZone(id="zone-uuid", name="ru.AZ-2"))
+
+    NetworkingManager(
+        session_factory,
+        cloud,
+        "project-id",
+        availability_zone_name="ru.AZ-2",
+    ).getNetworkFromPool(uuid4())
+
+    assert cloud.requests[0].availability_zone_id == "zone-uuid"
+
+
+def test_raises_for_unknown_availability_zone_name(session_factory) -> None:
+    cloud = Cloud()
+    cloud.zones.append(AvailabilityZone(id="zone-uuid", name="ru.AZ-2"))
+
+    with pytest.raises(LookupError, match="ru.AZ-9"):
+        NetworkingManager(
+            session_factory,
+            cloud,
+            "project-id",
+            availability_zone_name="ru.AZ-9",
+        ).getNetworkFromPool(uuid4())
+
+
+def test_raises_when_all_pool_cidrs_are_used(session_factory) -> None:
+    cloud = Cloud()
+    with session_factory.begin() as session:
+        for octet in range(0, 256, 16):
+            session.add(
+                NetworkPoolEntry(
+                    cidr=f"10.{octet}.0.0/20",
+                    cloud_subnet_id=str(octet),
+                    game_id=uuid4(),
+                )
+            )
+
+    with pytest.raises(NetworkPoolExhaustedError, match="network pool exhausted"):
+        NetworkingManager(session_factory, cloud, "project-id", availability_zone_id="zone-id").getNetworkFromPool(uuid4())

tests/test_vm_api_errors.py

@@ -0,0 +1,10 @@
+from cloud.vm_manager import VMManager
+
+
+def test_error_response_keeps_non_json_body() -> None:
+    payload = VMManager._decode_error_response(b"<html>upstream error</html>")
+
+    assert payload == {"raw_response": "<html>upstream error</html>"}
+    assert VMManager._error_message(payload, "Bad Gateway") == (
+        "Bad Gateway: <html>upstream error</html>"
+    )

tests/test_vm_manager.py

@@ -0,0 +1,102 @@
+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")

tests/test_vpc_network_manager.py

@@ -0,0 +1,49 @@
+import json
+from unittest.mock import patch
+
+import pytest
+
+from cloud import CreateVPCRequest, VPCAPIError, VPCNetworkManager
+
+
+class Response:
+    def __init__(self, payload: dict) -> None:
+        self.payload = payload
+
+    def read(self) -> bytes:
+        return json.dumps(self.payload).encode()
+
+    def __enter__(self) -> "Response":
+        return self
+
+    def __exit__(self, *args: object) -> None:
+        return None
+
+
+@patch("cloud.vpc_network_manager.urlopen")
+def test_create_vpc_sends_bearer_request(urlopen) -> None:
+    urlopen.return_value = Response({"id": "operation-id"})
+    manager = VPCNetworkManager("token", base_url="https://api.example")
+
+    operation = manager.create_vpc(
+        CreateVPCRequest(projectId="project-id", name="game-vpc")
+    )
+    assert operation.id == "operation-id"
+
+    request = urlopen.call_args.args[0]
+    assert request.full_url == "https://api.example/v1/vpcs"
+    assert request.method == "POST"
+    assert request.get_header("Authorization") == "Bearer token"
+    assert json.loads(request.data) == {"projectId": "project-id", "name": "game-vpc"}
+
+
+def test_manager_rejects_empty_token() -> None:
+    with pytest.raises(ValueError, match="token must not be empty"):
+        VPCNetworkManager("")
+
+
+def test_api_error_keeps_status_and_details() -> None:
+    error = VPCAPIError(403, "forbidden", [{"reason": "denied"}])
+
+    assert error.status == 403
+    assert error.details == [{"reason": "denied"}]