tests/test_python.py

   1 #!/usr/bin/env python3
   2 # nbdkit
   3 # Copyright (C) 2019 Red Hat Inc.
   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 os
  40 import sys
  41 import nbd
  42 import unittest
  43 import pickle
  44 import base64
  45
  46 class Test (unittest.TestCase):
  47     def setUp (self):
  48         self.h = nbd.NBD ()
  49
  50     def tearDown (self):
  51         del self.h
  52
  53     def connect (self, cfg):
  54         cfg = base64.b64encode (pickle.dumps (cfg)).decode()
  55         cmd = ["nbdkit", "-v", "-s", "--exit-with-parent",
  56                "python", "test-python-plugin.py", "cfg=" + cfg]
  57         self.h.connect_command (cmd)
  58
  59     def test_none (self):
  60         """
  61         Test we can send an empty pickled test configuration and do
  62         nothing else.  This is just to ensure the machinery of the
  63         test works.
  64         """
  65         self.connect ({})
  66
  67     def test_size_512 (self):
  68         """Test the size."""
  69         self.connect ({"size": 512})
  70         assert self.h.get_size() == 512
  71
  72     def test_size_1m (self):
  73         """Test the size."""
  74         self.connect ({"size": 1024*1024})
  75         assert self.h.get_size() == 1024*1024
  76
  77     # Test each flag call.
  78     def test_is_rotational_true (self):
  79         self.connect ({"size": 512, "is_rotational": True})
  80         assert self.h.is_rotational()
  81
  82     def test_is_rotational_false (self):
  83         self.connect ({"size": 512, "is_rotational": False})
  84         assert not self.h.is_rotational()
  85
  86     def test_can_multi_conn_true (self):
  87         self.connect ({"size": 512, "can_multi_conn": True})
  88         assert self.h.can_multi_conn()
  89
  90     def test_can_multi_conn_false (self):
  91         self.connect ({"size": 512, "can_multi_conn": False})
  92         assert not self.h.can_multi_conn()
  93
  94     def test_read_write (self):
  95         self.connect ({"size": 512, "can_write": True})
  96         assert not self.h.is_read_only()
  97
  98     def test_read_only (self):
  99         self.connect ({"size": 512, "can_write": False})
 100         assert self.h.is_read_only()
 101
 102     def test_can_flush_true (self):
 103         self.connect ({"size": 512, "can_flush": True})
 104         assert self.h.can_flush()
 105
 106     def test_can_flush_false (self):
 107         self.connect ({"size": 512, "can_flush": False})
 108         assert not self.h.can_flush()
 109
 110     def test_can_trim_true (self):
 111         self.connect ({"size": 512, "can_trim": True})
 112         assert self.h.can_trim()
 113
 114     def test_can_trim_false (self):
 115         self.connect ({"size": 512, "can_trim": False})
 116         assert not self.h.can_trim()
 117
 118     # nbdkit can always zero because it emulates it.
 119     #self.connect ({"size": 512, "can_zero": True})
 120     #assert self.h.can_zero()
 121     #self.connect ({"size": 512, "can_zero": False})
 122     #assert not self.h.can_zero()
 123
 124     def test_can_fast_zero_true (self):
 125         self.connect ({"size": 512, "can_fast_zero": True})
 126         assert self.h.can_fast_zero()
 127
 128     def test_can_fast_zero_false (self):
 129         self.connect ({"size": 512, "can_fast_zero": False})
 130         assert not self.h.can_fast_zero()
 131
 132     def test_can_fua_none (self):
 133         self.connect ({"size": 512, "can_fua": "none"})
 134         assert not self.h.can_fua()
 135
 136     def test_can_fua_emulate (self):
 137         self.connect ({"size": 512, "can_fua": "emulate"})
 138         assert self.h.can_fua()
 139
 140     def test_can_fua_native (self):
 141         self.connect ({"size": 512, "can_fua": "native"})
 142         assert self.h.can_fua()
 143
 144     def test_can_cache_none (self):
 145         self.connect ({"size": 512, "can_cache": "none"})
 146         assert not self.h.can_cache()
 147
 148     def test_can_cache_emulate (self):
 149         self.connect ({"size": 512, "can_cache": "emulate"})
 150         assert self.h.can_cache()
 151
 152     def test_can_cache_native (self):
 153         self.connect ({"size": 512, "can_cache": "native"})
 154         assert self.h.can_cache()
 155
 156     # Not yet implemented: can_extents.
 157
 158     def test_pread (self):
 159         """Test pread."""
 160         self.connect ({"size": 512})
 161         buf = self.h.pread (512, 0)
 162         assert buf == bytearray (512)
 163
 164     # Test pwrite + flags.
 165     def test_pwrite (self):
 166         self.connect ({"size": 512})
 167         buf = bytearray (512)
 168         self.h.pwrite (buf, 0)
 169
 170     def test_pwrite_fua (self):
 171         self.connect ({"size": 512,
 172                        "can_fua": "native",
 173                        "pwrite_expect_fua": True})
 174         buf = bytearray (512)
 175         self.h.pwrite (buf, 0, nbd.CMD_FLAG_FUA)
 176
 177     def test_flush (self):
 178         """Test flush."""
 179         self.connect ({"size": 512, "can_flush": True})
 180         self.h.flush ()
 181
 182     # Test trim + flags.
 183     def test_trim (self):
 184         self.connect ({"size": 512, "can_trim": True})
 185         self.h.trim (512, 0)
 186
 187     def test_trim_fua (self):
 188         self.connect ({"size": 512,
 189                        "can_trim": True,
 190                        "can_fua": "native",
 191                        "trim_expect_fua": True})
 192         self.h.trim (512, 0, nbd.CMD_FLAG_FUA)
 193
 194     # Test zero + flags.
 195     def test_zero (self):
 196         self.connect ({"size": 512, "can_zero": True})
 197         self.h.zero (512, 0, nbd.CMD_FLAG_NO_HOLE)
 198
 199     def test_zero_fua (self):
 200         self.connect ({"size": 512,
 201                        "can_zero": True,
 202                        "can_fua": "native",
 203                        "zero_expect_fua": True})
 204         self.h.zero (512, 0, nbd.CMD_FLAG_NO_HOLE | nbd.CMD_FLAG_FUA)
 205
 206     def test_zero_may_trim (self):
 207         self.connect ({"size": 512,
 208                        "can_zero": True,
 209                        "zero_expect_may_trim": True})
 210         self.h.zero (512, 0, 0) # absence of nbd.CMD_FLAG_NO_HOLE
 211
 212     def test_zero_fast_zero (self):
 213         self.connect ({"size": 512,
 214                        "can_zero": True,
 215                        "can_fast_zero": True,
 216                        "zero_expect_fast_zero": True})
 217         self.h.zero (512, 0, nbd.CMD_FLAG_NO_HOLE | nbd.CMD_FLAG_FAST_ZERO)
 218
 219     def test_cache (self):
 220         """Test cache."""
 221         self.connect ({"size": 512, "can_cache": "native"})
 222         self.h.cache (512, 0)