Add converter. Add push. Add tests

catalystwan/api/administration.py

@@ -1,4 +1,4 @@
-# Copyright 2022 Cisco Systems, Inc. and its affiliates
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 
 from __future__ import annotations
 
@@ -34,6 +34,7 @@
     UserUpdateRequest,
 )
 from catalystwan.exceptions import CatalystwanDeprecationWarning, CatalystwanException
+from catalystwan.models.settings import ThreadGridApi
 from catalystwan.typed_list import DataSequence
 from catalystwan.utils.creation_tools import asdict, create_dataclass
 
@@ -384,7 +385,14 @@ def update(self, payload: Vbond) -> bool:
     def update(self, payload: Organization) -> bool:
         ...
 
-    def update(self, payload: Union[Organization, Certificate, Password, Vbond]) -> bool:
+    @overload
+    def update(self, payload: ThreadGridApi) -> bool:
+        ...
+
+    def update(self, payload: Union[Organization, Certificate, Password, Vbond, ThreadGridApi]) -> bool:
+        if isinstance(payload, ThreadGridApi):
+            dataseq = self.__update_thread_grid_api(payload)
+            return True
         json_payload = asdict(payload)  # type: ignore
         if isinstance(payload, Organization):
             response = self.__update_organization(json_payload)
@@ -414,6 +422,9 @@ def __update_vbond(self, payload: dict) -> Response:
         endpoint = "/dataservice/settings/configuration/device"
         return self.session.post(endpoint, json=payload)
 
+    def __update_thread_grid_api(self, payload: ThreadGridApi) -> DataSequence[ThreadGridApi]:
+        return self.session.endpoints.configuration_settings.create_threat_grid_api_key(payload)
+
     @deprecated(
         "Use .endpoints.configuration_settings.edit_organizations() instead", category=CatalystwanDeprecationWarning
     )

catalystwan/endpoints/configuration_settings.py

@@ -1,4 +1,4 @@
-# Copyright 2023 Cisco Systems, Inc. and its affiliates
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 
 # mypy: disable-error-code="empty-body"
 import datetime
@@ -7,6 +7,7 @@
 from pydantic import BaseModel, ConfigDict, Field, IPvAnyAddress, field_validator
 
 from catalystwan.endpoints import JSON, APIEndpoints, get, post, put, view
+from catalystwan.models.settings import ThreadGridApi
 from catalystwan.typed_list import DataSequence
 from catalystwan.utils.session_type import ProviderView, SingleTenantView
 
@@ -718,3 +719,7 @@ def edit_cloud_credentials(self, payload: CloudCredentials) -> DataSequence[Clou
     @post("/settings/configuration/cloudProviderSetting", "data")
     def create_cloud_credentials(self, payload: CloudCredentials) -> DataSequence[CloudCredentials]:
         ...
+
+    @post("/settings/configuration/threatGridApiKey", "data")
+    def create_threat_grid_api_key(self, payload: ThreadGridApi) -> DataSequence[ThreadGridApi]:
+        ...

catalystwan/integration_tests/test_settings.py

@@ -0,0 +1,28 @@
+from typing import Optional
+
+from catalystwan.integration_tests.base import TestCaseBase
+from catalystwan.models.settings import ThreadGridApi
+
+
+class TestThreadGridApi(TestCaseBase):
+    created_thread: Optional[ThreadGridApi] = None
+
+    def test_thread_grid_api(self):
+        # Arrange
+        thread = ThreadGridApi()
+        thread.set_region_api_key("eur", "1234567890")
+        thread.set_region_api_key("nam", "0987654321")
+        # Act
+        response = self.session.endpoints.configuration_settings.create_threat_grid_api_key(thread)
+        self.created_thread = response.single_or_default()
+        # Assert
+        assert self.created_thread is not None
+        assert self.created_thread.entries[0].region == "nam"
+        assert self.created_thread.entries[0].apikey == "0987654321"
+        assert self.created_thread.entries[1].region == "eur"
+        assert self.created_thread.entries[1].apikey == "1234567890"
+
+    def tearDown(self) -> None:
+        if self.created_thread:
+            self.session.endpoints.configuration_settings.create_threat_grid_api_key(ThreadGridApi())
+        return super().tearDown()

catalystwan/models/configuration/config_migration.py

@@ -37,6 +37,7 @@
     LoggingEntry,
     ZoneToNoZoneInternet,
 )
+from catalystwan.models.settings import ThreadGridApi
 from catalystwan.models.templates import FeatureTemplateInformation, TemplateInformation
 from catalystwan.version import parse_api_version
 
@@ -213,6 +214,7 @@ class UX2Config(BaseModel):
     feature_profiles: List[TransformedFeatureProfile] = Field(default_factory=list)
     profile_parcels: List[TransformedParcel] = Field(default_factory=list)
     cloud_credentials: Optional[CloudCredentials] = Field(default=None)
+    thread_grid_api: Optional[ThreadGridApi] = Field(default=None)
 
     @model_validator(mode="before")
     @classmethod
@@ -555,6 +557,7 @@ class PolicyConvertContext:
     security_policy_residues: Dict[UUID, SecurityPolicyResidues] = field(default_factory=dict)
     qos_map_residues: Dict[UUID, List[QoSMapResidues]] = field(default_factory=dict)
     as_path_list_num_mapping: Dict[str, int] = field(default_factory=dict)
+    thread_grid_api: Optional[ThreadGridApi] = None
 
     def get_vpn_id_to_vpn_name_map(self) -> Dict[Union[str, int], List[str]]:
         vpn_map: Dict[Union[str, int], List[str]] = {}

catalystwan/models/settings.py

