152 lines
4.7 KiB
Python
152 lines
4.7 KiB
Python
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())
|