qemu-e2k/tests/qemu-iotests/207
Max Reitz 8eb5e6746f iotests: Use Python byte strings where appropriate
Since byte strings are no longer the default in Python 3, we have to
explicitly use them where we need to, which is mostly when working with
structures.  It also means that we need to open a file in binary mode
when we want to use structures.

On the other hand, we have to accomodate for the fact that some
functions (still) work with byte strings but we want to use unicode
strings (in Python 3 at least, and it does not matter in Python 2).
This includes base64 encoding, but it is most notable when working with
the subprocess module: Either we set universal_newlines to True so that
the default streams are opened in text mode (hence this parameter is
aliased as "text" as of 3.7), or, if that is not possible, we have to
decode the output to a normal string.

Signed-off-by: Max Reitz <mreitz@redhat.com>
Reviewed-by: Eduardo Habkost <ehabkost@redhat.com>
Message-Id: <20181022135307.14398-4-mreitz@redhat.com>
Signed-off-by: Eduardo Habkost <ehabkost@redhat.com>
2018-10-30 21:11:52 -03:00

218 lines
7.6 KiB
Python
Executable File

#!/usr/bin/env python
#
# Test ssh image creation
#
# Copyright (C) 2018 Red Hat, Inc.
#
# Creator/Owner: Kevin Wolf <kwolf@redhat.com>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
import iotests
import subprocess
import re
iotests.verify_image_format(supported_fmts=['raw'])
iotests.verify_protocol(supported=['ssh'])
def filter_hash(msg):
return re.sub("'hash': '[0-9a-f]+'", "'hash': HASH", msg)
def blockdev_create(vm, options):
result = vm.qmp_log('blockdev-create', job_id='job0', options=options,
filters=[iotests.filter_testfiles, filter_hash])
if 'return' in result:
assert result['return'] == {}
vm.run_job('job0')
iotests.log("")
with iotests.FilePath('t.img') as disk_path, \
iotests.VM() as vm:
remote_path = iotests.remote_filename(disk_path)
#
# Successful image creation (defaults)
#
iotests.log("=== Successful image creation (defaults) ===")
iotests.log("")
vm.launch()
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
}
},
'size': 4194304 })
vm.shutdown()
iotests.img_info_log(remote_path, filter_path=disk_path)
iotests.log("")
iotests.img_info_log(disk_path)
#
# Test host-key-check options
#
iotests.log("=== Test host-key-check options ===")
iotests.log("")
vm.launch()
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'none'
}
},
'size': 8388608 })
vm.shutdown()
iotests.img_info_log(remote_path, filter_path=disk_path)
vm.launch()
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'known_hosts'
}
},
'size': 4194304 })
vm.shutdown()
iotests.img_info_log(remote_path, filter_path=disk_path)
md5_key = subprocess.check_output(
'ssh-keyscan -t rsa 127.0.0.1 2>/dev/null | grep -v "\\^#" | ' +
'cut -d" " -f3 | base64 -d | md5sum -b | cut -d" " -f1',
shell=True).rstrip().decode('ascii')
vm.launch()
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'hash',
'type': 'md5',
'hash': 'wrong',
}
},
'size': 2097152 })
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'hash',
'type': 'md5',
'hash': md5_key,
}
},
'size': 8388608 })
vm.shutdown()
iotests.img_info_log(remote_path, filter_path=disk_path)
sha1_key = subprocess.check_output(
'ssh-keyscan -t rsa 127.0.0.1 2>/dev/null | grep -v "\\^#" | ' +
'cut -d" " -f3 | base64 -d | sha1sum -b | cut -d" " -f1',
shell=True).rstrip().decode('ascii')
vm.launch()
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'hash',
'type': 'sha1',
'hash': 'wrong',
}
},
'size': 2097152 })
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'hash',
'type': 'sha1',
'hash': sha1_key,
}
},
'size': 4194304 })
vm.shutdown()
iotests.img_info_log(remote_path, filter_path=disk_path)
#
# Invalid path and user
#
iotests.log("=== Invalid path and user ===")
iotests.log("")
vm.launch()
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': '/this/is/not/an/existing/path',
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'none'
}
},
'size': 4194304 })
blockdev_create(vm, { 'driver': 'ssh',
'location': {
'path': disk_path,
'user': 'invalid user',
'server': {
'host': '127.0.0.1',
'port': '22'
},
'host-key-check': {
'mode': 'none'
}
},
'size': 4194304 })
vm.shutdown()