@@ -0,0 +1,33 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, ConfigDict, Field, SerializationInfo, SerializerFunctionWrapHandler, model_serializer
+
+Region = Literal["nam", "eur"]
+
+
+class ThreadGridApiEntires(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    region: Region
+    apikey: Optional[str] = Field(default="")
+
+
+class ThreadGridApi(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    entries: List[ThreadGridApiEntires] = Field(
+        default_factory=lambda: [
+            ThreadGridApiEntires(region="nam"),
+            ThreadGridApiEntires(region="eur"),
+        ]
+    )
+
+    def set_region_api_key(self, region: Region, apikey: str) -> None:
+        for entry in self.entries:
+            if entry.region == region:
+                entry.apikey = apikey
+                return
+        raise ValueError(f"Region {region} not found in ThreadGridApi")
+
+    @model_serializer(mode="wrap")
+    def envelope_data(self, handler: SerializerFunctionWrapHandler, info: SerializationInfo) -> Dict[str, Any]:
+        return {"data": [handler(self)]}

catalystwan/tests/config_migration/policy_converters/test_thread_grid_api.py

@@ -0,0 +1,31 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
+import unittest
+
+from catalystwan.models.configuration.config_migration import PolicyConvertContext
+from catalystwan.models.policy.list.threat_grid_api_key import ThreatGridApiKeyEntry, ThreatGridApiKeyList
+from catalystwan.utils.config_migration.converters.policy.policy_lists import convert
+
+
+class TestThreadGridApiConverter(unittest.TestCase):
+    def setUp(self) -> None:
+        self.context = PolicyConvertContext()
+
+    def test_thread_grid_api_conversion(self):
+        # Arrange
+        policy = ThreatGridApiKeyList(
+            name="ThreatGridApiKeyList",
+            description="ThreatGridApiKeyList",
+            entries=[
+                ThreatGridApiKeyEntry(region="eur", api_key="123"),
+                ThreatGridApiKeyEntry(region="nam", api_key="456"),
+            ],
+        )
+        # Act -- This action adds object to the context
+        convert(policy, context=self.context)
+        thread = self.context.thread_grid_api
+        # Assert
+        assert len(thread.entries) == 2
+        assert thread.entries[0].region == "nam"
+        assert thread.entries[0].apikey == "456"
+        assert thread.entries[1].region == "eur"
+        assert thread.entries[1].apikey == "123"

catalystwan/utils/config_migration/converters/policy/policy_lists.py

@@ -1,3 +1,4 @@
+# Copyright 2024 Cisco Systems, Inc. and its affiliates
 from logging import getLogger
 from re import match
 from typing import Any, Callable, Dict, List, Mapping, Type, TypeVar, cast
@@ -72,7 +73,9 @@
 from catalystwan.models.policy.list.local_app import LocalAppList
 from catalystwan.models.policy.list.region import RegionList, RegionListInfo
 from catalystwan.models.policy.list.site import SiteList, SiteListInfo
+from catalystwan.models.policy.list.threat_grid_api_key import ThreatGridApiKeyList
 from catalystwan.models.policy.list.vpn import VPNList, VPNListInfo
+from catalystwan.models.settings import ThreadGridApi
 
 logger = getLogger(__name__)
 
@@ -420,6 +423,14 @@ def local_app_list(in_: LocalAppList, context: PolicyConvertContext) -> ConvertR
     return ConvertResult[SecurityApplicationListParcel](output=out, status="complete")
 
 
+def threat_grid_api(in_: ThreatGridApiKeyList, context: PolicyConvertContext) -> ConvertResult[None]:
+    out = ThreadGridApi()
+    for entry in in_.entries:
+        out.set_region_api_key(region=entry.region, apikey=entry.api_key)
+    context.thread_grid_api = out
+    return ConvertResult[None](status="complete")
+
+
 OPL = TypeVar("OPL", AnyPolicyObjectParcel, None)
 Input = AnyPolicyList
 Output = ConvertResult[OPL]
@@ -452,6 +463,7 @@ def local_app_list(in_: LocalAppList, context: PolicyConvertContext) -> ConvertR
     SiteList: site,
     SLAClassList: sla_class,
     TLOCList: tloc,
+    ThreatGridApiKeyList: threat_grid_api,
     URLAllowList: url_allow,
     URLBlockList: url_block,
     VPNList: vpn,

catalystwan/utils/config_migration/creators/config_pusher.py

@@ -79,6 +79,7 @@ def _create_config_map(self, ux2_config: UX2Config) -> ConfigurationMapping:
 
     def push(self) -> UX2ConfigPushResult:
         self._create_cloud_credentials()
+        self._create_thread_grid_api()
         self._create_config_groups()
         self._groups_of_interests_pusher.push()
         self._localized_policy_feature_pusher.push()
@@ -99,6 +100,15 @@ def _create_cloud_credentials(self):
         except ManagerHTTPError as e:
             logger.error(f"Error occured during credentials migration: {e}")
 
+    def _create_thread_grid_api(self):
+        thread_grid_api = self._ux2_config.thread_grid_api
+        if thread_grid_api is None:
+            return
+        try:
+            self._session.api.administration_settings.update(thread_grid_api)
+        except ManagerHTTPError as e:
+            logger.error(f"Error occured during thread grid api migration: {e}")
+
     def _create_config_groups(self):
         config_groups = self._ux2_config.config_groups
         config_groups_length = len(config_groups)

catalystwan/workflows/config_migration.py

@@ -545,6 +545,9 @@ def transform(ux1: UX1Config, add_suffix: bool = False) -> ConfigTransformResult
         )
     )
 
+    # Add additional objects emmited by the conversion
+    ux2.thread_grid_api = policy_context.thread_grid_api
+
     ux2 = merge_parcels(ux2)
     transform_result.ux2_config = ux2
     if add_suffix: