From 971259f4054281a5e8ba80c3f6f4ca00cf6de13e Mon Sep 17 00:00:00 2001 From: PrasadhNanjundan Date: Thu, 6 Nov 2025 16:54:22 +0530 Subject: [PATCH] CCLA-1180: sdk methods for tunnel crud operation --- msa_sdk/pops.py | 81 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_pops.py | 84 ++++++++++++++++++++++++++++++++++++++++++++++ tests/util.py | 12 +++++++ 3 files changed, 177 insertions(+) create mode 100644 tests/test_pops.py diff --git a/msa_sdk/pops.py b/msa_sdk/pops.py index 17f15700..c3be692b 100644 --- a/msa_sdk/pops.py +++ b/msa_sdk/pops.py @@ -1,4 +1,5 @@ """Module Pops.""" +import json from msa_sdk.msa_api import MSA_API @@ -45,4 +46,84 @@ def remove_pop(self, entity_type, vendor, name): """ self.action = 'Remove Pop' self.path = '{}?entityType={}&vendor={}&name={}'.format(self.api_path, entity_type, vendor, name) + self._call_delete() + + def save_tunnel(self, data: dict) -> None: + """ + Create/Save a tunnel. + + Parameters + ---------- + data : create tunnel data in json + + Returns + ------- + None + """ + self.action = "Save Tunnel" + self.path = '{}/tunnel'.format(self.api_path) + self._call_post(data) + + def update_tunnel(self, cpe_device_id: int, pop_vendor: str, pop_identifier: str, data: dict) -> None: + """ + Update a tunnel identified by (cpeDeviceId, popVendor, popIdentifier). + + Parameters + ---------- + cpe_device_id : int + pop_vendor : str + pop_identifier : str + data : update tunnel data in json + + Returns + ------- + None + """ + self.action = "Update Tunnel" + self.path = '{}/tunnel?cpeDeviceId={}&popVendor={}&popIdentifier={}'.format(self.api_path, cpe_device_id, + pop_vendor, + pop_identifier) + self._call_put(json.dumps(data)) + + def list_tunnels(self, tenant_prefix: str, location_id=None): + """ + Get tunnels in GeoJSON for a tenant prefix (and optional location). + + Parameters + ---------- + tenant_prefix : str + location_id : int + + Returns + ------- + dict + GeoJSON FeatureCollection + """ + self.action = "List Tunnels" + if location_id is not None: + self.path = '{}/tunnels?tenantPrefix={}&locationId={}'.format( + self.api_path, tenant_prefix, location_id + ) + else: + self.path = '{}/tunnels?tenantPrefix={}'.format(self.api_path, tenant_prefix) + return self._call_get() + + def remove_tunnel(self, cpe_device_id: int, pop_vendor: str, pop_identifier: str) -> None: + """ + Delete a tunnel identified by (cpeDeviceId, popVendor, popIdentifier). + + Parameters + ---------- + cpe_device_id : int + pop_vendor : str + pop_identifier : str + + Returns + ------- + None + """ + self.action = "Remove Tunnel" + self.path = '{}/tunnel?cpeDeviceId={}&popVendor={}&popIdentifier={}'.format(self.api_path, cpe_device_id, + pop_vendor, + pop_identifier) self._call_delete() \ No newline at end of file diff --git a/tests/test_pops.py b/tests/test_pops.py new file mode 100644 index 00000000..c0b463d9 --- /dev/null +++ b/tests/test_pops.py @@ -0,0 +1,84 @@ +import json +from unittest.mock import patch + +from util import pops_fixture + + +def test_save_pops_saves_data_correctly(pops_fixture): + pops = pops_fixture + test_data = {"key": "value"} + + with patch('msa_sdk.msa_api.MSA_API._call_post') as mock_call_post: + mock_call_post.return_value.status_code = 200 + pops.save_pops(test_data) + assert pops.path == '/sase/pops' + mock_call_post.assert_called_once_with(test_data) + + +def test_remove_pop_removes_correct_pop(pops_fixture): + pops = pops_fixture + entity_type = "type1" + vendor = "vendor1" + name = "pop1" + + with patch('requests.delete') as mock_call_delete: + mock_call_delete.return_value.status_code = 200 + pops.remove_pop(entity_type, vendor, name) + assert pops.path == '/sase/pops?entityType={}&vendor={}&name={}'.format(entity_type, vendor, name) + mock_call_delete.assert_called_once() + + +def test_save_tunnel_saves_tunnel_data(pops_fixture): + pops = pops_fixture + test_data = {"tunnel": "data"} + + with patch('msa_sdk.msa_api.MSA_API._call_post') as mock_call_post: + mock_call_post.return_value.status_code = 200 + pops.save_tunnel(test_data) + assert pops.path == '/sase/pops/tunnel' + mock_call_post.assert_called_once_with(test_data) + + +def test_update_tunnel_updates_correct_tunnel(pops_fixture): + pops = pops_fixture + cpe_device_id = 123 + pop_vendor = "vendor1" + pop_identifier = "identifier1" + test_data = {"update": "data"} + + with patch('msa_sdk.msa_api.MSA_API._call_put') as mock_call_put: + mock_call_put.return_value.status_code = 200 + pops.update_tunnel(cpe_device_id, pop_vendor, pop_identifier, test_data) + assert pops.path == '/sase/pops/tunnel?cpeDeviceId={}&popVendor={}&popIdentifier={}'.format(cpe_device_id, + pop_vendor, + pop_identifier) + mock_call_put.assert_called_once_with(json.dumps(test_data)) + + +def test_list_tunnels_returns_tunnels(pops_fixture): + pops = pops_fixture + tenant_prefix = "tenant1" + location_id = 456 + test_response = {"tunnels": []} + + with patch('msa_sdk.msa_api.MSA_API._call_get') as mock_call_get: + mock_call_get.return_value = test_response + result = pops.list_tunnels(tenant_prefix, location_id) + assert pops.path == '/sase/pops/tunnels?tenantPrefix={}&locationId={}'.format( tenant_prefix, location_id) + assert result == test_response + mock_call_get.assert_called_once() + + +def test_remove_tunnel_removes_correct_tunnel(pops_fixture): + pops = pops_fixture + cpe_device_id = 123 + pop_vendor = "vendor1" + pop_identifier = "identifier1" + + with patch('requests.delete') as mock_call_delete: + mock_call_delete.return_value.status_code = 200 + pops.remove_tunnel(cpe_device_id, pop_vendor, pop_identifier) + assert pops.path == '/sase/pops/tunnel?cpeDeviceId={}&popVendor={}&popIdentifier={}'.format(cpe_device_id, + pop_vendor, + pop_identifier) + mock_call_delete.assert_called_once() diff --git a/tests/util.py b/tests/util.py index ff9dd7df..90413419 100644 --- a/tests/util.py +++ b/tests/util.py @@ -12,6 +12,7 @@ from msa_sdk.orchestration import Orchestration from msa_sdk.order import Order from msa_sdk.order_stack import OrderStack +from msa_sdk.pops import Pops from msa_sdk.profile import Profile from msa_sdk.repository import Repository @@ -287,3 +288,14 @@ def conf_backup_fixture(): mock_host_port.return_value = ('api_hostname', '8080') conf_backup = ConfBackup() return conf_backup + +@pytest.fixture +def pops_fixture(): + """Pops fixture.""" + with patch('requests.post') as mock_post: + mock_post.return_value.json.return_value = {'token': '12345qwert'} + + with patch('msa_sdk.msa_api.host_port') as mock_host_port: + mock_host_port.return_value = ('api_hostname', '8080') + pops = Pops() + return pops