fix import workflows tests

gso/api/v1/subscriptions.py

 """:term:`API` endpoint for fetching different types of subscriptions."""
+
 from typing import Any
 
 from fastapi import Depends, Response, status
@@ -19,17 +20,6 @@ router = APIRouter(
 )
 
 
-# class MySubscriptionDomainModelSchema(SubscriptionDomainModelSchema):
-#     model_config = ConfigDict(
-#         extra="allow",
-#         json_encoders={
-#             # datetime: lambda dt: dt.timestamp(),
-#             ipaddress.IPv4Address: lambda v: 1/0,
-#             ipaddress.IPv6Address: lambda v: str(v),
-#         }
-#     )
-
-
 @router.get(
     "/routers",
     status_code=status.HTTP_200_OK,

gso/auth/oidc_policy_helper.py

@@ -245,7 +245,7 @@ class OIDCUser(HTTPBearer):
             return
 
         response = await async_request.get(self.openid_url + "/.well-known/openid-configuration")
-        self.openid_config = OIDCConfig.parse_obj(response.json())
+        self.openid_config = OIDCConfig.model_validate(response.json())
 
     async def userinfo(self, async_request: AsyncClient, token: str) -> OIDCUserModel:
         """Get the userinfo from the openid server.

gso/middlewares.py

@@ -93,9 +93,9 @@ class ModifyProcessEndpointResponse(BaseHTTPMiddleware):
                     if callback_result and isinstance(callback_result, str):
                         callback_result = json.loads(callback_result)
                     if callback_result.get("output") and len(callback_result["output"]) > max_output_length:
-                        callback_result[
-                            "output"
-                        ] = f'{request.base_url}api/v1/processes/steps/{step["step_id"]}/callback-results{token}'
+                        callback_result["output"] = (
+                            f'{request.base_url}api/v1/processes/steps/{step["step_id"]}/callback-results{token}'
+                        )
                     step["state"]["callback_result"] = callback_result
         except (AttributeError, KeyError, TypeError):
             pass

gso/migrations/versions/2024-04-02_1ec810b289c0_add_orchestrator_2_1_2_migrations.py

@@ -5,6 +5,8 @@ Revises:
 Create Date: 2024-04-02 10:21:08.539591
 
 """
+from alembic import op
+from orchestrator.migrations.helpers import create_workflow, delete_workflow
 
 # revision identifiers, used by Alembic.
 revision = '1ec810b289c0'
@@ -13,11 +15,47 @@ branch_labels = None
 # TODO: check it carefuly
 depends_on = '048219045729'  # in this revision, SURF has added a new columns to the workflow table like delted_at, so we need to add a dependency on the revision that added the columns to the workflow table.
 
+new_workflows = [
+    {
+        "name": "import_site",
+        "target": "SYSTEM",
+        "description": "Import a site without provisioning it.",
+        "product_type": "Site"
+    },
+    {
+        "name": "import_router",
+        "target": "SYSTEM",
+        "description": "Import a router without provisioning it.",
+        "product_type": "Router"
+    },
+    {
+        "name": "import_iptrunk",
+        "target": "SYSTEM",
+        "description": "Import an IP trunk without provisioning it.",
+        "product_type": "Iptrunk"
+    },
+    {
+        "name": "import_super_pop_switch",
+        "target": "SYSTEM",
+        "description": "Import a Super PoP switch without provisioning it.",
+        "product_type": "SuperPopSwitch"
+    },
+    {
+        "name": "import_office_router",
+        "target": "SYSTEM",
+        "description": "Import an office router without provisioning it.",
+        "product_type": "OfficeRouter"
+    },
+]
+
 
 def upgrade() -> None:
-    pass
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        create_workflow(conn, workflow)
 
 
 def downgrade() -> None:
-    pass
-
+    conn = op.get_bind()
+    for workflow in new_workflows:
+        delete_workflow(conn, workflow["name"])

gso/monkeypatches.py

@@ -3,6 +3,7 @@
 This adjustment is typically done to extend or modify the functionality of the original
 oauth2_lib package to meet specific requirements of the gso application.
 """
+
 from datetime import datetime
 from ipaddress import IPv4Address, IPv6Address
 

gso/products/product_blocks/office_router.py

 """Product block for :class:`office router` products."""
 
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 

gso/products/product_blocks/site.py

@@ -44,7 +44,7 @@ LatitudeCoordinate: type[float] = Annotated[
         example="40.7128",
         description="A latitude coordinate, modeled as a string. The coordinate must match the format conforming to the latitude range of -90 to +90 degrees. It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 90, -90, 0.",
     ),
-    AfterValidator(validate_latitude)
+    AfterValidator(validate_latitude),
 ]
 LongitudeCoordinate = Annotated[
     float,
@@ -54,7 +54,7 @@ LongitudeCoordinate = Annotated[
         example="74.0060",
         description="A longitude coordinate, modeled as a string. The coordinate must match the format conforming to the longitude range of -180 to +180 degrees. It can be a floating-point number or an integer. Valid examples: 40.7128, -74.0060, 180, -180, 0.",
     ),
-    AfterValidator(validate_longitude)
+    AfterValidator(validate_longitude),
 ]
 
 

gso/products/product_blocks/super_pop_switch.py

 """Product block for :class:`Super PoP Switch` products."""
 
-
 from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle
 

gso/utils/helpers.py

@@ -215,7 +215,6 @@ class BaseSiteValidatorModel(BaseModel):
     site_latitude: LatitudeCoordinate
     site_longitude: LongitudeCoordinate
 
-
     @field_validator("site_ts_address")
     def validate_ts_address(cls, site_ts_address: str) -> str:
         """Validate that a terminal server address is valid."""

gso/utils/shared_enums.py

 """Shared choices for the different models."""
+
 import ipaddress
 from typing import Annotated
 
@@ -31,7 +32,7 @@ FancyIPV4Address = Annotated[
 ]
 
 FancyIPV6Address = Annotated[
-    ipaddress.IPv6Address, PlainSerializer(lambda ip: str(ip),  return_type=str, when_used="always")
+    ipaddress.IPv6Address, PlainSerializer(lambda ip: str(ip), return_type=str, when_used="always")
 ]
 
 
@@ -40,4 +41,3 @@ class ConnectionStrategy(strEnum):
 
     IN_BAND = "IN BAND"
     OUT_OF_BAND = "OUT OF BAND"
-

gso/workflows/__init__.py

@@ -3,17 +3,15 @@
 from orchestrator.services.subscriptions import WF_USABLE_MAP
 from orchestrator.workflows import LazyWorkflowInstance
 
-WF_USABLE_MAP.update(
-    {
-        "cancel_subscription": ["initial"],
-        "redeploy_base_config": ["provisioning", "active"],
-        "update_ibgp_mesh": ["provisioning", "active"],
-        "activate_router": ["provisioning"],
-        "deploy_twamp": ["provisioning", "active"],
-        "modify_trunk_interface": ["provisioning", "active"],
-        "activate_iptrunk": ["provisioning"],
-    }
-)
+WF_USABLE_MAP.update({
+    "cancel_subscription": ["initial"],
+    "redeploy_base_config": ["provisioning", "active"],
+    "update_ibgp_mesh": ["provisioning", "active"],
+    "activate_router": ["provisioning"],
+    "deploy_twamp": ["provisioning", "active"],
+    "modify_trunk_interface": ["provisioning", "active"],
+    "activate_iptrunk": ["provisioning"],
+})
 
 LazyWorkflowInstance("gso.workflows.iptrunk.activate_iptrunk", "activate_iptrunk")
 LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")

gso/workflows/iptrunk/create_iptrunk.py

@@ -84,7 +84,9 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
     user_input_router_side_a = yield SelectRouterSideA
     router_a = user_input_router_side_a.side_a_node_id.name
 
-    JuniperAeMembers = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)]
+    JuniperAeMembers = Annotated[
+        list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)
+    ]
 
     if get_router_vendor(router_a) == Vendor.NOKIA:
 
@@ -94,7 +96,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
                 initial_user_input.iptrunk_speed,
             )
 
-        ae_members_side_a = Annotated[list[NokiaLAGMemberA], AfterValidator(validate_unique_list), Len(min_length=initial_user_input.iptrunk_minimum_links)]
+        ae_members_side_a = Annotated[
+            list[NokiaLAGMemberA],
+            AfterValidator(validate_unique_list),
+            Len(min_length=initial_user_input.iptrunk_minimum_links),
+        ]
     else:
         ae_members_side_a = JuniperAeMembers  # type: ignore[assignment]
 
@@ -137,7 +143,13 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
                 initial_user_input.iptrunk_speed,
             )
 
-        ae_members_side_b = Annotated[list[NokiaLAGMemberB], AfterValidator(validate_unique_list), Len(min_length=len(user_input_side_a.side_a_ae_members), max_length= len(user_input_side_a.side_a_ae_members))]
+        ae_members_side_b = Annotated[
+            list[NokiaLAGMemberB],
+            AfterValidator(validate_unique_list),
+            Len(
+                min_length=len(user_input_side_a.side_a_ae_members), max_length=len(user_input_side_a.side_a_ae_members)
+            ),
+        ]
     else:
         ae_members_side_b = JuniperAeMembers  # type: ignore[assignment]
 

gso/workflows/iptrunk/migrate_iptrunk.py

@@ -118,10 +118,23 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
                 subscription.iptrunk.iptrunk_speed,
             )
 
-        ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members), max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members))]
+        ae_members = Annotated[
+            list[NokiaLAGMember],
+            AfterValidator(validate_unique_list),
+            Len(
+                min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members),
+                max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members),
+            ),
+        ]
     else:
-
-        ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members), max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members))]
+        ae_members = Annotated[
+            list[LAGMember],
+            AfterValidator(validate_unique_list),
+            Len(
+                min_length=len(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_members),
+                max_length=len(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_members),
+            ),
+        ]
 
     replace_index = (
         0
@@ -133,7 +146,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         LAGMember(
             interface_name=iface.interface_name,
             interface_description=iface.interface_description,
-    )
+        )
         for iface in subscription.iptrunk.iptrunk_sides[replace_index].iptrunk_side_ae_members
     ]
 

gso/workflows/iptrunk/modify_trunk_interface.py

@@ -60,10 +60,13 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
                 )
             )
 
-        ae_members = Annotated[list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)]
+        ae_members = Annotated[
+            list[NokiaLAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)
+        ]
     else:
-
-        ae_members = Annotated[list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)]
+        ae_members = Annotated[
+            list[LAGMember], AfterValidator(validate_unique_list), Len(min_length=iptrunk_minimum_link)
+        ]
     return ae_members  # type: ignore[return-value]
 
 
@@ -83,8 +86,12 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         iptrunk_speed: PhysicalPortCapacity = subscription.iptrunk.iptrunk_speed
         iptrunk_minimum_links: int = subscription.iptrunk.iptrunk_minimum_links
         iptrunk_isis_metric: ReadOnlyField(subscription.iptrunk.iptrunk_isis_metric, default_type=int)
-        iptrunk_ipv4_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv4_network), default_type=FancyIPV4Address)
-        iptrunk_ipv6_network: ReadOnlyField(str(subscription.iptrunk.iptrunk_ipv6_network), default_type=FancyIPV6Address)
+        iptrunk_ipv4_network: ReadOnlyField(
+            str(subscription.iptrunk.iptrunk_ipv4_network), default_type=FancyIPV4Address
+        )
+        iptrunk_ipv6_network: ReadOnlyField(
+            str(subscription.iptrunk.iptrunk_ipv6_network), default_type=FancyIPV6Address
+        )
 
         @field_validator("tt_number")
         def validate_tt_number(cls, tt_number: str) -> str:
@@ -96,7 +103,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     class ModifyIptrunkSideAForm(FormPage):
         model_config = ConfigDict(title="Provide subscription details for side A of the trunk.")
 
-        side_a_node: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str)
+        side_a_node: ReadOnlyField(
+            subscription.iptrunk.iptrunk_sides[0].iptrunk_side_node.router_fqdn, default_type=str
+        )
         side_a_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_iface, default_type=str)
         side_a_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sides[0].iptrunk_side_ae_geant_a_sid
         side_a_ae_members: ae_members_side_a = (  # type: ignore[valid-type]
@@ -120,7 +129,9 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
     class ModifyIptrunkSideBForm(FormPage):
         model_config = ConfigDict(title="Provide subscription details for side B of the trunk.")
 
-        side_b_node: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str)
+        side_b_node: ReadOnlyField(
+            subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_fqdn, default_type=str
+        )
         side_b_ae_iface: ReadOnlyField(subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_iface, default_type=str)
         side_b_ae_geant_a_sid: str = subscription.iptrunk.iptrunk_sides[1].iptrunk_side_ae_geant_a_sid
         side_b_ae_members: ae_members_side_b = (  # type: ignore[valid-type]

gso/workflows/router/create_router.py

@@ -52,7 +52,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         @model_validator(mode="after")
         def hostname_must_be_available(self) -> Self:
             router_site = self.router_site
-            if not router_site: # TODO Test on UI
+            if not router_site:  # TODO Test on UI
                 msg = "Please select a site before setting the hostname."
                 raise ValueError(msg)
 

gso/workflows/router/update_ibgp_mesh.py

@@ -34,7 +34,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         tt_number: str
 
         @model_validator(mode="before")
-        def router_has_a_trunk(self ) -> Self:
+        def router_has_a_trunk(self) -> Self:
             terminating_trunks = get_trunks_that_terminate_on_router(
                 subscription_id, SubscriptionLifecycle.PROVISIONING
             ) + get_trunks_that_terminate_on_router(subscription_id, SubscriptionLifecycle.ACTIVE)

gso/workflows/site/create_site.py

@@ -23,7 +23,6 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         model_config = ConfigDict(title=product_name)
         partner: ReadOnlyField("GEANT", default_type=str)
 
-
     user_input = yield CreateSiteForm
 
     return user_input.dict()

gso/workflows/tasks/import_iptrunk.py

@@ -146,7 +146,7 @@ def update_ipam_stub_for_subscription(
 @workflow(
     "Import iptrunk",
     initial_input_form=initial_input_form_generator,
-    target=Target.CREATE,
+    target=Target.SYSTEM,
 )
 def import_iptrunk() -> StepList:
     """Import an IP trunk without provisioning it."""

gso/workflows/tasks/import_office_router.py

 """A creation workflow that adds existing office routers to the coreDB."""
 
-
 from orchestrator import workflow
 from orchestrator.forms import FormPage
 from orchestrator.targets import Target

gso/workflows/tasks/import_router.py

 """A creation workflow that adds an existing router to the service database."""
 
-
 from orchestrator import workflow
 from orchestrator.forms import FormPage
 from orchestrator.targets import Target
@@ -89,7 +88,7 @@ def initialize_subscription(
 @workflow(
     "Import router",
     initial_input_form=initial_input_form_generator,
-    target=Target.CREATE,
+    target=Target.SYSTEM,
 )
 def import_router() -> StepList:
     """Import a router without provisioning it."""