Skip to content
Snippets Groups Projects
Commit 38a42444 authored by Erik Reid's avatar Erik Reid
Browse files

follow the various pipeline style checks

parent 1672f815
Branches
Tags
No related merge requests found
Pipeline #83526 passed
......@@ -18,25 +18,20 @@ class InterfaceAllocationState(Enum):
def _dummy_router_interfaces():
return {
'lags': [],
'physical': [
{
'name': f'ifc-{x}',
'state': InterfaceAllocationState.AVAILABLE
} for x in range(250)]
"lags": [],
"physical": [{"name": f"ifc-{x}", "state": InterfaceAllocationState.AVAILABLE} for x in range(250)],
}
_DUMMY_INVENTORY = {
'fqdn-a': _dummy_router_interfaces(),
'fqdn-b': _dummy_router_interfaces(),
'fqdn-c': _dummy_router_interfaces(),
'fqdn-d': _dummy_router_interfaces()
"fqdn-a": _dummy_router_interfaces(),
"fqdn-b": _dummy_router_interfaces(),
"fqdn-c": _dummy_router_interfaces(),
"fqdn-d": _dummy_router_interfaces(),
}
def import_new_router(
new_router_fqdn: str, oss_params=settings.OSSParams):
def import_new_router(new_router_fqdn: str, oss_params=settings.OSSParams):
# TODO: this is a dummy implementation
# TODO: specifiy if this should be an error (and if now, what it means)
......@@ -51,90 +46,79 @@ def next_lag(router_fqdn: str, oss_params=settings.OSSParams) -> str:
lag_idx = 0
while True:
lag_name = f'ae-{lag_idx}'
if lag_name not in _DUMMY_INVENTORY[router_fqdn]['lags']:
_DUMMY_INVENTORY[router_fqdn]['lags'].append(lag_name)
lag_name = f"ae-{lag_idx}"
if lag_name not in _DUMMY_INVENTORY[router_fqdn]["lags"]:
_DUMMY_INVENTORY[router_fqdn]["lags"].append(lag_name)
return lag_name
lag_idx += 1
def available_physical_interfaces(
router_fqdn: str, oss_params=settings.OSSParams) -> List[str]:
def available_physical_interfaces(router_fqdn: str, oss_params=settings.OSSParams) -> List[str]:
# TODO: this is a dummy implementation
assert router_fqdn in _DUMMY_INVENTORY
return [
ifc['name'] for ifc in _DUMMY_INVENTORY[router_fqdn]['physical']
if ifc['state'] == InterfaceAllocationState.AVAILABLE
ifc["name"]
for ifc in _DUMMY_INVENTORY[router_fqdn]["physical"]
if ifc["state"] == InterfaceAllocationState.AVAILABLE
]
def _find_physical(router_fqdn: str, interface_name: str) -> dict:
assert router_fqdn in _DUMMY_INVENTORY
for ifc in _DUMMY_INVENTORY[router_fqdn]['physical']:
if ifc['name'] == interface_name:
for ifc in _DUMMY_INVENTORY[router_fqdn]["physical"]:
if ifc["name"] == interface_name:
return ifc
assert False, f'interface {interface_name} not found on {router_fqdn}'
raise AssertionError(f"interface {interface_name} not found on {router_fqdn}")
def reserve_physical_interface(
router_fqdn: str,
interface_name: str,
oss_params=settings.OSSParams):
def reserve_physical_interface(router_fqdn: str, interface_name: str, oss_params=settings.OSSParams):
# TODO: this is a dummy implementation
ifc = _find_physical(router_fqdn, interface_name)
assert ifc['state'] == InterfaceAllocationState.AVAILABLE, \
f'interface {router_fqdn}:{interface_name} is not available'
ifc['state'] = InterfaceAllocationState.RESERVED
assert (
ifc["state"] == InterfaceAllocationState.AVAILABLE
), f"interface {router_fqdn}:{interface_name} is not available"
ifc["state"] = InterfaceAllocationState.RESERVED
def allocate_physical_interface(
router_fqdn: str,
interface_name: str,
oss_params=settings.OSSParams):
def allocate_physical_interface(router_fqdn: str, interface_name: str, oss_params=settings.OSSParams):
# TODO: this is a dummy implementation
ifc = _find_physical(router_fqdn, interface_name)
# TODO: is there a use case for moving
# directly from AVAILABLE to ALLOCATED?
assert ifc['state'] == InterfaceAllocationState.RESERVED, \
f'interface {router_fqdn}:{interface_name} is not reserved'
ifc['state'] = InterfaceAllocationState.RESERVED
assert (
ifc["state"] == InterfaceAllocationState.RESERVED
), f"interface {router_fqdn}:{interface_name} is not reserved"
ifc["state"] = InterfaceAllocationState.RESERVED
def free_physical_interface(
router_fqdn: str,
interface_name: str,
oss_params=settings.OSSParams):
def free_physical_interface(router_fqdn: str, interface_name: str, oss_params=settings.OSSParams):
# TODO: this is a dummy implementation
ifc = _find_physical(router_fqdn, interface_name)
# TODO: is this really an error that should be handled?
# ... or is it ok to ignore this?
assert ifc['state'] != InterfaceAllocationState.AVAILABLE, \
f'interface {router_fqdn}:{interface_name} is already available'
ifc['state'] = InterfaceAllocationState.AVAILABLE
assert (
ifc["state"] != InterfaceAllocationState.AVAILABLE
), f"interface {router_fqdn}:{interface_name} is already available"
ifc["state"] = InterfaceAllocationState.AVAILABLE
def free_lag(
router_fqdn: str,
lag_name: str,
oss_params=settings.OSSParams):
def free_lag(router_fqdn: str, lag_name: str, oss_params=settings.OSSParams):
# TODO: is this a use case that should be handled?
# e.g. who keeps track of the bundled physical interfaces?
pass
def remove_router(
router_fqdn: str, oss_params=settings.OSSParams):
def remove_router(router_fqdn: str, oss_params=settings.OSSParams):
# TODO: is this a use case that should be handled?
pass
def all_lags(
router_fqdn: str, oss_params=settings.OSSParams) -> List[str]:
def all_lags(router_fqdn: str, oss_params=settings.OSSParams) -> List[str]:
# TODO: is this a use case that should be handled?
assert router_fqdn in _DUMMY_INVENTORY
return _DUMMY_INVENTORY[router_fqdn]['lags']
return _DUMMY_INVENTORY[router_fqdn]["lags"]
"""GSO settings, ensuring that the required parameters are set correctly."""
"""GSO settings.
Ensuring that the required parameters are set correctly.
"""
import ipaddress
import json
import logging
......@@ -76,7 +80,8 @@ class ProvisioningProxyParams(BaseSettings):
class ResourceManagementParams(BaseSettings):
"""TO DO: resource management parameters"""
"""TO DO: resource management parameters."""
todo: str
......
......@@ -14,9 +14,7 @@ def configuration_data() -> dict:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
yield {
"GENERAL": {"public_hostname": "https://gap.geant.org"},
"RESOURCE_MANAGEMENT": {
'todo': 'todo'
},
"RESOURCE_MANAGEMENT": {"todo": "todo"},
"IPAM": {
"INFOBLOX": {
"scheme": "https",
......
......@@ -4,12 +4,11 @@ import string
from gso.services import resource_manager
def _random_string(
n=None,
letters=string.ascii_letters + string.digits + string.punctuation):
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)
return ''.join(random.choices(letters, k=n))
n = random.randint(1, 20) # noqa: S311
return "".join(random.choices(letters, k=n)) # noqa: S311
def test_new_router():
......@@ -21,36 +20,28 @@ def test_new_router():
def test_new_lag():
router_name = list(resource_manager._DUMMY_INVENTORY.keys())[0]
new_lags = {
resource_manager.next_lag(router_fqdn=router_name)
for _ in range(10)
}
new_lags = {resource_manager.next_lag(router_fqdn=router_name) for _ in range(10)}
assert len(new_lags) == 10
assert new_lags <= set(
resource_manager._DUMMY_INVENTORY[router_name]['lags'])
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)
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, interface_name=interface_name)
resource_manager.reserve_physical_interface(router_fqdn=router_name, 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)
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
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment