diff options
Diffstat (limited to 'tests/qemu-iotests/qcow2.py')
| -rwxr-xr-x | tests/qemu-iotests/qcow2.py | 226 |
1 files changed, 67 insertions, 159 deletions
diff --git a/tests/qemu-iotests/qcow2.py b/tests/qemu-iotests/qcow2.py index b95a837759..77ca59cc66 100755 --- a/tests/qemu-iotests/qcow2.py +++ b/tests/qemu-iotests/qcow2.py @@ -1,168 +1,55 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 +# +# Manipulations with qcow2 image +# +# Copyright (C) 2012 Red Hat, Inc. +# +# 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/>. +# -from __future__ import print_function import sys -import struct -import string - -class QcowHeaderExtension: - - def __init__(self, magic, length, data): - if length % 8 != 0: - padding = 8 - (length % 8) - data += "\0" * padding - - self.magic = magic - self.length = length - self.data = data - - @classmethod - def create(cls, magic, data): - return QcowHeaderExtension(magic, len(data), data) - -class QcowHeader: - - uint32_t = 'I' - uint64_t = 'Q' - - fields = [ - # Version 2 header fields - [ uint32_t, '%#x', 'magic' ], - [ uint32_t, '%d', 'version' ], - [ uint64_t, '%#x', 'backing_file_offset' ], - [ uint32_t, '%#x', 'backing_file_size' ], - [ uint32_t, '%d', 'cluster_bits' ], - [ uint64_t, '%d', 'size' ], - [ uint32_t, '%d', 'crypt_method' ], - [ uint32_t, '%d', 'l1_size' ], - [ uint64_t, '%#x', 'l1_table_offset' ], - [ uint64_t, '%#x', 'refcount_table_offset' ], - [ uint32_t, '%d', 'refcount_table_clusters' ], - [ uint32_t, '%d', 'nb_snapshots' ], - [ uint64_t, '%#x', 'snapshot_offset' ], - - # Version 3 header fields - [ uint64_t, '%#x', 'incompatible_features' ], - [ uint64_t, '%#x', 'compatible_features' ], - [ uint64_t, '%#x', 'autoclear_features' ], - [ uint32_t, '%d', 'refcount_order' ], - [ uint32_t, '%d', 'header_length' ], - ]; - - fmt = '>' + ''.join(field[0] for field in fields) - - def __init__(self, fd): - - buf_size = struct.calcsize(QcowHeader.fmt) - - fd.seek(0) - buf = fd.read(buf_size) - - header = struct.unpack(QcowHeader.fmt, buf) - self.__dict__ = dict((field[2], header[i]) - for i, field in enumerate(QcowHeader.fields)) - - self.set_defaults() - self.cluster_size = 1 << self.cluster_bits - - fd.seek(self.header_length) - self.load_extensions(fd) - - if self.backing_file_offset: - fd.seek(self.backing_file_offset) - self.backing_file = fd.read(self.backing_file_size) - else: - self.backing_file = None - - def set_defaults(self): - if self.version == 2: - self.incompatible_features = 0 - self.compatible_features = 0 - self.autoclear_features = 0 - self.refcount_order = 4 - self.header_length = 72 - - def load_extensions(self, fd): - self.extensions = [] - - if self.backing_file_offset != 0: - end = min(self.cluster_size, self.backing_file_offset) - else: - end = self.cluster_size - - while fd.tell() < end: - (magic, length) = struct.unpack('>II', fd.read(8)) - if magic == 0: - break - else: - padded = (length + 7) & ~7 - data = fd.read(padded) - self.extensions.append(QcowHeaderExtension(magic, length, data)) - - def update_extensions(self, fd): - - fd.seek(self.header_length) - extensions = self.extensions - extensions.append(QcowHeaderExtension(0, 0, "")) - for ex in extensions: - buf = struct.pack('>II', ex.magic, ex.length) - fd.write(buf) - fd.write(ex.data) - - if self.backing_file != None: - self.backing_file_offset = fd.tell() - fd.write(self.backing_file) - - if fd.tell() > self.cluster_size: - raise Exception("I think I just broke the image...") - - def update(self, fd): - header_bytes = self.header_length +from qcow2_format import ( + QcowHeader, + QcowHeaderExtension +) - self.update_extensions(fd) - fd.seek(0) - header = tuple(self.__dict__[f] for t, p, f in QcowHeader.fields) - buf = struct.pack(QcowHeader.fmt, *header) - buf = buf[0:header_bytes-1] - fd.write(buf) +is_json = False - def dump(self): - for f in QcowHeader.fields: - print("%-25s" % f[2], f[1] % self.__dict__[f[2]]) - print("") - def dump_extensions(self): - for ex in self.extensions: - - data = ex.data[:ex.length] - if all(c in string.printable for c in data): - data = "'%s'" % data - else: - data = "<binary>" - - print("Header extension:") - print("%-25s %#x" % ("magic", ex.magic)) - print("%-25s %d" % ("length", ex.length)) - print("%-25s %s" % ("data", data)) - print("") +def cmd_dump_header(fd): + h = QcowHeader(fd) + h.dump(is_json) + print() + h.dump_extensions(is_json) -def cmd_dump_header(fd): +def cmd_dump_header_exts(fd): h = QcowHeader(fd) - h.dump() - h.dump_extensions() + h.dump_extensions(is_json) + def cmd_set_header(fd, name, value): try: value = int(value, 0) - except: + except ValueError: print("'%s' is not a valid number" % value) sys.exit(1) fields = (field[2] for field in QcowHeader.fields) - if not name in fields: + if name not in fields: print("'%s' is not a known header field" % name) sys.exit(1) @@ -170,25 +57,29 @@ def cmd_set_header(fd, name, value): h.__dict__[name] = value h.update(fd) + def cmd_add_header_ext(fd, magic, data): try: magic = int(magic, 0) - except: + except ValueError: print("'%s' is not a valid magic number" % magic) sys.exit(1) h = QcowHeader(fd) - h.extensions.append(QcowHeaderExtension.create(magic, data)) + h.extensions.append(QcowHeaderExtension.create(magic, + data.encode('ascii'))) h.update(fd) + def cmd_add_header_ext_stdio(fd, magic): data = sys.stdin.read() cmd_add_header_ext(fd, magic, data) + def cmd_del_header_ext(fd, magic): try: magic = int(magic, 0) - except: + except ValueError: print("'%s' is not a valid magic number" % magic) sys.exit(1) @@ -206,12 +97,13 @@ def cmd_del_header_ext(fd, magic): h.update(fd) + def cmd_set_feature_bit(fd, group, bit): try: bit = int(bit, 0) if bit < 0 or bit >= 64: raise ValueError - except: + except ValueError: print("'%s' is not a valid bit number in range [0, 64)" % bit) sys.exit(1) @@ -223,20 +115,27 @@ def cmd_set_feature_bit(fd, group, bit): elif group == 'autoclear': h.autoclear_features |= 1 << bit else: - print("'%s' is not a valid group, try 'incompatible', 'compatible', or 'autoclear'" % group) + print("'%s' is not a valid group, try " + "'incompatible', 'compatible', or 'autoclear'" % group) sys.exit(1) h.update(fd) + cmds = [ - [ 'dump-header', cmd_dump_header, 0, 'Dump image header and header extensions' ], - [ 'set-header', cmd_set_header, 2, 'Set a field in the header'], - [ 'add-header-ext', cmd_add_header_ext, 2, 'Add a header extension' ], - [ 'add-header-ext-stdio', cmd_add_header_ext_stdio, 1, 'Add a header extension, data from stdin' ], - [ 'del-header-ext', cmd_del_header_ext, 1, 'Delete a header extension' ], - [ 'set-feature-bit', cmd_set_feature_bit, 2, 'Set a feature bit'], + ['dump-header', cmd_dump_header, 0, + 'Dump image header and header extensions'], + ['dump-header-exts', cmd_dump_header_exts, 0, + 'Dump image header extensions'], + ['set-header', cmd_set_header, 2, 'Set a field in the header'], + ['add-header-ext', cmd_add_header_ext, 2, 'Add a header extension'], + ['add-header-ext-stdio', cmd_add_header_ext_stdio, 1, + 'Add a header extension, data from stdin'], + ['del-header-ext', cmd_del_header_ext, 1, 'Delete a header extension'], + ['set-feature-bit', cmd_set_feature_bit, 2, 'Set a feature bit'], ] + def main(filename, cmd, args): fd = open(filename, "r+b") try: @@ -253,16 +152,25 @@ def main(filename, cmd, args): finally: fd.close() + def usage(): - print("Usage: %s <file> <cmd> [<arg>, ...]" % sys.argv[0]) + print("Usage: %s <file> <cmd> [<arg>, ...] [<key>, ...]" % sys.argv[0]) print("") print("Supported commands:") for name, handler, num_args, desc in cmds: print(" %-20s - %s" % (name, desc)) + print("") + print("Supported keys:") + print(" %-20s - %s" % ('-j', 'Dump in JSON format')) + if __name__ == '__main__': if len(sys.argv) < 3: usage() sys.exit(1) + is_json = '-j' in sys.argv + if is_json: + sys.argv.remove('-j') + main(sys.argv[1], sys.argv[2], sys.argv[3:]) |