summaryrefslogtreecommitdiffstats
path: root/app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py
diff options
context:
space:
mode:
Diffstat (limited to 'app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py')
-rw-r--r--app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py90
1 files changed, 88 insertions, 2 deletions
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)