Add summary view to the bunch of workflows.

gso/workflows/iptrunk/create_iptrunk.py

@@ -45,6 +45,7 @@ from gso.utils.types.interfaces import JuniperLAGMember, LAGMember, LAGMemberLis
 from gso.utils.types.netbox_router import NetboxEnabledRouter
 from gso.utils.types.tt_number import TTNumber
 from gso.utils.workflow_steps import prompt_sharepoint_checklist_url
+from gso.workflows.shared import create_summary_form
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -160,8 +161,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         side_b_ae_members: ae_members_side_b
 
     user_input_side_b = yield CreateIptrunkSideBForm
-
-    return (
+    input_forms_data = (
         initial_user_input.model_dump()
         | verify_minimum_links.model_dump()
         | user_input_router_side_a.model_dump()
@@ -169,6 +169,25 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         | user_input_router_side_b.model_dump()
         | user_input_side_b.model_dump()
     )
+    summary_form_data = input_forms_data | {"side_a_node": router_a_fqdn, "side_b_node": router_b_fqdn}
+    summary_fields = [
+        "geant_s_sid",
+        "iptrunk_type",
+        "iptrunk_speed",
+        "iptrunk_description",
+        "iptrunk_minimum_links",
+        "side_a_node",
+        "side_a_ae_iface",
+        "side_a_ae_members",
+        "side_a_ae_geant_a_sid",
+        "side_b_node",
+        "side_b_ae_iface",
+        "side_b_ae_members",
+        "side_b_ae_geant_a_sid",
+    ]
+    yield from create_summary_form(summary_form_data, product_name, summary_fields)
+
+    return input_forms_data
 
 
 @step("Create subscription")

gso/workflows/iptrunk/modify_isis_metric.py

@@ -13,6 +13,7 @@ from orchestrator.workflows.utils import wrap_modify_initial_input_form
 from gso.products.product_types.iptrunk import Iptrunk
 from gso.services.lso_client import LSOState, lso_interaction
 from gso.utils.types.tt_number import TTNumber
+from gso.workflows.shared import modify_summary_form
 
 
 def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -24,8 +25,11 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
         isis_metric: int = subscription.iptrunk.iptrunk_isis_metric
 
     user_input = yield ModifyIptrunkForm
-
-    return user_input.model_dump()
+    user_input = user_input.model_dump()
+    yield from modify_summary_form(
+        {"iptrunk_isis_metric": user_input["isis_metric"]}, subscription.iptrunk, ["iptrunk_isis_metric"]
+    )
+    return user_input
 
 
 @step("Update subscription")

gso/workflows/router/create_router.py

@@ -33,6 +33,7 @@ from gso.utils.workflow_steps import (
     prompt_sharepoint_checklist_url,
     run_checks_after_base_config,
 )
+from gso.workflows.shared import create_summary_form
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -65,8 +66,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
             return self
 
     user_input = yield CreateRouterForm
+    user_input = user_input.model_dump()
+    summary_fields = ["hostname", "router_site", "vendor", "ts_port", "router_role"]
+    yield from create_summary_form(user_input, product_name, summary_fields)
 
-    return user_input.model_dump()
+    return user_input
 
 
 @step("Create subscription")

gso/workflows/shared.py

+"""Shared functions for the workflows."""
+
+from collections.abc import Generator
+from typing import TypeAlias, cast
+
+from orchestrator.domain.base import ProductBlockModel
+from orchestrator.forms import FormPage
+from orchestrator.forms.validators import MigrationSummary, migration_summary
+from pydantic import ConfigDict
+
+
+def summary_form(product_name: str, summary_data: dict) -> Generator:
+    """Generate a summary form for the product."""
+    ProductSummary: TypeAlias = cast(type[MigrationSummary], migration_summary(summary_data))  # type: ignore[no-redef]
+
+    class SummaryForm(FormPage):
+        model_config = ConfigDict(title=f"{product_name} summary")
+
+        product_summary: ProductSummary
+
+    yield SummaryForm
+
+
+def create_summary_form(user_input: dict, product_name: str, fields: list[str]) -> Generator:
+    """Create a summary form for the product."""
+    columns = [[str(user_input[nm]) for nm in fields]]
+    yield from summary_form(product_name, {"labels": fields, "columns": columns})
+
+
+def modify_summary_form(user_input: dict, block: ProductBlockModel, fields: list[str]) -> Generator:
+    """Modify the summary form for the product."""
+    before = [str(getattr(block, nm)) for nm in fields]
+    after = [str(user_input[nm]) for nm in fields]
+    yield from summary_form(
+        block.subscription.product.name,
+        {
+            "labels": fields,
+            "headers": ["Before", "After"],
+            "columns": [before, after],
+        },
+    )

gso/workflows/site/create_site.py

@@ -15,6 +15,7 @@ from gso.services.partners import get_partner_by_name
 from gso.utils.types.base_site import BaseSiteValidatorModel
 from gso.utils.types.coordinates import LatitudeCoordinate, LongitudeCoordinate
 from gso.utils.types.ip_address import IPAddress
+from gso.workflows.shared import create_summary_form
 
 
 def initial_input_form_generator(product_name: str) -> FormGenerator:
@@ -25,8 +26,23 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
         partner: ReadOnlyField("GEANT", default_type=str)  # type: ignore[valid-type]
 
     user_input = yield CreateSiteForm
-
-    return user_input.model_dump()
+    user_input = user_input.model_dump()
+    summary_fields = [
+        "site_name",
+        "site_bgp_community_id",
+        "site_internal_id",
+        "site_tier",
+        "site_ts_address",
+        "site_country_code",
+        "site_city",
+        "site_country",
+        "site_latitude",
+        "site_longitude",
+        "partner",
+    ]
+    yield from create_summary_form(user_input, product_name, summary_fields)
+
+    return user_input
 
 
 @step("Create subscription")

test/workflows/iptrunk/test_create_iptrunk.py

@@ -90,6 +90,7 @@ def input_form_wizard_data(request, router_subscription_factory, faker):
         "side_b_ae_geant_a_sid": faker.geant_sid(),
         "side_b_ae_members": side_b_members,
     }
