run ruff

gso/api/v1/__init__.py

@@ -3,11 +3,9 @@
 from fastapi import APIRouter
 
 from gso.api.v1.network import router as network_router
-from gso.api.v1.processes import router as processes_router
 from gso.api.v1.subscriptions import router as subscriptions_router
 
 router = APIRouter()
 
 router.include_router(subscriptions_router)
-router.include_router(processes_router)
 router.include_router(network_router)

gso/auth/oidc.py

@@ -63,6 +63,8 @@ class OIDCAuthentication(OIDCAuth):
         :return: OIDCUserModel: OIDC user model from openid server
 
         """
+        assert self.openid_config is not None, "OpenID config is not loaded"  # noqa: S101
+
         intercepted_token = await self.introspect_token(async_request, token)
         client_id = intercepted_token.get("client_id")
         if _is_client_credentials_token(intercepted_token):
@@ -106,6 +108,8 @@ class OIDCAuthentication(OIDCAuth):
         :param token: the access_token
         :return: dict from openid server
         """
+        assert self.openid_config is not None, "OpenID config is not loaded"  # noqa: S101
+
         endpoint = self.openid_config.introspect_endpoint or self.openid_config.introspection_endpoint or ""
         response = await async_request.post(
             endpoint,

gso/workflows/__init__.py

@@ -10,19 +10,17 @@ ALL_ALIVE_STATES: list[str] = [
     SubscriptionLifecycle.ACTIVE,
 ]
 
-WF_USABLE_MAP.update(
-    {
-        "redeploy_base_config": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
-        "update_ibgp_mesh": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
-        "activate_router": [SubscriptionLifecycle.PROVISIONING],
-        "deploy_twamp": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
-        "modify_trunk_interface": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
-        "activate_iptrunk": [SubscriptionLifecycle.PROVISIONING],
-        "terminate_site": ALL_ALIVE_STATES,
-        "terminate_router": ALL_ALIVE_STATES,
-        "terminate_iptrunk": ALL_ALIVE_STATES,
-    }
-)
+WF_USABLE_MAP.update({
+    "redeploy_base_config": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+    "update_ibgp_mesh": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+    "activate_router": [SubscriptionLifecycle.PROVISIONING],
+    "deploy_twamp": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+    "modify_trunk_interface": [SubscriptionLifecycle.PROVISIONING, SubscriptionLifecycle.ACTIVE],
+    "activate_iptrunk": [SubscriptionLifecycle.PROVISIONING],
+    "terminate_site": ALL_ALIVE_STATES,
+    "terminate_router": ALL_ALIVE_STATES,
+    "terminate_iptrunk": ALL_ALIVE_STATES,
+})
 
 #  IP trunk workflows
 LazyWorkflowInstance("gso.workflows.iptrunk.activate_iptrunk", "activate_iptrunk")

gso/workflows/iptrunk/create_imported_iptrunk.py

@@ -123,12 +123,10 @@ def initialize_subscription(
         subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members.append(
             IptrunkInterfaceBlockInactive.new(subscription_id=uuid4(), **member),
         )
-    side_names = sorted(
-        [
-            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
-            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
-        ]
-    )
+    side_names = sorted([
+        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
+        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
+    ])
     subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, geant_s_sid:{geant_s_sid}"
     return {"subscription": subscription}
 

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -229,12 +229,10 @@ def modify_iptrunk_subscription(
             IptrunkInterfaceBlock.new(subscription_id=uuid4(), **member),
         )
 
-    side_names = sorted(
-        [
-            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
-            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
-        ]
-    )
+    side_names = sorted([
+        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
+        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
+    ])
     subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, geant_s_sid:{geant_s_sid}"
 
     return {

test/auth/test_oidc.py

@@ -135,13 +135,13 @@ async def test_userinfo_success(oidc_user, mock_async_client):
     mock_response_introspect_token = {
         "active": True,
         "scope": "openid profile email aarc",
-        "client_id": "APP-775F0BD8-B1D7-4936-BE2C-A300A6509F0B",
+        "client_id": "APP-775F0BD8-B1D7-4936-BE2C-A300A6509F00",
         "exp": 1721395275,
         "iat": 1721391675,
         "sub": "ed145263-b652-3d4x-8f96-4abae9c98124@aai.geant.org",
         "iss": "https://proxy.aai.geant.org",
         "token_type": "Bearer",
-        "aud": ["APP-775F0BD8-B1D7-4936-BE2C-A300A6509F0B"],
+        "aud": ["APP-775F0BD8-B1D7-4936-BE2C-A300A6509F00"],
     }
     mock_response_userinfo = {"sub": "1234", "name": "John Doe", "email": "johndoe@example.com"}
 

test/workflows/iptrunk/test_create_iptrunk.py

@@ -137,12 +137,10 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
     subscription_id = state["subscription_id"]
     subscription = Iptrunk.from_subscription(subscription_id)
 
-    sorted_sides = sorted(
-        [
-            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
-            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
-        ]
-    )
+    sorted_sides = sorted([
+        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
+        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
+    ])
     assert subscription.status == "provisioning"
     assert subscription.description == (
         f"IP trunk {sorted_sides[0]} {sorted_sides[1]}, geant_s_sid:{input_form_wizard_data[0]["geant_s_sid"]}"

test/workflows/iptrunk/test_modify_trunk_interface.py

@@ -158,12 +158,10 @@ def test_iptrunk_modify_trunk_interface_success(
     assert mocked_detach_interfaces_from_lag.call_count == num_lag_ifaces  # 1 time per nokia side
 
     # Assert all subscription properties have been updated correctly
-    side_names = sorted(
-        [
-            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
-            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
-        ]
-    )
+    side_names = sorted([
+        subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_site.site_name,
+        subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
+    ])
     assert subscription.description == f"IP trunk {side_names[0]} {side_names[1]}, geant_s_sid:{new_sid}"
     assert subscription.iptrunk.geant_s_sid == input_form_iptrunk_data[1]["geant_s_sid"]
     assert subscription.iptrunk.iptrunk_description == input_form_iptrunk_data[1]["iptrunk_description"]