2278 lines
106 KiB
Python
2278 lines
106 KiB
Python
# Copyright (c) 2017 Ericsson AB
|
|
# Copyright (c) 2017-2023 Wind River Systems, Inc.
|
|
# All Rights Reserved.
|
|
#
|
|
# 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 oslo_utils import timeutils
|
|
|
|
import base64
|
|
import copy
|
|
import json
|
|
import keyring
|
|
import mock
|
|
import six
|
|
from six.moves import http_client
|
|
import webtest
|
|
|
|
from dccommon import consts as dccommon_consts
|
|
from dcmanager.api.controllers.v1 import subclouds
|
|
from dcmanager.common import consts
|
|
from dcmanager.common import prestage
|
|
from dcmanager.common import utils as cutils
|
|
from dcmanager.db.sqlalchemy import api as db_api
|
|
from dcmanager.rpc import client as rpc_client
|
|
|
|
from dcmanager.tests.unit.api import test_root_controller as testroot
|
|
from dcmanager.tests.unit.api.v1.controllers.mixins import APIMixin
|
|
from dcmanager.tests.unit.api.v1.controllers.mixins import PostMixin
|
|
from dcmanager.tests.unit.common import fake_subcloud
|
|
from dcmanager.tests import utils
|
|
|
|
from tsconfig.tsconfig import SW_VERSION
|
|
|
|
SAMPLE_SUBCLOUD_NAME = 'SubcloudX'
|
|
SAMPLE_SUBCLOUD_DESCRIPTION = 'A Subcloud of mystery'
|
|
|
|
FAKE_ID = fake_subcloud.FAKE_ID
|
|
FAKE_URL = fake_subcloud.FAKE_URL
|
|
WRONG_URL = fake_subcloud.WRONG_URL
|
|
FAKE_HEADERS = fake_subcloud.FAKE_HEADERS
|
|
FAKE_SUBCLOUD_DATA = fake_subcloud.FAKE_SUBCLOUD_DATA
|
|
FAKE_BOOTSTRAP_VALUE = fake_subcloud.FAKE_BOOTSTRAP_VALUE
|
|
FAKE_SUBCLOUD_INSTALL_VALUES = fake_subcloud.FAKE_SUBCLOUD_INSTALL_VALUES
|
|
FAKE_SUBCLOUD_INSTALL_VALUES_WITH_PERSISTENT_SIZE = \
|
|
fake_subcloud.FAKE_SUBCLOUD_INSTALL_VALUES_WITH_PERSISTENT_SIZE
|
|
FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD = fake_subcloud.FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD
|
|
OAM_FLOATING_IP = '10.10.10.12'
|
|
|
|
FAKE_PATCH = {
|
|
"value": {
|
|
"patchstate": "Partial-Apply"
|
|
}
|
|
}
|
|
|
|
health_report_no_alarm = \
|
|
"System Health:\n \
|
|
All hosts are provisioned: [Fail]\n \
|
|
1 Unprovisioned hosts\n \
|
|
All hosts are unlocked/enabled: [OK]\n \
|
|
All hosts have current configurations: [OK]\n \
|
|
All hosts are patch current: [OK]\n \
|
|
No alarms: [OK]\n \
|
|
All kubernetes nodes are ready: [OK]\n \
|
|
All kubernetes control plane pods are ready: [OK]"
|
|
|
|
|
|
health_report_no_mgmt_alarm = \
|
|
"System Health:\n" \
|
|
"All hosts are provisioned: [OK]\n" \
|
|
"All hosts are unlocked/enabled: [OK]\n" \
|
|
"All hosts have current configurations: [OK]\n" \
|
|
"All hosts are patch current: [OK]\n" \
|
|
"Ceph Storage Healthy: [OK]\n" \
|
|
"No alarms: [Fail]\n" \
|
|
"[1] alarms found, [0] of which are management affecting\n" \
|
|
"All kubernetes nodes are ready: [OK]\n" \
|
|
"All kubernetes control plane pods are ready: [OK]"
|
|
|
|
|
|
health_report_mgmt_alarm = \
|
|
"System Health:\n" \
|
|
"All hosts are provisioned: [OK]\n" \
|
|
"All hosts are unlocked/enabled: [OK]\n" \
|
|
"All hosts have current configurations: [OK]\n" \
|
|
"All hosts are patch current: [OK]\n" \
|
|
"Ceph Storage Healthy: [OK]\n" \
|
|
"No alarms: [Fail]\n" \
|
|
"[1] alarms found, [1] of which are management affecting\n" \
|
|
"All kubernetes nodes are ready: [OK]\n" \
|
|
"All kubernetes control plane pods are ready: [OK]"
|
|
|
|
|
|
class Subcloud(object):
|
|
def __init__(self, data, is_online):
|
|
self.id = data['id']
|
|
self.name = data['name']
|
|
self.description = data['description']
|
|
self.location = data['location']
|
|
self.management_state = dccommon_consts.MANAGEMENT_UNMANAGED
|
|
if is_online:
|
|
self.availability_status = dccommon_consts.AVAILABILITY_ONLINE
|
|
else:
|
|
self.availability_status = dccommon_consts.AVAILABILITY_OFFLINE
|
|
self.deploy_status = data['deploy_status']
|
|
self.management_subnet = data['management_subnet']
|
|
self.management_gateway_ip = data['management_gateway_address']
|
|
self.management_start_ip = data['management_start_address']
|
|
self.management_end_ip = data['management_end_address']
|
|
self.external_oam_subnet = data['external_oam_subnet']
|
|
self.external_oam_gateway_address = \
|
|
data['external_oam_gateway_address']
|
|
self.external_oam_floating_address = \
|
|
data['external_oam_floating_address']
|
|
self.systemcontroller_gateway_ip = \
|
|
data['systemcontroller_gateway_address']
|
|
self.created_at = timeutils.utcnow()
|
|
self.updated_at = timeutils.utcnow()
|
|
self.data_install = ''
|
|
self.data_upgrade = ''
|
|
|
|
|
|
class FakeAddressPool(object):
|
|
def __init__(self, pool_network, pool_prefix, pool_start, pool_end):
|
|
self.network = pool_network
|
|
self.prefix = pool_prefix
|
|
range = list()
|
|
range.append(pool_start)
|
|
range.append(pool_end)
|
|
self.ranges = list()
|
|
self.ranges.append(range)
|
|
|
|
|
|
class FakeOAMAddressPool(object):
|
|
def __init__(self, oam_subnet, oam_start_ip,
|
|
oam_end_ip, oam_c1_ip,
|
|
oam_c0_ip, oam_gateway_ip,
|
|
oam_floating_ip):
|
|
self.oam_start_ip = oam_start_ip
|
|
self.oam_end_ip = oam_end_ip
|
|
self.oam_c1_ip = oam_c1_ip
|
|
self.oam_c0_ip = oam_c0_ip
|
|
self.oam_subnet = oam_subnet
|
|
self.oam_gateway_ip = oam_gateway_ip
|
|
self.oam_floating_ip = oam_floating_ip
|
|
|
|
|
|
class SubcloudAPIMixin(APIMixin):
|
|
API_PREFIX = '/v1.0/subclouds'
|
|
RESULT_KEY = 'subclouds'
|
|
# todo: populate the entire expected fields
|
|
EXPECTED_FIELDS = ['id',
|
|
'name',
|
|
'description',
|
|
'location',
|
|
'management-state',
|
|
'created-at',
|
|
'updated-at']
|
|
|
|
FAKE_BOOTSTRAP_DATA = {
|
|
"system_mode": "simplex",
|
|
"name": "fake subcloud1",
|
|
"management_subnet": "192.168.101.0/24",
|
|
"management_start_address": "192.168.101.2",
|
|
"management_end_address": "192.168.101.50",
|
|
"management_gateway_address": "192.168.101.1",
|
|
"external_oam_subnet": "10.10.10.0/24",
|
|
"external_oam_gateway_address": "10.10.10.1",
|
|
"external_oam_floating_address": "10.10.10.12",
|
|
"systemcontroller_gateway_address": "192.168.204.101",
|
|
}
|
|
|
|
OPTIONAL_BOOTSTRAP_DATA = {
|
|
"location": "fake location",
|
|
"description": "fake description",
|
|
}
|
|
|
|
# based off MANDATORY_INSTALL_VALUES
|
|
# bmc_password must be passed as a param
|
|
FAKE_INSTALL_DATA = {
|
|
"bootstrap_interface": "fake interface",
|
|
"bootstrap_address": "10.10.10.12",
|
|
"bootstrap_address_prefix": "10.10.10.12",
|
|
"bmc_address": "128.224.64.1",
|
|
"bmc_username": "fake bmc user",
|
|
"install_type": 2,
|
|
}
|
|
|
|
list_of_post_files = subclouds.SUBCLOUD_ADD_MANDATORY_FILE
|
|
bootstrap_data = copy.copy(FAKE_BOOTSTRAP_DATA)
|
|
install_data = copy.copy(FAKE_INSTALL_DATA)
|
|
|
|
def setUp(self):
|
|
super(SubcloudAPIMixin, self).setUp()
|
|
|
|
def _get_test_subcloud_dict(self, **kw):
|
|
# id should not be part of the structure
|
|
subcloud = {
|
|
'name': kw.get('name', SAMPLE_SUBCLOUD_NAME),
|
|
'description': kw.get('description',
|
|
SAMPLE_SUBCLOUD_DESCRIPTION),
|
|
}
|
|
return subcloud
|
|
|
|
def _post_get_test_subcloud(self, **kw):
|
|
post_body = self._get_test_subcloud_dict(**kw)
|
|
return post_body
|
|
|
|
# The following methods are required for subclasses of APIMixin
|
|
def get_api_prefix(self):
|
|
return self.API_PREFIX
|
|
|
|
def get_result_key(self):
|
|
return self.RESULT_KEY
|
|
|
|
def get_expected_api_fields(self):
|
|
return self.EXPECTED_FIELDS
|
|
|
|
def get_omitted_api_fields(self):
|
|
return []
|
|
|
|
def _create_db_object(self, context, **kw):
|
|
creation_fields = self._get_test_subcloud_dict(**kw)
|
|
return db_api.subcloud_create(context, **creation_fields)
|
|
|
|
def get_post_params(self):
|
|
return copy.copy(FAKE_BOOTSTRAP_VALUE)
|
|
|
|
def set_list_of_post_files(self, value):
|
|
self.list_of_post_files = value
|
|
|
|
def get_post_upload_files(self):
|
|
fields = list()
|
|
for f in self.list_of_post_files:
|
|
fake_name = f + "_fake"
|
|
# The data in the bootstrap file needs to be dictionary syntax
|
|
if f == subclouds.BOOTSTRAP_VALUES:
|
|
fake_content = json.dumps(self.bootstrap_data).encode("utf-8")
|
|
elif f == subclouds.INSTALL_VALUES:
|
|
fake_content = json.dumps(self.install_data).encode("utf-8")
|
|
else:
|
|
fake_content = "fake content".encode("utf-8")
|
|
fields.append((f, fake_name, fake_content))
|
|
return fields
|
|
|
|
def get_post_object(self):
|
|
return self._post_get_test_subcloud()
|
|
|
|
def get_update_object(self):
|
|
update_object = {
|
|
'description': 'Updated description'
|
|
}
|
|
return update_object
|
|
|
|
|
|
# Combine Subcloud Group API with mixins to test post, get, update and delete
|
|
class TestSubcloudPost(testroot.DCManagerApiTest,
|
|
SubcloudAPIMixin,
|
|
PostMixin):
|
|
def setUp(self):
|
|
super(TestSubcloudPost, self).setUp()
|
|
self.list_of_post_files = subclouds.SUBCLOUD_ADD_MANDATORY_FILE
|
|
self.bootstrap_data = copy.copy(self.FAKE_BOOTSTRAP_DATA)
|
|
self.install_data = copy.copy(self.FAKE_INSTALL_DATA)
|
|
|
|
self.management_address_pool = FakeAddressPool('192.168.204.0', 24,
|
|
'192.168.204.2',
|
|
'192.168.204.100')
|
|
|
|
p = mock.patch.object(subclouds.SubcloudsController,
|
|
'_get_management_address_pool')
|
|
self.mock_get_management_address_pool = p.start()
|
|
self.mock_get_management_address_pool.return_value = \
|
|
self.management_address_pool
|
|
self.addCleanup(p.stop)
|
|
|
|
p = mock.patch.object(rpc_client, 'ManagerClient')
|
|
self.mock_rpc_client = p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
p = mock.patch.object(subclouds.SubcloudsController, 'get_ks_client')
|
|
self.mock_get_ks_client = p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
p = mock.patch.object(subclouds.PatchingClient, 'query')
|
|
self.mock_query = p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
p = mock.patch.object(rpc_client, 'SubcloudStateClient')
|
|
self.mock_rpc_state_client = p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
def _verify_post_failure(self, response, param, value):
|
|
self.assertEqual(http_client.BAD_REQUEST,
|
|
response.status_code,
|
|
message=("%s=%s returned %s instead of %s"
|
|
% (param,
|
|
value,
|
|
response.status_code,
|
|
http_client.BAD_REQUEST)))
|
|
# Note: response failures return 'text' rather than json
|
|
self.assertEqual('text/plain', response.content_type)
|
|
|
|
def _verify_post_success(self, response):
|
|
self.assertEqual(http_client.OK, response.status_code)
|
|
self.assertEqual('application/json', response.content_type)
|
|
self.assert_fields(response.json)
|
|
|
|
def test_post_subcloud_wrong_url(self):
|
|
"""Test POST operation rejected when going to the wrong URL."""
|
|
params = self.get_post_params()
|
|
upload_files = self.get_post_upload_files()
|
|
six.assertRaisesRegex(self,
|
|
webtest.app.AppError,
|
|
"404 *",
|
|
self.app.post,
|
|
WRONG_URL,
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
|
|
def test_post_no_body(self):
|
|
"""Test POST operation with nearly everything wrong with it."""
|
|
six.assertRaisesRegex(self,
|
|
webtest.app.AppError,
|
|
"400 *",
|
|
self.app.post,
|
|
self.get_api_prefix(),
|
|
params={},
|
|
headers=self.get_api_headers())
|
|
|
|
def test_post_subcloud_boostrap_entries_missing(self):
|
|
"""Test POST operation with some mandatory boostrap fields missing.
|
|
|
|
Example: name is a required field
|
|
"""
|
|
|
|
self.list_of_post_files = subclouds.SUBCLOUD_ADD_MANDATORY_FILE
|
|
params = self.get_post_params()
|
|
|
|
for key in self.FAKE_BOOTSTRAP_DATA:
|
|
self.bootstrap_data = copy.copy(self.FAKE_BOOTSTRAP_DATA)
|
|
del self.bootstrap_data[key]
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self._verify_post_failure(response, key, None)
|
|
|
|
# try with nothing removed and verify it works
|
|
self.bootstrap_data = copy.copy(self.FAKE_BOOTSTRAP_DATA)
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
self._verify_post_success(response)
|
|
|
|
def _test_post_param_inputs(self, param_key, bad_values, good_value):
|
|
upload_files = self.get_post_upload_files()
|
|
params = self.get_post_params()
|
|
|
|
# Test all the bad param values
|
|
for bad_value in bad_values:
|
|
params[param_key] = bad_value
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self._verify_post_failure(response, param_key, bad_value)
|
|
|
|
# Test that a good value will work
|
|
params[param_key] = good_value
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
self._verify_post_success(response)
|
|
|
|
def test_post_subcloud_bad_bootstrap_address(self):
|
|
"""Test POST operation with a bad bootstrap-address"""
|
|
|
|
param_key = "bootstrap-address"
|
|
# bootstrap-address must be valid IP address
|
|
bad_values = ["10.10.10.wut", # including letters in the IP
|
|
"10.10.10.276" # 276 is invalid
|
|
]
|
|
good_values = "10.10.10.3"
|
|
self._test_post_param_inputs(param_key,
|
|
bad_values,
|
|
good_values)
|
|
|
|
def test_post_subcloud_bad_IPv6_bootstrap_address(self):
|
|
"""Test POST operation with a bad bootstrap-address"""
|
|
|
|
param_key = "bootstrap-address"
|
|
# bootstrap-address must be valid IP address
|
|
bad_values = ["2620::10a:a103::1135", # more than one double colons
|
|
"2620:10a:a001:a103::wut", # invalid letter
|
|
"2620:10a:a001:a103:1135" # Incomplete IP
|
|
]
|
|
good_values = "2620:10a:a001:a103::1135"
|
|
self._test_post_param_inputs(param_key,
|
|
bad_values,
|
|
good_values)
|
|
|
|
def test_post_subcloud_bad_gateway(self):
|
|
"""Test POST with an invalid gateway."""
|
|
|
|
param_key = "systemcontroller_gateway_address"
|
|
# systemcontroller_gateway_address must be appropriate address within
|
|
# the management address pool which is
|
|
# 192.168.204.0/24 greater than 100
|
|
bad_values = ["192.168.205.101", # 205.xx not in the pool
|
|
"192.168.204.99", # 99 is reserved in the pool
|
|
"192.168.276.276", # 276 is not a valid IP address
|
|
"192.168.206.wut", # including letters in the IP
|
|
"192.168.204", # incomplete IP
|
|
]
|
|
good_value = "192.168.204.101"
|
|
self._test_post_param_inputs(param_key,
|
|
bad_values,
|
|
good_value)
|
|
|
|
def test_post_subcloud_bad_subnet(self):
|
|
"""Test POST with an invalid subnet."""
|
|
|
|
param_key = "management_subnet"
|
|
bad_values = ["192.168.101.0/32", # /32 would be just one IP
|
|
"192.168.101.0/33", # /33 is an invalid CIDR
|
|
"192.168.276.0/24", # 276 makes no sense as an IP
|
|
"192.168.206.wut/24", # including letters in the IP
|
|
"192.168.204/24", # incomplete CIDR
|
|
]
|
|
good_value = "192.168.101.0/24"
|
|
self._test_post_param_inputs(param_key,
|
|
bad_values,
|
|
good_value)
|
|
|
|
def test_post_subcloud_bad_start_ip(self):
|
|
"""Test POST with an invalid management_start_address.
|
|
|
|
The management_start_address cannot be after the end or too close
|
|
since there must be enough range to allocate the IPs.
|
|
"""
|
|
|
|
param_key = "management_start_address"
|
|
# subnet is 192.168.101.0/24
|
|
# end address is 192.168.101.50
|
|
bad_values = ["192.168.100.2", # xx.xx.100.xx is not in the subnet
|
|
"192.168.101.51", # start is higher than end
|
|
"192.168.101.48", # start is too close to end
|
|
"192.168.276.0", # 276 makes no sense as an IP
|
|
"192.168.206.wut", # including letters in the IP
|
|
"192.168.204", # incomplete IP
|
|
]
|
|
good_value = "192.168.101.2"
|
|
self._test_post_param_inputs(param_key,
|
|
bad_values,
|
|
good_value)
|
|
|
|
def test_post_subcloud_bad_end_ip(self):
|
|
"""Test POST with an invalid management_end_address.
|
|
|
|
The management_end_address cannot be less than the start or too close
|
|
since there must be enough range to allocate the IPs.
|
|
"""
|
|
|
|
param_key = "management_end_address"
|
|
# subnet is 192.168.101.0/24
|
|
# start address is 192.168.101.2
|
|
bad_values = ["192.168.100.50", # xx.xx.100.xx is not in the subnet
|
|
"192.168.101.1", # end is less than start
|
|
"192.168.101.4", # end is too close to start
|
|
"192.168.276.50", # 276 makes no sense as an IP
|
|
"192.168.206.wut", # including letters in the IP
|
|
"192.168.204", # incomplete IP
|
|
]
|
|
good_value = "192.168.101.50"
|
|
self._test_post_param_inputs(param_key,
|
|
bad_values,
|
|
good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_values(self, mock_vault_files):
|
|
"""Test POST operation with install values is supported by the API."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
|
|
# pass a different "install" list of files for this POST
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
upload_files = self.get_post_upload_files()
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8")})
|
|
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
self._verify_post_success(response)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_without_release_parameter(self, mock_vault_files):
|
|
"""Test POST operation without release parameter."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
upload_files = self.get_post_upload_files()
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8")})
|
|
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
self._verify_post_success(response)
|
|
# Verify that the subcloud installed with the active release
|
|
# when no release parameter provided.
|
|
self.assertEqual(SW_VERSION, response.json['software-version'])
|
|
|
|
def test_post_subcloud_release_not_match_install_values_sw(self):
|
|
"""Release parameter not match software_version in the install_values."""
|
|
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
upload_files = self.get_post_upload_files()
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password and release to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8"),
|
|
'release': '21.12'})
|
|
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
|
|
# Verify the request was rejected
|
|
self.assertEqual(response.status_code, http_client.BAD_REQUEST)
|
|
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_k8s_version')
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_with_release_parameter(self, mock_vault_files,
|
|
mock_validate_k8s_version):
|
|
"""Test POST operation with release parameter."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
software_version = '21.12'
|
|
# Update the software_version value to match the release parameter value,
|
|
# otherwise, the request will be rejected
|
|
self.install_data['software_version'] = software_version
|
|
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
upload_files = self.get_post_upload_files()
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password and release to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8"),
|
|
'release': software_version})
|
|
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
|
|
self.assertEqual(response.status_code, http_client.OK)
|
|
self.assertEqual(software_version, response.json['software-version'])
|
|
|
|
# Revert the software_version value
|
|
self.install_data['software_version'] = SW_VERSION
|
|
|
|
@mock.patch.object(subclouds.PatchingClient, 'query')
|
|
def test_post_subcloud_when_partial_applied_patch(self, mock_query):
|
|
"""Test POST operation when there is a partial-applied patch."""
|
|
|
|
upload_files = self.get_post_upload_files()
|
|
params = self.get_post_params()
|
|
mock_query.return_value = FAKE_PATCH
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self.assertEqual(http_client.UNPROCESSABLE_ENTITY, response.status_code)
|
|
self.assertEqual('text/plain', response.content_type)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_values_no_bmc_password(self, mock_vault_files):
|
|
"""Test POST operation with install values is supported by the API."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
|
|
# pass a different "install" list of files for this POST
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
upload_files = self.get_post_upload_files()
|
|
|
|
params = self.get_post_params()
|
|
# for this unit test, omit adding bmc_password to params
|
|
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self._verify_post_failure(response, "bmc_password", None)
|
|
|
|
# add the bmc_password and verify that now it works
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8")})
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
self._verify_post_success(response)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_missing_image(self, mock_vault_files):
|
|
"""Test POST operation without image in install values and vault files."""
|
|
|
|
mock_vault_files.return_value = (None, None)
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8")})
|
|
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
self.install_data = copy.copy(self.FAKE_INSTALL_DATA)
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_code, http_client.BAD_REQUEST)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_values_missing(self, mock_vault_files):
|
|
"""Test POST operation with install values fails if data missing."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8")})
|
|
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
# for each entry in install content, try with one key missing
|
|
for key in self.FAKE_INSTALL_DATA:
|
|
self.install_data = copy.copy(self.FAKE_INSTALL_DATA)
|
|
del self.install_data[key]
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self._verify_post_failure(response, key, None)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
@mock.patch.object(cutils, 'get_playbook_for_software_version')
|
|
@mock.patch.object(cutils, 'get_value_from_yaml_file')
|
|
def test_post_subcloud_bad_kubernetes_version(self,
|
|
mock_get_value_from_yaml_file,
|
|
mock_get_playbook_for_software_version,
|
|
mock_vault_files):
|
|
"""Test POST operation with bad kubernetes_version."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
|
|
software_version = '21.12'
|
|
# Update the software_version value to match the release parameter value,
|
|
# otherwise, the request will be rejected
|
|
self.install_data['software_version'] = software_version
|
|
|
|
params = self.get_post_params()
|
|
# add bmc_password to params
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8"),
|
|
'release': software_version})
|
|
|
|
# Add kubernetes version to bootstrap_data
|
|
self.bootstrap_data['kubernetes_version'] = '1.21.8'
|
|
mock_get_value_from_yaml_file.return_value = '1.23.1'
|
|
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
self.install_data = copy.copy(self.FAKE_INSTALL_DATA)
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_code, http_client.BAD_REQUEST)
|
|
|
|
# Revert the change of bootstrap_data
|
|
del self.bootstrap_data['kubernetes_version']
|
|
|
|
def _test_post_input_value_inputs(self,
|
|
setup_overrides,
|
|
required_overrides,
|
|
param_key, bad_values, good_value):
|
|
"""This utility checks for test permutions.
|
|
|
|
The setup_overrides are the initial modifications to the install data
|
|
The required_overrides are all tested to see that if any of them are
|
|
missing, the 'good' value will not work.
|
|
The param_key is tested with the list of bad_values to ensure they fail
|
|
The param_key is tested with the good value to ensure it passes.
|
|
"""
|
|
params = self.get_post_params()
|
|
params.update(
|
|
{'bmc_password':
|
|
base64.b64encode('fake pass'.encode("utf-8")).decode("utf-8")})
|
|
self.set_list_of_post_files(subclouds.SUBCLOUD_ADD_GET_FILE_CONTENTS)
|
|
|
|
# Setup starting install data
|
|
# Note: upload_files are populated based on the install values data.
|
|
starting_data = copy.copy(self.FAKE_INSTALL_DATA)
|
|
for key, val in setup_overrides.items():
|
|
starting_data[key] = val
|
|
starting_data['image'] = 'fake image'
|
|
|
|
# Test all the bad param values
|
|
for bad_value in bad_values:
|
|
self.install_data = copy.copy(starting_data)
|
|
# Apply all required_overrides
|
|
for key, val in required_overrides.items():
|
|
self.install_data[key] = val
|
|
# Apply the bad value
|
|
self.install_data[param_key] = bad_value
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self._verify_post_failure(response, param_key, bad_value)
|
|
|
|
# Test that any missing override required to use with the good value
|
|
# will cause a failure
|
|
for missing_override in required_overrides:
|
|
self.install_data = copy.copy(starting_data)
|
|
# We cannot simply delete the missing override, but we can skip it
|
|
for key, val in required_overrides.items():
|
|
if key != missing_override:
|
|
self.install_data[key] = val
|
|
# The 'good' value should still fail if a required override missing
|
|
self.install_data[param_key] = good_value
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers(),
|
|
expect_errors=True)
|
|
self._verify_post_failure(response, param_key, bad_value)
|
|
|
|
# Test that a good value and all required overrides works
|
|
self.install_data = copy.copy(starting_data)
|
|
for key, val in required_overrides.items():
|
|
self.install_data[key] = val
|
|
self.install_data[param_key] = good_value
|
|
upload_files = self.get_post_upload_files()
|
|
response = self.app.post(self.get_api_prefix(),
|
|
params=params,
|
|
upload_files=upload_files,
|
|
headers=self.get_api_headers())
|
|
self._verify_post_success(response)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_values_invalid_type(self, mock_vault_files):
|
|
"""Test POST with an invalid type specified in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {}
|
|
required_overrides = {}
|
|
# the install_type must a number 0 <= X <=5
|
|
install_key = "install_type"
|
|
bad_values = [-1, # negative
|
|
6, # too big
|
|
"3", # alphbetical
|
|
"w", # really alphbetical
|
|
"", # empty
|
|
None, # None
|
|
]
|
|
good_value = 3
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_bad_bootstrap_ip(self, mock_vault_files):
|
|
"""Test POST with invalid boostrap ip specified in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {}
|
|
required_overrides = {}
|
|
install_key = "bootstrap_address"
|
|
bad_values = ["192.168.1.256", # 256 is not valid
|
|
"192.168.206.wut", # including letters in the IP
|
|
None, # None
|
|
]
|
|
# Note: an incomplete IP address is 10.10.10 is considered valid
|
|
good_value = "10.10.10.12"
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_bad_bmc_ip(self, mock_vault_files):
|
|
"""Test POST with invalid bmc ip specified in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {}
|
|
required_overrides = {}
|
|
install_key = "bmc_address"
|
|
bad_values = ["128.224.64.256", # 256 is not valid
|
|
"128.224.64.wut", # including letters in the IP
|
|
None, # None
|
|
]
|
|
good_value = "128.224.64.1"
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_bad_persistent_size(self, mock_vault_files):
|
|
"""Test POST with invalid persistent_size specified in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {}
|
|
required_overrides = {}
|
|
install_key = "persistent_size"
|
|
bad_values = ["4000o", # not an integer
|
|
"20000", # less than 30000
|
|
40000.1, # fraction
|
|
None, # None
|
|
]
|
|
good_value = 40000
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_bad_nexthop_gateway(self, mock_vault_files):
|
|
"""Test POST with invalid nexthop_gateway in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {}
|
|
required_overrides = {}
|
|
# nexthop_gateway is not required. but if provided, it must be valid
|
|
install_key = "nexthop_gateway"
|
|
bad_values = ["128.224.64.256", # 256 is not valid
|
|
"128.224.64.wut", # including letters in the IP
|
|
None, # None
|
|
]
|
|
good_value = "192.168.1.2"
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_bad_network_address(self, mock_vault_files):
|
|
"""Test POST with invalid network_address in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {}
|
|
# The nexthop_gateway is required when network_address is present
|
|
# The network mask is required when network address is present
|
|
required_overrides = {
|
|
"nexthop_gateway": "192.168.1.2",
|
|
"network_mask": 32, # Note: this netmask is validated when used
|
|
}
|
|
# network_address is not required. but if provided, it must be valid
|
|
install_key = "network_address"
|
|
# todo(abailey): None will cause the API to fail
|
|
bad_values = ["fd01:6::0", # mis-match ipv6 vs ipv4
|
|
]
|
|
good_value = "192.168.101.10" # ipv4
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_bad_network_mask(self, mock_vault_files):
|
|
"""Test POST with invalid network_mask in install values."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
# network_address is not required. but if provided a valid network_mask
|
|
# is needed
|
|
setup_overrides = {
|
|
"nexthop_gateway": "192.168.1.2",
|
|
"network_address": "192.168.101.10"
|
|
}
|
|
required_overrides = {}
|
|
|
|
install_key = "network_mask"
|
|
bad_values = [None, # None
|
|
64, # network_mask cannot really be greater than 32
|
|
-1, # network_mask cannot really be negative
|
|
"junk", # network_mask cannot be a junk string
|
|
]
|
|
good_value = 32
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_diff_bmc_ip_version(self, mock_vault_files):
|
|
"""Test POST install values with mismatched(ipv4/ipv6) bmc ip."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
setup_overrides = {
|
|
"bootstrap_address": "192.168.1.2"
|
|
}
|
|
required_overrides = {}
|
|
# bootstrap address ip version must match bmc_address. default ipv4
|
|
install_key = "bmc_address"
|
|
bad_values = ["fd01:6::7", # ipv6
|
|
None, # None
|
|
"192.168.-1.1", # bad ipv4
|
|
]
|
|
good_value = "192.168.1.7" # ipv4
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_diff_bmc_ip_version_ipv6(self, mock_vault_files):
|
|
"""Test POST install values with mismatched(ipv6/ipv4) bmc ip."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
# version of bootstrap address must be same as bmc_address
|
|
setup_overrides = {
|
|
"bootstrap_address": "fd01:6::7"
|
|
}
|
|
required_overrides = {}
|
|
install_key = "bmc_address"
|
|
bad_values = ["192.168.1.7", # ipv4
|
|
None, # None
|
|
"fd01:6:-1", # bad ipv6
|
|
]
|
|
good_value = "fd01:6::7" # ipv6
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_diff_nexthop_ip_version(self, mock_vault_files):
|
|
"""Test POST install values mismatched(ipv4/ipv6) nexthop_gateway."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
# ip version of bootstrap address must be same as nexthop_gateway
|
|
# All required addresses (like bmc address) much match bootstrap
|
|
# default bmc address is ipv4
|
|
setup_overrides = {
|
|
"bootstrap_address": "192.168.1.5"
|
|
}
|
|
required_overrides = {}
|
|
install_key = "nexthop_gateway"
|
|
bad_values = ["fd01:6::7", ] # ipv6
|
|
good_value = "192.168.1.7" # ipv4
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_post_subcloud_install_diff_nexthop_ip_version_ipv6(self,
|
|
mock_vault_files):
|
|
"""Test POST install values with mismatched(ipv6/ipv4) bmc ip."""
|
|
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
# version of bootstrap address must be same as nexthop_gateway
|
|
# All required addresses must also be setup ipv6 such as bmc_address
|
|
# default bmc address is ipv4
|
|
setup_overrides = {
|
|
"bootstrap_address": "fd01:6::6"
|
|
}
|
|
required_overrides = {
|
|
"bmc_address": "fd01:6::7"
|
|
}
|
|
|
|
install_key = "nexthop_gateway"
|
|
bad_values = ["192.168.1.7", ] # ipv4
|
|
good_value = "fd01:6::8" # ipv6
|
|
self._test_post_input_value_inputs(setup_overrides, required_overrides,
|
|
install_key, bad_values, good_value)
|
|
|
|
|
|
class TestSubcloudAPIOther(testroot.DCManagerApiTest):
|
|
|
|
"""Test GET, delete and patch API calls"""
|
|
def setUp(self):
|
|
super(TestSubcloudAPIOther, self).setUp()
|
|
self.ctx = utils.dummy_context()
|
|
|
|
p = mock.patch.object(rpc_client, 'SubcloudStateClient')
|
|
self.mock_rpc_state_client = p.start()
|
|
self.addCleanup(p.stop)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_delete_subcloud(self, mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
delete_url = FAKE_URL + '/' + str(subcloud.id)
|
|
mock_rpc_client().delete_subcloud.return_value = True
|
|
response = self.app.delete_json(delete_url, headers=FAKE_HEADERS)
|
|
mock_rpc_client().delete_subcloud.assert_called_once_with(
|
|
mock.ANY, mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_delete_wrong_request(self, mock_rpc_client):
|
|
delete_url = WRONG_URL + '/' + FAKE_ID
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "404 *",
|
|
self.app.delete_json, delete_url,
|
|
headers=FAKE_HEADERS)
|
|
|
|
@mock.patch.object(subclouds.SubcloudsController,
|
|
'_get_oam_addresses')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_subcloud(self,
|
|
mock_rpc_client,
|
|
mock_get_oam_addresses):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
get_url = FAKE_URL + '/' + str(subcloud.id)
|
|
response = self.app.get(get_url, headers=FAKE_HEADERS)
|
|
self.assertEqual(response.content_type, 'application/json')
|
|
self.assertEqual(response.status_code, http_client.OK)
|
|
self.assertEqual(response.json.get('oam_floating_ip', None), None)
|
|
self.assertEqual(response.json['name'], subcloud.name)
|
|
|
|
@mock.patch.object(subclouds.SubcloudsController,
|
|
'_get_oam_addresses')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_online_subcloud_with_additional_detail(self,
|
|
mock_rpc_client,
|
|
mock_get_oam_addresses):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
updated_subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE)
|
|
|
|
get_url = FAKE_URL + '/' + str(updated_subcloud.id) + '/detail'
|
|
oam_addresses = FakeOAMAddressPool('10.10.10.254',
|
|
'10.10.10.1',
|
|
'10.10.10.254',
|
|
'10.10.10.4',
|
|
'10.10.10.3',
|
|
'10.10.10.1',
|
|
'10.10.10.2')
|
|
mock_get_oam_addresses.return_value = oam_addresses
|
|
response = self.app.get(get_url, headers=FAKE_HEADERS)
|
|
self.assertEqual(response.content_type, 'application/json')
|
|
self.assertEqual(response.status_code, http_client.OK)
|
|
self.assertEqual('10.10.10.2', response.json['oam_floating_ip'])
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_offline_subcloud_with_additional_detail(self,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
get_url = FAKE_URL + '/' + str(subcloud.id) + '/detail'
|
|
response = self.app.get(get_url, headers=FAKE_HEADERS)
|
|
self.assertEqual(response.content_type, 'application/json')
|
|
self.assertEqual(response.status_code, http_client.OK)
|
|
self.assertEqual('unavailable', response.json['oam_floating_ip'])
|
|
|
|
@mock.patch.object(subclouds.SubcloudsController,
|
|
'_get_oam_addresses')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_subcloud_oam_ip_unavailable(self,
|
|
mock_rpc_client,
|
|
mock_get_oam_addresses):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
updated_subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE)
|
|
|
|
get_url = FAKE_URL + '/' + str(updated_subcloud.id) + '/detail'
|
|
mock_get_oam_addresses.return_value = None
|
|
response = self.app.get(get_url, headers=FAKE_HEADERS)
|
|
self.assertEqual(response.content_type, 'application/json')
|
|
self.assertEqual(response.status_code, http_client.OK)
|
|
self.assertEqual('unavailable', response.json['oam_floating_ip'])
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_wrong_request(self, mock_rpc_client):
|
|
get_url = WRONG_URL + '/' + FAKE_ID
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "404 *",
|
|
self.app.get, get_url,
|
|
headers=FAKE_HEADERS)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_subcloud_all(self, mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
get_url = FAKE_URL
|
|
response = self.app.get(get_url, headers=FAKE_HEADERS)
|
|
self.assertEqual(response.json['subclouds'][0]['name'], subcloud.name)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud(self, mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'management-state': dccommon_consts.MANAGEMENT_UNMANAGED}
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = data
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
# Verify subcloud was updated with correct values
|
|
updated_subcloud = db_api.subcloud_get_by_name(self.ctx, subcloud.name)
|
|
self.assertEqual(dccommon_consts.MANAGEMENT_UNMANAGED,
|
|
updated_subcloud.management_state)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_update_subcloud_group_value(self,
|
|
mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
good_values = [1, "1"]
|
|
expected_group_id = 1
|
|
for x in good_values:
|
|
data = {'group_id': x}
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = data
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
self.assertEqual(response.status_int, 200)
|
|
# Verify subcloud was updated with correct values
|
|
updated_subcloud = db_api.subcloud_get_by_name(self.ctx,
|
|
subcloud.name)
|
|
self.assertEqual(expected_group_id,
|
|
updated_subcloud.group_id)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_update_subcloud_group_value_by_name(self,
|
|
mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
expected_group_id = 1
|
|
data = {'group_id': 'Default'}
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = data
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
# Verify subcloud was updated with correct values
|
|
updated_subcloud = db_api.subcloud_get_by_name(self.ctx, subcloud.name)
|
|
self.assertEqual(expected_group_id,
|
|
updated_subcloud.group_id)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_update_subcloud_group_bad_value(self,
|
|
mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
# There is only 1 subcloud group 'Default' which has id '1'
|
|
# This should test that boolean, zero, negative, float and bad values
|
|
# all get rejected
|
|
bad_values = [0, -1, 2, "0", "-1", 0.5, "BadName", "False", "True"]
|
|
for x in bad_values:
|
|
data = {'group_id': x}
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = data
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data,
|
|
expect_errors=True)
|
|
self.assertEqual(response.status_int, 400)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_update_subcloud_install_values_persistent_size(self, mock_vault_files,
|
|
mock_get_patch_data,
|
|
mock_rpc_client):
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx, data_install=None)
|
|
payload = {}
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES_WITH_PERSISTENT_SIZE)
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
data = {'bmc_password': encoded_password}
|
|
payload.update({'install_values': install_data})
|
|
payload.update(data)
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = payload
|
|
|
|
fake_content = "fake content".encode("utf-8")
|
|
response = self.app.patch(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data,
|
|
upload_files=[("install_values",
|
|
"fake_name",
|
|
fake_content)])
|
|
install_data.update({'bmc_password': encoded_password})
|
|
mock_rpc_client().update_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
management_state=None,
|
|
description=None,
|
|
location=None,
|
|
group_id=None,
|
|
data_install=json.dumps(install_data),
|
|
force=None)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_admin_network_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud_admin_values(self, mock_get_patch_data,
|
|
mock_validate_admin_network_config,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
db_api.subcloud_update(self.ctx, subcloud.id,
|
|
availability_status=dccommon_consts.AVAILABILITY_ONLINE)
|
|
fake_password = (
|
|
base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
payload = {'sysadmin_password': fake_password,
|
|
'admin_subnet': "192.168.102.0/24",
|
|
'admin_start_address': "192.168.102.5",
|
|
'admin_end_address': "192.168.102.49",
|
|
'admin_gateway_ip': "192.168.102.1"}
|
|
|
|
mock_rpc_client().update_subcloud_with_network_reconfig.return_value = True
|
|
mock_get_patch_data.return_value = payload
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=payload)
|
|
self.assertEqual(response.status_int, 200)
|
|
mock_validate_admin_network_config.assert_called_once()
|
|
mock_rpc_client().update_subcloud_with_network_reconfig.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
payload)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_patch_subcloud_install_values(self, mock_vault_files,
|
|
mock_get_patch_data,
|
|
mock_rpc_client):
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx, data_install=None)
|
|
payload = {}
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
data = {'bmc_password': encoded_password}
|
|
payload.update({'install_values': install_data})
|
|
payload.update(data)
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = payload
|
|
|
|
fake_content = "fake content".encode("utf-8")
|
|
response = self.app.patch(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data,
|
|
upload_files=[("install_values",
|
|
"fake_name",
|
|
fake_content)])
|
|
install_data.update({'bmc_password': encoded_password})
|
|
mock_rpc_client().update_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
management_state=None,
|
|
description=None,
|
|
location=None,
|
|
group_id=None,
|
|
data_install=json.dumps(install_data),
|
|
force=None)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
@mock.patch('dcmanager.common.utils.get_vault_load_files')
|
|
def test_patch_subcloud_install_values_with_existing_data_install(
|
|
self, mock_vault_files, mock_get_patch_data, mock_rpc_client):
|
|
mock_vault_files.return_value = ('fake_iso', 'fake_sig')
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
subcloud = fake_subcloud.create_fake_subcloud(
|
|
self.ctx, data_install=json.dumps(install_data))
|
|
install_data.update({"install_type": 2})
|
|
payload = {}
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
data = {'bmc_password': encoded_password}
|
|
payload.update({'install_values': install_data})
|
|
payload.update(data)
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = payload
|
|
|
|
fake_content = "fake content".encode("utf-8")
|
|
response = self.app.patch(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=data,
|
|
upload_files=[("install_values",
|
|
"fake_name",
|
|
fake_content)])
|
|
install_data.update({'bmc_password': encoded_password})
|
|
mock_rpc_client().update_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
management_state=None,
|
|
description=None,
|
|
location=None,
|
|
group_id=None,
|
|
data_install=json.dumps(install_data),
|
|
force=None)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud_no_body(self, mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {}
|
|
mock_get_patch_data.return_value = data
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json,
|
|
FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud_bad_status(self, mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'management-state': 'bad-status'}
|
|
mock_get_patch_data.return_value = data
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json,
|
|
FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud_bad_force_value(self, mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'management-state': dccommon_consts.MANAGEMENT_MANAGED,
|
|
'force': 'bad-value'}
|
|
mock_get_patch_data.return_value = data
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json,
|
|
FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud_forced_unmanaged(self, mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'management-state': dccommon_consts.MANAGEMENT_UNMANAGED,
|
|
'force': True}
|
|
mock_get_patch_data.return_value = data
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json,
|
|
FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_patch_data')
|
|
def test_patch_subcloud_forced_manage(self, mock_get_patch_data,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
payload = {'management-state': dccommon_consts.MANAGEMENT_MANAGED,
|
|
'force': True}
|
|
mock_rpc_client().update_subcloud.return_value = True
|
|
mock_get_patch_data.return_value = payload
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id),
|
|
headers=FAKE_HEADERS,
|
|
params=payload)
|
|
mock_rpc_client().update_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
mock.ANY,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED,
|
|
description=None,
|
|
location=None,
|
|
group_id=None,
|
|
data_install=None,
|
|
force=True)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_reconfig_payload')
|
|
def test_reconfigure_subcloud(self, mock_get_reconfig_payload,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password}
|
|
|
|
mock_rpc_client().reconfigure_subcloud.return_value = True
|
|
mock_get_reconfig_payload.return_value = data
|
|
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id) +
|
|
'/reconfigure',
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
mock_rpc_client().reconfigure_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_reconfig_payload')
|
|
def test_reconfigure_subcloud_no_body(self, mock_get_reconfig_payload,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
# Pass an empty request body
|
|
data = {}
|
|
mock_get_reconfig_payload.return_value = data
|
|
mock_rpc_client().reconfigure_subcloud.return_value = True
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reconfigure',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_reconfig_payload')
|
|
def test_reconfigure_subcloud_bad_password(self, mock_get_reconfig_payload,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
# Pass a sysadmin_password which is not base64 encoded
|
|
data = {'sysadmin_password': 'not_base64'}
|
|
mock_get_reconfig_payload.return_value = data
|
|
mock_rpc_client().reconfigure_subcloud.return_value = True
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reconfigure',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_reconfig_payload')
|
|
def test_reconfigure_invalid_deploy_status(self,
|
|
mock_get_reconfig_payload,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(
|
|
self.ctx,
|
|
deploy_status=consts.DEPLOY_STATE_BOOTSTRAP_FAILED)
|
|
fake_password = base64.b64encode('testpass'.encode("utf-8")).decode("utf-8")
|
|
data = {'sysadmin_password': fake_password}
|
|
mock_get_reconfig_payload.return_value = data
|
|
mock_rpc_client().reconfigure_subcloud.return_value = True
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reconfigure',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(rpc_client, 'SubcloudStateClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_updatestatus_payload')
|
|
def test_subcloud_updatestatus(self, mock_get_updatestatus_payload,
|
|
mock_rpc_state_client, _):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'endpoint': 'dc-cert', 'status': 'in-sync'}
|
|
mock_get_updatestatus_payload.return_value = data
|
|
|
|
mock_rpc_state_client().update_subcloud_endpoint_status.return_value = True
|
|
response = self.app.patch_json(
|
|
FAKE_URL + '/' + str(subcloud.id) + '/update_status',
|
|
data, headers=FAKE_HEADERS)
|
|
|
|
mock_rpc_state_client().update_subcloud_endpoint_status.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.name,
|
|
'dc-cert',
|
|
'in-sync')
|
|
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_updatestatus_payload')
|
|
def test_subcloud_updatestatus_invalid_endpoint(
|
|
self, mock_get_updatestatus_payload,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'endpoint': 'any-other-endpoint', 'status': 'in-sync'}
|
|
mock_get_updatestatus_payload.return_value = data
|
|
|
|
mock_rpc_client().update_subcloud_endpoint_status.return_value = True
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/update_status',
|
|
headers=FAKE_HEADERS, params=data)
|
|
mock_rpc_client().update_subcloud_endpoint_status.assert_not_called()
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_updatestatus_payload')
|
|
def test_subcloud_updatestatus_invalid_status(
|
|
self, mock_get_updatestatus_payload,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'endpoint': 'dc-cert', 'status': 'not-sure'}
|
|
mock_get_updatestatus_payload.return_value = data
|
|
|
|
mock_rpc_client().update_subcloud_endpoint_status.return_value = True
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/update_status',
|
|
headers=FAKE_HEADERS, params=data)
|
|
mock_rpc_client().update_subcloud_endpoint_status.assert_not_called()
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_get_config_file_path(self, mock_rpc_client):
|
|
sc = subclouds.SubcloudsController()
|
|
bootstrap_file = sc._get_config_file_path("subcloud1")
|
|
install_values = sc._get_config_file_path("subcloud1", "install_values")
|
|
deploy_config = sc._get_config_file_path("subcloud1", consts.DEPLOY_CONFIG)
|
|
self.assertEqual(bootstrap_file,
|
|
"/var/opt/dc/ansible/subcloud1.yml")
|
|
self.assertEqual(install_values,
|
|
"/var/opt/dc/ansible/subcloud1/install_values.yml")
|
|
self.assertEqual(deploy_config,
|
|
"/var/opt/dc/ansible/subcloud1_deploy_config.yml")
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_format_ip_address(self, mock_rpc_client):
|
|
sc = subclouds.SubcloudsController()
|
|
fake_payload = dict()
|
|
good_values = {
|
|
'10.10.10.3': '10.10.10.3',
|
|
'2620:10a:a001:a103::1135': '2620:10a:a001:a103::1135',
|
|
'2620:10A:A001:A103::1135': '2620:10a:a001:a103::1135', # with upper case letters
|
|
'2620:010a:a001:a103::1135': '2620:10a:a001:a103::1135', # with leading zeros
|
|
'2620:10a:a001:a103:0000::1135': '2620:10a:a001:a103::1135' # with a string of zeros
|
|
}
|
|
|
|
for k, v in good_values.items():
|
|
fake_payload.update({'bootstrap-address': k})
|
|
sc._format_ip_address(fake_payload)
|
|
self.assertEqual(fake_payload['bootstrap-address'], v)
|
|
|
|
fake_payload[subclouds.INSTALL_VALUES] = dict()
|
|
for k, v in good_values.items():
|
|
fake_payload[subclouds.INSTALL_VALUES].update({'bmc_address': k})
|
|
sc._format_ip_address(fake_payload)
|
|
self.assertEqual(fake_payload[subclouds.INSTALL_VALUES]['bmc_address'], v)
|
|
|
|
fake_payload.update({'othervalues1': 'othervalues1'})
|
|
fake_payload[subclouds.INSTALL_VALUES].update({'othervalues2': 'othervalues2'})
|
|
sc._format_ip_address(fake_payload)
|
|
self.assertEqual(fake_payload['othervalues1'], 'othervalues1')
|
|
self.assertEqual(fake_payload[subclouds.INSTALL_VALUES]['othervalues2'], 'othervalues2')
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(keyring, 'get_password')
|
|
def test_get_subcloud_db_install_values(
|
|
self, mock_keyring, mock_rpc_client):
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
test_subcloud = copy.copy(FAKE_SUBCLOUD_DATA)
|
|
subcloud_info = Subcloud(test_subcloud, False)
|
|
subcloud_info.data_install = json.dumps(install_data)
|
|
|
|
sc = subclouds.SubcloudsController()
|
|
actual_result = sc._get_subcloud_db_install_values(subcloud_info)
|
|
actual_result.update({
|
|
'admin_password': 'adminpass'
|
|
})
|
|
install_data.update({
|
|
'ansible_become_pass': consts.TEMP_SYSADMIN_PASSWORD,
|
|
'ansible_ssh_pass': consts.TEMP_SYSADMIN_PASSWORD,
|
|
'admin_password': 'adminpass'
|
|
})
|
|
self.assertEqual(
|
|
json.loads(json.dumps(install_data)),
|
|
json.loads(json.dumps(actual_result)))
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(keyring, 'get_password')
|
|
def test_get_subcloud_db_install_values_without_bmc_password(
|
|
self, mock_keyring, mock_rpc_client):
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
subcloud = fake_subcloud.create_fake_subcloud(
|
|
self.ctx, data_install=json.dumps(install_data))
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS)
|
|
|
|
@mock.patch.object(cutils, 'get_vault_load_files')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_oam_network_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_upload_deploy_config_file')
|
|
def test_reinstall_subcloud(
|
|
self, mock_upload_deploy_config_file,
|
|
mock_get_request_data, mock_validate_oam_network_config,
|
|
mock_get_subcloud_db_install_values, mock_rpc_client,
|
|
mock_get_vault_load_files):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
reinstall_data = copy.copy(FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD)
|
|
mock_get_request_data.return_value = reinstall_data
|
|
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
mock_get_vault_load_files.return_value = ('iso_file_path', 'sig_file_path')
|
|
|
|
response = self.app.patch_json(
|
|
FAKE_URL + '/' + str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params=reinstall_data)
|
|
|
|
mock_validate_oam_network_config.assert_called_once()
|
|
mock_rpc_client().reinstall_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
mock_upload_deploy_config_file.assert_called_once()
|
|
self.assertEqual(SW_VERSION, response.json['software-version'])
|
|
|
|
@mock.patch.object(cutils, 'get_vault_load_files')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController,
|
|
'_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_oam_network_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_upload_deploy_config_file')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_k8s_version')
|
|
def test_reinstall_subcloud_with_release_parameter(
|
|
self, mock_validate_k8s_version, mock_upload_deploy_config_file,
|
|
mock_get_request_data, mock_validate_oam_network_config,
|
|
mock_get_subcloud_db_install_values, mock_rpc_client,
|
|
mock_get_vault_load_files):
|
|
|
|
software_version = '21.12'
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
reinstall_data = copy.copy(FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD)
|
|
reinstall_data['release'] = software_version
|
|
mock_get_request_data.return_value = reinstall_data
|
|
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
mock_get_vault_load_files.return_value = ('iso_file_path', 'sig_file_path')
|
|
|
|
response = self.app.patch_json(
|
|
FAKE_URL + '/' + str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params=reinstall_data)
|
|
|
|
mock_validate_oam_network_config.assert_called_once()
|
|
mock_rpc_client().reinstall_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
subcloud.id,
|
|
mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
mock_validate_k8s_version.assert_called_once()
|
|
mock_upload_deploy_config_file.assert_called_once()
|
|
self.assertEqual(software_version, response.json['software-version'])
|
|
self.assertIn(software_version,
|
|
json.loads(response.json['data_install'])['software_version'])
|
|
|
|
@mock.patch.object(cutils, 'get_vault_load_files')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_oam_network_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
def test_reinstall_subcloud_no_body(
|
|
self, mock_get_request_data, mock_validate_oam_network_config,
|
|
mock_get_subcloud_db_install_values, mock_rpc_client,
|
|
mock_get_vault_load_files):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
mock_get_request_data.return_value = {}
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
|
|
mock_validate_oam_network_config.assert_not_called()
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params={})
|
|
|
|
@mock.patch.object(cutils, 'get_vault_load_files')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_oam_network_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
def test_reinstall_online_subcloud(
|
|
self, mock_get_request_data, mock_validate_oam_network_config,
|
|
mock_get_subcloud_db_install_values, mock_rpc_client,
|
|
mock_get_vault_load_files):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
db_api.subcloud_update(self.ctx,
|
|
subcloud.id,
|
|
availability_status=dccommon_consts.AVAILABILITY_ONLINE)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
reinstall_data = copy.copy(FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD)
|
|
mock_get_request_data.return_value = reinstall_data
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
|
|
mock_validate_oam_network_config.assert_not_called()
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params={})
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
def test_reinstall_subcloud_missing_required_value(
|
|
self, mock_get_request_data, mock_get_subcloud_db_install_values,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
|
|
for k in ['name', 'system_mode', 'external_oam_subnet',
|
|
'external_oam_gateway_address', 'external_oam_floating_address',
|
|
'sysadmin_password']:
|
|
reinstall_data = copy.copy(FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD)
|
|
del reinstall_data[k]
|
|
mock_get_request_data.return_value = reinstall_data
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params=reinstall_data)
|
|
|
|
@mock.patch.object(cutils, 'get_vault_load_files')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_oam_network_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
def test_reinstall_subcloud_missing_stored_value(
|
|
self, mock_get_request_data, mock_validate_oam_network_config,
|
|
mock_get_subcloud_db_install_values, mock_rpc_client,
|
|
mock_get_vault_load_files):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
mock_get_vault_load_files.return_value = ('iso_file_path', 'sig_file_path')
|
|
|
|
for k in ['management_subnet', 'management_start_address', 'management_end_address',
|
|
'management_gateway_address', 'systemcontroller_gateway_address']:
|
|
reinstall_data = copy.copy(FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD)
|
|
del reinstall_data[k]
|
|
mock_get_request_data.return_value = reinstall_data
|
|
response = self.app.patch_json(
|
|
FAKE_URL + '/' + str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params=reinstall_data)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(cutils, 'get_vault_load_files')
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_subcloud_db_install_values')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_validate_subcloud_config')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_request_data')
|
|
def test_reinstall_subcloud_stored_value_not_match(
|
|
self, mock_get_request_data, mock_validate_subcloud_config,
|
|
mock_get_subcloud_db_install_values, mock_rpc_client,
|
|
mock_get_vault_load_files):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
install_data = copy.copy(FAKE_SUBCLOUD_INSTALL_VALUES)
|
|
|
|
encoded_password = base64.b64encode(
|
|
'bmc_password'.encode("utf-8")).decode('utf-8')
|
|
bmc_password = {'bmc_password': encoded_password}
|
|
install_data.update(bmc_password)
|
|
mock_get_subcloud_db_install_values.return_value = install_data
|
|
|
|
mock_rpc_client().reinstall_subcloud.return_value = True
|
|
mock_get_vault_load_files.return_value = ('iso_file_path', 'sig_file_path')
|
|
|
|
for k in ['management_subnet', 'management_start_address', 'management_end_address',
|
|
'management_gateway_address', 'systemcontroller_gateway_address']:
|
|
reinstall_data = copy.copy(FAKE_SUBCLOUD_BOOTSTRAP_PAYLOAD)
|
|
reinstall_data[k] = 'wrong_value'
|
|
mock_get_request_data.return_value = reinstall_data
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/reinstall',
|
|
headers=FAKE_HEADERS, params=reinstall_data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(prestage, '_get_prestage_subcloud_info')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_validate_detailed(self, mock_get_prestage_payload,
|
|
mock_prestage_subcloud_info,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED)
|
|
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password,
|
|
'force': False}
|
|
mock_controller_upgrade.return_value = list()
|
|
mock_prestage_subcloud_info.return_value = consts.SYSTEM_MODE_SIMPLEX, \
|
|
health_report_no_alarm, \
|
|
OAM_FLOATING_IP
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id) +
|
|
'/prestage',
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
mock_rpc_client().prestage_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
def test_prestage_subcloud_unmanaged(self, mock_controller_upgrade,
|
|
mock_get_prestage_payload,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_UNMANAGED)
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password}
|
|
mock_controller_upgrade.return_value = list()
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
def test_prestage_subcloud_offline(self, mock_controller_upgrade,
|
|
mock_get_prestage_payload,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_OFFLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED)
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password}
|
|
mock_controller_upgrade.return_value = list()
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(prestage, '_get_prestage_subcloud_info')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_duplex(self, mock_get_prestage_payload,
|
|
mock_prestage_subcloud_info,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED)
|
|
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password,
|
|
'force': False}
|
|
mock_controller_upgrade.return_value = list()
|
|
mock_prestage_subcloud_info.return_value = consts.SYSTEM_MODE_DUPLEX, \
|
|
health_report_no_alarm, \
|
|
OAM_FLOATING_IP
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(prestage, '_get_prestage_subcloud_info')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_non_mgmt_alarm(self, mock_get_prestage_payload,
|
|
mock_prestage_subcloud_info,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED)
|
|
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password,
|
|
'force': False}
|
|
mock_controller_upgrade.return_value = list()
|
|
mock_prestage_subcloud_info.return_value = consts.SYSTEM_MODE_SIMPLEX, \
|
|
health_report_no_mgmt_alarm, \
|
|
OAM_FLOATING_IP
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id) +
|
|
'/prestage',
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
mock_rpc_client().prestage_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(prestage, '_get_prestage_subcloud_info')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_mgmt_alarm(self, mock_get_prestage_payload,
|
|
mock_prestage_subcloud_info,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED)
|
|
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password,
|
|
'force': False}
|
|
mock_controller_upgrade.return_value = list()
|
|
mock_prestage_subcloud_info.return_value = consts.SYSTEM_MODE_SIMPLEX, \
|
|
health_report_mgmt_alarm, \
|
|
OAM_FLOATING_IP
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(prestage, '_get_prestage_subcloud_info')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_mgmt_alarm_force(self, mock_get_prestage_payload,
|
|
mock_prestage_subcloud_info,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(
|
|
self.ctx, subcloud.id, availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED)
|
|
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password,
|
|
'force': True}
|
|
mock_controller_upgrade.return_value = list()
|
|
mock_prestage_subcloud_info.return_value = consts.SYSTEM_MODE_SIMPLEX, \
|
|
health_report_mgmt_alarm, \
|
|
OAM_FLOATING_IP
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
response = self.app.patch_json(FAKE_URL + '/' + str(subcloud.id) +
|
|
'/prestage',
|
|
headers=FAKE_HEADERS,
|
|
params=data)
|
|
mock_rpc_client().prestage_subcloud.assert_called_once_with(
|
|
mock.ANY,
|
|
mock.ANY)
|
|
self.assertEqual(response.status_int, 200)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(prestage, '_get_prestage_subcloud_info')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_not_allowed_state(self, mock_get_prestage_payload,
|
|
mock_prestage_subcloud_info,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
subcloud = db_api.subcloud_update(self.ctx, subcloud.id,
|
|
availability_status=dccommon_consts.AVAILABILITY_ONLINE,
|
|
management_state=dccommon_consts.MANAGEMENT_MANAGED,
|
|
deploy_status='NotAllowedState')
|
|
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password,
|
|
'force': False}
|
|
mock_controller_upgrade.return_value = list()
|
|
mock_prestage_subcloud_info.return_value = consts.SYSTEM_MODE_SIMPLEX, \
|
|
health_report_no_alarm, \
|
|
OAM_FLOATING_IP
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_controller_upgrading(self, mock_get_prestage_payload,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
fake_password = (base64.b64encode('testpass'.encode("utf-8"))).decode('ascii')
|
|
data = {'sysadmin_password': fake_password}
|
|
mock_controller_upgrade.return_value = list('upgrade')
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_no_password(self, mock_get_prestage_payload,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {}
|
|
mock_controller_upgrade.return_value = list()
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
@mock.patch.object(prestage, '_get_system_controller_upgrades')
|
|
@mock.patch.object(subclouds.SubcloudsController, '_get_prestage_payload')
|
|
def test_prestage_subcloud_password_not_encoded(self, mock_get_prestage_payload,
|
|
mock_controller_upgrade,
|
|
mock_rpc_client):
|
|
|
|
subcloud = fake_subcloud.create_fake_subcloud(self.ctx)
|
|
data = {'sysadmin_password': 'notencoded'}
|
|
mock_controller_upgrade.return_value = list()
|
|
|
|
mock_rpc_client().prestage_subcloud.return_value = True
|
|
mock_get_prestage_payload.return_value = data
|
|
|
|
six.assertRaisesRegex(self, webtest.app.AppError, "400 *",
|
|
self.app.patch_json, FAKE_URL + '/' +
|
|
str(subcloud.id) + '/prestage',
|
|
headers=FAKE_HEADERS, params=data)
|
|
|
|
def test_get_management_subnet(self):
|
|
payload = {
|
|
'management_subnet': "192.168.204.0/24"
|
|
}
|
|
self.assertEqual(cutils.get_management_subnet(payload), payload['management_subnet'])
|
|
|
|
def test_get_management_subnet_return_admin(self):
|
|
payload = {
|
|
'admin_subnet': "192.168.205.0/24",
|
|
'management_subnet': "192.168.204.0/24"
|
|
}
|
|
self.assertEqual(cutils.get_management_subnet(payload), payload['admin_subnet'])
|
|
|
|
def test_get_management_start_address(self):
|
|
payload = {
|
|
'management_start_address': "192.168.204.2"
|
|
}
|
|
self.assertEqual(cutils.get_management_start_address(payload), payload['management_start_address'])
|
|
|
|
def test_get_management_start_address_return_admin(self):
|
|
payload = {
|
|
'admin_start_address': "192.168.205.2",
|
|
'management_start_address': "192.168.204.2"
|
|
}
|
|
self.assertEqual(cutils.get_management_start_address(payload), payload['admin_start_address'])
|
|
|
|
def test_get_management_end_address(self):
|
|
payload = {
|
|
'management_end_address': "192.168.204.50"
|
|
}
|
|
self.assertEqual(cutils.get_management_end_address(payload), payload['management_end_address'])
|
|
|
|
def test_get_management_end_address_return_admin(self):
|
|
payload = {
|
|
'admin_end_address': "192.168.205.50",
|
|
'management_end_address': "192.168.204.50"
|
|
}
|
|
self.assertEqual(cutils.get_management_end_address(payload), payload['admin_end_address'])
|
|
|
|
def test_get_management_gateway_address(self):
|
|
payload = {
|
|
'management_gateway_address': "192.168.204.1"
|
|
}
|
|
self.assertEqual(cutils.get_management_gateway_address(payload), payload['management_gateway_address'])
|
|
|
|
def test_get_management_gateway_address_return_admin(self):
|
|
payload = {
|
|
'admin_gateway_address': "192.168.205.1",
|
|
'management_gateway_address': "192.168.204.1"
|
|
}
|
|
self.assertEqual(cutils.get_management_gateway_address(payload), payload['admin_gateway_address'])
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_validate_admin_config_subnet_small(self, mock_rpc_client):
|
|
|
|
admin_subnet = "192.168.205.0/32"
|
|
admin_start_address = "192.168.205.2"
|
|
admin_end_address = "192.168.205.50"
|
|
admin_gateway_address = "192.168.205.1"
|
|
|
|
six.assertRaisesRegex(self,
|
|
Exception,
|
|
"Subnet too small*",
|
|
subclouds.SubcloudsController().
|
|
_validate_admin_network_config,
|
|
admin_subnet,
|
|
admin_start_address,
|
|
admin_end_address,
|
|
admin_gateway_address,
|
|
existing_networks=None)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_validate_admin_config_start_address_outOfSubnet(self, mock_rpc_client):
|
|
|
|
admin_subnet = "192.168.205.0/28"
|
|
admin_start_address = "192.168.205.200"
|
|
admin_end_address = "192.168.205.50"
|
|
admin_gateway_address = "192.168.205.1"
|
|
|
|
six.assertRaisesRegex(self,
|
|
Exception,
|
|
"Address must be in subnet*",
|
|
subclouds.SubcloudsController().
|
|
_validate_admin_network_config,
|
|
admin_subnet,
|
|
admin_start_address,
|
|
admin_end_address,
|
|
admin_gateway_address,
|
|
existing_networks=None)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_validate_admin_config_end_address_outOfSubnet(self, mock_rpc_client):
|
|
|
|
admin_subnet = "192.168.205.0/28"
|
|
admin_start_address = "192.168.205.1"
|
|
admin_end_address = "192.168.205.50"
|
|
admin_gateway_address = "192.168.205.1"
|
|
|
|
six.assertRaisesRegex(self,
|
|
Exception,
|
|
"Address must be in subnet*",
|
|
subclouds.SubcloudsController().
|
|
_validate_admin_network_config,
|
|
admin_subnet,
|
|
admin_start_address,
|
|
admin_end_address,
|
|
admin_gateway_address,
|
|
existing_networks=None)
|
|
|
|
@mock.patch.object(rpc_client, 'ManagerClient')
|
|
def test_validate_admin_config_gateway_address_outOfSubnet(self, mock_rpc_client):
|
|
|
|
admin_subnet = "192.168.205.0/28"
|
|
admin_start_address = "192.168.205.1"
|
|
admin_end_address = "192.168.205.12"
|
|
admin_gateway_address = "192.168.205.50"
|
|
|
|
six.assertRaisesRegex(self,
|
|
Exception,
|
|
"Address must be in subnet*",
|
|
subclouds.SubcloudsController().
|
|
_validate_admin_network_config,
|
|
admin_subnet,
|
|
admin_start_address,
|
|
admin_end_address,
|
|
admin_gateway_address,
|
|
existing_networks=None)
|