Add EIGRP model. Add unit test. Add integration test.

catalystwan/integration_tests/feature_profile/sdwan/test_service.py

@@ -2,11 +2,17 @@
 
 from catalystwan.api.configuration_groups.parcel import Global, as_global, as_variable
 from catalystwan.integration_tests.feature_profile.sdwan.base import TestFeatureProfileModels
+from catalystwan.models.configuration.feature_profile.common import Prefix
 from catalystwan.models.configuration.feature_profile.sdwan.service.dhcp_server import (
     AddressPool,
     LanVpnDhcpServerParcel,
     SubnetMask,
 )
+from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import (
+    AddressFamily,
+    EigrpParcel,
+    SummaryAddress,
+)
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ethernet import InterfaceEthernetParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import BasicGre, InterfaceGreParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ipsec import (
@@ -104,6 +110,27 @@ def test_when_default_ospfv3_ipv6_expect_successful_post(self):
         # Assert
         assert parcel_id
 
+    def test_when_default_values_eigrp_parcel_expect_successful_post(self):
+        eigrp_parcel = EigrpParcel(
+            parcel_name="TestEigrpParcel",
+            parcel_description="Test Eigrp Parcel",
+            as_number=Global[int](value=1),
+            address_family=AddressFamily(
+                network=[
+                    SummaryAddress(
+                        prefix=Prefix(
+                            address=as_global("10.3.2.1"),
+                            mask=as_global("255.255.255.0"),
+                        )
+                    )
+                ]
+            ),
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, eigrp_parcel).id
+        # Assert
+        assert parcel_id
+
     def tearDown(self) -> None:
         self.api.delete_profile(self.profile_uuid)
         self.session.close()

catalystwan/models/configuration/feature_profile/sdwan/service/eigrp.py

@@ -26,20 +26,22 @@
 
 
 class KeychainDetails(BaseModel):
-    key_id: Union[Global[int], Variable, Default[None]] = Field(serialization_alias="keyId", validation_alias="keyId")
-    keystring: Union[Global[str], Variable, Default[None]]
+    model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
+
+    key_id: Union[Global[int], Variable, Default[None]] = Field(
+        default=Default[None](value=None), serialization_alias="keyId", validation_alias="keyId"
+    )
+    keystring: Union[Global[str], Variable, Default[None]] = Field(default=Default[None](value=None))
 
 
 class EigrpAuthentication(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
 
-    auth_type: Union[Global[EigrpAuthType], Variable, Default[None]] = Field(
-        serialization_alias="type", validation_alias="type"
-    )
+    auth_type: Union[Global[EigrpAuthType], Variable, Default[None]] = Default[None](value=None)
     auth_key: Optional[Union[Global[str], Variable, Default[None]]] = Field(
-        serialization_alias="authKey", validation_alias="authKey"
+        serialization_alias="authKey", validation_alias="authKey", default=Default[None](value=None)
     )
-    key: Optional[List[KeychainDetails]] = Field(serialization_alias="key", validation_alias="key")
+    key: Optional[List[KeychainDetails]] = None
 
 
 class TableMap(BaseModel):
@@ -59,9 +61,9 @@ class IPv4StaticRoute(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
 
     name: Union[Global[str], Variable]
-    shutdown: Optional[Union[Global[int], Variable, Default[bool]]] = Default[bool](value=False)
-    summary_address: Optional[List[SummaryAddress]] = Field(
-        serialization_alias="summaryAddress", validation_alias="summaryAddress"
+    shutdown: Optional[Union[Global[bool], Variable, Default[bool]]] = Default[bool](value=False)
+    summary_address: List[SummaryAddress] = Field(
+        serialization_alias="summaryAddress", validation_alias="summaryAddress", default_factory=list
     )
 
 
@@ -76,7 +78,7 @@ class AddressFamily(BaseModel):
     model_config = ConfigDict(arbitrary_types_allowed=True, populate_by_name=True, extra="forbid")
 
     redistribute: Optional[List[RedistributeIntoEigrp]] = None
-    network: List[SummaryAddress]
+    network: List[SummaryAddress] = Field(min_length=1)
 
 
 class EigrpParcel(_ParcelBase):

catalystwan/tests/test_feature_profile_api.py

@@ -19,6 +19,7 @@
     LanVpnParcel,
     OspfParcel,
 )
+from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import EigrpParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.gre import BasicGre
 from catalystwan.models.configuration.feature_profile.sdwan.service.lan.ipsec import IpsecAddress, IpsecTunnelMode
 from catalystwan.models.configuration.feature_profile.sdwan.service.ospfv3 import Ospfv3IPv4Parcel, Ospfv3IPv6Parcel
@@ -108,6 +109,7 @@ def test_update_method_with_valid_arguments(self, parcel, expected_path):
     OspfParcel: "routing/ospf",
     Ospfv3IPv4Parcel: "routing/ospfv3/ipv4",
     Ospfv3IPv6Parcel: "routing/ospfv3/ipv6",
+    EigrpParcel: "routing/eigrp",
 }
 
 service_interface_parcels = [

catalystwan/utils/config_migration/converters/feature_template/eigrp.py

@@ -1,24 +1,129 @@
 from copy import deepcopy
+from typing import List, Optional
 
+from catalystwan.api.configuration_groups.parcel import Default, as_default, as_global, as_variable
+from catalystwan.models.configuration.feature_profile.common import Prefix
 from catalystwan.models.configuration.feature_profile.sdwan.service import EigrpParcel
+from catalystwan.models.configuration.feature_profile.sdwan.service.eigrp import (
+    AddressFamily,
+    EigrpAuthentication,
+    IPv4StaticRoute,
+    RedistributeIntoEigrp,
+    RedistributeProtocol,
+    SummaryAddress,
+    TableMap,
+)
 
 
 class EigrpTemplateConverter:
     supported_template_types = ("eigrp",)
 
     delete_keys = ("as_num",)
 
+    # Default values
+    lan_eigrp_auto_syst_id = "{{lan_eigrp_auto_syst_id}}"
+    lan_eigrp_addr_fami_netw_1_ip = "{{lan_eigrp_addr_fami_netw_1_ip}}"
+    lan_eigrp_addr_fami_netw_1_mask = "{{lan_eigrp_addr_fami_netw_1_mask}}"
+
     def create_parcel(self, name: str, description: str, template_values: dict) -> EigrpParcel:
+        print(template_values)
         values = self.prepare_values(template_values)
         self.configure_as_number(values)
+        self.configure_address_family_interface(values)
+        self.configure_address_family(values)
+        self.configure_authentication(values)
+        self.configure_table_map(values)
         self.cleanup_keys(values)
         return EigrpParcel(parcel_name=name, parcel_description=description, **values)
 
     def prepare_values(self, template_values: dict) -> dict:
         return deepcopy(template_values)["eigrp"]
 
     def configure_as_number(self, values: dict) -> None:
-        values["as_number"] = values.pop("as_num")
+        values["as_number"] = values.pop("as_num", as_variable(self.lan_eigrp_auto_syst_id))
+
+    def configure_address_family(self, values: dict) -> None:
+        address_family = values.get("address_family", [])  # feature template sends list instead of dict
+        if not address_family:
+            return
+        address_family = address_family[0]
+        values["address_family"] = AddressFamily(
+            redistribute=self._set_redistribute(address_family),
+            network=self._set_adress_family_addresses(address_family),
+        )
+
+    def _set_adress_family_addresses(self, values: dict) -> List[SummaryAddress]:
+        summary_address = values.get("network", [])
+        if not summary_address:
+            return [
+                SummaryAddress(
+                    prefix=Prefix(
+                        address=as_variable(self.lan_eigrp_addr_fami_netw_1_ip),
+                        mask=as_variable(self.lan_eigrp_addr_fami_netw_1_mask),
+                    )
+                )
+            ]
+        return [self._set_summary_address(addr) for addr in summary_address]
+
+    def _set_redistribute(self, values: dict) -> Optional[List[RedistributeIntoEigrp]]:
+        redistributes = values.get("topology", {}).get("base", {}).get("redistribute", [])
+        if not redistributes:
+            return None
+        return [
+            RedistributeIntoEigrp(
+                protocol=as_global(redistribute["protocol"].value, RedistributeProtocol),
+                # route_policy=redistribute.get("route_policy", None),
+                # route polict is represented as a string in feature template and as UUID in model
+            )
+            for redistribute in redistributes
+        ]
+
+    def configure_address_family_interface(self, values: dict) -> None:
+        interfaces = values.get("af_interface", [])
+        if not interfaces:
+            return
+        interfaces_list = []
+        for interface in interfaces:
+            interfaces_list.append(
+                IPv4StaticRoute(
+                    name=interface["name"],
+                    shutdown=interface.get("shutdown", as_default(False)),
+                    summary_address=self._set_summary_addresses(interface),
+                )
+            )
+        values["af_interface"] = interfaces_list
+
+    def _set_summary_addresses(self, values: dict) -> List[SummaryAddress]:
+        summary_address = values.get("summary_address", [])
+        return [self._set_summary_address(addr) for addr in summary_address]
+
+    def _set_summary_address(self, addr: dict) -> SummaryAddress:
+        return SummaryAddress(
+            prefix=Prefix(
+                address=as_global(addr["prefix"].value.network.network_address),
+                mask=as_global(str(addr["prefix"].value.netmask)),
+            )
+        )
+
+    def configure_authentication(self, values: dict) -> None:
+        auth = values.get("authentication", None)
+        if not auth:
+            return
+        values["authentication"] = EigrpAuthentication(
+            auth_type=auth.get("type", Default[None](value=None)),
+            auth_key=auth.get("key", Default[None](value=None)),
+            key=auth.get("keychain", {}).get("key", None),
+            # There should be more keys
+        )
+
+    def configure_table_map(self, values: dict) -> None:
+        table_map = values.get("table_map", None)
+        if not table_map:
+            return
+        values["table_map"] = TableMap(
+            # name=table_map.get("name", Default[None](value=None)), this should be Global[UUID] not Global[int]
+            filter=table_map.get("filter", as_default(False)),
+        )
 
     def cleanup_keys(self, values: dict) -> None:
         for key in self.delete_keys: