Release ONAP SDKv10.2

Issue-ID: INT-2150
Signed-off-by: Michal Jagiello <michal.jagiello@t-mobile.pl>
Change-Id: I650047c599a5aae6de7c6b42d38e34aea88578e2

1 files changed, 532 insertions, 0 deletions

diff --git a/tests/test_clamp.py b/tests/test_clamp.py
new file mode 100644
index 0000000..5768341
--- /dev/null
+++ b/tests/test_clamp.py
@@ -0,0 +1,532 @@
+"""Test clamp module."""
+#   Copyright 2022 Orange, Deutsche Telekom AG
+#
+#   Licensed under the Apache License, Version 2.0 (the "License");
+#   you may not use this file except in compliance with the License.
+#   You may obtain a copy of the License at
+#
+#       http://www.apache.org/licenses/LICENSE-2.0
+#
+#   Unless required by applicable law or agreed to in writing, software
+#   distributed under the License is distributed on an "AS IS" BASIS,
+#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#   See the License for the specific language governing permissions and
+#   limitations under the License.
+
+from unittest import mock
+import pytest
+
+from onapsdk.clamp.clamp_element import Clamp
+from onapsdk.clamp.loop_instance import LoopInstance
+from onapsdk.exceptions import ParameterError, ResourceNotFound
+from onapsdk.sdc.service import Service
+
+#examples
+TEMPLATES = [
+    {
+        "name" : "test_template",
+        "modelService" : {
+            "serviceDetails" : {
+                "name" : "test"
+            }
+        }
+    }
+]
+
+POLICIES = [
+    {
+        "policyModelType" : "onap.policies.controlloop.Test",
+        "version" : "1.0.0",
+        "policyAcronym" : "Test",
+        "createdDate" : "2020-04-30T09:03:30.362897Z",
+        "updatedDate" : "2020-04-30T09:03:30.362897Z",
+        "updatedBy" : "Not found",
+        "createdBy" : "Not found"
+    }
+]
+
+LOOP_DETAILS = {
+    "name" : "LOOP_test",
+    "globalPropertiesJson": {
+        "dcaeDeployParameters" : {
+            "uniqueBlueprintParameters" : {
+                "policy_id" : "Microservice12345"
+            }
+        }
+    },
+    "components" : {
+        "POLICY" : {
+            "componentState" : {
+                "stateName" : "UNKNOWN"
+            }
+        },
+        "DCAE" : {
+            "componentState" : {
+                "stateName" : "BLUEPRINT_DEPLOYED"
+            }
+        }
+    },
+    "modelService" : {
+        "resourceDetails": {
+            "VFModule" : {
+                "resourceID" : {
+                    "vfModuleModelName" : "resourceID",
+                    "vfModuleModelInvariantUUID" : "InvariantUUID",
+                    "vfModuleModelUUID" : "UUID",
+                    "vfModuleModelVersion" : "1.0",
+                    "vfModuleModelCustomizationUUID" : "CustomizationUUID"
+                }
+            }
+        }
+    },
+    "operationalPolicies" : [
+        {
+            "name" : "MICROSERVICE_test"
+        }
+    ],
+    "microServicePolicies" : [
+        {
+            "name" : "MICROSERVICE_test"
+        }
+    ]
+}
+
+#for policy deploy to policy engine
+SUBMITED_POLICY = {
+        "components" : {
+        "POLICY" : {
+            "componentState" : {
+                "stateName" : "SENT_AND_DEPLOYED"
+            }
+        }
+    }
+}
+
+NOT_SUBMITED_POLICY = {
+        "components" : {
+        "POLICY" : {
+            "componentState" : {
+                "stateName" : "SENT"
+            }
+        }
+    }
+}
+
+#for the deploy to DCAE
+SUBMITED = {
+        "components" : {
+        "DCAE" : {
+            "componentState" : {
+                "stateName" : "MICROSERVICE_INSTALLED_SUCCESSFULLY"
+            }
+        }
+    }
+}
+
+NOT_SUBMITED = {
+        "components" : {
+        "DCAE" : {
+            "componentState" : {
+                "stateName" : "MICROSERVICE_INSTALLATION_FAILED"
+            }
+        }
+    }
+}
+#end of examples
+
+
+def test_initialization():
+    """Class initialization test."""
+    clamp = Clamp()
+    assert isinstance(clamp, Clamp)
+
+
+@mock.patch.object(Clamp, 'send_message_json')
+def test_check_loop_template(mock_send_message_json):
+    """Test Clamp's class method."""
+    svc = Service(name='test')
+    mock_send_message_json.return_value = TEMPLATES
+    template = Clamp.check_loop_template(service=svc)
+    mock_send_message_json.assert_called_once_with('GET',
+                                                   'Get Loop Templates',
+                                                   (f"{Clamp.base_url()}/templates/"))
+    assert template == "test_template"
+
+
+@mock.patch.object(Clamp, 'send_message_json')
+def test_check_loop_template_none(mock_send_message_json):
+    """Test Clamp's class method."""
+    svc = Service(name='test')
+    mock_send_message_json.return_value = {}
+    with pytest.raises(ResourceNotFound) as exc:
+        template = Clamp.check_loop_template(service=svc)
+        assert template is None
+    assert exc.type is ResourceNotFound
+
+
+@mock.patch.object(Clamp, 'send_message_json')
+def test_check_policies(mock_send_message_json):
+    mock_send_message_json.return_value = POLICIES
+    exists = Clamp.check_policies(policy_name="Test", req_policies=1)
+    mock_send_message_json.\
+        assert_called_once_with('GET',
+                                'Get stocked policies',
+                                (f"{Clamp.base_url()}/policyToscaModels/"))
+    assert exists
+
+
+@mock.patch.object(Clamp, 'send_message_json')
+def test_check_policies_none(mock_send_message_json):
+    mock_send_message_json.return_value = POLICIES
+    exists = Clamp.check_policies(policy_name="Test")
+    mock_send_message_json.\
+            assert_called_once_with('GET',
+                                    'Get stocked policies',
+                                    (f"{Clamp.base_url()}/policyToscaModels/"))
+    assert not exists
+
+
+def test_cl_initialization():
+    """Class initialization test."""
+    loop = LoopInstance(template="template", name="LOOP_name", details={})
+    assert isinstance(loop, LoopInstance)
+
+
+@mock.patch.object(LoopInstance, '_update_loop_details')
+def test_details(mock_update):
+    """Test loop instace details gette."""
+    loop = LoopInstance(template="template", name="LOOP_name", details={})
+    mock_update.return_value = {"name" : "test"}
+    details = loop.details
+    assert details == {}
+
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_update_loop_details(mock_send_message_json):
+    """Test Loop instance methode."""
+    loop = LoopInstance(template="template", name="test", details={})
+    mock_send_message_json.return_value = LOOP_DETAILS
+    loop.details = loop._update_loop_details()
+    mock_send_message_json.assert_called_once_with('GET', 'Get loop details',
+         (f"{loop.base_url()}/loop/LOOP_test"))
+    assert loop.details == LOOP_DETAILS
+
+
+@mock.patch('time.sleep', return_value=False)
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_refresh_status(mock_send_message_json,mock_timer):
+    """Test Loop instance methode."""
+    loop = LoopInstance(template="template", name="test", details={})
+    mock_send_message_json.return_value = LOOP_DETAILS
+    loop.refresh_status()
+    mock_send_message_json.assert_called_once_with('GET', 'Get loop status',
+         (f"{loop.base_url()}/loop/getstatus/LOOP_test"))
+    assert loop.details == LOOP_DETAILS
+
+
+def test_validate():
+    """Test Loop instance details validation."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    valid = loop.validate_details()
+    assert  valid
+
+
+def test_validate_details():
+    """Test Loop instance details validation."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    loop.details = {"test":"test"}
+    valid = loop.validate_details()
+    assert  not valid
+
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_create(mock_send_message_json):
+    """Test Loop instance creation."""
+    instance = LoopInstance(template="template", name="test", details={})
+    mock_send_message_json.return_value = LOOP_DETAILS
+    instance.create()
+    mock_send_message_json.assert_called_once_with('POST', 'Create Loop Instance',
+         (f"{instance.base_url()}/loop/create/LOOP_test?templateName=template"))
+    assert instance.name == "LOOP_test"
+    assert len(instance.details["microServicePolicies"]) > 0
+
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_add_operational_policy(mock_send_message_json):
+    """Test adding an op policy."""
+    loop = LoopInstance(template="template", name="test", details={})
+    loop.details = {
+        "name" : "LOOP_test",
+        "operationalPolicies" : None,
+        "microServicePolicies" : [
+            {
+                "name" : "MICROSERVICE_test"
+            }
+        ]
+    }
+    mock_send_message_json.return_value = LOOP_DETAILS
+    loop.add_operational_policy(policy_type="FrequencyLimiter", policy_version="1.0.0")
+    mock_send_message_json.assert_called_once_with('PUT', 'Create Operational Policy',
+        (f"{loop.base_url()}/loop/addOperationaPolicy/{loop.name}/policyModel/FrequencyLimiter/1.0.0"))
+    assert loop.name == "LOOP_test"
+    assert len(loop.details["operationalPolicies"]) > 0
+
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_not_add_operational_policy_parameter_error(mock_send_message_json):
+    """Test adding an op policy - mistaken policy version."""
+    loop = LoopInstance(template="template", name="test", details={})
+    loop.details = {
+        "name" : "LOOP_test",
+        "operationalPolicies" : [],
+        "microServicePolicies" : [
+            {
+                "name" : "MICROSERVICE_test"
+            }
+        ]
+    }
+    with pytest.raises(ParameterError) as exc:
+        mock_send_message_json.return_value = loop.details
+        loop.add_operational_policy(policy_type="FrequencyLimiter", policy_version="not_correct")
+        mock_send_message_json.assert_called_once_with('PUT', 'Create Operational Policy',
+            (f"{loop.base_url()}/loop/addOperationaPolicy/{loop.name}/policyModel/FrequencyLimiter/not_correct"))
+        assert len(loop.details["operationalPolicies"]) == 0
+        assert exc.type is ParameterError
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_add_operational_policy_key_parameter_error(mock_send_message_json):
+    """Test adding an op policy - key doesn't exist."""
+    loop = LoopInstance(template="template", name="test", details={})
+    loop.details = {}
+    with pytest.raises(ParameterError) as exc:
+        mock_send_message_json.return_value = loop.details
+        loop.add_operational_policy(policy_type="FrequencyLimiter", policy_version="not_correct")
+        mock_send_message_json.assert_called_once_with('PUT', 'Create Operational Policy',
+                (f"{loop.base_url()}/loop/addOperationaPolicy/{loop.name}/policyModel/FrequencyLimiter/not_correct"))
+        assert exc.type is ParameterError
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_add_operational_policy_condition_parameter_error(mock_send_message_json):
+    """Test adding an op policy - response cintains more policies."""
+
+    key = "operationalPolicies"
+
+    response_policies = ["one"]           # N policies
+    current_policies = ["one", "two"]   # N+1 policies
+
+    details = {key: current_policies}
+    response = {key: response_policies}
+
+    loop = LoopInstance(template="template", name="test", details=details)
+
+    assert len(response_policies) < len(current_policies)  # raising condition
+    with pytest.raises(ParameterError) as exc:
+        mock_send_message_json.return_value = response
+        loop.add_operational_policy(policy_type="FrequencyLimiter", policy_version="not_correct")
+    assert exc.type is ParameterError
+
+
+@mock.patch.object(LoopInstance, 'send_message_json')
+def test_remove_operational_policy(mock_send_message_json):
+    """Test remove an op policy."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message_json.return_value = {
+        "name" : "LOOP_test",
+        "operationalPolicies" : [],
+        "microServicePolicies" : [
+            {
+                "name" : "MICROSERVICE_test"
+            }
+        ]
+    }
+    loop.remove_operational_policy(policy_type="FrequencyLimiter", policy_version="1.0.0")
+    mock_send_message_json.assert_called_once_with('PUT', 'Remove Operational Policy',
+        (f"{loop.base_url()}/loop/removeOperationaPolicy/{loop.name}/policyModel/FrequencyLimiter/1.0.0"))
+    assert len(loop.details["operationalPolicies"]) == 0
+
+
+@mock.patch.object(LoopInstance, 'send_message')
+def test_update_microservice_policy(mock_send_message):
+    """Test Loop Instance add TCA configuration."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message.return_value = True
+    loop.update_microservice_policy()
+    mock_send_message.assert_called_once()
+    method, description, url = mock_send_message.call_args[0]
+    assert method == "POST"
+    assert description == "ADD TCA config"
+    assert url == (f"{loop.base_url()}/loop/updateMicroservicePolicy/{loop.name}")
+
+
+@mock.patch.object(LoopInstance, 'send_message')
+def test_update_microservice_policy_none(mock_send_message):
+    """Test Loop Instance add TCA configuration."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message.return_value = False
+    loop.update_microservice_policy()
+    mock_send_message.assert_called_once()
+
+
+def test_extract_operational_policy_name():
+    """Test Loop Instance extract operational policy name."""
+    loop = LoopInstance(template="template", name="test", details={})
+    loop.details = {"operationalPolicies":[{"name":"test","policyModel":{"policyAcronym":"Drools"}}]}
+    policy_name = loop.extract_operational_policy_name(policy_type="Drools")
+    assert policy_name=='test'
+
+
+def test_extract_none():
+    """Test Loop Instance extract operational policy name."""
+    loop = LoopInstance(template="template", name="test", details={})
+    loop.details = {"operationalPolicies":[]}
+    with pytest.raises(ParameterError) as exc:
+        policy_name = loop.extract_operational_policy_name(policy_type="Drools")
+        assert policy_name == None
+    assert exc.type is ParameterError
+
+
+@mock.patch.object(LoopInstance, 'extract_operational_policy_name')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_add_drools_policy_config(mock_send_message, mock_extract):
+    """Test Loop Instance add op policy configuration."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message.return_value = True
+    loop.add_op_policy_config(loop.add_drools_conf)
+    mock_send_message.assert_called_once()
+    method, description, url = mock_send_message.call_args[0]
+    assert method == "POST"
+    assert description == "ADD operational policy config"
+    assert url == (f"{loop.base_url()}/loop/updateOperationalPolicies/{loop.name}")
+
+
+@mock.patch.object(LoopInstance, 'extract_operational_policy_name')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_add_minmax_config(mock_send_message, mock_extract):
+    """Test Loop Instance add op policy configuration."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message.return_value = True
+    loop.add_op_policy_config(loop.add_minmax_config)
+    mock_send_message.assert_called_once()
+    method, description, url = mock_send_message.call_args[0]
+    assert method == "POST"
+    assert description == "ADD operational policy config"
+    assert url == (f"{loop.base_url()}/loop/updateOperationalPolicies/{loop.name}")
+
+
+@mock.patch.object(LoopInstance, 'extract_operational_policy_name')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_add_frequency_policy_config(mock_send_message, mock_extract):
+    """Test Loop Instance add op policy configuration."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message.return_value = True
+    loop.add_op_policy_config(loop.add_frequency_limiter)
+    mock_send_message.assert_called_once()
+    method, description, url = mock_send_message.call_args[0]
+    assert method == "POST"
+    assert description == "ADD operational policy config"
+    assert url == (f"{loop.base_url()}/loop/updateOperationalPolicies/{loop.name}")
+
+@mock.patch.object(LoopInstance, 'send_message')
+@mock.patch.object(LoopInstance, 'add_minmax_config')
+@mock.patch.object(LoopInstance, 'add_frequency_limiter')
+def test_add_two_policies_config(mock_freq, mock_min, mock_send_message):
+    """Test Loop Instance add op policy configuration."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_min.return_value = '[{"test1":"test1"}]'
+    mock_freq.return_value = '[{"test2":"test2"}]'
+    loop.add_op_policy_config(loop.add_minmax_config)
+    mock_min.assert_called_once()
+    mock_send_message.assert_called_once()
+    loop.add_op_policy_config(loop.add_frequency_limiter)
+    mock_freq.assert_called_once()
+    assert loop.operational_policies == '[{"test1":"test1"},{"test2":"test2"}]'
+
+
+@mock.patch.object(LoopInstance, 'refresh_status')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_submit_policy(mock_send_message, mock_refresh):
+    """Test submit policies to policy engine."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    action = loop.act_on_loop_policy(loop.submit)
+    mock_send_message.assert_called_once_with('PUT',
+                                            'submit policy',
+                                            (f"{loop.base_url()}/loop/submit/LOOP_test"))
+    mock_refresh.assert_called_once()
+    loop.details = SUBMITED_POLICY
+    assert loop.details["components"]["POLICY"]["componentState"]["stateName"] == "SENT_AND_DEPLOYED"
+
+
+@mock.patch.object(LoopInstance, 'refresh_status')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_stop_policy(mock_send_message, mock_refresh):
+    """Test submit policies to policy engine."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    action = loop.act_on_loop_policy(loop.stop)
+    mock_send_message.assert_called_once_with('PUT',
+                                            'stop policy',
+                                            (f"{loop.base_url()}/loop/stop/LOOP_test"))
+    mock_refresh.assert_called_once()
+    loop.details = {"components":{"POLICY":{"componentState":{"stateName":"SENT"}}}}
+    assert loop.details["components"]["POLICY"]["componentState"]["stateName"] == "SENT"
+
+
+@mock.patch.object(LoopInstance, 'refresh_status')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_restart_policy(mock_send_message, mock_refresh):
+    """Test submit policies to policy engine."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    action = loop.act_on_loop_policy(loop.restart)
+    mock_send_message.assert_called_once_with('PUT',
+                                            'restart policy',
+                                            (f"{loop.base_url()}/loop/restart/LOOP_test"))
+    mock_refresh.assert_called_once()
+    loop.details = SUBMITED_POLICY
+    assert loop.details["components"]["POLICY"]["componentState"]["stateName"] == "SENT_AND_DEPLOYED"
+
+
+@mock.patch.object(LoopInstance, 'refresh_status')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_not_submited_policy(mock_send_message, mock_refresh):
+    """Test submit policies to policy engine."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_refresh.return_value = NOT_SUBMITED_POLICY
+    action = loop.act_on_loop_policy(loop.submit)
+    mock_send_message.assert_called_once_with('PUT',
+                                            'submit policy',
+                                            (f"{loop.base_url()}/loop/submit/LOOP_test"))
+    mock_refresh.assert_called_once()
+    loop.details = NOT_SUBMITED_POLICY
+    assert loop.details["components"]["POLICY"]["componentState"]["stateName"] == "SENT"
+
+
+@mock.patch('time.sleep', return_value=False)
+@mock.patch.object(LoopInstance, 'send_message_json')
+@mock.patch.object(LoopInstance, 'send_message')
+def test_deploy_microservice_to_dcae(mock_send_message, mock_send_message_json, mock_timer):
+    """Test stop microservice."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    mock_send_message_json.return_value = SUBMITED
+    state = loop.deploy_microservice_to_dcae()
+    mock_send_message.assert_called_once_with('PUT',
+                                            'Deploy microservice to DCAE',
+                                            (f"{loop.base_url()}/loop/deploy/LOOP_test"))
+    assert state
+
+
+@mock.patch.object(LoopInstance, 'send_message')
+def test_undeploy_microservice_from_dcae(mock_send_message):
+    """Test stop microservice."""
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    request = loop.undeploy_microservice_from_dcae()
+    mock_send_message.assert_called_once_with('PUT',
+                                            'Undeploy microservice from DCAE',
+                                            (f"{loop.base_url()}/loop/undeploy/LOOP_test"))
+
+
+@mock.patch.object(LoopInstance, 'send_message')
+def test_delete(mock_send_message):
+    loop = LoopInstance(template="template", name="test", details=LOOP_DETAILS)
+    request = loop.delete()
+    mock_send_message.assert_called_once_with('PUT',
+                                            'Delete loop instance',
+                                            (f"{loop.base_url()}/loop/delete/{loop.name}"))