new file mode 100644
@@ -0,0 +1,374 @@
+import kvm_test_utils
+from autotest_lib.client.common_lib import error
+
+def run_qmp_basic(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 not key 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:
+ value = int(qmp_dict[key])
+ except:
+ 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))
+ check_key_is_dict(resp["error"], "data")
+ if datadict and resp["error"]["data"] != datadict:
+ raise error.TestFail("got data dict '%s' expected '%s'"
+ % (resp["error"]["data"], datadict))
+
+ 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.send(cmd)
+ check_error_resp(resp, "JSONParsing")
+
+
+ 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 = "kvm-autotest"
+ resp = monitor.cmd_qmp("eject", { "foobar": True }, id=id)
+ check_error_resp(resp)
+ check_str_key(resp, "id", id)
+
+ # The "id" key must be echoed back in success responses
+ resp = monitor.cmd_qmp("query-status", id=id)
+ check_success_resp(resp)
+ check_str_key(resp, "id", id)
+
+ # The "id" key can be any json-object
+ for id in [ True, 1234, "string again!", [1, [], {}, True, "foo"],
+ { "key": {} } ]:
+ resp = monitor.cmd_qmp("query-status", id=id)
+ check_success_resp(resp)
+ if resp["id"] != id:
+ raise error.TestFail("expected id '%s' but got '%s'"
+ % (str(id), 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, "QMPExtraInputObjectMember",
+ { "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, "QMPBadInputObjectMember",
+ { "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, "QMPBadInputObjectMember",
+ { "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, "QMPBadInputObject", { "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)
+
+ id = "1234foo"
+ resp = monitor.cmd_obj({ "id": id, "execute": "query-version",
+ "arguments": {} })
+ check_success_resp(resp)
+ check_str_key(resp, "id", id)
+
+ # 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, "InvalidParameter", { "name": "foo" })
+
+ # required argument omitted
+ resp = monitor.cmd_qmp("screendump")
+ check_error_resp(resp, "MissingParameter", { "name": "filename" })
+
+ # 'bar' is not a valid argument
+ resp = monitor.cmd_qmp("screendump", { "filename": "outfile",
+ "bar": "bar" })
+ check_error_resp(resp, "InvalidParameter", { "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, "InvalidParameterType",
+ { "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, "InvalidParameterType",
+ { "name": "force", "expected": "bool" })
+
+ # val argument must ba a json-int
+ for arg in [ {}, [], True, "foo" ]:
+ resp = monitor.cmd_qmp("memsave", { "val": arg, "filename": "foo",
+ "size": 10 })
+ check_error_resp(resp, "InvalidParameterType",
+ { "name": "val", "expected": "int" })
+
+ # value argument must ba a json-number
+ for arg in [ {}, [], True, "foo" ]:
+ resp = monitor.cmd_qmp("migrate_set_speed", { "value": arg })
+ check_error_resp(resp, "InvalidParameterType",
+ { "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, "PropertyNotFound",
+ {"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 = kvm_test_utils.get_living_vm(env, params.get("main_vm"))
+
+ # Run all suites
+ greeting_suite(vm.monitor)
+ input_object_suite(vm.monitor)
+ argument_checker_suite(vm.monitor)
+ unknown_commands_suite(vm.monitor)
+ json_parsing_errors_suite(vm.monitor)
+
+ # check if QMP is still alive
+ if not vm.monitor.is_responsive():
+ raise error.TestFail('QEMU is not alive after QMP testing')
@@ -456,6 +456,9 @@ variants:
- fmt_raw:
image_format_stg = raw
+ - qmp_basic: install setup unattended_install.cdrom
+ type = qmp_basic
+
- vlan: install setup unattended_install.cdrom
type = vlan
# subnet should not be used by host