qmp_basic.py

from autotest.client.shared import error
from virttest import qemu_monitor


def run(test, params, env):
    """
    QMP Specification test-suite: this checks if the *basic* protocol conforms
    to its specification, which is file QMP/qmp-spec.txt in QEMU's source tree.

    IMPORTANT NOTES:

        o Most tests depend heavily on QMP's error information (eg. classes),
          this might have bad implications as the error interface is going to
          change in QMP

        o Command testing is *not* covered in this suite. Each command has its
          own specification and should be tested separately

        o We use the same terminology as used by the QMP specification,
          specially with regard to JSON types (eg. a Python dict is called
          a json-object)

        o This is divided in sub test-suites, please check the bottom of this
          file to check the order in which they are run

    TODO:

        o Finding which test failed is not as easy as it should be

        o Are all those check_*() functions really needed? Wouldn't a
          specialized class (eg. a Response class) do better?
    """
    def fail_no_key(qmp_dict, key):
        if not isinstance(qmp_dict, dict):
            raise error.TestFail("qmp_dict is not a dict (it's '%s')" %
                                 type(qmp_dict))
        if key not in qmp_dict:
            raise error.TestFail("'%s' key doesn't exist in dict ('%s')" %
                                 (key, str(qmp_dict)))

    def check_dict_key(qmp_dict, key, keytype):
        """
        Performs the following checks on a QMP dict key:

        1. qmp_dict is a dict
        2. key exists in qmp_dict
        3. key is of type keytype

        If any of these checks fails, error.TestFail is raised.
        """
        fail_no_key(qmp_dict, key)
        if not isinstance(qmp_dict[key], keytype):
            raise error.TestFail("'%s' key is not of type '%s', it's '%s'" %
                                 (key, keytype, type(qmp_dict[key])))

    def check_key_is_dict(qmp_dict, key):
        check_dict_key(qmp_dict, key, dict)

    def check_key_is_list(qmp_dict, key):
        check_dict_key(qmp_dict, key, list)

    def check_key_is_str(qmp_dict, key):
        check_dict_key(qmp_dict, key, unicode)

    def check_str_key(qmp_dict, keyname, value=None):
        check_dict_key(qmp_dict, keyname, unicode)
        if value and value != qmp_dict[keyname]:
            raise error.TestFail("'%s' key value '%s' should be '%s'" %
                                 (keyname, str(qmp_dict[keyname]), str(value)))

    def check_key_is_int(qmp_dict, key):
        fail_no_key(qmp_dict, key)
        try:
            int(qmp_dict[key])
        except Exception:
            raise error.TestFail("'%s' key is not of type int, it's '%s'" %
                                 (key, type(qmp_dict[key])))

    def check_bool_key(qmp_dict, keyname, value=None):
        check_dict_key(qmp_dict, keyname, bool)
        if value and value != qmp_dict[keyname]:
            raise error.TestFail("'%s' key value '%s' should be '%s'" %
                                 (keyname, str(qmp_dict[keyname]), str(value)))

    def check_success_resp(resp, empty=False):
        """
        Check QMP OK response.

        :param resp: QMP response
        :param empty: if True, response should not contain data to return
        """
        check_key_is_dict(resp, "return")
        if empty and len(resp["return"]) > 0:
            raise error.TestFail("success response is not empty ('%s')" %
                                 str(resp))

    def check_error_resp(resp, classname=None, datadict=None):
        """
        Check QMP error response.

        :param resp: QMP response
        :param classname: Expected error class name
        :param datadict: Expected error data dictionary
        """
        check_key_is_dict(resp, "error")
        check_key_is_str(resp["error"], "class")
        if classname and resp["error"]["class"] != classname:
            raise error.TestFail("got error class '%s' expected '%s'" %
                                 (resp["error"]["class"], classname))

    def test_version(version):
        """
        Check the QMP greeting message version key which, according to QMP's
        documentation, should be:

        { "qemu": { "major": json-int, "minor": json-int, "micro": json-int }
          "package": json-string }
        """
        check_key_is_dict(version, "qemu")
        for key in ("major", "minor", "micro"):
            check_key_is_int(version["qemu"], key)
        check_key_is_str(version, "package")

    def test_greeting(greeting):
        check_key_is_dict(greeting, "QMP")
        check_key_is_dict(greeting["QMP"], "version")
        check_key_is_list(greeting["QMP"], "capabilities")

    def greeting_suite(monitor):
        """
        Check the greeting message format, as described in the QMP
        specfication section '2.2 Server Greeting'.

        { "QMP": { "version": json-object, "capabilities": json-array } }
        """
        greeting = monitor.get_greeting()
        test_greeting(greeting)
        test_version(greeting["QMP"]["version"])

    def json_parsing_errors_suite(monitor):
        """
        Check that QMP's parser is able to recover from parsing errors, please
        check the JSON spec for more info on the JSON syntax (RFC 4627).
        """
        # We're quite simple right now and the focus is on parsing errors that
        # have already biten us in the past.
        #
        # TODO: The following test-cases are missing:
        #
        #   - JSON numbers, strings and arrays
        #   - More invalid characters or malformed structures
        #   - Valid, but not obvious syntax, like zillion of spaces or
        #     strings with unicode chars (different suite maybe?)
        bad_json = []

        # A JSON value MUST be an object, array, number, string, true, false,
        # or null
        #
        # NOTE: QMP seems to ignore a number of chars, like: | and ?
        bad_json.append(":")
        bad_json.append(",")

        # Malformed json-objects
        #
        # NOTE: sending only "}" seems to break QMP
        # NOTE: Duplicate keys are accepted (should it?)
        bad_json.append("{ \"execute\" }")
        bad_json.append("{ \"execute\": \"query-version\", }")
        bad_json.append("{ 1: \"query-version\" }")
        bad_json.append("{ true: \"query-version\" }")
        bad_json.append("{ []: \"query-version\" }")
        bad_json.append("{ {}: \"query-version\" }")

        for cmd in bad_json:
            resp = monitor.cmd_raw(cmd)
            check_error_resp(resp, "GenericError")

    def test_id_key(monitor):
        """
        Check that QMP's "id" key is correctly handled.
        """
        # The "id" key must be echoed back in error responses
        id_key = "virt-test"
        resp = monitor.cmd_qmp("eject", {"foobar": True}, q_id=id_key)
        check_error_resp(resp)
        check_str_key(resp, "id", id_key)

        # The "id" key must be echoed back in success responses
        resp = monitor.cmd_qmp("query-status", q_id=id_key)
        check_success_resp(resp)
        check_str_key(resp, "id", id_key)

        # The "id" key can be any json-object
        for id_key in (True, 1234, "string again!", [1, [], {}, True, "foo"],
                       {"key": {}}):
            resp = monitor.cmd_qmp("query-status", q_id=id_key)
            check_success_resp(resp)
            if resp["id"] != id_key:
                raise error.TestFail("expected id '%s' but got '%s'" %
                                     (str(id_key), str(resp["id"])))

    def test_invalid_arg_key(monitor):
        """
        Currently, the only supported keys in the input object are: "execute",
        "arguments" and "id". Although expansion is supported, invalid key
        names must be detected.
        """
        resp = monitor.cmd_obj({"execute": "eject", "foobar": True})
        check_error_resp(resp, "GenericError", {"member": "foobar"})

    def test_bad_arguments_key_type(monitor):
        """
        The "arguments" key must be an json-object.

        We use the eject command to perform the tests, but that's a random
        choice, any command that accepts arguments will do, as the command
        doesn't get called.
        """
        for item in (True, [], 1, "foo"):
            resp = monitor.cmd_obj({"execute": "eject", "arguments": item})
            check_error_resp(resp, "GenericError",
                             {"member": "arguments", "expected": "object"})

    def test_bad_execute_key_type(monitor):
        """
        The "execute" key must be a json-string.
        """
        for item in (False, 1, {}, []):
            resp = monitor.cmd_obj({"execute": item})
            check_error_resp(resp, "GenericError",
                             {"member": "execute", "expected": "string"})

    def test_no_execute_key(monitor):
        """
        The "execute" key must exist, we also test for some stupid parsing
        errors.
        """
        for cmd in ({}, {"execut": "qmp_capabilities"},
                    {"executee": "qmp_capabilities"}, {"foo": "bar"}):
            resp = monitor.cmd_obj(cmd)
            check_error_resp(resp)  # XXX: check class and data dict?

    def test_bad_input_obj_type(monitor):
        """
        The input object must be... an json-object.
        """
        for cmd in ("foo", [], True, 1):
            resp = monitor.cmd_obj(cmd)
            check_error_resp(resp, "GenericError", {"expected": "object"})

    def test_good_input_obj(monitor):
        """
        Basic success tests for issuing QMP commands.
        """
        # NOTE: We don't use the cmd_qmp() method here because the command
        # object is in a 'random' order
        resp = monitor.cmd_obj({"execute": "query-version"})
        check_success_resp(resp)

        resp = monitor.cmd_obj({"arguments": {}, "execute": "query-version"})
        check_success_resp(resp)

        idd = "1234foo"
        resp = monitor.cmd_obj({"id": idd, "execute": "query-version",
                                "arguments": {}})
        check_success_resp(resp)
        check_str_key(resp, "id", idd)

        # TODO: would be good to test simple argument usage, but we don't have
        # a read-only command that accepts arguments.

    def input_object_suite(monitor):
        """
        Check the input object format, as described in the QMP specfication
        section '2.3 Issuing Commands'.

        { "execute": json-string, "arguments": json-object, "id": json-value }
        """
        test_good_input_obj(monitor)
        test_bad_input_obj_type(monitor)
        test_no_execute_key(monitor)
        test_bad_execute_key_type(monitor)
        test_bad_arguments_key_type(monitor)
        test_id_key(monitor)
        test_invalid_arg_key(monitor)

    def argument_checker_suite(monitor):
        """
        Check that QMP's argument checker is detecting all possible errors.

        We use a number of different commands to perform the checks, but the
        command used doesn't matter much as QMP performs argument checking
        _before_ calling the command.
        """
        # stop doesn't take arguments
        resp = monitor.cmd_qmp("stop", {"foo": 1})
        check_error_resp(resp, "GenericError", {"name": "foo"})

        # required argument omitted
        resp = monitor.cmd_qmp("screendump")
        check_error_resp(resp, "GenericError", {"name": "filename"})

        # 'bar' is not a valid argument
        resp = monitor.cmd_qmp("screendump", {"filename": "outfile",
                                              "bar": "bar"})
        check_error_resp(resp, "GenericError", {"name": "bar"})

        # test optional argument: 'force' is omitted, but it's optional, so
        # the handler has to be called. Test this happens by checking an
        # error that is generated by the handler itself.
        resp = monitor.cmd_qmp("eject", {"device": "foobar"})
        check_error_resp(resp, "DeviceNotFound")

        # filename argument must be a json-string
        for arg in ({}, [], 1, True):
            resp = monitor.cmd_qmp("screendump", {"filename": arg})
            check_error_resp(resp, "GenericError",
                             {"name": "filename", "expected": "string"})

        # force argument must be a json-bool
        for arg in ({}, [], 1, "foo"):
            resp = monitor.cmd_qmp("eject", {"force": arg, "device": "foo"})
            check_error_resp(resp, "GenericError",
                             {"name": "force", "expected": "bool"})

        # val argument must be a json-int
        for arg in ({}, [], True, "foo"):
            resp = monitor.cmd_qmp("memsave", {"val": arg, "filename": "foo",
                                               "size": 10})
            check_error_resp(resp, "GenericError",
                             {"name": "val", "expected": "int"})

        # value argument must be a json-number
        for arg in ({}, [], True, "foo"):
            resp = monitor.cmd_qmp("migrate_set_speed", {"value": arg})
            check_error_resp(resp, "GenericError",
                             {"name": "value", "expected": "number"})

        # qdev-type commands have their own argument checker, all QMP does
        # is to skip its checking and pass arguments through. Check this
        # works by providing invalid options to device_add and expecting
        # an error message from qdev
        resp = monitor.cmd_qmp("device_add", {"driver": "e1000", "foo": "bar"})
        check_error_resp(resp, "GenericError",
                               {"device": "e1000", "property": "foo"})

    def unknown_commands_suite(monitor):
        """
        Check that QMP handles unknown commands correctly.
        """
        # We also call a HMP-only command, to be sure it will fail as expected
        for cmd in ("bar", "query-", "query-foo", "q", "help"):
            resp = monitor.cmd_qmp(cmd)
            check_error_resp(resp, "CommandNotFound", {"name": cmd})

    vm = env.get_vm(params["main_vm"])
    vm.verify_alive()

    # Look for the first qmp monitor available, otherwise, fail the test
    qmp_monitor = vm.get_monitors_by_type("qmp")
    if qmp_monitor:
        qmp_monitor = qmp_monitor[0]
    else:
        raise error.TestError('Could not find a QMP monitor, aborting test')

    # Run all suites
    greeting_suite(qmp_monitor)
    input_object_suite(qmp_monitor)
    argument_checker_suite(qmp_monitor)
    unknown_commands_suite(qmp_monitor)
    json_parsing_errors_suite(qmp_monitor)

    # check if QMP is still alive
    if not qmp_monitor.is_responsive():
        raise error.TestFail('QMP monitor is not responsive after testing')