CCLA-1180: sdk methods for tunnel crud operation

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()

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()

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