aboutsummaryrefslogtreecommitdiffstats
path: root/app/test/fetch/db_fetch
diff options
context:
space:
mode:
Diffstat (limited to 'app/test/fetch/db_fetch')
-rw-r--r--app/test/fetch/db_fetch/mock_cursor.py24
-rw-r--r--app/test/fetch/db_fetch/test_data/db_fetch_host_network_agents.py8
-rw-r--r--app/test/fetch/db_fetch/test_data/db_fetch_oteps.py6
-rw-r--r--app/test/fetch/db_fetch/test_data/db_fetch_vedges_ovs.py55
-rw-r--r--app/test/fetch/db_fetch/test_db_fetch_oteps.py2
-rw-r--r--app/test/fetch/db_fetch/test_db_fetch_vedges_ovs.py90
6 files changed, 175 insertions, 10 deletions
diff --git a/app/test/fetch/db_fetch/mock_cursor.py b/app/test/fetch/db_fetch/mock_cursor.py
index 71efd3b..10c67e1 100644
--- a/app/test/fetch/db_fetch/mock_cursor.py
+++ b/app/test/fetch/db_fetch/mock_cursor.py
@@ -7,19 +7,39 @@
# which accompanies this distribution, and is available at #
# http://www.apache.org/licenses/LICENSE-2.0 #
###############################################################################
+
+
+def require_open(method):
+ def wrapped(self, *args, **kwargs):
+ if self.closed:
+ raise ValueError("Cursor is closed")
+ return method(self, *args, **kwargs)
+ return wrapped
+
+
class MockCursor:
def __init__(self, result):
self.result = result
self.current = 0
+ self.closed = False
+ @require_open
def __next__(self):
if self.current < len(self.result):
- next = self.result[self.current]
+ nxt = self.result[self.current]
self.current += 1
- return next
+ return nxt
else:
raise StopIteration
+ @require_open
def __iter__(self):
return self
+
+ @require_open
+ def fetchall(self):
+ return self.result
+
+ def close(self):
+ self.closed = True
diff --git a/app/test/fetch/db_fetch/test_data/db_fetch_host_network_agents.py b/app/test/fetch/db_fetch/test_data/db_fetch_host_network_agents.py
index 6188ddf..b6d344c 100644
--- a/app/test/fetch/db_fetch/test_data/db_fetch_host_network_agents.py
+++ b/app/test/fetch/db_fetch/test_data/db_fetch_host_network_agents.py
@@ -37,13 +37,13 @@ NETWORK_AGENT = [
NETWORK_AGENT_WITH_MECHANISM_DRIVERS_IN_CONFIG_RESULTS = [
{
'configurations': {},
- 'id': 'OVS-1764430c-c09e-4717-86fa-c04350b1fcbb',
+ 'id': 'neutron-openvswitch-agent-1764430c-c09e-4717-86fa-c04350b1fcbb',
'binary': 'neutron-openvswitch-agent',
'name': 'neutron-openvswitch-agent'
},
{
'configurations': {},
- 'id': 'OVS-2c2ddfee-91f9-47da-bd65-aceecd998b7c',
+ 'id': 'neutron-dhcp-agent-2c2ddfee-91f9-47da-bd65-aceecd998b7c',
'binary': 'neutron-dhcp-agent',
'name': 'neutron-dhcp-agent'
}
@@ -52,13 +52,13 @@ NETWORK_AGENT_WITH_MECHANISM_DRIVERS_IN_CONFIG_RESULTS = [
NETWORK_AGENT_WITHOUT_MECHANISM_DRIVERS_IN_CONFIG_RESULTS = [
{
'configurations': {},
- 'id': 'network_agent-1764430c-c09e-4717-86fa-c04350b1fcbb',
+ 'id': 'neutron-openvswitch-agent-1764430c-c09e-4717-86fa-c04350b1fcbb',
'binary': 'neutron-openvswitch-agent',
'name': 'neutron-openvswitch-agent'
},
{
'configurations': {},
- 'id': 'network_agent-2c2ddfee-91f9-47da-bd65-aceecd998b7c',
+ 'id': 'neutron-dhcp-agent-2c2ddfee-91f9-47da-bd65-aceecd998b7c',
'binary': 'neutron-dhcp-agent',
'name': 'neutron-dhcp-agent'
}
diff --git a/app/test/fetch/db_fetch/test_data/db_fetch_oteps.py b/app/test/fetch/db_fetch/test_data/db_fetch_oteps.py
index a5bc63d..2bd1784 100644
--- a/app/test/fetch/db_fetch/test_data/db_fetch_oteps.py
+++ b/app/test/fetch/db_fetch/test_data/db_fetch_oteps.py
@@ -34,10 +34,12 @@ VEDGE_WITHOUT_TUNNEL_TYPES = {
}
}
NON_ICEHOUSE_CONFIGS = {
- "distribution": "Mirantis-8.0"
+ "distribution": "Mirantis",
+ "distribution_version": "8.0"
}
ICEHOUSE_CONFIGS = {
- "distribution": "Canonical-icehouse"
+ "distribution": "Canonical",
+ "distribution_version": "icehouse"
}
HOST = {
"host": "node-5.cisco.com",
diff --git a/app/test/fetch/db_fetch/test_data/db_fetch_vedges_ovs.py b/app/test/fetch/db_fetch/test_data/db_fetch_vedges_ovs.py
index 818704c..c1f9d4f 100644
--- a/app/test/fetch/db_fetch/test_data/db_fetch_vedges_ovs.py
+++ b/app/test/fetch/db_fetch/test_data/db_fetch_vedges_ovs.py
@@ -166,3 +166,58 @@ DOC_TO_GET_OVERLAY = {
"agent_type": "Open vSwitch agent",
"configurations": {"tunneling_ip": "192.168.2.3"},
}
+
+LIST_IFACES_LINES = [
+ "eth0",
+ "p",
+ "t"
+]
+LIST_IFACES_NAMES = LIST_IFACES_LINES
+LIST_IFACES_LINES_MIRANTIS = {
+ "eth0--br-eth0",
+ "phy-eth0"
+}
+LIST_IFACES_NAMES_MIRANTIS = ["eth0"]
+
+VEDGE_CONFIGURATIONS_MIRANTIS = {
+ "bridge_mappings": {
+ "br-prv": "eth0"
+ }
+}
+VEDGE_CONFIGURATIONS = {
+ "bridge_mappings": {
+ "physnet1": "eth0",
+ "physnet2": "p",
+ "physnet3": "t",
+ "physnet4": "p",
+ "physnet5": "p"
+ }
+}
+
+VEDGE_MIRANTIS = {
+ 'host': HOST['host'],
+ 'ports': {
+ "eth0": {"name": "eth0", "id": "eth0-port_id"}
+ },
+ 'configurations': VEDGE_CONFIGURATIONS_MIRANTIS
+}
+VEDGE = {
+ 'host': HOST['host'],
+ 'ports': {
+ "eth0": {"name": "eth0", "id": "eth0-port_id"},
+ "p": {"name": "p", "id": "p-port_id"},
+ "t": {"name": "t", "id": "t-port_id"}
+ },
+ 'configurations': VEDGE_CONFIGURATIONS
+}
+
+ANOTHER_DIST = "another distribution"
+
+PNICS_MIRANTS = {
+ "eth0": {"name": "eth0", "mac_address": "eth0 mac_address"}
+}
+PNICS = {
+ "eth0": {"name": "eth0", "mac_address": "eth0 mac_address"},
+ "p": {"name": "p", "mac_address": "p mac_address"},
+ "t": {"name": "t", "mac_address": "t mac_address"}
+}
diff --git a/app/test/fetch/db_fetch/test_db_fetch_oteps.py b/app/test/fetch/db_fetch/test_db_fetch_oteps.py
index 7d29622..a161e03 100644
--- a/app/test/fetch/db_fetch/test_db_fetch_oteps.py
+++ b/app/test/fetch/db_fetch/test_db_fetch_oteps.py
@@ -32,11 +32,13 @@ class TestDbFetchOteps(TestFetch):
original_get_vconnector = self.fetcher.get_vconnector
self.fetcher.get_vconnector = MagicMock()
self.fetcher.inv.get_by_id = MagicMock(side_effect=[vedge, host])
+ original_get_env_config = self.fetcher.config.get_env_config
self.fetcher.config.get_env_config = MagicMock(return_value=config)
self.fetcher.get_objects_list_for_id = MagicMock(return_value=oteps_from_db)
results = self.fetcher.get(VEDGE_ID)
self.assertEqual(results, expected_results, err_msg)
self.fetcher.get_vconnector = original_get_vconnector
+ self.fetcher.config.get_env_config = original_get_env_config
def test_get(self):
test_cases = [
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)