diff options
Diffstat (limited to 'tests/test_msb_k8s.py')
-rw-r--r-- | tests/test_msb_k8s.py | 378 |
1 files changed, 378 insertions, 0 deletions
diff --git a/tests/test_msb_k8s.py b/tests/test_msb_k8s.py new file mode 100644 index 0000000..37be189 --- /dev/null +++ b/tests/test_msb_k8s.py @@ -0,0 +1,378 @@ +# 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 + +from onapsdk.msb.k8s import Definition, ConnectivityInfo, Instance + + +CONNECTIVITY_INFO = { + "cloud-region": "test_cloud_region", + "cloud-owner": "test_cloud_owner", + "other-connectivity-list": {}, + "kubeconfig": "test_kubeconfig" +} + + +DEFINITION = { + "rb-name": "test_rb_name_0", + "rb-version": "test_rb_version_0" +} + + +DEFINITIONS = [ + DEFINITION, + { + "rb-name": "test_rb_name_1", + "rb-version": "test_rb_version_1", + "chart-name": "test_chart_name_1", + "description": "test_description_1", + "labels": {} + } +] + + +PROFILE = { + "rb-name": "test_rb_name", + "rb-version": "test_rb_version", + "profile-name": "test_profile_name", + "namespace": "test_namespace" +} + + +PROFILES = [ + PROFILE, + { + "rb-name": "test_rb_name_1", + "rb-version": "test_rb_version_1", + "profile-name": "test_profile_name_1", + "namespace": "test_namespace_1" + } +] + + +CONFIGURATION_TEMPLATE = { + "template-name": "test_configuration_template_name", + "description": "test_configuration_template_description" +} + + +CONFIGURATION_TEMPLATES = [ + CONFIGURATION_TEMPLATE, + { + "template-name": "test_configuration_template_name_0" + } +] + + +INSTANCE = { + "id": "ID_GENERATED_BY_K8SPLUGIN", + "namespace": "NAMESPACE_WHERE_INSTANCE_HAS_BEEN_DEPLOYED_AS_DERIVED_FROM_PROFILE", + "release-name": "RELEASE_NAME_AS_COMPUTED_BASED_ON_INSTANTIATION_REQUEST_AND_PROFILE_DEFAULT", + "request": { + "rb-name": "test-rbdef", + "rb-version": "v1", + "profile-name": "p1", + "release-name": "release-x", + "cloud-region": "krd", + "override-values": { + "optionalDictOfParameters": "andTheirValues, like", + "global.name": "dummy-name" + }, + "labels": { + "optionalLabelForInternalK8spluginInstancesMetadata": "dummy-value" + }, + }, + "resources": [ + { + "GVK": { + "Group": "", + "Kind": "ConfigMap", + "Version": "v1" + }, + "Name": "test-cm" + }, + { + "GVK": { + "Group": "", + "Kind": "Service", + "Version": "v1" + }, + "Name": "test-svc" + }, + { + "GVK": { + "Group": "apps", + "Kind": "Deployment", + "Version": "v1" + }, + "Name": "test-dep" + } + ] +} + + +INSTANCES = [ + INSTANCE +] + + +@mock.patch.object(ConnectivityInfo, "send_message_json") +def test_get_connectivity_info_by_region_id(mock_send_message_json): + mock_send_message_json.return_value = CONNECTIVITY_INFO + conn_info: ConnectivityInfo = ConnectivityInfo.get_connectivity_info_by_region_id("test_cloud_region_id") + assert conn_info.cloud_region_id == "test_cloud_region" + assert conn_info.cloud_owner == "test_cloud_owner" + assert conn_info.other_connectivity_list == {} + assert conn_info.kubeconfig == "test_kubeconfig" + + +@mock.patch.object(ConnectivityInfo, "send_message") +@mock.patch.object(ConnectivityInfo, "send_message_json") +def test_connectivity_info_create_delete(mock_send_message_json, mock_send_message): + mock_send_message_json.return_value = CONNECTIVITY_INFO + conn_info: ConnectivityInfo = ConnectivityInfo.create("test_cloud_region", "test_cloud_owner", b"kubeconfig") + assert conn_info.cloud_region_id == "test_cloud_region" + assert conn_info.cloud_owner == "test_cloud_owner" + assert conn_info.other_connectivity_list == {} + assert conn_info.kubeconfig == "test_kubeconfig" + conn_info.delete() + + +@mock.patch.object(Definition, "send_message_json") +def test_definition_get_all(mock_send_message_json): + mock_send_message_json.return_value = [] + assert len(list(Definition.get_all())) == 0 + + mock_send_message_json.return_value = DEFINITIONS + definitions = list(Definition.get_all()) + assert len(definitions) == 2 + + def_0, def_1 = definitions + assert def_0.rb_name == "test_rb_name_0" + assert def_0.rb_version == "test_rb_version_0" + assert def_0.chart_name is None + assert def_0.description is None + assert def_0.labels is None + + assert def_1.rb_name == "test_rb_name_1" + assert def_1.rb_version == "test_rb_version_1" + assert def_1.chart_name == "test_chart_name_1" + assert def_1.description == "test_description_1" + assert def_1.labels == {} + + +@mock.patch.object(Definition, "send_message_json") +def test_get_definition_by_name_version(mock_send_message_json): + mock_send_message_json.return_value = DEFINITION + def_0 = Definition.get_definition_by_name_version("rb_name", "rb_version") + assert def_0.rb_name == "test_rb_name_0" + assert def_0.rb_version == "test_rb_version_0" + assert def_0.chart_name is None + assert def_0.description is None + assert def_0.labels is None + + +@mock.patch.object(Definition, "send_message_json") +@mock.patch.object(Definition, "send_message") +def test_create_definition(mock_send_message, mock_send_message_json): + mock_send_message_json.return_value = DEFINITION + def_0 = Definition.create( + rb_name="test_rb_name_0", + rb_version="test_rb_version_0" + ) + assert def_0.rb_name == "test_rb_name_0" + assert def_0.rb_version == "test_rb_version_0" + assert def_0.chart_name is None + assert def_0.description is None + assert def_0.labels is None + + +@mock.patch.object(Definition, "send_message_json") +@mock.patch.object(Definition, "send_message") +def test_definition_create_profile(mock_send_message, mock_send_message_json): + mock_send_message_json.return_value = PROFILE + deff = Definition( + rb_name="test_rb_name", + rb_version="test_rb_version", + chart_name="test_chart_name", + description="test_description", + labels={} + ) + profile = deff.create_profile( + profile_name="test_profile_name", + namespace="test_namespace", + kubernetes_version="test_k8s_version" + ) + assert profile.rb_name == "test_rb_name" + assert profile.rb_version == "test_rb_version" + assert profile.profile_name == "test_profile_name" + assert profile.namespace == "test_namespace" + assert profile.kubernetes_version is None + assert profile.labels == {} + assert profile.release_name == "test_profile_name" + + +@mock.patch.object(Definition, "send_message_json") +def test_definition_get_profile_by_name(mock_send_message_json): + mock_send_message_json.return_value = PROFILE + deff = Definition( + rb_name="test_rb_name", + rb_version="test_rb_version", + chart_name="test_chart_name", + description="test_description", + labels={} + ) + profile = deff.get_profile_by_name("test_profile_name") + assert profile.rb_name == "test_rb_name" + assert profile.rb_version == "test_rb_version" + assert profile.profile_name == "test_profile_name" + assert profile.namespace == "test_namespace" + assert profile.kubernetes_version is None + assert profile.labels == {} + assert profile.release_name == "test_profile_name" + + +@mock.patch.object(Definition, "send_message_json") +def test_definition_get_all_profiles(mock_send_message_json): + mock_send_message_json.return_value = [] + deff = Definition( + rb_name="test_rb_name", + rb_version="test_rb_version", + chart_name="test_chart_name", + description="test_description", + labels={} + ) + assert len(list(deff.get_all_profiles())) == 0 + + mock_send_message_json.return_value = PROFILES + profiles = list(deff.get_all_profiles()) + assert len(profiles) == 2 + prof_0, prof_1 = profiles + + assert prof_0.rb_name == "test_rb_name" + assert prof_0.rb_version == "test_rb_version" + assert prof_0.profile_name == "test_profile_name" + assert prof_0.namespace == "test_namespace" + assert prof_0.kubernetes_version is None + assert prof_0.labels == {} + assert prof_0.release_name == "test_profile_name" + + assert prof_1.rb_name == "test_rb_name_1" + assert prof_1.rb_version == "test_rb_version_1" + assert prof_1.profile_name == "test_profile_name_1" + assert prof_1.namespace == "test_namespace_1" + assert prof_1.kubernetes_version is None + assert prof_1.labels == {} + assert prof_1.release_name == "test_profile_name_1" + + +@mock.patch.object(Definition, "send_message_json") +def test_definition_get_configuration_template_by_name(mock_send_message_json): + mock_send_message_json.return_value = CONFIGURATION_TEMPLATE + deff = Definition( + rb_name="test_rb_name", + rb_version="test_rb_version", + chart_name="test_chart_name", + description="test_description", + labels={} + ) + configuration_tmpl = deff.get_configuration_template_by_name( + template_name="test_configuration_template_name" + ) + assert configuration_tmpl.rb_name == deff.rb_name + assert configuration_tmpl.rb_version == deff.rb_version + assert configuration_tmpl.template_name == "test_configuration_template_name" + assert configuration_tmpl.description == "test_configuration_template_description" + + +@mock.patch.object(Definition, "send_message_json") +@mock.patch.object(Definition, "send_message") +def test_definition_create_configuration_template(mock_send_message, mock_send_message_json): + mock_send_message_json.return_value = CONFIGURATION_TEMPLATE + deff = Definition( + rb_name="test_rb_name", + rb_version="test_rb_version", + chart_name="test_chart_name", + description="test_description", + labels={} + ) + configuration_tmpl = deff.create_configuration_template( + template_name="test_configuration_template_name", + description="test_configuration_template_description" + ) + assert configuration_tmpl.rb_name == deff.rb_name + assert configuration_tmpl.rb_version == deff.rb_version + assert configuration_tmpl.template_name == "test_configuration_template_name" + assert configuration_tmpl.description == "test_configuration_template_description" + assert configuration_tmpl.url == f"{deff.base_url}/{deff.rb_name}/{deff.rb_version}/config-template/test_configuration_template_name" + + +@mock.patch.object(Definition, "send_message_json") +def test_definition_get_all_configuration_templates(mock_send_message_json): + mock_send_message_json.return_value = [] + deff = Definition( + rb_name="test_rb_name", + rb_version="test_rb_version", + chart_name="test_chart_name", + description="test_description", + labels={} + ) + assert len(list(deff.get_all_configuration_templates())) == 0 + + mock_send_message_json.return_value = CONFIGURATION_TEMPLATES + configuration_tmplts = list(deff.get_all_configuration_templates()) + assert len(configuration_tmplts) == 2 + + tmpl_0, tmpl_1 = configuration_tmplts + assert tmpl_0.rb_name == deff.rb_name + assert tmpl_0.rb_version == deff.rb_version + assert tmpl_0.template_name == "test_configuration_template_name" + assert tmpl_0.description == "test_configuration_template_description" + + assert tmpl_1.rb_name == deff.rb_name + assert tmpl_1.rb_version == deff.rb_version + assert tmpl_1.template_name == "test_configuration_template_name_0" + assert tmpl_1.description is None + + +@mock.patch.object(Instance, "send_message_json") +def test_instance_get_all(mock_send_message_json): + mock_send_message_json.return_value = [] + assert len(list(Instance.get_all())) == 0 + + mock_send_message_json.return_value = INSTANCES + assert len(list(Instance.get_all())) == 1 + + +@mock.patch.object(Instance, "send_message_json") +def test_instance_create(mock_send_message_json): + mock_send_message_json.return_value = INSTANCE + instance = Instance.create( + "test_cloud_region_id", + "test_profile_name", + "test_rb_name", + "test_rb_version" + ) + assert instance.instance_id == "ID_GENERATED_BY_K8SPLUGIN" + assert instance.namespace == "NAMESPACE_WHERE_INSTANCE_HAS_BEEN_DEPLOYED_AS_DERIVED_FROM_PROFILE" + + +@mock.patch.object(Instance, "send_message_json") +@mock.patch.object(Instance, "send_message") +def test_instance_get_by_id(mock_send_message, mock_send_message_json): + mock_send_message_json.return_value = INSTANCE + instance = Instance.get_by_id("ID_GENERATED_BY_K8SPLUGIN") + assert instance.instance_id == "ID_GENERATED_BY_K8SPLUGIN" + assert instance.namespace == "NAMESPACE_WHERE_INSTANCE_HAS_BEEN_DEPLOYED_AS_DERIVED_FROM_PROFILE" + instance.delete() |