test_resource_manager.py

import random
import string

from gso.services import resource_manager


def _random_string(n=None, letters=string.ascii_letters + string.digits + string.punctuation):
    # ignoring S311 because this is a test
    if not n:
        n = random.randint(1, 20)  # noqa: S311
    return "".join(random.choices(letters, k=n))  # noqa: S311


def test_new_router():
    router_name = _random_string(10)
    assert router_name not in resource_manager._DUMMY_INVENTORY
    resource_manager.import_new_router(new_router_fqdn=router_name, subscription_id=_random_string(10))
    assert router_name in resource_manager._DUMMY_INVENTORY


def test_new_lag():
    router_name = list(resource_manager._DUMMY_INVENTORY.keys())[0]
    new_lags = {
        resource_manager.next_lag(router_fqdn=router_name, subscription_id=_random_string(10)) for _ in range(10)
    }
    assert len(new_lags) == 10
    assert new_lags <= set(resource_manager._DUMMY_INVENTORY[router_name]["lags"])


def test_physical_allocation_lifecycle_happy():
    router_name = list(resource_manager._DUMMY_INVENTORY.keys())[0]

    def _interfaces():
        return resource_manager.available_physical_interfaces(router_fqdn=router_name)

    initial_available = _interfaces()

    interface_name = initial_available[0]

    resource_manager.reserve_physical_interface(
        router_fqdn=router_name, subscription_id=_random_string(10), interface_name=interface_name
    )

    current_available = _interfaces()
    assert interface_name not in current_available

    resource_manager.allocate_physical_interface(router_fqdn=router_name, interface_name=interface_name)
    resource_manager.free_physical_interface(router_fqdn=router_name, interface_name=interface_name)

    current_available = _interfaces()
    assert interface_name in current_available