Management vpn. API create method. unit test. integration test (#200)

.pre-commit-config.yaml

@@ -33,7 +33,7 @@ repos:
   # Code analysis (only checks staged files)
 
   - repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.2.0
+    rev: v1.9.0
     hooks:
       - id: mypy
         name: mypy

catalystwan/api/feature_profile_api.py

@@ -10,10 +10,12 @@
 from catalystwan.endpoints.configuration.feature_profile.sdwan.other import OtherFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.service import ServiceFeatureProfile
 from catalystwan.endpoints.configuration.feature_profile.sdwan.system import SystemFeatureProfile
+from catalystwan.endpoints.configuration.feature_profile.sdwan.transport import TransportFeatureProfile
 from catalystwan.models.configuration.feature_profile.sdwan.other import AnyOtherParcel
 from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.url import URLParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service import AnyServiceParcel
 from catalystwan.models.configuration.feature_profile.sdwan.service.multicast import MulticastParcel
+from catalystwan.models.configuration.feature_profile.sdwan.transport import AnyTransportParcel
 from catalystwan.typed_list import DataSequence
 
 if TYPE_CHECKING:
@@ -83,6 +85,7 @@ def __init__(self, session: ManagerSession):
         self.system = SystemFeatureProfileAPI(session=session)
         self.other = OtherFeatureProfileAPI(session=session)
         self.service = ServiceFeatureProfileAPI(session=session)
+        self.transport = TransportFeatureProfileAPI(session=session)
 
 
 class FeatureProfileAPI(Protocol):
@@ -135,6 +138,46 @@ def delete(self, fp_id: str) -> None:
         self.endpoint.delete_cli_feature_profile(cli_fp_id=fp_id)
 
 
+class TransportFeatureProfileAPI:
+    """
+    SDWAN Feature Profile Transport APIs
+    """
+
+    def __init__(self, session: ManagerSession):
+        self.session = session
+        self.endpoint = TransportFeatureProfile(session)
+
+    def get_profiles(
+        self, limit: Optional[int] = None, offset: Optional[int] = None
+    ) -> DataSequence[FeatureProfileInfo]:
+        """
+        Get all Transport Feature Profiles
+        """
+        payload = GetFeatureProfilesPayload(limit=limit if limit else None, offset=offset if offset else None)
+
+        return self.endpoint.get_transport_feature_profiles(payload)
+
+    def create_profile(self, name: str, description: str) -> FeatureProfileCreationResponse:
+        """
+        Create Transport Feature Profile
+        """
+        payload = FeatureProfileCreationPayload(name=name, description=description)
+        return self.endpoint.create_transport_feature_profile(payload)
+
+    def delete_profile(self, profile_id: UUID) -> None:
+        """
+        Delete Transport Feature Profile
+        """
+        self.endpoint.delete_transport_feature_profile(profile_id)
+
+    def create_parcel(self, profile_id: UUID, payload: AnyTransportParcel) -> ParcelCreationResponse:
+        """
+        Create Transport Parcel for selected profile_id based on payload type
+        """
+
+        return self.endpoint.create_transport_parcel(profile_id, payload._get_parcel_type(), payload)
+
+
 class OtherFeatureProfileAPI:
     """
     SDWAN Feature Profile System APIs

catalystwan/endpoints/configuration/feature_profile/sdwan/transport.py

@@ -16,8 +16,11 @@
     ParcelId,
     SchemaTypeQuery,
 )
-from catalystwan.models.configuration.feature_profile.sdwan.management.vpn import ManagementVPN
-from catalystwan.models.configuration.feature_profile.sdwan.transport import CellularControllerParcel
+from catalystwan.models.configuration.feature_profile.sdwan.transport import (
+    AnyTransportParcel,
+    CellularControllerParcel,
+)
+from catalystwan.models.configuration.feature_profile.sdwan.transport.vpn_management import ManagementVpn
 from catalystwan.typed_list import DataSequence
 
 
@@ -51,6 +54,13 @@ def edit_transport_feature_profile(
     def delete_transport_feature_profile(self, profile_id: str) -> None:
         ...
 
+    @versions(supported_versions=(">=20.13"), raises=False)
+    @post("/v1/feature-profile/sdwan/transport/{profile_id}/{parcel_type}")
+    def create_transport_parcel(
+        self, profile_id: str, parcel_type: str, payload: AnyTransportParcel
+    ) -> ParcelCreationResponse:
+        ...
+
     #
     # ManagementVPN parcel
     #
@@ -72,7 +82,7 @@ def create_management_vpn_parcel(self, profile_id: str, payload: _ParcelBase) ->
     @versions(supported_versions=(">=20.13"), raises=False)
     @put("/v1/feature-profile/sdwan/transport/{profile_id}/management/vpn/{parcel_id}")
     def edit_management_vpn_parcel(
-        self, profile_id: str, parcel_id: str, payload: ManagementVPN
+        self, profile_id: str, parcel_id: str, payload: ManagementVpn
     ) -> ParcelCreationResponse:
         ...
 

catalystwan/integration_tests/feature_profile/sdwan/base.py

@@ -7,14 +7,15 @@
     OtherFeatureProfileAPI,
     ServiceFeatureProfileAPI,
     SystemFeatureProfileAPI,
+    TransportFeatureProfileAPI,
 )
 from catalystwan.session import ManagerSession, create_manager_session
 
 
 class TestFeatureProfileModels(unittest.TestCase):
     session: ManagerSession
     profile_uuid: UUID
-    api: Union[SystemFeatureProfileAPI, ServiceFeatureProfileAPI, OtherFeatureProfileAPI]
+    api: Union[SystemFeatureProfileAPI, ServiceFeatureProfileAPI, OtherFeatureProfileAPI, TransportFeatureProfileAPI]
 
     @classmethod
     def setUpClass(cls) -> None:

catalystwan/integration_tests/feature_profile/sdwan/test_transport.py

@@ -0,0 +1,78 @@
+from ipaddress import IPv4Address, IPv6Address
+
+from catalystwan.api.configuration_groups.parcel import as_global
+from catalystwan.integration_tests.feature_profile.sdwan.base import TestFeatureProfileModels
+from catalystwan.models.configuration.feature_profile.sdwan.transport.vpn_management import (
+    DnsIpv4,
+    DnsIpv6,
+    Ipv4RouteItem,
+    Ipv6RouteItem,
+    ManagementVpn,
+    NewHostMappingItem,
+    NextHopItem,
+    OneOfIpRouteNull0,
+    Prefix,
+    SubnetMask,
+)
+
+
+class TestTransportFeatureProfileModels(TestFeatureProfileModels):
+    @classmethod
+    def setUpClass(cls) -> None:
+        super().setUpClass()
+        cls.api = cls.session.api.sdwan_feature_profiles.transport
+        cls.profile_uuid = cls.api.create_profile("TestProfileService", "Description").id
+
+    def test_when_fully_specified_management_vpn_parcel_expect_successful_post(self):
+        # Arrange
+        management_vpn_parcel = ManagementVpn(
+            parcel_name="FullySpecifiedManagementVpnParcel",
+            description="Description",
+            dns_ipv6=DnsIpv6(
+                primary_dns_address_ipv6=as_global(IPv6Address("67ca:c2df:edfe:c8ec:b6cb:f9f4:eab0:ece6")),
+                secondary_dns_address_ipv6=as_global(IPv6Address("8989:8d33:c00a:4d13:324d:8b23:8d77:a289")),
+            ),
+            dns_ipv4=DnsIpv4(
+                primary_dns_address_ipv4=as_global(IPv4Address("68.138.29.222")),
+                secondary_dns_address_ipv4=as_global(IPv4Address("122.89.114.112")),
+            ),
+            new_host_mapping=[
+                NewHostMappingItem(
+                    host_name=as_global("FullySpecifiedHost"),
+                    list_of_ip=as_global(
+                        [
+                            "165.16.181.116",
+                            "7a4c:1d87:8587:a6ec:21a6:48a7:00e8:1fef",
+                        ]
+                    ),
+                )
+            ],
+            ipv6_route=[
+                Ipv6RouteItem(
+                    prefix=as_global("0::/16"),
+                    one_of_ip_route=OneOfIpRouteNull0(),
+                )
+            ],
+            ipv4_route=[
+                Ipv4RouteItem(
+                    prefix=Prefix(
+                        ip_address=as_global(IPv4Address("202.153.165.234")),
+                        subnet_mask=as_global("255.255.255.0", SubnetMask),
+                    ),
+                    next_hop=[
+                        NextHopItem(
+                            address=as_global(IPv4Address("1.1.1.1")),
+                        )
+                    ],
+                )
+            ],
+        )
+        # Act
+        parcel_id = self.api.create_parcel(self.profile_uuid, management_vpn_parcel).id
+        # Assert
+        assert parcel_id
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.api.delete_profile(cls.profile_uuid)
+        super().tearDownClass()

catalystwan/models/configuration/feature_profile/sdwan/transport/__init__.py

@@ -7,10 +7,11 @@
 
 from .bgp import WanRoutingBgpParcel as BGPParcel
 from .cellular_controller import CellularControllerParcel
+from .vpn_management import ManagementVpn
 
-AnyTransportParcel = Annotated[Union[BGPParcel, CellularControllerParcel], Field(discriminator="type_")]
+AnyTransportParcel = Annotated[Union[BGPParcel, CellularControllerParcel, ManagementVpn], Field(discriminator="type_")]
 
-__all__ = ["BGPParcel", "CellularControllerParcel", "AnyTransportParcel"]
+__all__ = ["BGPParcel", "CellularControllerParcel", "ManagementVpn", "AnyTransportParcel"]
 
 
 def __dir__() -> "List[str]":