This repository has been archived by the owner on Nov 21, 2024. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 3
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
503517c
commit 5594193
Showing
7 changed files
with
97 additions
and
206 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
60 changes: 60 additions & 0 deletions
60
catalystwan/tests/config_migration/policy_converters/test_references_updater.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,60 @@ | ||
| # Copyright 2024 Cisco Systems, Inc. and its affiliates | ||
| import unittest | ||
| from typing import Dict | ||
| from uuid import UUID, uuid4 | ||
|
|
||
| from catalystwan.models.configuration.feature_profile.sdwan.policy_object.security.aip import ( | ||
| AdvancedInspectionProfileParcel, | ||
| ) | ||
| from catalystwan.utils.config_migration.creators.references_updater import update_parcel_references | ||
|
|
||
|
|
||
| class TestReferencesUpdater(unittest.TestCase): | ||
| def test_update_parcel_references(self): | ||
| intrusion_prevention_origin_ref = uuid4() | ||
| amp_origin_ref = uuid4() | ||
| url_filtering_origin_ref = uuid4() | ||
|
|
||
| pushed_objects: Dict[UUID, UUID] = { | ||
| intrusion_prevention_origin_ref: uuid4(), | ||
| amp_origin_ref: uuid4(), | ||
| url_filtering_origin_ref: uuid4(), | ||
| } | ||
|
|
||
| aip_parcel = AdvancedInspectionProfileParcel.create( | ||
| "aip1", | ||
| "description1", | ||
| intrusion_prevention=intrusion_prevention_origin_ref, | ||
| advanced_malware_protection=amp_origin_ref, | ||
| url_filtering=url_filtering_origin_ref, | ||
| ) | ||
|
|
||
| updated_parcel = update_parcel_references(aip_parcel, pushed_objects) | ||
|
|
||
| assert updated_parcel is not aip_parcel | ||
| assert updated_parcel.intrusion_prevention.ref_id.value == str(pushed_objects[intrusion_prevention_origin_ref]) | ||
| assert updated_parcel.advanced_malware_protection.ref_id.value == str(pushed_objects[amp_origin_ref]) | ||
| assert updated_parcel.url_filtering.ref_id.value == str(pushed_objects[url_filtering_origin_ref]) | ||
| assert updated_parcel.ssl_decryption_profile is None | ||
| assert updated_parcel.tls_decryption_action == aip_parcel.tls_decryption_action | ||
| assert updated_parcel.parcel_name == aip_parcel.parcel_name | ||
| assert updated_parcel.parcel_description == aip_parcel.parcel_description | ||
|
|
||
| def test_not_update_parcel_references(self): | ||
| intrusion_prevention_origin_ref = uuid4() | ||
| amp_origin_ref = uuid4() | ||
| url_filtering_origin_ref = uuid4() | ||
|
|
||
| pushed_objects: Dict[UUID, UUID] = {} | ||
|
|
||
| aip_parcel = AdvancedInspectionProfileParcel.create( | ||
| "aip1", | ||
| "description1", | ||
| intrusion_prevention=intrusion_prevention_origin_ref, | ||
| advanced_malware_protection=amp_origin_ref, | ||
| url_filtering=url_filtering_origin_ref, | ||
| ) | ||
|
|
||
| updated_parcel = update_parcel_references(aip_parcel, pushed_objects) | ||
|
|
||
| assert updated_parcel is aip_parcel |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
47 changes: 21 additions & 26 deletions
47
catalystwan/utils/config_migration/creators/references_updater.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,35 +1,30 @@ | ||
| from abc import ABC, abstractmethod | ||
| from dataclasses import dataclass | ||
| from typing import Dict, Mapping, Type, Union, cast | ||
| import logging | ||
| from typing import Dict, TypeVar | ||
| from uuid import UUID | ||
|
|
||
| from catalystwan.models.configuration.feature_profile.parcel import AnyParcel | ||
| from catalystwan.utils.config_migration.converters.exceptions import CatalystwanConverterCantConvertException | ||
| from pydantic import BaseModel, ValidationError | ||
| from pydantic_core import from_json | ||
|
|
||
| logger = logging.getLogger(__name__) | ||
|
|
||
| @dataclass | ||
| class ReferencesUpdater(ABC): | ||
| parcel: AnyParcel | ||
| pushed_objects_map: Dict[UUID, UUID] | ||
| T = TypeVar("T", bound=BaseModel) | ||
|
|
||
| @abstractmethod | ||
| def update_references(self): | ||
| pass | ||
|
|
||
| def get_target_uuid(self, origin_uuid: Union[str, UUID]) -> UUID: | ||
| _origin_uuid: UUID = cast(UUID, UUID(origin_uuid) if type(origin_uuid) is str else origin_uuid) | ||
| if target_uuid := self.pushed_objects_map.get(_origin_uuid): | ||
| return target_uuid | ||
| def update_parcel_references(parcel: T, uuid_map: Dict[UUID, UUID]) -> T: | ||
| t = type(parcel) | ||
| origin_dump = target_dump = parcel.model_dump_json(by_alias=True) | ||
| pattern = '"{}"' | ||
|
|
||
| raise CatalystwanConverterCantConvertException( | ||
| f"Cannot find transferred policy object based on v1 API id: {_origin_uuid}" | ||
| ) | ||
| for origin_uuid, target_uuid in uuid_map.items(): | ||
| origin_uuid_str = pattern.format(str(origin_uuid)) | ||
| target_uuid_str = pattern.format(str(target_uuid)) | ||
| target_dump = target_dump.replace(origin_uuid_str, target_uuid_str) | ||
|
|
||
| if origin_dump == target_dump: | ||
| return parcel | ||
|
|
||
| def update_parcels_references( | ||
| parcel: AnyParcel, | ||
| pushed_objects_map: Dict[UUID, UUID], | ||
| updaters_map: Mapping[Type[AnyParcel], Type[ReferencesUpdater]], | ||
| ) -> None: | ||
| if ref_updater := updaters_map.get(type(parcel)): | ||
| ref_updater(parcel, pushed_objects_map=pushed_objects_map).update_references() | ||
| try: | ||
| return t.model_validate(from_json(target_dump)) | ||
| except ValidationError as e: | ||
| logging.error(f"Cannot validate model after references update: {e}") | ||
| raise e |
Oops, something went wrong.