diff --git a/gso/services/resource_manager.py b/gso/services/resource_manager.py
index 3dadd19c08365d6b82ae6b91e45b483e775965a5..abd359ca4c8de9a6ca62e07c3c0f260d88c46419 100644
--- a/gso/services/resource_manager.py
+++ b/gso/services/resource_manager.py
@@ -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"]
diff --git a/gso/settings.py b/gso/settings.py
index 2d527ff1eddb19ef95ba1068d2deda6b604e9c4f..a576616c0b5003f27910afe983a29b4736eac253 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -1,4 +1,8 @@
-"""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
diff --git a/test/conftest.py b/test/conftest.py
index 1ef473a1c3cbcadb18a98932945ca3bb6c6b4ede..75ce00bc67ee6cee82d4b1ba305b7855daf9adf1 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -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",
diff --git a/test/test_resource_manager.py b/test/test_resource_manager.py
index b99d547f81084544630ffb766b886d2570fb80d3..5258f4b80e7f12f85d0d194a7c2df3cbf40e2cc5 100644
--- a/test/test_resource_manager.py
+++ b/test/test_resource_manager.py
@@ -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