NAT-329: merged develop

data/routers.json

+[
+  {
+    "customer": "GÉANT",
+    "router_site": "AMS",
+    "hostname": "rt1",
+    "ts_port": 22111,
+    "router_vendor": "juniper",
+    "router_role": "p",
+    "router_lo_ipv4_address": "62.40.119.2",
+    "router_lo_ipv6_address": "2001:798:1ab::2",
+    "router_lo_iso_address": "49.51e5.0001.0620.4011.9002.00",
+    "is_ias_connected": false
+  },
+  {
+    "customer": "GÉANT",
+    "router_site": "AMS2",
+    "hostname": "rt2",
+    "ts_port": 22111,
+    "router_vendor": "juniper",
+    "router_role": "p",
+    "router_lo_ipv4_address": "62.40.119.1",
+    "router_lo_ipv6_address": "2001:798:1ab::1",
+    "router_lo_iso_address": "49.51e5.0001.0620.4011.9001.00",
+    "is_ias_connected": false
+  }
+]

data/sites.json

+[
+  {
+    "site_name": "AMS2",
+    "site_city": "Amsterdam",
+    "site_country": "The Netherlands",
+    "site_country_code": "NL",
+    "site_latitude": 0,
+    "site_longitude": 0,
+    "site_bgp_community_id": 4,
+    "site_internal_id": 4,
+    "site_tier": "1",
+    "site_ts_address": "0.1.1.1",
+    "customer": "GÉANT"
+  }
+]
\ No newline at end of file

data/trunks.json

+[
+  {
+    "customer": "GÉANT",
+    "geant_s_sid": "12",
+    "iptrunk_type": "Dark_fiber",
+    "iptrunk_description": "Description",
+    "iptrunk_speed": "100G",
+    "iptrunk_minimum_links": 1,
+    "side_a_node_id": "",
+    "side_a_ae_iface": "string",
+    "side_a_ae_geant_a_sid": "string",
+    "side_a_ae_members": [
+      {
+        "interface_name": "string",
+        "interface_description": "string"
+      }
+    ],
+    "side_b_node_id": "string",
+    "side_b_ae_iface": "string",
+    "side_b_ae_geant_a_sid": "string",
+    "side_b_ae_members": [
+      {
+        "interface_name": "string",
+        "interface_description": "string"
+      }
+    ],
+    "iptrunk_ipv4_network": "string",
+    "iptrunk_ipv6_network": "string"
+  }
+]

gso/__init__.py

@@ -8,7 +8,6 @@ from orchestrator.cli.main import app as cli_app
 import gso.products
 import gso.workflows  # noqa: F401
 from gso.api import router as api_router
-from gso.cli import netbox
 
 
 def init_gso_app() -> OrchestratorCore:
@@ -25,8 +24,8 @@ def init_worker_app() -> OrchestratorCore:
 
 def init_cli_app() -> typer.Typer:
     """Initialise :term:`GSO` as a CLI application."""
-    from gso.cli import import_sites  # noqa: PLC0415
+    from gso.cli import imports, netbox  # noqa: PLC0415
 
-    cli_app.add_typer(import_sites.app, name="import_sites")
+    cli_app.add_typer(imports.app, name="import-cli")
     cli_app.add_typer(netbox.app, name="netbox-cli")
     return cli_app()

gso/cli/import_sites.py

-""":term:`CLI` command for importing sites."""
-
-import typer
-
-app: typer.Typer = typer.Typer()
-
-
-@app.command()
-def import_sites() -> None:
-    """Import sites from a source."""
-    # TODO: Implement this CLI command to import sites from a source.
-    typer.echo("Importing sites...")

gso/cli/imports.py

