add qdevice_format module

replace the details in qdevices.py by qdevice_format.

Signed-off-by: Houqi (Nick) Zuo <[email protected]>

virttest/qemu_devices/qdevice_format.py

@@ -0,0 +1,343 @@
+import logging
+import re
+
+from avocado.utils import process
+from virttest import utils_numeric
+
+LOG = logging.getLogger("avocado." + __name__)
+
+
+class _QDeviceFormatManagement(object):
+    def __init__(self):
+        self._format_func = {"json": self._json_format, "raw": self._raw_format}
+        self._special_args_in_json = {
+            "blockdev": {
+                "detect-zeroes": self._unchanged,
+                "offset": self._int,
+                "size": self._int,
+                "cache-size": self._int,
+                "timeout": self._int,
+            },
+            "device": {
+                "legacy": {
+                    "write-cache": self._unchanged,
+                    "disable-legacy": self._unchanged,
+                    "intremap": self._unchanged,
+                    "serial": self._unchanged,
+                    "eim": self._unchanged,
+                    # wwn needs to be presented as hexadecimal
+                    "wwn": self._int64,
+                },
+                "virtio-scsi-pci": {
+                    "virtqueue_size": self._int,
+                    "num_queues": self._int,
+                    "max_sectors": self._int,
+                },
+                "virtio-rng-pci": {
+                    "period": self._int,
+                    "max-bytes": self._int,
+                },
+                "scsi-hd": {
+                    "discard_granularity": self._int,
+                    "physical_block_size": self._int,
+                    "logical_block_size": self._int,
+                    "bootindex": self._int,
+                },
+                "virtio-blk-pci": {
+                    "max-write-zeroes-sectors": self._int,
+                    "queue-size": self._int,
+                    "max-discard-sectors": self._int,
+                    "num-queues": self._int,
+                    "discard_granularity": self._int,
+                },
+                "usb_driver": {
+                    "port": self._str,
+                    "serial": self._on,
+                    # min_io_size, opt_io_size from device ("driver": "usb-storage")
+                    "min_io_size": self._int,
+                    "opt_io_size": self._int,
+                },
+                "pcie-root-port": {
+                    "port": self._int64,
+                },
+                "nvdimm": {
+                    "label-size": self._normalize_data_size,
+                },
+                "pc-dimm": {
+                    "node": self._int,
+                },
+                "virtio-mem-pci": {
+                    "requested-size": self._normalize_data_size,
+                },
+                "intel-iommu": {
+                    "aw-bits": self._int,
+                },
+                "virtio-net-pci": {
+                    "acpi-index": self._int,
+                    "host_mtu": self._int,
+                    "speed": self._int,
+                    "vectors": self._int,
+                },
+                "virtio-balloon-ccw": {
+                    "guest-stats-polling-interval": self._int,
+                },
+                "pvpanic": {
+                    "events": self._int,
+                },
+            },
+            "object": {
+                "size": self._normalize_data_size,
+                "align": self._normalize_data_size,
+                "host-nodes": self._int_in_list,
+            },
+            "netdev": {
+                "fd": self._unchanged,
+                "vhostfd": self._unchanged,
+                "dnssearch": self._dict_in_list,
+                "hostfwd": self._dict_in_list,
+                "guestfwd": self._dict_in_list,
+                "sndbuf": self._normalize_data_size,
+            },
+        }
+        self._device_driver_checked = []
+        self._skip_args = ("addr",)
+        self._mandatory_assignment_args_type = {
+            "device": {
+                "pvpanic": {
+                    "ioport": self._int64,
+                },
+            }
+        }
+
+    def format(self, format_type, params, dev_type):
+        return self._format_func[format_type](params, dev_type)
+
+    def _json_format(self, params, dev_type):
+        return eval("self._" + dev_type + "_json_format")(params, dev_type)
+
+    def _raw_format(self, params):
+        pass
+
+    def _netdev_json_format(self, params, dev_type="netdev"):
+        args_in_json = self._special_args_in_json[dev_type]
+        new_args = dict()
+        for key, value in params.items():
+            if key in args_in_json.keys():
+                new_args[key] = args_in_json[key](value)
+            elif isinstance(value, str) and value.isdigit():
+                new_args[key] = int(value)
+            else:
+                self._bool_in_string_to_bool(value)
+
+            subs = key.split(".")
+            curr = new_args
+            for subk in subs[:-1]:
+                try:
+                    int(subk)
+                    subv = list()
+                except ValueError:
+                    subv = dict()
+                curr.setdefault(subk, subv)
+                curr = curr[subk]
+            curr[subs[-1]] = value
+
+        return new_args
+
+    def _object_json_format(self, params, dev_type="object"):
+        params["qom-type"] = params.pop("backend")
+        args_in_json = self._special_args_in_json[dev_type]
+        new_args = dict()
+        for key, value in params.items():
+            if key in args_in_json.keys():
+                new_args[key] = args_in_json[key](value)
+            else:
+                new_args[key] = self._bool_in_string_to_bool(value)
+
+        return new_args
+
+    def _device_json_format(self, params, dev_type="device"):
+        """
+        Convert args type to adapt output in json format.
+
+        :param params: Parameters of device.
+        :type params: Dict
+        :param dev_type: The dev_type used in mapping.
+        :type dev_type: String.
+        :return: The converted args.
+        :rtype: Dict
+        """
+        driver = params.get("driver")
+        driver = "usb_driver" if driver.startswith("usb-") else driver
+        if driver not in self._device_driver_checked:
+            self._device_driver_checked.append(driver)
+            self._update_args_type_from_qemu(driver)
+        device_args = self._special_args_in_json[dev_type]
+        new_args = dict()
+        # convert type
+        for key, value in params.items():
+            if key in device_args[driver]:
+                value = device_args[driver][key](value)
+                new_args[key] = value
+            if key in device_args["legacy"]:
+                new_args[key] = device_args["legacy"][key](value)
+            else:
+                new_args[key] = self._bool_in_string_to_bool(value)
+
+        return new_args
+
+    def _blockdev_json_format(self, params, dev_type="blockdev"):
+        """
+        Convert args type to adapt output in json format.
+
+        :param params: Parameters of device.
+        :type params: Dict
+        :param dev_type: The dev_type used in mapping.
+        :type dev_type: String.
+        :return: The converted args.
+        :rtype: Dict
+        """
+        args_in_json = self._special_args_in_json[dev_type]
+        new_args = dict()
+        for key, value in params.items():
+            new_args[key] = (
+                args_in_json[key](value)
+                if key in args_in_json.keys()
+                else self._bool_in_string_to_bool(value)
+            )
+            new_args = self._flat_to_dict(key, new_args)
+
+        return new_args
+
+    @staticmethod
+    def _unchanged(val):
+        return val
+
+    @staticmethod
+    def _int(val):
+        return int(val, 0)
+
+    @staticmethod
+    def _str(val):
+        return str(val)
+
+    @staticmethod
+    def _on(val):
+        return "on" if val == "NO_EQUAL_STRING" else val
+
+    @staticmethod
+    def _int64(val):
+        if isinstance(val, str) and val.startswith("0x"):
+            val = val[2:]
+        return int(val, 16)
+
+    @staticmethod
+    def _normalize_data_size(val):
+        return int(utils_numeric.normalize_data_size(val, "B"))
+
+    @staticmethod
+    def _int_in_list(val):
+        return list(map(int, val.split()))
+
+    @staticmethod
+    def _dict_in_list(val):
+        if isinstance(val, list):
+            return [{"str": v} for v in val]
+        return val
+
+    @staticmethod
+    def _bool_in_string_to_bool(val):
+        """
+        Convert the "on""off""yes""no""true""false" to boolean.
+        Note: If the val is a string except "on""off""yes""no""true""false",
+                just return val without any operations.
+
+        :param val: The value.
+        :type val: String.
+        :return: The value converted or original val.
+        :rtype: Boolean or the original type.
+        """
+        if isinstance(val, str) and val.lower() in (
+            "on",
+            "off",
+            "yes",
+            "no",
+            "true",
+            "false",
+        ):
+            return val in (
+                "on",
+                "yes",
+                "true",
+            )
+        return val
+
+    @staticmethod
+    def _flat_to_dict(key, args):
+        """
+        Convert the flat expression to multi-level expression.
+
+        :param key: The flat key.
+        :type key: String.
+        :param args: The structure including flat key.
+        :type args: Dict.
+
+        :return: The structure converted.
+        :rtype: Dict.
+        """
+        val = args[key]
+        parts = key.split(".")
+        if parts:
+            args.pop(key)
+        p = re.compile(r"server\.(?P<index>\d+)\.(?P<opt>.+)")
+        m = p.match(key)
+        if m:
+            # convert 'server.0.host', 'server.1.host'
+            # to {'server': [{'host':xx}, {'host':xx}]}
+            servers = args["server"] if "server" in args else []
+            servers.append({m.group("opt"): val})
+            args["server"] = servers.sort(key=lambda k: k[1])
+        else:
+            # convert 'cache.direct': 'true' to {'cache': {'direct': 'true'}}
+            tmp = args
+            for part in parts[:-1]:
+                if part not in tmp:
+                    tmp[part] = dict()
+                tmp = tmp[part]
+            tmp[parts[-1]] = val
+
+        return args
+
+    def _update_args_type_from_qemu(self, driver):
+        """
+        Update the args type from qemu.
+        Only update the following type: int16, int32, int64, bool, str
+        """
+        if driver not in self._special_args_in_json["device"]:
+            self._special_args_in_json["device"][driver] = dict()
+        cmd = "/usr/libexec/qemu-kvm --device %s,\\?" % driver
+        output = process.run(cmd, shell=True, verbose=False).stdout_text.strip()
+        args_list = re.findall("(.+)=(<[^>]+>+)", output)
+        for arg, arg_type in args_list:
+            arg = arg.strip()
+            if arg in self._skip_args:
+                continue
+            arg_type = arg_type.strip()
+            if driver not in self._mandatory_assignment_args_type["device"]:
+                self._mandatory_assignment_args_type["device"][driver] = dict()
+            if arg in self._mandatory_assignment_args_type["device"][driver]:
+                self._special_args_in_json["device"][driver][
+                    arg
+                ] = self._mandatory_assignment_args_type["device"][driver][arg]
+            elif "int16" in arg_type or "int32" in arg_type:
+                self._special_args_in_json["device"][driver][arg] = self._int
+            elif "bool" in arg_type:
+                self._special_args_in_json["device"][driver][
+                    arg
+                ] = self._bool_in_string_to_bool
+            elif "int64" in arg_type:
+                self._special_args_in_json["device"][driver][arg] = self._int64
+            elif "str" in arg_type:
+                self._special_args_in_json["device"][driver][arg] = self._str
+
+
+qdevice_format = _QDeviceFormatManagement()