tests/test_python.py

   1 #!/usr/bin/env python3
   2 # nbdkit
   3 # Copyright Red Hat
   4 #
   5 # Redistribution and use in source and binary forms, with or without
   6 # modification, are permitted provided that the following conditions are
   7 # met:
   8 #
   9 # * Redistributions of source code must retain the above copyright
  10 # notice, this list of conditions and the following disclaimer.
  11 #
  12 # * Redistributions in binary form must reproduce the above copyright
  13 # notice, this list of conditions and the following disclaimer in the
  14 # documentation and/or other materials provided with the distribution.
  15 #
  16 # * Neither the name of Red Hat nor the names of its contributors may be
  17 # used to endorse or promote products derived from this software without
  18 # specific prior written permission.
  19 #
  20 # THIS SOFTWARE IS PROVIDED BY RED HAT AND CONTRIBUTORS ''AS IS'' AND
  21 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
  22 # THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A
  23 # PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL RED HAT OR
  24 # CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25 # SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26 # LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
  27 # USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
  28 # ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
  29 # OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT
  30 # OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  31 # SUCH DAMAGE.
  32
  33 """
  34 This tests the Python plugin thoroughly by issuing client commands
  35 through libnbd and checking we get the expected results.  It uses an
  36 associated plugin (test-python-plugin.sh).
  37 """
  38
  39 import nbd
  40 import unittest
  41 import pickle
  42 import base64
  43
  44
  45 class Test(unittest.TestCase):
  46     def setUp(self):
  47         self.h = nbd.NBD()
  48
  49     def tearDown(self):
  50         del self.h
  51
  52     def connect(self, cfg):
  53         cfg = base64.b64encode(pickle.dumps(cfg)).decode()
  54         cmd = ["nbdkit", "-v", "-s", "--exit-with-parent",
  55                "python", "test-python-plugin.py", "cfg=" + cfg]
  56         self.h.connect_command(cmd)
  57
  58     def test_none(self):
  59         """
  60         Test we can send an empty pickled test configuration and do
  61         nothing else.  This is just to ensure the machinery of the
  62         test works.
  63         """
  64         self.connect({})
  65
  66     def test_size_512(self):
  67         """Test the size."""
  68         self.connect({"size": 512})
  69         assert self.h.get_size() == 512
  70
  71     def test_size_1m(self):
  72         """Test the size."""
  73         self.connect({"size": 1024*1024})
  74         assert self.h.get_size() == 1024*1024
  75
  76     # Test each flag call.
  77     def test_is_rotational_true(self):
  78         self.connect({"size": 512, "is_rotational": True})
  79         assert self.h.is_rotational()
  80
  81     def test_is_rotational_false(self):
  82         self.connect({"size": 512, "is_rotational": False})
  83         assert not self.h.is_rotational()
  84
  85     def test_can_multi_conn_true(self):
  86         self.connect({"size": 512, "can_multi_conn": True})
  87         assert self.h.can_multi_conn()
  88
  89     def test_can_multi_conn_false(self):
  90         self.connect({"size": 512, "can_multi_conn": False})
  91         assert not self.h.can_multi_conn()
  92
  93     def test_read_write(self):
  94         self.connect({"size": 512, "can_write": True})
  95         assert not self.h.is_read_only()
  96
  97     def test_read_only(self):
  98         self.connect({"size": 512, "can_write": False})
  99         assert self.h.is_read_only()
 100
 101     def test_can_flush_true(self):
 102         self.connect({"size": 512, "can_flush": True})
 103         assert self.h.can_flush()
 104
 105     def test_can_flush_false(self):
 106         self.connect({"size": 512, "can_flush": False})
 107         assert not self.h.can_flush()
 108
 109     def test_can_trim_true(self):
 110         self.connect({"size": 512, "can_trim": True})
 111         assert self.h.can_trim()
 112
 113     def test_can_trim_false(self):
 114         self.connect({"size": 512, "can_trim": False})
 115         assert not self.h.can_trim()
 116
 117     # nbdkit can always zero because it emulates it.
 118     # self.connect({"size": 512, "can_zero": True})
 119     # assert self.h.can_zero()
 120     # self.connect({"size": 512, "can_zero": False})
 121     # assert not self.h.can_zero()
 122
 123     def test_can_fast_zero_true(self):
 124         self.connect({"size": 512, "can_fast_zero": True})
 125         assert self.h.can_fast_zero()
 126
 127     def test_can_fast_zero_false(self):
 128         self.connect({"size": 512, "can_fast_zero": False})
 129         assert not self.h.can_fast_zero()
 130
 131     def test_can_fua_none(self):
 132         self.connect({"size": 512, "can_fua": "none"})
 133         assert not self.h.can_fua()
 134
 135     def test_can_fua_emulate(self):
 136         self.connect({"size": 512, "can_fua": "emulate"})
 137         assert self.h.can_fua()
 138
 139     def test_can_fua_native(self):
 140         self.connect({"size": 512, "can_fua": "native"})
 141         assert self.h.can_fua()
 142
 143     def test_can_cache_none(self):
 144         self.connect({"size": 512, "can_cache": "none"})
 145         assert not self.h.can_cache()
 146
 147     def test_can_cache_emulate(self):
 148         self.connect({"size": 512, "can_cache": "emulate"})
 149         assert self.h.can_cache()
 150
 151     def test_can_cache_native(self):
 152         self.connect({"size": 512, "can_cache": "native"})
 153         assert self.h.can_cache()
 154
 155     # In theory we could use a test like this, but nbdkit can
 156     # always synthesize base:allocation block_status responses
 157     # even if the plugin doesn't support them.
 158     #
 159     # def test_can_extents_true(self):
 160     #     self.h.add_meta_context("base:allocation")
 161     #     self.connect({"size": 512, "can_extents": True})
 162     #     assert self.h.can_meta_context("base:allocation")
 163     #
 164     # def test_can_extents_false(self):
 165     #     self.h.add_meta_context("base:allocation")
 166     #     self.connect({"size": 512, "can_extents": False})
 167     #     assert not self.h.can_meta_context("base:allocation")
 168
 169     def test_pread(self):
 170         """Test pread."""
 171         self.connect({"size": 512})
 172         buf = self.h.pread(512, 0)
 173         assert buf == bytearray(512)
 174
 175     # Test pwrite + flags.
 176     def test_pwrite(self):
 177         self.connect({"size": 512})
 178         buf = bytearray(512)
 179         self.h.pwrite(buf, 0)
 180
 181     def test_pwrite_fua(self):
 182         self.connect({"size": 512,
 183                       "can_fua": "native",
 184                       "pwrite_expect_fua": True})
 185         buf = bytearray(512)
 186         self.h.pwrite(buf, 0, nbd.CMD_FLAG_FUA)
 187
 188     def test_flush(self):
 189         """Test flush."""
 190         self.connect({"size": 512, "can_flush": True})
 191         self.h.flush()
 192
 193     def test_trim(self):
 194         """Test trim."""
 195         self.connect({"size": 512, "can_trim": True})
 196         self.h.trim(512, 0)
 197
 198     def test_trim_fua(self):
 199         """Test trim + flags."""
 200         self.connect({"size": 512,
 201                       "can_trim": True,
 202                       "can_fua": "native",
 203                       "trim_expect_fua": True})
 204         self.h.trim(512, 0, nbd.CMD_FLAG_FUA)
 205
 206     def test_zero(self):
 207         """Test zero."""
 208         self.connect({"size": 512, "can_zero": True})
 209         self.h.zero(512, 0, nbd.CMD_FLAG_NO_HOLE)
 210
 211     def test_zero_fua(self):
 212         """Test zero + flags."""
 213         self.connect({"size": 512,
 214                       "can_zero": True,
 215                       "can_fua": "native",
 216                       "zero_expect_fua": True})
 217         self.h.zero(512, 0, nbd.CMD_FLAG_NO_HOLE | nbd.CMD_FLAG_FUA)
 218
 219     def test_zero_may_trim(self):
 220         """Test zero + may trim."""
 221         self.connect({"size": 512,
 222                       "can_zero": True,
 223                       "zero_expect_may_trim": True})
 224         self.h.zero(512, 0, 0)  # absence of nbd.CMD_FLAG_NO_HOLE
 225
 226     def test_zero_fast_zero(self):
 227         """Test zero + fast zero."""
 228         self.connect({"size": 512,
 229                       "can_zero": True,
 230                       "can_fast_zero": True,
 231                       "zero_expect_fast_zero": True})
 232         self.h.zero(512, 0, nbd.CMD_FLAG_NO_HOLE | nbd.CMD_FLAG_FAST_ZERO)
 233
 234     def test_zero_large(self):
 235         """Test large zero."""
 236         self.connect({"size": 0x10_0000_0000,
 237                       "can_zero": True,
 238                       "create_disk": False})
 239         self.h.zero(0xf000_0000, 0, nbd.CMD_FLAG_NO_HOLE)
 240
 241     def test_cache(self):
 242         """Test cache."""
 243         self.connect({"size": 512, "can_cache": "native"})
 244         self.h.cache(512, 0)
 245
 246     # We don't have access to the magic constants defined in the
 247     # nbdkit module, so redefine them here.
 248     EXTENT_HOLE = 1
 249     EXTENT_ZERO = 2
 250
 251     def test_extents_1(self):
 252         """Test extents."""
 253
 254         offset = None
 255         entries = []
 256
 257         def f(meta_context, o, e, err):
 258             nonlocal offset, entries
 259             if meta_context != "base:allocation":
 260                 return
 261             offset = o
 262             entries = e
 263
 264         self.h.add_meta_context("base:allocation")
 265         self.connect({"size": 512,
 266                       "can_extents": True,
 267                       "extents":
 268                       [(0, 512, self.EXTENT_HOLE | self.EXTENT_ZERO)]})
 269
 270         self.h.block_status(512, 0, f)
 271         assert offset == 0
 272         assert entries == [512, self.EXTENT_HOLE | self.EXTENT_ZERO]
 273
 274     def test_extents_2(self):
 275         """Test extents."""
 276
 277         offset = None
 278         entries = []
 279
 280         def f(meta_context, o, e, err):
 281             nonlocal offset, entries
 282             if meta_context != "base:allocation":
 283                 return
 284             offset = o
 285             entries = e
 286
 287         self.h.add_meta_context("base:allocation")
 288         self.connect({"size": 2048,
 289                       "can_extents": True,
 290                       "extents":
 291                       [(0, 512, self.EXTENT_HOLE | self.EXTENT_ZERO),
 292                        (512, 512, 0),
 293                        (1024, 1024, self.EXTENT_ZERO)]})
 294
 295         self.h.block_status(2048, 0, f)
 296         assert offset == 0
 297         assert entries == [512, self.EXTENT_HOLE | self.EXTENT_ZERO,
 298                            512, 0,
 299                            1024, self.EXTENT_ZERO]
 300
 301         self.h.block_status(1024, 1024, f)
 302         assert offset == 1024
 303         assert entries == [1024, self.EXTENT_ZERO]