diff options
Diffstat (limited to 'tests/test_clamp.py')
| -rw-r--r-- | tests/test_clamp.py | 532 |
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}")) |