Move shared code that was not shared into single file

gso/workflows/l3_core_service/base_validate_prefix_list.py

-"""Prefix Validation workflow for L3 Core Service subscription objects."""
-
-from typing import Any
-
-from orchestrator.config.assignee import Assignee
-from orchestrator.domain import SubscriptionModel
-from orchestrator.forms import SubmitFormPage
-from orchestrator.workflow import inputstep, step
-from pydantic import Field
-from pydantic_forms.types import FormGenerator, State, UUIDstr
-from pydantic_forms.validators import Label
-
-from gso.services.lso_client import LSOState
-from gso.services.partners import get_partner_by_id
-from gso.utils.shared_enums import Vendor
-
-
-@step("Prepare list of all Access Ports")
-def build_fqdn_list(subscription_id: UUIDstr) -> State:
-    """Build the list of all FQDNs in the access ports of L3 Core Service subscription, excluding Juniper devices."""
-    subscription = SubscriptionModel.from_subscription(subscription_id)
-    ap_list = subscription.l3_core.ap_list  # type: ignore[attr-defined]
-    ap_fqdn_list = [
-        ap.sbp.edge_port.node.router_fqdn for ap in ap_list if ap.sbp.edge_port.node.vendor != Vendor.JUNIPER
-    ]
-    return {"ap_fqdn_list": ap_fqdn_list, "subscription": subscription}
-
-
-@step("[DRY RUN] Validate Prefix-Lists")
-def validate_prefix_lists_dry(subscription: dict[str, Any], process_id: UUIDstr, ap_fqdn_list: list[str]) -> LSOState:
-    """Workflow step for running a playbook that validates prefix-lists in dry run mode."""
-    extra_vars = {
-        "subscription": subscription,
-        "partner_name": get_partner_by_id(subscription["customer_id"]).name,
-        "dry_run": True,
-        "verb": "deploy",
-        "object": "prefix_list",
-        "is_verification_workflow": "true",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - Validate prefix-lists for {subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/validate_prefix_list.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(ap_fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("Evaluate validation of Prefix-Lists")
-def evaluate_result_has_diff(callback_result: dict) -> State:
-    """Evaluate the result of the playbook that validates prefix-lists."""
-    return {"callback_result": callback_result, "prefix_list_drift": bool(callback_result["return_code"] != 0)}
-
-
-@inputstep("Await operator confirmation", assignee=Assignee.SYSTEM)
-def await_operator() -> FormGenerator:
-    """Show a form for the operator to start redeploying the prefix list that has drifted."""
-
-    class AwaitOperatorForm(SubmitFormPage):
-        info_label_a: Label = Field("A drift has been detected for this prefix list!", exclude=True)
-        info_label_b: Label = Field("Please continue this workflow to redeploy the drifted prefix list.", exclude=True)
-
-    yield AwaitOperatorForm
-
-    return {}
-
-
-@step("[DRY RUN] Deploy Prefix-Lists")
-def deploy_prefix_lists_dry(subscription: dict[str, Any], process_id: UUIDstr, ap_fqdn_list: list[str]) -> LSOState:
-    """Workflow step for running a playbook that deploys prefix-lists in dry run mode."""
-    extra_vars = {
-        "subscription": subscription,
-        "partner_name": get_partner_by_id(subscription["customer_id"]).name,
-        "dry_run": True,
-        "verb": "deploy",
-        "object": "prefix_list",
-        "is_verification_workflow": "false",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - Deploy prefix-lists for {subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/deploy_prefix_list.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(ap_fqdn_list)}},
-        "extra_vars": extra_vars,
-    }
-
-
-@step("[REAL] Deploy Prefix-Lists")
-def deploy_prefix_lists_real(subscription: dict[str, Any], process_id: UUIDstr, ap_fqdn_list: list[str]) -> LSOState:
-    """Workflow step for running a playbook that deploys prefix-lists."""
-    extra_vars = {
-        "subscription": subscription,
-        "partner_name": get_partner_by_id(subscription["customer_id"]).name,
-        "dry_run": False,
-        "verb": "deploy",
-        "object": "prefix_list",
-        "is_verification_workflow": "false",
-        "commit_comment": f"GSO_PROCESS_ID: {process_id} - Deploy prefix-lists for {subscription["description"]}",
-    }
-
-    return {
-        "playbook_name": "gap_ansible/playbooks/deploy_prefix_list.yaml",
-        "inventory": {"all": {"hosts": dict.fromkeys(ap_fqdn_list)}},
-        "extra_vars": extra_vars,
-    }

gso/workflows/l3_core_service/geant_ip/validate_prefix_list.py

 """Prefix Validation workflow for GÉANT IP subscription objects."""
 
+from typing import Any
+
+from orchestrator.config.assignee import Assignee
+from orchestrator.domain import SubscriptionModel
+from orchestrator.forms import SubmitFormPage
 from orchestrator.targets import Target
-from orchestrator.workflow import StepList, begin, conditional, done, workflow
+from orchestrator.workflow import StepList, begin, conditional, done, inputstep, step, workflow
 from orchestrator.workflows.steps import resync, store_process_subscription, unsync
 from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import Field
+from pydantic_forms.types import FormGenerator, State, UUIDstr
+from pydantic_forms.validators import Label
 
-from gso.services.lso_client import anonymous_lso_interaction, lso_interaction
-from gso.workflows.l3_core_service.base_validate_prefix_list import (
-    await_operator,
-    build_fqdn_list,
-    deploy_prefix_lists_dry,
-    deploy_prefix_lists_real,
-    evaluate_result_has_diff,
-    validate_prefix_lists_dry,
-)
+from gso.services.lso_client import LSOState, anonymous_lso_interaction, lso_interaction
+from gso.services.partners import get_partner_by_id
+from gso.utils.shared_enums import Vendor
+
+
+@step("Prepare list of all Access Ports")
+def build_fqdn_list(subscription_id: UUIDstr) -> State:
+    """Build the list of all FQDNs in the access ports of L3 Core Service subscription, excluding Juniper devices."""
+    subscription = SubscriptionModel.from_subscription(subscription_id)
+    ap_list = subscription.l3_core.ap_list  # type: ignore[attr-defined]
+    ap_fqdn_list = [
+        ap.sbp.edge_port.node.router_fqdn for ap in ap_list if ap.sbp.edge_port.node.vendor != Vendor.JUNIPER
+    ]
+    return {"ap_fqdn_list": ap_fqdn_list, "subscription": subscription}
+
+
+@step("[DRY RUN] Validate Prefix-Lists")
+def validate_prefix_lists_dry(subscription: dict[str, Any], process_id: UUIDstr, ap_fqdn_list: list[str]) -> LSOState:
+    """Workflow step for running a playbook that validates prefix-lists in dry run mode."""
+    extra_vars = {
+        "subscription": subscription,
+        "partner_name": get_partner_by_id(subscription["customer_id"]).name,
+        "dry_run": True,
+        "verb": "deploy",
+        "object": "prefix_list",
+        "is_verification_workflow": "true",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - Validate prefix-lists for {subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/validate_prefix_list.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(ap_fqdn_list)}},
+        "extra_vars": extra_vars,
+    }
+
+
+@step("Evaluate validation of Prefix-Lists")
+def evaluate_result_has_diff(callback_result: dict) -> State:
+    """Evaluate the result of the playbook that validates prefix-lists."""
+    return {"callback_result": callback_result, "prefix_list_drift": bool(callback_result["return_code"] != 0)}
+
+
+@inputstep("Await operator confirmation", assignee=Assignee.SYSTEM)
+def await_operator() -> FormGenerator:
+    """Show a form for the operator to start redeploying the prefix list that has drifted."""
+
+    class AwaitOperatorForm(SubmitFormPage):
+        info_label_a: Label = Field("A drift has been detected for this prefix list!", exclude=True)
+        info_label_b: Label = Field("Please continue this workflow to redeploy the drifted prefix list.", exclude=True)
+
+    yield AwaitOperatorForm
+
+    return {}
+
+
+@step("[DRY RUN] Deploy Prefix-Lists")
+def deploy_prefix_lists_dry(subscription: dict[str, Any], process_id: UUIDstr, ap_fqdn_list: list[str]) -> LSOState:
+    """Workflow step for running a playbook that deploys prefix-lists in dry run mode."""
+    extra_vars = {
+        "subscription": subscription,
+        "partner_name": get_partner_by_id(subscription["customer_id"]).name,
+        "dry_run": True,
+        "verb": "deploy",
+        "object": "prefix_list",
+        "is_verification_workflow": "false",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - Deploy prefix-lists for {subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/deploy_prefix_list.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(ap_fqdn_list)}},
+        "extra_vars": extra_vars,
+    }
+
+
+@step("[REAL] Deploy Prefix-Lists")
+def deploy_prefix_lists_real(subscription: dict[str, Any], process_id: UUIDstr, ap_fqdn_list: list[str]) -> LSOState:
+    """Workflow step for running a playbook that deploys prefix-lists."""
+    extra_vars = {
+        "subscription": subscription,
+        "partner_name": get_partner_by_id(subscription["customer_id"]).name,
+        "dry_run": False,
+        "verb": "deploy",
+        "object": "prefix_list",
+        "is_verification_workflow": "false",
+        "commit_comment": f"GSO_PROCESS_ID: {process_id} - Deploy prefix-lists for {subscription["description"]}",
+    }
+
+    return {
+        "playbook_name": "gap_ansible/playbooks/deploy_prefix_list.yaml",
+        "inventory": {"all": {"hosts": dict.fromkeys(ap_fqdn_list)}},
+        "extra_vars": extra_vars,
+    }
 
 
 @workflow(