Skip to content
Snippets Groups Projects
Commit fb983cf1 authored by Aleksandr Kurbatov's avatar Aleksandr Kurbatov
Browse files

l3-services-imports-update

parent 2b68e037
No related branches found
No related tags found
No related merge requests found
Pipeline #93896 failed
......@@ -544,13 +544,13 @@ def import_edge_port(filepath: str = common_filepath_option) -> None:
successfully_imported_data = []
data = _read_data(Path(filepath))
for edge_port in data:
typer.echo(f"Importing Edge Port {edge_port["name"]} on {edge_port["node"]}. ")
typer.echo(f"Importing Edge Port {edge_port['name']} on {edge_port['node']}. ")
try:
edge_port["node"] = _get_router_subscription_id(edge_port["node"])
initial_data = EdgePortImportModel(**edge_port)
start_process("create_imported_edge_port", [initial_data.model_dump()])
successfully_imported_data.append(edge_port["name"])
typer.echo(f"Successfully imported Edge Port {edge_port["name"]} on {edge_port["node"]}.")
typer.echo(f"Successfully imported Edge Port {edge_port['name']} on {edge_port['node']}.")
except ValidationError as e:
typer.echo(f"Validation error: {e}")
......@@ -585,19 +585,19 @@ def import_iptrunks(filepath: str = common_filepath_option) -> None:
iptrunk_ipv4_network = ipv4_network_a
else:
# Handle the case where IPv4 networks are different
typer.echo(f"Error: IPv4 networks are different for trunk {trunk["id"]}.")
typer.echo(f"Error: IPv4 networks are different for trunk {trunk['id']}.")
continue
# Check if IPv6 networks are the same
if ipv6_network_a == ipv6_network_b:
iptrunk_ipv6_network = ipv6_network_a
else:
# Handle the case where IPv6 networks are different
typer.echo(f"Error: IPv6 networks are different for trunk {trunk["id"]}.")
typer.echo(f"Error: IPv6 networks are different for trunk {trunk['id']}.")
continue
typer.echo(
f"Importing IP Trunk: "
f'{get_active_subscriptions_by_field_and_value("router_fqdn", trunk["config"]["nodeA"]["name"])}',
f"{get_active_subscriptions_by_field_and_value('router_fqdn', trunk['config']['nodeA']['name'])}",
)
try:
initial_data = IptrunkImportModel(
......@@ -621,7 +621,7 @@ def import_iptrunks(filepath: str = common_filepath_option) -> None:
)
start_process("create_imported_iptrunk", [initial_data.model_dump()])
successfully_imported_data.append(trunk["id"])
typer.echo(f"Successfully imported IP Trunk: {trunk["id"]}")
typer.echo(f"Successfully imported IP Trunk: {trunk['id']}")
except ValidationError as e:
typer.echo(f"Validation error: {e}")
......@@ -672,18 +672,18 @@ def import_l3_core_service(filepath: str = common_filepath_option) -> None:
l3_core_service_list = _read_data(Path(filepath))
for l3_core_service in l3_core_service_list:
partner = l3_core_service[0]["partner"]
product_name = l3_core_service[0]["product_name"]
partner = l3_core_service["partner"]
product_name = l3_core_service["product_name"]
typer.echo(f"Creating imported {product_name} for {partner}")
try:
if product_name == ProductName.IAS.value:
initial_data = IASImportModel(**l3_core_service[0], **l3_core_service[1]).model_dump()
initial_data = IASImportModel(**l3_core_service).model_dump()
else:
initial_data = L3CoreServiceImportModel(**l3_core_service[0], **l3_core_service[1]).model_dump()
initial_data = L3CoreServiceImportModel(**l3_core_service).model_dump()
start_process(L3_CREAT_IMPORTED_WF_MAP[product_name], [initial_data])
edge_ports = [sbp["edge_port"] for sbp in l3_core_service[0]["service_binding_ports"]]
start_process("create_imported_ias", [initial_data])
edge_ports = [sbp["edge_port"] for sbp in l3_core_service["service_binding_ports"]]
successfully_imported_data.append(edge_ports)
typer.echo(f"Successfully created imported {product_name} for {partner}")
except ValidationError as e:
......
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment