From d32f75145676bacefde0d08a14680a5984623451 Mon Sep 17 00:00:00 2001 From: Koren Lev Date: Fri, 29 Sep 2017 01:38:18 +0300 Subject: release 1.0 calipso for opnfv apex Change-Id: I3e63cd27c5f4d3756e67a07c749863a68e84dde2 Signed-off-by: Koren Lev --- .../fetch/db_fetch/test_db_fetch_vedges_ovs.py | 90 +++++++++++++++++++++- 1 file changed, 88 insertions(+), 2 deletions(-) (limited to 'app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py') diff --git a/app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py b/app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py index 0cfb500..9916e5d 100644 --- a/app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py +++ b/app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py @@ -7,6 +7,8 @@ # which accompanies this distribution, and is available at # # http://www.apache.org/licenses/LICENSE-2.0 # ############################################################################### +import copy + from discover.fetchers.db.db_fetch_vedges_ovs import DbFetchVedgesOvs from test.fetch.test_fetch import TestFetch from test.fetch.db_fetch.test_data.db_fetch_vedges_ovs import * @@ -20,6 +22,12 @@ class TestDbFetchVedgesOvs(TestFetch): self.configure_environment() self.fetcher = DbFetchVedgesOvs() self.fetcher.set_env(self.env) + self.original_inv_set = self.fetcher.inv.set + self.fetcher.inv.set = MagicMock() + + def tearDown(self): + super().tearDown() + self.fetcher.inv.set = self.original_inv_set def check_get_result(self, objects_from_db, host, @@ -32,7 +40,8 @@ class TestDbFetchVedgesOvs(TestFetch): original_fetch_ports = self.fetcher.fetch_ports original_get_overlay_tunnels = self.fetcher.get_overlay_tunnels - self.fetcher.get_objects_list_for_id = MagicMock(return_value=objects_from_db) + self.fetcher.get_objects_list_for_id = \ + MagicMock(return_value=objects_from_db) self.fetcher.inv.get_by_id = MagicMock(return_value=host) self.fetcher.run_fetch_lines = MagicMock(return_value=vsctl_lines) self.fetcher.fetch_ports = MagicMock(return_value=ports) @@ -96,7 +105,7 @@ class TestDbFetchVedgesOvs(TestFetch): results = self.fetcher.fetch_ports_from_dpctl(HOST['id']) self.fetcher.run_fetch_lines = original_run_fetch_lines self.assertEqual(results, DPCTL_RESULTS, - "Can' t get correct ports info from dpctl lines") + "Can't get correct ports info from dpctl lines") def test_fetch_port_tags_from_vsctl(self): ports = self.fetcher.fetch_port_tags_from_vsctl(VSCTL_LINES, @@ -108,3 +117,80 @@ class TestDbFetchVedgesOvs(TestFetch): results = self.fetcher.get_overlay_tunnels(DOC_TO_GET_OVERLAY, VSCTL_LINES) self.assertEqual(results, TUNNEL_PORTS) + + @staticmethod + def get_test_pnic_for_interface_mirantis(search: dict, + get_single: bool=True): + if not get_single: + # we're only supposed to get calls with get_single == True + return [] + return PNICS_MIRANTS.get(search.get('name'), {}) + + @staticmethod + def get_test_pnic_for_interface(search: dict, + get_single: bool=True): + if not get_single: + # we're only supposed to get calls with get_single == True + return [] + return PNICS.get(search.get('name'), {}) + + @staticmethod + def get_expected_results_for_get_pnics(test_pnics: dict, ports: dict, + ifaces_names: list) -> dict: + expected_results = {} + for p in test_pnics.values(): + if p.get("name") not in ifaces_names: + continue + p1 = copy.deepcopy(p) + name = p1["name"] + port = ports[name] + p1["port_id"] = port["id"] + expected_results[name] = p1 + return expected_results + + def test_get_pnics(self): + expected_results = \ + self.get_expected_results_for_get_pnics(PNICS_MIRANTS, + VEDGE_MIRANTIS["ports"], + LIST_IFACES_NAMES_MIRANTIS) + self.check_get_pnics_for_dist(VEDGE_MIRANTIS, + LIST_IFACES_LINES_MIRANTIS, + LIST_IFACES_NAMES_MIRANTIS, + expected_results, + self.get_test_pnic_for_interface_mirantis, + self.fetcher.MIRANTIS_DIST, + ver="6.0", + msg="Incorrect get_pnics result " + "(Mirantis)") + expected_results = \ + self.get_expected_results_for_get_pnics(PNICS, + VEDGE["ports"], + LIST_IFACES_NAMES) + self.check_get_pnics_for_dist(VEDGE, + LIST_IFACES_LINES, + LIST_IFACES_NAMES, + expected_results, + self.get_test_pnic_for_interface, + ANOTHER_DIST, + msg="Incorrect get_pnics result") + + def check_get_pnics_for_dist(self, test_vedge, + ifaces_list_output, ifaces_list_clear, + expected_results, + pnic_find_func, + dist, ver=None, msg=None): + self.fetcher.configuration.environment = { + "distribution": dist, + "distribution_version": ver + } + original_run_fetch_lines = self.fetcher.run_fetch_lines + self.fetcher.run_fetch_lines = \ + MagicMock(return_value=ifaces_list_output) + original_find_items = self.fetcher.inv.find_items + self.fetcher.inv.find_items = pnic_find_func + vedge = copy.deepcopy(test_vedge) + results = self.fetcher.get_pnics(vedge) + self.fetcher.run_fetch_lines = original_run_fetch_lines + self.fetcher.inv.find_items = original_find_items + self.assertTrue(vedge.get("pnic") in ifaces_list_clear) + self.assertEqual(results, expected_results, msg) -- cgit 1.2.3-korg