tests/test-python-plugin.py

   1 # nbdkit test plugin
   2 # Copyright (C) 2019-2020 Red Hat Inc.
   3 #
   4 # Redistribution and use in source and binary forms, with or without
   5 # modification, are permitted provided that the following conditions are
   6 # met:
   7 #
   8 # * Redistributions of source code must retain the above copyright
   9 # notice, this list of conditions and the following disclaimer.
  10 #
  11 # * Redistributions in binary form must reproduce the above copyright
  12 # notice, this list of conditions and the following disclaimer in the
  13 # documentation and/or other materials provided with the distribution.
  14 #
  15 # * Neither the name of Red Hat nor the names of its contributors may be
  16 # used to endorse or promote products derived from this software without
  17 # specific prior written permission.
  18 #
  19 # THIS SOFTWARE IS PROVIDED BY RED HAT AND CONTRIBUTORS ''AS IS'' AND
  20 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  21 # THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  22 # PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL RED HAT OR
  23 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  24 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  25 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  26 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  27 # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  28 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  29 # OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  30 # SUCH DAMAGE.
  31
  32 """See test-python.py."""
  33
  34 import nbdkit
  35 import sys
  36 import pickle
  37 import base64
  38
  39 API_VERSION = 2
  40
  41 cfg = {}
  42
  43 def config (k, v):
  44     global cfg
  45     if k == "cfg":
  46         cfg = pickle.loads (base64.b64decode (v.encode()))
  47
  48 def config_complete ():
  49     print ("set_error = %r" % nbdkit.set_error)
  50
  51 def open (readonly):
  52     return {
  53         'disk': bytearray (cfg.get ('size', 0))
  54     }
  55
  56 def get_size (h):
  57     return len (h['disk'])
  58
  59 def is_rotational (h):
  60     return cfg.get ('is_rotational', False)
  61
  62 def can_multi_conn (h):
  63     return cfg.get ('can_multi_conn', False)
  64
  65 def can_write (h):
  66     return cfg.get ('can_write', True)
  67
  68 def can_flush (h):
  69     return cfg.get ('can_flush', False)
  70
  71 def can_trim (h):
  72     return cfg.get ('can_trim', False)
  73
  74 def can_zero (h):
  75     return cfg.get ('can_zero', False)
  76
  77 def can_fast_zero (h):
  78     return cfg.get ('can_fast_zero', False)
  79
  80 def can_fua (h):
  81     fua = cfg.get ('can_fua', "none")
  82     if fua == "none":
  83         return nbdkit.FUA_NONE
  84     elif fua == "emulate":
  85         return nbdkit.FUA_EMULATE
  86     elif fua == "native":
  87         return nbdkit.FUA_NATIVE
  88
  89 def can_cache (h):
  90     cache = cfg.get ('can_cache', "none")
  91     if cache == "none":
  92         return nbdkit.CACHE_NONE
  93     elif cache == "emulate":
  94         return nbdkit.CACHE_EMULATE
  95     elif cache == "native":
  96         return nbdkit.CACHE_NATIVE
  97
  98 def init_sparse (h):
  99     return cfg.get ('init_sparse', False)
 100
 101 def init_zero (h):
 102     return cfg.get ('init_zero', False)
 103
 104 def pread (h, buf, offset, flags):
 105     assert flags == 0
 106     end = offset + len(buf)
 107     buf[:] = h['disk'][offset:end]
 108
 109 def pwrite (h, buf, offset, flags):
 110     expect_fua = cfg.get ('pwrite_expect_fua', False)
 111     actual_fua = bool (flags & nbdkit.FLAG_FUA)
 112     assert expect_fua == actual_fua
 113     end = offset + len(buf)
 114     h['disk'][offset:end] = buf
 115
 116 def flush (h, flags):
 117     assert flags == 0
 118
 119 def trim (h, count, offset, flags):
 120     expect_fua = cfg.get ('trim_expect_fua', False)
 121     actual_fua = bool (flags & nbdkit.FLAG_FUA)
 122     assert expect_fua == actual_fua
 123     h['disk'][offset:offset+count] = bytearray(count)
 124
 125 def zero (h, count, offset, flags):
 126     expect_fua = cfg.get ('zero_expect_fua', False)
 127     actual_fua = bool (flags & nbdkit.FLAG_FUA)
 128     assert expect_fua == actual_fua
 129     expect_may_trim = cfg.get ('zero_expect_may_trim', False)
 130     actual_may_trim = bool (flags & nbdkit.FLAG_MAY_TRIM)
 131     assert expect_may_trim == actual_may_trim
 132     expect_fast_zero = cfg.get ('zero_expect_fast_zero', False)
 133     actual_fast_zero = bool (flags & nbdkit.FLAG_FAST_ZERO)
 134     assert expect_fast_zero == actual_fast_zero
 135     h['disk'][offset:offset+count] = bytearray(count)
 136
 137 def cache (h, count, offset, flags):
 138     assert flags == 0
 139     # do nothing