[pytest] Add test_gen_spytestbed.py to generate Spytest testbed file

ansible/library/conn_graph_facts.py

@@ -188,6 +188,8 @@ def get_host_vlan(self, hostname):
             self.vlanport[hostname] = {}
             for port in self.links[hostname]:
                 peerdevice = self.links[hostname][port]['peerdevice']
+                if self.devices[peerdevice]["Type"].lower() == "devsonic":
+                    continue
                 peerport = self.links[hostname][port]['peerport']
                 peerportmode = self.vlanport[peerdevice][peerport]['mode']
                 peervlanids = self.vlanport[peerdevice][peerport]['vlanids']

tests/templates/spytest_testbed.yaml.j2

@@ -0,0 +1,41 @@
+version: 2.0
+
+services: {default: !include sonic_services.yaml}
+
+params: !include sonic_params.yaml
+builds: !include sonic_builds.yaml
+speeds: !include sonic_speeds.yaml
+errors: !include sonic_errors.yaml
+instrument: !include sonic_instrument.yaml
+
+configs:
+    default: !include sonic_configs.yaml
+    empty: {current: [], restore: []}
+
+devices:
+    {% for hostname, login_info in devices %}
+    {{ hostname }}:
+        device_type: DevSonic
+        access: {{ login_info.login_access }}
+        credentials: {{ login_info.login_credentials }}
+        properties: {config: default, build: default, services: default, params: def_dut, speed: default, errors: default}
+    {% endfor %}
+    ptf-01:
+        device_type: TGEN
+        properties: {type: scapy, version: 1.0, ip: {{ testbed.ptf_ip }}, params: def_tg}
+
+topology:
+    ptf-01:
+        interfaces:
+            {% for ptf_conn in ptf_connections %}
+            {{ ptf_conn.start_port }}: {{ "{" }}EndDevice: {{ ptf_conn.end_device }}, EndPort: {{ ptf_conn.end_port }}, params: def_tg_link{{ "}" }}
+            {% endfor %}
+    {% for hostname in testbed.duts %}
+    {% if hostname in dev_connections %}
+    {{ hostname }}:
+        interfaces:
+            {% for dev_conn in dev_connections[hostname] %}
+            {{ dev_conn.start_port }}: {{ "{" }}EndDevice: {{ dev_conn.end_device }}, EndPort: {{ dev_conn.end_port }}, params: def_tg_link{{ "}" }}
+            {% endfor %}
+    {% endif %}
+    {% endfor %}

tests/testbed_setup/test_gen_spy_testbed.py

@@ -0,0 +1,109 @@
+import logging
+import json
+import os
+import pytest
+import yaml
+
+from collections import defaultdict
+from jinja2 import Template
+
+
+TESTBED_TEMPLATE = "templates/spytest_testbed.yaml.j2"
+PTF_INTERFACE_TEMPLATE = "1/%d"
+
+pytestmark = [
+    pytest.mark.topology("util"),
+    pytest.mark.sanity_check(skip_sanity=True)
+]
+
+
+@pytest.fixture(scope="function")
+def hostvars(duthosts):
+    """Return host variables dicts for DUTs defined in testbed."""
+    if not duthosts:
+        return {}
+    var_manager = duthosts[-1].host.options["variable_manager"]
+    hostvars_all = var_manager.get_vars()["hostvars"]
+    return {duthost.hostname: hostvars_all[duthost.hostname]
+            for duthost in duthosts}
+
+
+def test_gen_spy_testbed(conn_graph_facts_multi_duts, hostvars, testbed,
+                         pytestconfig):
+    """Generate spytest testbed file."""
+
+    def _interface_key(interface):
+        """Get interface key to sort."""
+        return list(map(int, interface.lstrip("Ethernet").split("/")))
+
+    def _to_string(obj):
+        """Convert unicodes in obj to strings"""
+        return yaml.safe_load(json.dumps(obj))
+
+    hostnames = testbed["duts"]
+    device_conn = _to_string(conn_graph_facts_multi_duts["device_conn"])
+    connections = dict(
+        zip(hostnames, conn_graph_facts_multi_duts["device_conn"]))
+
+    # devices section
+    devices = []
+    for hostname in hostnames:
+        hostvar = hostvars[hostname]
+        login_info = {}
+        login_info["login_access"] = \
+            json.dumps(hostvar["login_access"]).replace('"', '')
+        login_info["login_credentials"] = \
+            json.dumps(hostvar["login_credentials"]).replace('"', '')
+        devices.append((hostname, login_info))
+
+    # topology section
+    ptf_connections = []
+    intf = 1
+    for hostname in hostnames:
+        end_device = hostname
+        conns = connections[hostname]
+        end_ports = sorted(
+            (_ for _ in conns if conns[_]['peerdevice'] not in connections),
+            key=_interface_key)
+        for end_port in end_ports:
+            ptf_conn = {
+                "start_port": PTF_INTERFACE_TEMPLATE % intf,
+                "end_device": hostname,
+                "end_port": end_port
+            }
+            ptf_connections.append(ptf_conn)
+            conns.pop(end_port)
+            intf += 1
+
+    dev_connections = defaultdict(list)
+    for hostname in hostnames:
+        conns = connections[hostname]
+        for start_port in sorted(conns.keys(), key=_interface_key):
+            end_device = conns[start_port]["peerdevice"]
+            end_port = conns[start_port]["peerport"]
+            dev_connections[hostname].append(
+                {
+                    "start_port": start_port,
+                    "end_device": end_device,
+                    "end_port": end_port
+                }
+            )
+            connections[end_device].pop(end_port)
+
+    # write to testbed dest file
+    with open(TESTBED_TEMPLATE) as tmpl_fd:
+        testbed_tmpl = Template(
+            tmpl_fd.read(), trim_blocks=True, lstrip_blocks=True)
+    testbed_file = os.path.join(str(pytestconfig.rootdir),
+                                "../spytest/testbeds/spytest_testbed.yaml")
+    testbed_file = os.path.normpath(testbed_file)
+    logging.info("testbed save path: %s", testbed_file)
+    if os.path.exists(testbed_file):
+        logging.warn("testbed file(%s) exists, overwrite!", testbed_file)
+    testbed_stream = testbed_tmpl.stream(
+        devices=devices,
+        testbed=testbed,
+        ptf_connections=ptf_connections,
+        dev_connections=dev_connections
+    )
+    testbed_stream.dump(testbed_file)