+    summary_view_step = {}
 
     return [
         create_ip_trunk_step,
@@ -98,6 +99,7 @@ def input_form_wizard_data(request, router_subscription_factory, faker):
         create_ip_trunk_side_a_step,
         create_ip_trunk_side_b_router_name,
         create_ip_trunk_side_b_step,
+        summary_view_step,
     ]
 
 

test/workflows/iptrunk/test_modify_isis_metric.py

@@ -26,6 +26,7 @@ def test_iptrunk_modify_isis_metric_success(
     initial_iptrunk_data = [
         {"subscription_id": product_id},
         {"tt_number": faker.tt_number(), "isis_metric": new_isis_metric},
+        {},
     ]
     result, process_stat, step_log = run_workflow("modify_isis_metric", initial_iptrunk_data)
 

test/workflows/router/test_create_router.py

@@ -68,7 +68,7 @@ def test_create_nokia_router_success(
     mock_sharepoint_client.return_value = MockedSharePointClient
 
     #  Run workflow
-    initial_router_data = [{"product": product_id}, router_creation_input_form_data]
+    initial_router_data = [{"product": product_id}, router_creation_input_form_data, {}]
     result, process_stat, step_log = run_workflow("create_router", initial_router_data)
 
     state = extract_state(result)
@@ -176,7 +176,7 @@ def test_create_nokia_router_lso_failure(
 
     #  Run workflow
     product_id = get_product_id_by_name(ProductName.ROUTER)
-    initial_router_data = [{"product": product_id}, router_creation_input_form_data]
+    initial_router_data = [{"product": product_id}, router_creation_input_form_data, {}]
     result, process_stat, step_log = run_workflow("create_router", initial_router_data)
 
     result, step_log = assert_lso_interaction_success(result, process_stat, step_log)

test/workflows/site/test_create_site.py

@@ -25,6 +25,7 @@ def test_create_site(responses, faker):
             "site_tier": SiteTier.TIER1,
             "site_ts_address": faker.ipv4(),
         },
+        {},
     ]
     result, _, _ = run_workflow("create_site", initial_site_data)
     assert_complete(result)
@@ -66,6 +67,7 @@ def test_site_name_is_incorrect(responses, faker):
             "site_ts_address": faker.ipv4(),
             "partner": "GEANT",
         },
+        {},
     ]
 
     with pytest.raises(FormValidationError, match=expected_exception_msg):

test/workflows/site/test_modify_site.py

@@ -15,6 +15,7 @@ def test_modify_site(responses, site_subscription_factory, faker):
             "site_internal_id": faker.pyint(),
             "site_ts_address": faker.ipv4(),
         },
+        {},
     ]
     result, _, _ = run_workflow("modify_site", initial_site_data)
     assert_complete(result)
@@ -39,6 +40,7 @@ def test_modify_site_with_duplicate_bgp_community_id(faker, site_subscription_fa
         {
             "site_bgp_community_id": duplicate_bgp_community_id,
         },
+        {},
     ]
 
     with pytest.raises(FormValidationError, match="site_bgp_community_id must be unique"):
@@ -57,6 +59,7 @@ def test_modify_site_with_duplicate_internal_id(faker, site_subscription_factory
         {
             "site_internal_id": duplicate_internal_id,
         },
+        {},
     ]
 
     with pytest.raises(FormValidationError, match="site_internal_id must be unique"):
@@ -75,6 +78,7 @@ def test_modify_site_with_duplicate_ts_address(faker, site_subscription_factory)
         {
             "site_ts_address": duplicate_ts_address,
         },
+        {},
     ]
 
     with pytest.raises(FormValidationError, match="site_ts_address must be unique"):