diff options
Diffstat (limited to 'app/test/fetch')
7 files changed, 533 insertions, 32 deletions
diff --git a/app/test/fetch/api_fetch/test_api_access.py b/app/test/fetch/api_fetch/test_api_access.py index 0effc0e..440b730 100644 --- a/app/test/fetch/api_fetch/test_api_access.py +++ b/app/test/fetch/api_fetch/test_api_access.py @@ -7,9 +7,9 @@ # which accompanies this distribution, and is available at # # http://www.apache.org/licenses/LICENSE-2.0 # ############################################################################### -from unittest.mock import MagicMock, Mock - +import copy import requests +from unittest.mock import MagicMock, Mock from discover.fetchers.api.api_access import ApiAccess from test.fetch.api_fetch.test_data.api_access import * @@ -35,38 +35,45 @@ class TestApiAccess(TestFetch): def test_parse_illegal_time(self): time = self.api_access.parse_time(ILLEGAL_TIME) - self.assertEqual(time, None, "Can't get None when the time format is wrong") + self.assertEqual(time, None, + "Can't get None when the time format is wrong") def test_get_existing_token(self): self.api_access.tokens = VALID_TOKENS token = self.api_access.get_existing_token(PROJECT) - self.assertNotEqual(token, VALID_TOKENS[PROJECT], "Can't get existing token") + self.assertNotEqual(token, VALID_TOKENS[PROJECT], + "Can't get existing token") def test_get_nonexistent_token(self): self.api_access.tokens = EMPTY_TOKENS token = self.api_access.get_existing_token(TEST_PROJECT) - self.assertEqual(token, None, "Can't get None when the token doesn't " + - "exist in tokens") + self.assertEqual(token, None, + "Can't get None when the token doesn't exist " + "in tokens") def test_v2_auth(self): self.api_access.get_existing_token = MagicMock(return_value=None) self.response.json = Mock(return_value=CORRECT_AUTH_CONTENT) # mock authentication info from OpenStack Api - token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, TEST_BODY) + token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, + TEST_BODY) self.assertNotEqual(token_details, None, "Can't get the token details") def test_v2_auth_with_error_content(self): self.api_access.get_existing_token = MagicMock(return_value=None) self.response.json = Mock(return_value=ERROR_AUTH_CONTENT) # authentication content from OpenStack Api will be incorrect - token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, TEST_BODY) - self.assertIs(token_details, None, "Can't get None when the content is wrong") + token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, + TEST_BODY) + self.assertIs(token_details, None, + "Can't get None when the content is wrong") def test_v2_auth_with_error_token(self): self.response.status_code = requests.codes.bad_request self.response.json = Mock(return_value=ERROR_TOKEN_CONTENT) # authentication info from OpenStack Api will not contain token info - token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, TEST_BODY) + token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, + TEST_BODY) self.assertIs(token_details, None, "Can't get None when the content " + "doesn't contain any token info") @@ -78,12 +85,13 @@ class TestApiAccess(TestFetch): # the time will not be parsed self.api_access.parse_time = MagicMock(return_value=None) - token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, TEST_BODY) + token_details = self.api_access.v2_auth(TEST_PROJECT, TEST_HEADER, + TEST_BODY) # reset original parse_time method self.api_access.parse_time = original_method - self.assertIs(token_details, None, "Can't get None when the time in token " + - "can't be parsed") + self.assertIs(token_details, None, + "Can't get None when the time in token can't be parsed") def test_v2_auth_pwd(self): self.response.json = Mock(return_value=CORRECT_AUTH_CONTENT) @@ -92,20 +100,30 @@ class TestApiAccess(TestFetch): self.assertNotEqual(token, None, "Can't get token") def test_get_url(self): - self.response.json = Mock(return_value=GET_CONTENT) + get_response = copy.deepcopy(self.response) + get_response.status_code = requests.codes.ok + self.requests_get = requests.get + requests.get = MagicMock(return_value=get_response) + get_response.json = Mock(return_value=GET_CONTENT) result = self.api_access.get_url(TEST_URL, TEST_HEADER) # check whether it returns content message when the response is correct self.assertNotEqual(result, None, "Can't get content when the " "response is correct") + requests.get = self.requests_get def test_get_url_with_error_response(self): - self.response.status_code = requests.codes.bad_request - self.response.json = Mock(return_value=None) - self.response.text = "Bad request" + get_response = copy.deepcopy(self.response) + get_response.status_code = requests.codes.bad_request + get_response.text = "Bad request" + get_response.json = Mock(return_value=GET_CONTENT) + self.requests_get = requests.get + requests.get = MagicMock(return_value=get_response) + # the response will be wrong result = self.api_access.get_url(TEST_URL, TEST_HEADER) self.assertEqual(result, None, "Result returned" + "when the response status is not 200") + requests.get = self.requests_get def test_get_region_url(self): region_url = self.api_access.get_region_url(REGION_NAME, SERVICE_NAME) @@ -120,23 +138,30 @@ class TestApiAccess(TestFetch): def test_get_region_url_without_service_endpoint(self): # error service doesn't exist in region service endpoints - region_url = self.api_access.get_region_url(REGION_NAME, ERROR_SERVICE_NAME) - self.assertIs(region_url, None, "Can't get None with wrong service name") + region_url = self.api_access.get_region_url(REGION_NAME, + ERROR_SERVICE_NAME) + self.assertIs(region_url, None, + "Can't get None with wrong service name") def test_region_url_nover(self): - # mock return value of get_region_url, which has something starting from v2 + # mock return value of get_region_url, + # which has something starting from v2 self.api_access.get_region_url = MagicMock(return_value=REGION_URL) - region_url = self.api_access.get_region_url_nover(REGION_NAME, SERVICE_NAME) + region_url = self.api_access.get_region_url_nover(REGION_NAME, + SERVICE_NAME) # get_region_nover will remove everything from v2 - self.assertNotIn("v2", region_url, "Can't get region url without v2 info") + self.assertNotIn("v2", region_url, + "Can't get region url without v2 info") def test_get_service_region_endpoints(self): region = REGIONS[REGION_NAME] - result = self.api_access.get_service_region_endpoints(region, SERVICE_NAME) + result = self.api_access.get_service_region_endpoints(region, + SERVICE_NAME) self.assertNotEqual(result, None, "Can't get service endpoint") def test_get_service_region_endpoints_with_nonexistent_service(self): region = REGIONS[REGION_NAME] - result = self.api_access.get_service_region_endpoints(region, ERROR_SERVICE_NAME) + get_endpoints = self.api_access.get_service_region_endpoints + result = get_endpoints(region, ERROR_SERVICE_NAME) self.assertIs(result, None, "Can't get None when the service name " + "doesn't exist in region's services") diff --git a/app/test/fetch/api_fetch/test_api_fetch_project_hosts.py b/app/test/fetch/api_fetch/test_api_fetch_project_hosts.py index da3df17..784079e 100644 --- a/app/test/fetch/api_fetch/test_api_fetch_project_hosts.py +++ b/app/test/fetch/api_fetch/test_api_fetch_project_hosts.py @@ -7,6 +7,7 @@ # which accompanies this distribution, and is available at # # http://www.apache.org/licenses/LICENSE-2.0 # ############################################################################### +import copy from unittest.mock import MagicMock from discover.fetchers.api.api_fetch_project_hosts import ApiFetchProjectHosts from test.fetch.test_fetch import TestFetch @@ -36,23 +37,28 @@ class TestApiFetchProjectHosts(TestFetch): "type in host_type") def test_add_host_type_with_existent_host_type(self): + fetch_host_os_details = self.fetcher.fetch_host_os_details + self.fetcher.fetch_host_os_details = MagicMock() # 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") + self.assertEqual(len(HOST_DOC['host_type']), 1, + "Add duplicate host type") + self.fetcher.fetch_host_os_details = fetch_host_os_details def test_add_compute_host_type(self): - HOST_DOC['host_type'] = [] + doc = copy.deepcopy(HOST_DOC) + doc['host_type'] = [] # clear zone - HOST_DOC['zone'] = None + doc['zone'] = None # add compute host type - self.fetcher.add_host_type(HOST_DOC, COMPUTE_TYPE, HOST_ZONE) + self.fetcher.add_host_type(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") + self.assertEqual(doc['zone'], HOST_ZONE, + "Can't update zone name for compute node") + self.assertEqual(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 @@ -78,16 +84,24 @@ class TestApiFetchProjectHosts(TestFetch): def test_get_host_details(self): # test node have nova-conductor attribute, controller type will be added + fetch_host_os_details = self.fetcher.fetch_host_os_details + self.fetcher.fetch_host_os_details = MagicMock() 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") + self.fetcher.fetch_host_os_details = fetch_host_os_details def test_get_hosts_from_az(self): + fetch_host_os_details = self.fetcher.fetch_host_os_details + self.fetcher.fetch_host_os_details = MagicMock() result = self.fetcher.get_hosts_from_az(AVAILABILITY_ZONE) self.assertNotEqual(result, [], "Can't get hosts information from " "availability zone") + self.fetcher.fetch_host_os_details = fetch_host_os_details def test_get_for_region(self): + fetch_host_os_details = self.fetcher.fetch_host_os_details + self.fetcher.fetch_host_os_details = MagicMock() # mock region url for nova node self.fetcher.get_region_url = MagicMock(return_value=REGION_URL) # mock the response from OpenStack Api @@ -96,6 +110,7 @@ class TestApiFetchProjectHosts(TestFetch): result = self.fetcher.get_for_region(self.region, TOKEN) self.assertNotEqual(result, [], "Can't get hosts information for region") + self.fetcher.fetch_host_os_details = fetch_host_os_details def test_get_for_region_without_token(self): self.fetcher.get_region_url = MagicMock(return_value=REGION_URL) @@ -112,6 +127,8 @@ class TestApiFetchProjectHosts(TestFetch): self.assertEqual(result, [], "Can't get [] when the response is wrong") def test_get_for_region_with_error_hypervisors_response(self): + fetch_host_os_details = self.fetcher.fetch_host_os_details + self.fetcher.fetch_host_os_details = MagicMock() 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] @@ -120,6 +137,7 @@ class TestApiFetchProjectHosts(TestFetch): result = self.fetcher.get_for_region(self.region, TOKEN) self.assertNotEqual(result, [], "Can't get hosts information when " + "the hypervisors response is wrong") + self.fetcher.fetch_host_os_details = fetch_host_os_details def test_get(self): original_method = self.fetcher.get_for_region @@ -140,6 +158,15 @@ class TestApiFetchProjectHosts(TestFetch): result = self.fetcher.get(PROJECT_NAME) self.assertEqual(result, [], "Can't get [] when the token is invalid") + def test_fetch_host_os_details(self): + original_method = self.fetcher.run + self.fetcher.run = MagicMock(return_value=OS_DETAILS_INPUT) + doc = {'host': 'host1'} + self.fetcher.fetch_host_os_details(doc) + self.assertEqual(doc.get('OS', {}), OS_DETAILS) + self.fetcher.run = original_method + + def tearDown(self): super().tearDown() ApiFetchProjectHosts.v2_auth_pwd = self._v2_auth_pwd diff --git a/app/test/fetch/api_fetch/test_data/api_fetch_host_project_hosts.py b/app/test/fetch/api_fetch/test_data/api_fetch_host_project_hosts.py index 3ef1ac7..ba42590 100644 --- a/app/test/fetch/api_fetch/test_data/api_fetch_host_project_hosts.py +++ b/app/test/fetch/api_fetch/test_data/api_fetch_host_project_hosts.py @@ -223,3 +223,24 @@ GET_FOR_REGION_INFO = [ "zone": "osdna-zone" } ] + +OS_DETAILS_INPUT = """ +NAME="Ubuntu" +VERSION="16.04 LTS (Xenial Xerus)" +ID=ubuntu +ID_LIKE=debian +PRETTY_NAME="Ubuntu 16.04 LTS" +VERSION_ID="16.04" +HOME_URL="http://www.ubuntu.com/" +SUPPORT_URL="http://help.ubuntu.com/" +BUG_REPORT_URL="http://bugs.launchpad.net/ubuntu/" +UBUNTU_CODENAME=xenial +ARCHITECURE=x86_64 +""" +OS_DETAILS = { + 'name': 'Ubuntu', + 'version': '16.04 LTS (Xenial Xerus)', + 'ID': 'ubuntu', + 'ID_LIKE': 'debian', + 'architecure': 'x86_64' +} diff --git a/app/test/fetch/link_finders/__init__.py b/app/test/fetch/link_finders/__init__.py new file mode 100644 index 0000000..b0637e9 --- /dev/null +++ b/app/test/fetch/link_finders/__init__.py @@ -0,0 +1,9 @@ +############################################################################### +# 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 # +############################################################################### diff --git a/app/test/fetch/link_finders/test_data/__init__.py b/app/test/fetch/link_finders/test_data/__init__.py new file mode 100644 index 0000000..b0637e9 --- /dev/null +++ b/app/test/fetch/link_finders/test_data/__init__.py @@ -0,0 +1,9 @@ +############################################################################### +# 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 # +############################################################################### diff --git a/app/test/fetch/link_finders/test_data/test_find_implicit_links.py b/app/test/fetch/link_finders/test_data/test_find_implicit_links.py new file mode 100644 index 0000000..aef20f6 --- /dev/null +++ b/app/test/fetch/link_finders/test_data/test_find_implicit_links.py @@ -0,0 +1,303 @@ +############################################################################### +# 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 # +############################################################################### +ENV = 'env1' +CLIQUE_CONSTRAINTS = [ + { + 'focal_point_type': 'instance', + 'constraints': ['network'] + }, + { + 'focal_point_type': 'dummy1', + 'constraints': [] + }, + { + 'focal_point_type': 'dummy2', + 'constraints': ['network', 'dummy_constraint'] + }, + { + 'focal_point_type': 'dummy3', + 'constraints': ['dummy_constraint2'] + } +] +CONSTRAINTS = ['network', 'dummy_constraint', 'dummy_constraint2'] + +LINK_ATTRIBUTES_NONE = {} +LINK_ATTRIBUTES_NONE_2 = {} +LINK_ATTRIBUTES_EMPTY = {'attributes': []} +LINK_ATTR_V1 = {'attributes': {'network': 'v1'}} +LINK_ATTR_V1_2 = {'attributes': {'network': 'v1'}} +LINK_ATTR_V2 = {'attributes': {'network': 'v2'}} +LINK_ATTR_V1_AND_A2V2 = {'attributes': {'network': 'v1', 'attr2': 'v2'}} + +LINK_TYPE_1 = { + 'link_type': 'instance-vnic', + 'source_id': 'instance1', + 'target_id': 'vnic1' +} +LINK_TYPE_1_REVERSED = { + 'link_type': 'instance-vnic', + 'source_id': 'vnic1', + 'target_id': 'instance1' +} +LINK_TYPE_1_2 = { + 'link_type': 'instance-vnic', + 'source_id': 'instance1', + 'target_id': 'vnic2' +} +LINK_TYPE_2 = { + 'link_type': 'vnic-vconnector', + 'source_id': 'vnic1', + 'target_id': 'vconnector1' +} +LINK_TYPE_3 = { + 'implicit': True, + 'link_type': 'instance-vconnector', + 'source_id': 'instance1', + 'target_id': 'vconnector1' +} +LINK_TYPE_4_NET1 = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'instance-host_pnic', + 'source': 'instance1_dbid', + 'source_id': 'instance1', + 'target': 'host_pnic1_dbid', + 'target_id': 'host_pnic1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_TYPE_5_NET2 = { + 'environment': ENV, + 'link_type': 'host_pnic-switch', + 'source_id': 'host_pnic1', + 'target': 'switch1_dbid', + 'target_id': 'switch1', + 'host': 'host2', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID2'} +} +LINK_TYPE_6_NET1 = { + 'environment': ENV, + 'link_type': 'host_pnic-switch', + 'source': 'host_pnic1_dbid', + 'source_id': 'host_pnic1', + 'target': 'switch2_dbid', + 'target_id': 'switch2', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_TYPE_7_NET1 = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'instance-switch', + 'source': 'instance1_dbid', + 'source_id': 'instance1', + 'target': 'switch2_dbid', + 'target_id': 'switch2', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} + +LINK_FULL_A2B = { + 'environment': ENV, + 'link_type': 'instance-vnic', + 'source': 'instance1_dbid', + 'source_id': 'instance1', + 'target': 'vnic1_dbid', + 'target_id': 'vnic1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_B2C = { + 'environment': ENV, + 'link_type': 'vnic-vconnector', + 'source': 'vnic1_dbid', + 'source_id': 'vnic1', + 'target': 'vconnector1_dbid', + 'target_id': 'vconnector1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_C2D = { + 'environment': ENV, + 'link_type': 'vconnector-vedge', + 'source': 'vconnector1_dbid', + 'source_id': 'vconnector1', + 'target': 'vedge1_dbid', + 'target_id': 'vedge1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_D2E = { + 'environment': ENV, + 'link_type': 'vedge-otep', + 'source': 'vedge1_dbid', + 'source_id': 'vedge1', + 'target': 'otep1_dbid', + 'target_id': 'otep1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_A2C = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'instance-vconnector', + 'source': 'instance1_dbid', + 'source_id': 'instance1', + 'target': 'vconnector1_dbid', + 'target_id': 'vconnector1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_B2D = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'vnic-vedge', + 'source': 'vnic1_dbid', + 'source_id': 'vnic1', + 'target': 'vedge1_dbid', + 'target_id': 'vedge1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_C2E = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'vconnector-otep', + 'source': 'vconnector1_dbid', + 'source_id': 'vconnector1', + 'target': 'otep1_dbid', + 'target_id': 'otep1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_A2D = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'instance-vedge', + 'source': 'instance1_dbid', + 'source_id': 'instance1', + 'target': 'vedge1_dbid', + 'target_id': 'vedge1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_B2E = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'vnic-otep', + 'source': 'vnic1_dbid', + 'source_id': 'vnic1', + 'target': 'otep1_dbid', + 'target_id': 'otep1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +LINK_FULL_A2E = { + 'environment': ENV, + 'implicit': True, + 'link_type': 'instance-otep', + 'source': 'instance1_dbid', + 'source_id': 'instance1', + 'target': 'otep1_dbid', + 'target_id': 'otep1', + 'host': 'host1', + 'link_name': '', + 'state': 'up', + 'source_label': '', + 'target_label': '', + 'link_weight': 0, + 'attributes': {'network': 'netID1'} +} +BASE_LINKS = [ + {'pass': 0, 'link': LINK_FULL_A2B}, + {'pass': 0, 'link': LINK_FULL_B2C}, + {'pass': 0, 'link': LINK_FULL_C2D}, + {'pass': 0, 'link': LINK_FULL_D2E}, +] +IMPLICIT_LINKS = [ + [ + {'pass': 1, 'link': LINK_FULL_A2C}, + {'pass': 1, 'link': LINK_FULL_B2D}, + {'pass': 1, 'link': LINK_FULL_C2E}, + ], + [ + {'pass': 2, 'link': LINK_FULL_A2D}, + {'pass': 2, 'link': LINK_FULL_B2E}, + ], + [ + {'pass': 3, 'link': LINK_FULL_A2E}, + ], + [] +] diff --git a/app/test/fetch/link_finders/test_find_implicit_links.py b/app/test/fetch/link_finders/test_find_implicit_links.py new file mode 100644 index 0000000..9931688 --- /dev/null +++ b/app/test/fetch/link_finders/test_find_implicit_links.py @@ -0,0 +1,107 @@ +############################################################################### +# 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 # +############################################################################### +import bson + +from discover.link_finders.find_implicit_links import FindImplicitLinks +from test.fetch.test_fetch import TestFetch +from unittest.mock import MagicMock +from test.fetch.link_finders.test_data.test_find_implicit_links import * + +from utils.inventory_mgr import InventoryMgr + + +class TestFindImplicitLinks(TestFetch): + + def setUp(self): + super().setUp() + self.configure_environment() + self.fetcher = FindImplicitLinks() + self.fetcher.set_env(ENV) + self.fetcher.constraint_attributes = ['network'] + self.original_write_link = self.inv.write_link + self.inv.write_link = lambda x: x + self.original_objectid = bson.ObjectId + bson.ObjectId = lambda x: x + + def tearDown(self): + super().tearDown() + bson.ObjectId = self.original_objectid + self.inv.write_link = self.original_write_link + + def test_get_constraint_attributes(self): + original_find = InventoryMgr.find + InventoryMgr.find = MagicMock(return_value=CLIQUE_CONSTRAINTS) + constraint_types = self.fetcher.get_constraint_attributes() + self.assertEqual(sorted(constraint_types), sorted(CONSTRAINTS)) + InventoryMgr.find = original_find + + def test_constraints_match(self): + matcher = self.fetcher.constraints_match + self.assertTrue(matcher(LINK_ATTRIBUTES_NONE, LINK_ATTRIBUTES_NONE_2)) + self.assertTrue(matcher(LINK_ATTRIBUTES_NONE, LINK_ATTRIBUTES_EMPTY)) + self.assertTrue(matcher(LINK_ATTRIBUTES_NONE, LINK_ATTR_V1)) + self.assertTrue(matcher(LINK_ATTRIBUTES_EMPTY, LINK_ATTR_V1)) + self.assertTrue(matcher(LINK_ATTR_V1, LINK_ATTR_V1_2)) + self.assertTrue(matcher(LINK_ATTR_V1, + LINK_ATTR_V1_AND_A2V2)) + self.assertFalse(matcher(LINK_ATTR_V1, LINK_ATTR_V2)) + + def test_links_match(self): + matcher = self.fetcher.links_match + self.assertFalse(matcher(LINK_TYPE_1, LINK_TYPE_1_2)) + self.assertFalse(matcher(LINK_TYPE_1, LINK_TYPE_1_REVERSED)) + self.assertFalse(matcher(LINK_TYPE_4_NET1, LINK_TYPE_5_NET2)) + self.assertFalse(matcher(LINK_TYPE_1_2, LINK_TYPE_2)) + self.assertTrue(matcher(LINK_TYPE_1, LINK_TYPE_2)) + + def test_get_link_constraint_attributes(self): + getter = self.fetcher.get_link_constraint_attributes + self.assertEqual(getter(LINK_TYPE_1, LINK_TYPE_1_2), {}) + self.assertEqual(getter(LINK_TYPE_1, LINK_TYPE_4_NET1), + LINK_TYPE_4_NET1.get('attributes')) + self.assertEqual(getter(LINK_TYPE_4_NET1, LINK_TYPE_1), + LINK_TYPE_4_NET1.get('attributes')) + self.assertEqual(getter(LINK_TYPE_1, LINK_TYPE_5_NET2), + LINK_TYPE_5_NET2.get('attributes')) + self.assertEqual(getter(LINK_TYPE_4_NET1, LINK_TYPE_6_NET1), + LINK_TYPE_4_NET1.get('attributes')) + + def test_get_attr(self): + getter = self.fetcher.get_attr + self.assertIsNone(getter('host', {}, {})) + self.assertIsNone(getter('host', {'host': 'v1'}, {'host': 'v2'})) + self.assertEqual(getter('host', {'host': 'v1'}, {}), 'v1') + self.assertEqual(getter('host', {}, {'host': 'v2'}), 'v2') + self.assertEqual(getter('host', {'host': 'v1'}, {'host': 'v1'}), 'v1') + + def test_add_implicit_link(self): + original_write_link = self.inv.write_link + self.inv.write_link = lambda x: x + original_objectid = bson.ObjectId + bson.ObjectId = lambda x: x + add_func = self.fetcher.add_implicit_link + self.assertEqual(add_func(LINK_TYPE_4_NET1, LINK_TYPE_6_NET1), + LINK_TYPE_7_NET1) + bson.ObjectId = original_objectid + self.inv.write_link = original_write_link + + def test_get_transitive_closure(self): + self.fetcher.links = [ + {'pass': 0, 'link': LINK_FULL_A2B}, + {'pass': 0, 'link': LINK_FULL_B2C}, + {'pass': 0, 'link': LINK_FULL_C2D}, + {'pass': 0, 'link': LINK_FULL_D2E}, + ] + self.fetcher.get_transitive_closure() + for pass_no in range(1, len(IMPLICIT_LINKS)): + implicit_links = [l for l in self.fetcher.links + if l['pass'] == pass_no] + self.assertEqual(implicit_links, IMPLICIT_LINKS[pass_no-1], + 'incorrect links for pass #{}'.format(pass_no)) |