Merge "Add a new test to check for duplicate IPs in an env"
diff --git a/test_set/cvp-sanity/global_config.yaml b/test_set/cvp-sanity/global_config.yaml
index 4faddaa..b1af7a4 100644
--- a/test_set/cvp-sanity/global_config.yaml
+++ b/test_set/cvp-sanity/global_config.yaml
@@ -65,6 +65,12 @@
"skipped_ifaces": ["bonding_masters", "lo", "veth", "tap", "cali", "qv", "qb", "br-int", "vxlan"]}
# mask for interfaces to skip
+# test duplicate ips
+# do not comment this section
+test_duplicate_ips:
+ {
+ "skipped_ifaces": ["lo", "virbr0", "docker_gwbridge", "docker0"]}
+
# specify what mcp version (tag) is deployed
drivetrain_version: ''
diff --git a/test_set/cvp-sanity/tests/test_duplicate_ips.py b/test_set/cvp-sanity/tests/test_duplicate_ips.py
new file mode 100644
index 0000000..50bdcba
--- /dev/null
+++ b/test_set/cvp-sanity/tests/test_duplicate_ips.py
@@ -0,0 +1,50 @@
+import pytest
+from collections import Counter
+from pprint import pformat
+import os
+
+import utils
+
+
+def get_duplicate_ifaces(nodes, ips):
+ dup_ifaces = {}
+ for node in nodes:
+ for iface in nodes[node]['ip4_interfaces']:
+ if set(nodes[node]['ip4_interfaces'][iface]) & set(ips):
+ dup_ifaces[node] = {iface: nodes[node]['ip4_interfaces'][iface]}
+ return dup_ifaces
+
+def test_duplicate_ips(local_salt_client):
+ active_nodes = utils.get_active_nodes()
+
+ testname = os.path.basename(__file__).split('.')[0]
+ config = utils.get_configuration()
+ skipped_ifaces = config.get(testname)["skipped_ifaces"]
+
+ local_salt_client.cmd('L@'+','.join(active_nodes),
+ 'saltutil.refresh_grains',
+ expr_form='compound')
+ nodes = local_salt_client.cmd('L@'+','.join(active_nodes),
+ 'grains.item',
+ ['ip4_interfaces'],
+ expr_form='compound')
+
+ ipv4_list = []
+ for node in nodes:
+ for iface in nodes[node]['ip4_interfaces']:
+ # Omit 'ip-less' ifaces
+ if not nodes[node]['ip4_interfaces'][iface]:
+ continue
+ if iface in skipped_ifaces:
+ continue
+ ipv4_list.extend(nodes[node]['ip4_interfaces'][iface])
+ no_dups = (len(ipv4_list) == len(set(ipv4_list)))
+ if not no_dups:
+ ips_count = Counter(ipv4_list).most_common()
+ dup_ips = filter(lambda x: x[1] > 1, ips_count)
+ dup_ifaces = get_duplicate_ifaces(nodes, [v[0] for v in dup_ips])
+
+ msg = ("\nDuplicate IP addresses found:\n{}"
+ "\n\nThe following interfaces are affected:\n{}"
+ "".format(pformat(dup_ips), pformat(dup_ifaces)))
+ assert no_dups, msg