def test_rebuild_empty(self, fake_sanlock):
        with make_volume() as vol:
            # Add underlying sanlock resources
            for i in [3, 4, 6]:
                resource = "%04d" % i
                offset = xlease.USER_RESOURCE_BASE + xlease.SLOT_SIZE * i
                fake_sanlock.write_resource(
                    vol.lockspace, resource, [(vol.path, offset)])
            # The index is empty
            assert vol.leases() == {}

            # After rebuilding the index it should contain all the underlying
            # resources.
            file = xlease.DirectFile(vol.path)
            with utils.closing(file):
                xlease.rebuild_index(vol.lockspace, file)
            expected = {
                "0003": {
                    "offset": xlease.USER_RESOURCE_BASE + xlease.SLOT_SIZE * 3,
                    "updating": False,
                },
                "0004": {
                    "offset": xlease.USER_RESOURCE_BASE + xlease.SLOT_SIZE * 4,
                    "updating": False,
                },
                "0006": {
                    "offset": xlease.USER_RESOURCE_BASE + xlease.SLOT_SIZE * 6,
                    "updating": False,
                },
            }
            file = xlease.DirectFile(vol.path)
            with utils.closing(file):
                vol = xlease.LeasesVolume(file)
                with utils.closing(vol):
                    assert vol.leases() == expected

def write_records(records, file):
    index = xlease.VolumeIndex()
    with utils.closing(index):
        index.load(file)
        for recnum, record in records:
            block = index.copy_record_block(recnum)
            with utils.closing(block):
                block.write_record(recnum, record)
                block.dump(file)

 def test_add_write_failure(self):
     with make_volume() as base:
         file = FailingWriter(base.path)
         with utils.closing(file):
             vol = xlease.LeasesVolume(file)
             with utils.closing(vol):
                 lease_id = make_uuid()
                 with pytest.raises(WriteError):
                     vol.add(lease_id)
                 # Must succeed becuase writng to storage failed
                 assert lease_id not in vol.leases()

 def test_pread_short(self, tmpdir, direct_file):
     data = b"a" * 1024
     path = tmpdir.join("file")
     path.write(data)
     file = direct_file(str(path))
     with utils.closing(file):
         buf = mmap.mmap(-1, 1024)
         with utils.closing(buf):
             n = file.pread(512, buf)
             assert n == 512
             assert buf[:n] == data[512:]

def make_volume(*records):
    with make_leases() as path:
        lockspace = os.path.basename(os.path.dirname(path))
        file = xlease.DirectFile(path)
        with utils.closing(file):
            xlease.format_index(lockspace, file)
            if records:
                write_records(records, file)
            vol = xlease.LeasesVolume(file)
            with utils.closing(vol):
                yield vol

 def test_pread(self, tmpdir, direct_file, offset, size):
     data = b"a" * 512 + b"b" * 512 + b"c" * 512 + b"d" * 512
     path = tmpdir.join("file")
     path.write(data)
     file = direct_file(str(path))
     with utils.closing(file):
         buf = mmap.mmap(-1, size)
         with utils.closing(buf):
             n = file.pread(offset, buf)
             assert n == size
             assert buf[:] == data[offset:offset + size]

 def test_remove_write_failure(self):
     record = xlease.Record(make_uuid(), 0, updating=True)
     with make_volume((42, record)) as base:
         file = FailingWriter(base.path)
         with utils.closing(file):
             vol = xlease.LeasesVolume(file)
             with utils.closing(vol):
                 with pytest.raises(WriteError):
                     vol.remove(record.resource)
                 # Must succeed becuase writng to storage failed
                 assert record.resource in vol.leases()

def _external_leases_volume(path):
    """
    Context manager returning the external leases volume.

    The caller is responsible for holding the external_leases_lock in the
    correct mode.
    """
    backend = xlease.DirectFile(path)
    with utils.closing(backend):
        vol = xlease.LeasesVolume(backend)
        with utils.closing(vol):
            yield vol

 def test_pwrite(self, tmpdir, direct_file, offset, size):
     # Create a file full of "a"s
     path = tmpdir.join("file")
     path.write(b"a" * 2048)
     buf = mmap.mmap(-1, size)
     with utils.closing(buf):
         # Write "b"s
         buf.write(b"b" * size)
         file = direct_file(str(path))
         with utils.closing(file):
             file.pwrite(offset, buf)
     data = path.read()
     expected = ("a" * offset +
                 "b" * size +
                 "a" * (2048 - offset - size))
     assert data == expected

    def pread(self, offset, buf):
        """
        Read len(buf) bytes from storage at offset into mmap buf.

        Returns:
            The number bytes read (int).
        """
        self._file.seek(offset, os.SEEK_SET)
        pos = 0
        if six.PY2:
            # There is no way to create a writable memoryview on mmap object in
            # python 2, so we must read into a temporary buffer and copy into
            # the given buffer.
            rbuf = mmap.mmap(-1, len(buf), mmap.MAP_SHARED)
            with utils.closing(rbuf, log=log.name):
                while pos < len(buf):
                    # TODO: Handle EOF
                    nread = uninterruptible(self._file.readinto, rbuf)
                    if nread == 0:
                        break  # EOF
                    buf.write(rbuf[:nread])
                    pos += nread
        else:
            # In python 3 we can read directly into the underlying buffer
            # without any copies using a memoryview.
            while pos < len(buf):
                rbuf = memoryview(buf)[pos:]
                # TODO: Handle EOF
                nread = uninterruptible(self._file.readinto, rbuf)
                if nread == 0:
                    break  # EOF
                pos += nread
        return pos

