Merge pull request #2279 from zhencliu/curl_access

Curl image access test

qemu/tests/boot_with_remote_readonly_image.py

+from virttest import qemu_storage
+from virttest import data_dir
+from virttest import utils_misc
+from virttest import utils_test
+
+from avocado import fail_on
+
+from avocado.utils import process
+
+from provider import qemu_img_utils as img_utils
+
+
+def run(test, params, env):
+    """
+    1) Convert remote readonly system image to the local image
+    2) Start VM from the local image,
+       with the remote iso image as its cdrom
+    3) Log into VM
+    4) Check readable cdrom
+    """
+    def _convert_image():
+        source = params['images'].split()[0]
+        target = params['convert_target']
+        source_params = params.object_params(source)
+        target_params = params.object_params(target)
+        source_image = qemu_storage.QemuImg(source_params, None, source)
+
+        # Convert source to target
+        fail_on((process.CmdError,))(source_image.convert)(
+            target_params, data_dir.get_data_dir())
+
+    _convert_image()
+    vm = img_utils.boot_vm_with_images(test, params, env,
+                                       (params['convert_target'],))
+    session = vm.wait_for_login(
+        timeout=params.get_numeric("login_timeout", 360))
+    cdroms = utils_misc.wait_for(
+        lambda: (utils_test.get_readable_cdroms(params, session)),
+        timeout=params.get_numeric("timeout", 10)
+    )
+    session.close()
+    if not cdroms:
+        test.fail("None readable cdrom found in vm.")

qemu/tests/cfg/remote_readonly_storage.cfg

+# Network storage backends:
+#   libcurl
+# The following testing scenarios are covered:
+# - boot_with_readonly_image
+#   boot vm with an iso image from https server
+# - access_with_qemu_info
+#   access image with qemu-img info
+# - cookie_with_secret_object
+#   check cookie data in tcpdump output
+
+- remote_readonly_storage:
+    only libcurl
+    virt_test_type = qemu
+    start_vm = 'no'
+
+    # cd1 is the libcurl iso image
+    remote_image_tag = 'cd1'
+    cdroms = "${remote_image_tag}"
+    cdrom_cd1 = Fedora-Server-dvd-x86_64-30-1.2.iso
+    image_format_cd1 = 'raw'
+    curl_server_cd1 = archives.fedoraproject.org
+    curl_protocol_cd1 = https
+    curl_path_cd1 = pub/archive/fedora/linux/releases/30/Server/x86_64/iso/Fedora-Server-dvd-x86_64-30-1.2.iso
+    curl_sslverify_cd1 = ''
+    curl_username_cd1 = ''
+    curl_password_cd1 = ''
+    curl_cookie_secret_cd1 = ''
+    curl_timeout_cd1 = 30
+    variants:
+        - cookie_with_secret_object:
+            only Linux
+            type = curl_cookie_with_secret
+            tcpdump_cmd = 'tcpdump -U -c 5 -A host {server} -w {dump_file} &'
+            curl_protocol_cd1 = http
+            curl_cookie_secret_cd1 = 'curl_cookie_data=redhat'
+        - boot_with_readonly_image:
+            only Linux
+            type = boot_with_remote_readonly_image
+            kill_vm = 'yes'
+
+            # stg0 is the local image, where VM starts
+            convert_target = 'stg0'
+            bootindex_stg0 = 0
+            image_name_stg0 = images/stg0
+            images += ' ${convert_target}'
+            enable_curl_stg0 = 'no'
+            storage_type_stg0 ='filesystem'
+            remove_image_stg0 = 'yes'
+            image_readonly_stg0 = 'no'
+        - access_with_qemu_info:
+            type = remote_image_qemu_info_access
+            ascii_symbol = '%5f'
+            replace_symbol = '_'

qemu/tests/curl_cookie_with_secret.py

+import os
+import signal
+import logging
+
+from virttest import utils_misc
+from virttest import qemu_storage
+from virttest import error_context
+
+from avocado.utils import process
+
+
+@error_context.context_aware
+def run(test, params, env):
+    """
+    1) Start tcpdump to capture the request
+    2) Access libcurl image by qemu-img
+    3) Wait till tcpdump finished
+    4) tcpdump should catch the cookie data
+
+    :param test: QEMU test object
+    :param params: Dictionary with the test parameters
+    :param env: Dictionary with test environment.
+    """
+    def _get_tcpdump_pid(dump_file):
+        cmd = ("ps -ef|grep tcpdump|grep %s|grep -v grep|awk '{print $2}'"
+               % dump_file)
+        return process.system_output(cmd, shell=True,
+                                     ignore_status=True).strip()
+
+    def _wait_for_tcpdump_done(dump_file):
+        response_timeout = params.get_numeric('response_timeout', 10)
+        if not utils_misc.wait_for(lambda: not _get_tcpdump_pid(dump_file),
+                                   response_timeout, 0, 1):
+            test.fail('tcpdump is running unexpectedly')
+
+    def _cleanup(dump_file):
+        if os.path.exists(dump_file):
+            os.unlink(dump_file)
+
+        pid = _get_tcpdump_pid(dump_file)
+        if pid:
+            os.kill(int(pid), signal.SIGKILL)
+
+    tag = params['remote_image_tag']
+    img_params = params.object_params(tag)
+    img_obj = qemu_storage.QemuImg(img_params, None, tag)
+    dump_file = utils_misc.generate_tmp_file_name('%s_access_tcpdump' % tag,
+                                                  'out')
+
+    logging.info('start tcpdump, save packets in %s' % dump_file)
+    process.system(
+        params['tcpdump_cmd'].format(server=img_params['curl_server'],
+                                     dump_file=dump_file),
+        shell=True, ignore_status=True, ignore_bg_processes=True
+    )
+
+    try:
+        img_obj.info()
+        _wait_for_tcpdump_done(dump_file)
+        with open(dump_file, 'rb') as fd:
+            for line in fd:
+                line = line.decode('utf-8', 'ignore')
+                if 'Cookie: %s' % img_params['curl_cookie_secret'] in line:
+                    logging.info('get "%s" from "%s"'
+                                 % (img_params['curl_cookie_secret'], line))
+                    break
+            else:
+                test.fail('Failed to get cookie data from tcpdump output')
+    finally:
+        _cleanup(dump_file)

qemu/tests/remote_image_qemu_info_access.py

+import logging
+
+from virttest import qemu_storage
+from virttest import error_context
+
+
+@error_context.context_aware
+def run(test, params, env):
+    """
+    1) Access remote image by qemu-img info
+    2) Check url in output for libcurl backend
+    3) Replace '_' with '%5f' in image name,
+       access image
+    4) Check url in output for libcurl backend
+
+    :param test: QEMU test object
+    :param params: Dictionary with the test parameters
+    :param env: Dictionary with test environment.
+    """
+    img_params = params.object_params(params['remote_image_tag'])
+    image_name_list = [
+        img_params['curl_path'],
+        img_params['curl_path'].replace(params['replace_symbol'],
+                                        params['ascii_symbol'])
+    ]
+
+    for image_name in image_name_list:
+        img_params['curl_path'] = image_name
+        img_obj = qemu_storage.QemuImg(img_params, None,
+                                       params['remote_image_tag'])
+
+        logging.info('Access image: %s' % img_obj.image_filename)
+        out = img_obj.info()
+
+        if img_obj.image_filename not in out:
+            test.fail('Failed to get url(%s) from output(%s)'
+                      % (img_obj.image_filename, out))