+""":term:`CLI` command for importing data to coreDB."""
+
+import ipaddress
+import json
+from pathlib import Path
+from typing import TypeVar
+
+import typer
+import yaml  # type: ignore[import-untyped]
+from pydantic import ValidationError
+
+from gso.api.v1.imports import (
+    IptrunkImportModel,
+    RouterImportModel,
+    SiteImportModel,
+    import_iptrunk,
+    import_router,
+    import_site,
+)
+from gso.services.subscriptions import get_active_subscriptions_by_field_and_value
+
+app: typer.Typer = typer.Typer()
+
+T = TypeVar("T", SiteImportModel, RouterImportModel, IptrunkImportModel)
+
+common_filepath_option = typer.Option(
+    default="data.json",
+    help="Path to the file",
+)
+
+
+def read_data(filepath: str) -> dict:
+    """Read data from a JSON or YAML file."""
+    typer.echo(f"Starting import from {filepath}")
+    file_path = Path(filepath)
+    file_extension = file_path.suffix.lower()
+
+    with file_path.open("r") as f:
+        supported_extensions = {".json", ".yaml", ".yml"}
+
+        if file_extension == ".json":
+            return json.load(f)
+        if file_extension in supported_extensions:
+            return yaml.safe_load(f)
+
+        typer.echo(f"Unsupported file format: {file_extension}")
+        raise typer.Exit(code=1)
+
+
+def generic_import_data(
+    filepath: str,
+    import_model: type[T],
+    import_function: callable,  # type: ignore[valid-type]
+    name_key: str,
+) -> None:
+    """Import data from a JSON or YAML file."""
+    successfully_imported_data = []
+    data = read_data(filepath)
+    for details in data:
+        details["customer"] = "GÉANT"
+        typer.echo(f"Importing {name_key}: {details[name_key]}")
+        try:
+            initial_data = import_model(**details)
+            import_function(initial_data)  # type: ignore[misc]
+            successfully_imported_data.append(getattr(initial_data, name_key))
+            typer.echo(
+                f"Successfully imported {name_key}: {getattr(initial_data, name_key)}",
+            )
+        except ValidationError as e:
+            typer.echo(f"Validation error: {e}")
+
+    if successfully_imported_data:
+        typer.echo(f"Successfully imported {name_key}s:")
+        for item in successfully_imported_data:
+            typer.echo(f"- {item}")
+
+
+@app.command()
+def import_sites(filepath: str = common_filepath_option) -> None:
+    """Import sites into GSO."""
+    # Use the import_data function to handle common import logic
+    generic_import_data(filepath, SiteImportModel, import_site, "site_name")
+
+
+@app.command()
+def import_routers(filepath: str = common_filepath_option) -> None:
+    """Import routers into GSO."""
+    # Use the import_data function to handle common import logic
+    generic_import_data(filepath, RouterImportModel, import_router, "hostname")
+
+
+def get_router_subscription_id(node_name: str) -> str | None:
+    """Get the subscription id for a router by its node name."""
+    subscriptions = get_active_subscriptions_by_field_and_value(
+        "router_fqdn",
+        node_name,
+    )
+    if subscriptions:
+        return str(subscriptions[0].subscription_id)
+    return None
+
+
+@app.command()
+def import_iptrunks(filepath: str = common_filepath_option) -> None:
+    """Import IP trunks into GSO."""
+    successfully_imported_data = []
+    data = read_data(filepath)
+    for trunk in data:
+        ipv4_network_a = ipaddress.ip_network(
+            trunk["config"]["nodeA"]["ipv4_address"],
+            strict=False,
+        )
+        ipv4_network_b = ipaddress.ip_network(
+            trunk["config"]["nodeB"]["ipv4_address"],
+            strict=False,
+        )
+        ipv6_network_a = ipaddress.ip_network(
+            trunk["config"]["nodeA"]["ipv6_address"],
+            strict=False,
+        )
+        ipv6_network_b = ipaddress.ip_network(
+            trunk["config"]["nodeB"]["ipv6_address"],
+            strict=False,
+        )
+        # Check if IPv4 networks are equal
+        if ipv4_network_a == ipv4_network_b:
+            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']}.")
+            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']}.")
+            continue
+
+        typer.echo(
+            f"Importing IP Trunk: "
+            f'{get_active_subscriptions_by_field_and_value("router_fqdn", trunk["config"]["nodeA"]["name"])}',
+        )
+        try:
+            initial_data = IptrunkImportModel(
+                customer="GÉANT",
+                geant_s_sid=trunk["id"],
+                iptrunk_type=trunk["config"]["common"]["type"],
+                iptrunk_description=trunk["config"]["common"].get("description", ""),
+                iptrunk_speed=trunk["config"]["common"]["link_speed"],
+                iptrunk_minimum_links=trunk["config"]["common"]["minimum_links"],
+                side_a_node_id=get_router_subscription_id(
+                    trunk["config"]["nodeA"]["name"],
+                )
+                or "",
+                side_a_ae_iface=trunk["config"]["nodeA"]["ae_name"],
+                side_a_ae_geant_a_sid=trunk["config"]["nodeA"]["port_sid"],
+                side_a_ae_members=trunk["config"]["nodeA"]["members"],
+                side_b_node_id=get_router_subscription_id(
+                    trunk["config"]["nodeB"]["name"],
+                )
+                or "",
+                side_b_ae_iface=trunk["config"]["nodeB"]["ae_name"],
+                side_b_ae_geant_a_sid=trunk["config"]["nodeB"]["port_sid"],
+                side_b_ae_members=trunk["config"]["nodeB"]["members"],
+                iptrunk_ipv4_network=iptrunk_ipv4_network,  # type:ignore[arg-type]
+                iptrunk_ipv6_network=iptrunk_ipv6_network,  # type:ignore[arg-type]
+            )
+            import_iptrunk(initial_data)
+            successfully_imported_data.append(trunk["id"])
+            typer.echo(f"Successfully imported IP Trunk: {trunk['id']}")
+        except ValidationError as e:
+            typer.echo(f"Validation error: {e}")
+
+        if successfully_imported_data:
+            typer.echo("Successfully imported IP Trunks:")
+            for item in successfully_imported_data:
+                typer.echo(f"- {item}")