def dump_chains(*args):
    """
    dump-volume-chains
    Query VDSM about the existing structure of image volumes and prints
    them in an ordered fashion with optional additional info per volume.
    Alternatively, dumps the volumes information in json format without
    analysis.
    """
    parsed_args = _parse_args(args)
    cli = client.connect(parsed_args.host, parsed_args.port,
                         use_tls=parsed_args.use_ssl)
    with utils.closing(cli):
        volumes_info = _get_volumes_info(cli, parsed_args.sd_uuid)
        if parsed_args.output == 'text':
            # perform analysis and print in human readable format
            image_chains = _get_volumes_chains(volumes_info)
            _print_volume_chains(image_chains, volumes_info)
        elif parsed_args.output == 'json':
            # no analysis, dump chains in json format
            json.dump(volumes_info, sys.stdout, indent=2)
        elif parsed_args.output == 'sqlite':
            # no analysis, dump chains in sql format
            _dump_sql(volumes_info, parsed_args.sqlite_file)
        else:
            raise ValueError('unknown output format %s' % parsed_args.output)

def main(args=None):
    schema = find_schema()
    namespaces = create_namespaces(schema)
    parser = option_parser(namespaces)
    args = parser.parse_args(args)
    try:
        if args.method_args and args.file is not None:
            raise UsageError("Conflicting command line parameters: %r and "
                             "file option: %r" % (args.method_args, args.file))

        namespace = args.namespace
        method = args.method

        if args.file:
            request_params = parse_file(args.file)
        else:
            request_params = parse_params(args.method_args)

        cli = client.connect(args.host, port=args.port, use_tls=args.use_tls,
                             timeout=args.timeout,
                             gluster_enabled=args.gluster_enabled)

        with utils.closing(cli):
            command = getattr(getattr(cli, namespace), method)
            result = command(**request_params)
            print(json.dumps(result, indent=4))
    except UsageError as e:
        parser.error(str(e))
    except Exception as e:
        fail(e)

def main(*args):
    """
    This tool is used to check and optionally repair broken volume leases.
    """
    parsed_args = _parse_args(args)

    if not parsed_args.repair and not _confirm_check_leases():
        return

    cli = client.connect(parsed_args.host, parsed_args.port,
                         use_tls=parsed_args.use_ssl)
    with utils.closing(cli):
        print()
        print("Checking active storage domains. This can take several "
              "minutes, please wait.")
        broken_leases = _get_leases_to_repair(cli)
        if not broken_leases:
            print()
            print("There are no leases to repair.")
            return

    print()
    _print_broken_leases(broken_leases)
    if not parsed_args.repair and not _confirm_repair_leases():
        return

    _repair(broken_leases)

    def test_write(self, cmd, recv_out, recv_err):
        text = bytes('Hello World')
        received = bytearray()

        def recv_data(buffer):
            # cannot use received += buffer with a variable
            # defined in the parent function.
            operator.iadd(received, buffer)

        c = self._startCommand(cmd)
        p = utils.CommandStream(
            c,
            recv_data if recv_out else self.assertUnexpectedCall,
            recv_data if recv_err else self.assertUnexpectedCall)
        with utils.closing(p):
            c.stdin.write(text)
            c.stdin.flush()
            c.stdin.close()

            while not p.closed:
                p.receive()

        retcode = c.wait()

        self.assertEqual(retcode, 0)
        self.assertEqual(text, str(received))

def resume_paused_vm(vm_id):
    unpause_file = MARK_FOR_UNPAUSE_PATH % vm_id
    if os.path.isfile(unpause_file):
        use_tls = config.getboolean('vars', 'ssl')
        cli = client.connect('localhost', use_tls=use_tls)
        with utils.closing(cli):
            cli.VM.cont(vmID=vm_id)
        os.remove(unpause_file)

def converted_size(filename, compat):
    converted = filename + ".qcow2"
    operation = qemuimg.convert(filename,
                                converted,
                                srcFormat=qemuimg.FORMAT.RAW,
                                dstFormat=qemuimg.FORMAT.QCOW2,
                                dstQcow2Compat=compat)
    with utils.closing(operation):
        operation.wait_for_completion()
    return os.stat(converted).st_size

 def _run(self):
     self.log.info("Merging subchain %s", self.subchain)
     with guarded.context(self.subchain.locks):
         self.subchain.validate()
         with self.subchain.prepare(), self.subchain.volume_operation():
             self.operation = qemuimg.commit(
                 self.subchain.top_vol.getVolumePath(),
                 topFormat=sc.fmt2str(self.subchain.top_vol.getFormat()),
                 base=self.subchain.base_vol.getVolumePath())
             with utils.closing(self.operation):
                 self.operation.wait_for_completion()

    def rebuild_external_leases(self):
        """
        Rebuild the external leases volume index from volume contents.

        Must be called only on the SPM.
        """
        with self._manifest.external_leases_lock.exclusive:
            path = self.external_leases_path()
            backend = xlease.DirectFile(path)
            with utils.closing(backend):
                xlease.rebuild_index(self.sdUUID, backend)

    def _write_record(self, recnum, record):
        """
        Write record recnum to storage atomically.

        Copy the block where the record is located, modify it and write the
        block to storage. If this succeeds, write the record to the index.
        """
        block = self._index.copy_record_block(recnum)
        with utils.closing(block):
            block.write_record(recnum, record)
            block.dump(self._file)
        self._index.write_record(recnum, record)

    def external_leases_volume(self):
        """
        Context manager returning the external leases volume.

        The caller is responsible for holding the external_leases_lock in the
        correct mode.
        """
        path = self.external_leases_path()
        with self.external_leases_backend(self.sdUUID, path) as backend:
            vol = xlease.LeasesVolume(backend)
            with utils.closing(vol):
                yield vol