1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
###############################################################################
# Copyright (c) 2017 Koren Lev (Cisco Systems), Yaron Yogev (Cisco Systems) #
# and others #
# #
# All rights reserved. This program and the accompanying materials #
# are made available under the terms of the Apache License, Version 2.0 #
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
from unittest.mock import MagicMock
from discover.fetchers.api.api_fetch_project_hosts import ApiFetchProjectHosts
from test.fetch.test_fetch import TestFetch
from test.fetch.api_fetch.test_data.api_fetch_host_project_hosts import *
from test.fetch.api_fetch.test_data.token import TOKEN
from test.fetch.api_fetch.test_data.regions import REGIONS
class TestApiFetchProjectHosts(TestFetch):
def setUp(self):
super().setUp()
self.configure_environment()
self._v2_auth_pwd = ApiFetchProjectHosts.v2_auth_pwd
ApiFetchProjectHosts.v2_auth_pwd = MagicMock(return_value=TOKEN)
self.fetcher = ApiFetchProjectHosts()
self.set_regions_for_fetcher(self.fetcher)
self.region = REGIONS[REGION_NAME]
def test_add_host_type_with_nonexistent_type(self):
# clear host type
HOST_DOC["host_type"] = []
self.fetcher.add_host_type(HOST_DOC, NONEXISTENT_TYPE, HOST_ZONE)
self.assertIn(NONEXISTENT_TYPE, HOST_DOC["host_type"], "Can't put nonexistent " +
"type in host_type")
def test_add_host_type_with_existent_host_type(self):
# add nonexistent host type to host type
HOST_DOC["host_type"] = [NONEXISTENT_TYPE]
# try to add existing host type
self.fetcher.add_host_type(HOST_DOC, NONEXISTENT_TYPE, HOST_ZONE)
self.assertEqual(len(HOST_DOC['host_type']), 1, "Add duplicate host type")
def test_add_compute_host_type(self):
HOST_DOC['host_type'] = []
# clear zone
HOST_DOC['zone'] = None
# add compute host type
self.fetcher.add_host_type(HOST_DOC, COMPUTE_TYPE, HOST_ZONE)
# for compute host type, zone information will be added
self.assertEqual(HOST_DOC['zone'], HOST_ZONE, "Can't update zone " +
"name for compute node")
self.assertEqual(HOST_DOC['parent_id'], HOST_ZONE, "Can't update parent_id " +
"for compute node")
def test_fetch_compute_node_ip_address(self):
# mock ip address information fetched from DB
self.fetcher.get_objects_list_for_id = MagicMock(return_value=IP_ADDRESS_RESPONSE)
self.fetcher.fetch_compute_node_ip_address(HOST_TO_BE_FETCHED_IP,
HOST_TO_BE_FETCHED_IP["host"])
self.assertIn("ip_address", HOST_TO_BE_FETCHED_IP, "Can't update ip address " +
"for the compute host")
def test_fetch_network_node_details(self):
# mock NETWORKS_DETAILS_RESPONSE fetched from DB
self.fetcher.get_objects_list = MagicMock(return_value=NETWORKS_DETAILS_RESPONSE)
self.fetcher.fetch_network_node_details(HOSTS_TO_BE_FETCHED_NETWORK_DETAILS)
# get the network node document from HOSTS
NETWORK_NODE_DOC = [doc for doc in HOSTS_TO_BE_FETCHED_NETWORK_DETAILS
if doc['host'] == HOST_NAME][0]
# check if the network node document has been updated
self.assertIn("Network", NETWORK_NODE_DOC['host_type'], "Can't put Network in " +
"the network node host_type")
self.assertIn("config", NETWORK_NODE_DOC, "Can't put config in the network node")
def test_get_host_details(self):
# test node have nova-conductor attribute, controller type will be added
result = self.fetcher.get_host_details(AVAILABILITY_ZONE, HOST_NAME)
self.assertIn("Controller", result['host_type'], "Can't put controller type " +
"in the compute node host_type")
def test_get_hosts_from_az(self):
result = self.fetcher.get_hosts_from_az(AVAILABILITY_ZONE)
self.assertNotEqual(result, [], "Can't get hosts information from "
"availability zone")
def test_get_for_region(self):
# mock region url for nova node
self.fetcher.get_region_url = MagicMock(return_value=REGION_URL)
# mock the response from OpenStack Api
side_effect = [AVAILABILITY_ZONE_RESPONSE, HYPERVISORS_RESPONSE]
self.fetcher.get_url = MagicMock(side_effect=side_effect)
result = self.fetcher.get_for_region(self.region, TOKEN)
self.assertNotEqual(result, [], "Can't get hosts information for region")
def test_get_for_region_without_token(self):
self.fetcher.get_region_url = MagicMock(return_value=REGION_URL)
result = self.fetcher.get_for_region(self.region, None)
self.assertEqual(result, [], "Can't get [] when the token is invalid")
def test_get_for_region_with_error_availability_response(self):
self.fetcher.get_region_url = MagicMock(return_value=REGION_URL)
# mock error availability zone response from OpenStack Api
side_effect = [AVAILABILITY_ERROR_RESPONSE, None]
self.fetcher.get_url = MagicMock(side_effect=side_effect)
result = self.fetcher.get_for_region(self.region, TOKEN)
self.assertEqual(result, [], "Can't get [] when the response is wrong")
def test_get_for_region_with_error_hypervisors_response(self):
self.fetcher.get_region_url = MagicMock(return_value=REGION_URL)
# mock error hypervisors response from OpenStack Api
side_effect = [AVAILABILITY_ZONE_RESPONSE, HYPERVISORS_ERROR_RESPONSE]
self.fetcher.get_url = MagicMock(side_effect=side_effect)
result = self.fetcher.get_for_region(self.region, TOKEN)
self.assertNotEqual(result, [], "Can't get hosts information when " +
"the hypervisors response is wrong")
def test_get(self):
original_method = self.fetcher.get_for_region
self.fetcher.get_for_region = MagicMock(return_value=GET_FOR_REGION_INFO)
result = self.fetcher.get(PROJECT_NAME)
self.fetcher.get_for_region = original_method
self.assertNotEqual(result, [], "Can't get hosts info for the project")
def test_get_with_wrong_project_name(self):
result = self.fetcher.get(TEST_PROJECT_NAME)
self.assertEqual(result, [], "Can't get [] when the project name is not admin")
def test_get_with_wrong_token(self):
self.fetcher.v2_auth_pwd = MagicMock(return_value=[])
result = self.fetcher.get(PROJECT_NAME)
self.assertEqual(result, [], "Can't get [] when the token is invalid")
def tearDown(self):
super().tearDown()
ApiFetchProjectHosts.v2_auth_pwd = self._v2_auth_pwd
self.reset_regions_for_fetcher(self.fetcher)
|