gso/products/product_blocks/router.py

@@ -6,6 +6,7 @@ from orchestrator.domain.base import ProductBlockModel
 from orchestrator.types import SubscriptionLifecycle, strEnum
 from pydantic import ConstrainedInt
 
+from gso import settings
 from gso.products.product_blocks.site import (
     SiteBlock,
     SiteBlockInactive,
@@ -58,7 +59,8 @@ class RouterBlockInactive(
 
 def generate_fqdn(hostname: str, site_name: str, country_code: str) -> str:
     """Generate an :term:`FQDN` from a hostname, site name, and a country code."""
-    return f"{hostname}.{site_name.lower()}.{country_code.lower()}.geant.net"
+    oss = settings.load_oss_params()
+    return f"{hostname}.{site_name.lower()}.{country_code.lower()}{oss.IPAM.LO.domain_name}"
 
 
 class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):

gso/services/netbox_client.py

@@ -125,7 +125,9 @@ class NetboxClient:
     def delete_interface(self, device_name: str, iface_name: str) -> None:
         """Delete an interface from a device by name."""
         interface = self.get_interface_by_name_and_device(iface_name, device_name)
-        return interface.delete()
+        if interface:
+            return interface.delete()
+        return None
 
     def create_device_type(self, manufacturer: str, model: str, slug: str) -> DeviceTypes:
         """Create a new device type in Netbox."""

gso/workflows/tasks/import_router.py

@@ -25,12 +25,12 @@ def _get_site_by_name(site_name: str) -> Site:
     :param site_name: The name of the site.
     :type site_name: str
     """
-    subscription = subscriptions.get_active_subscriptions_by_field_and_value("site_name", site_name)[0]
+    subscription = subscriptions.get_active_subscriptions_by_field_and_value("site_name", site_name)
     if not subscription:
         msg = f"Site with name {site_name} not found."
         raise ValueError(msg)
 
-    return Site.from_subscription(subscription.subscription_id)
+    return Site.from_subscription(subscription[0].subscription_id)
 
 
 @step("Create subscription")

test/conftest.py

@@ -130,7 +130,7 @@ def configuration_data() -> dict:
                         "networks": ["dead:beef::/80"],
                         "mask": 128,
                     },
-                    "domain_name": ".lo",
+                    "domain_name": ".geant.net",
                     "dns_view": "default",
                 },
                 "TRUNK": {