1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
|
import unittest
import math
import overrides_hack
import gi
gi.require_version('BlockDev', '3.0')
from gi.repository import BlockDev
from utils import TestTags, tag_test, required_plugins
@required_plugins(("crypto", "dm", "loop", "lvm", "mdraid", "part", "swap"))
class OverridesTest(unittest.TestCase):
# all plugins except for 'btrfs', 'fs' and 'mpath' -- these don't have all
# the dependencies on CentOS/Debian and we don't need them for this test
requested_plugins = BlockDev.plugin_specs_from_names(("crypto", "dm",
"loop", "lvm",
"mdraid", "part", "swap"))
@classmethod
def setUpClass(cls):
if not BlockDev.is_initialized():
BlockDev.init(cls.requested_plugins, None)
else:
BlockDev.reinit(cls.requested_plugins, True, None)
class OverridesTestCase(OverridesTest):
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_error_proxy(self):
"""Verify that the error proxy works as expected"""
# calls via the error proxy has to be done as
# e.g. BlockDev.swap.swapon() instead of BlockDev.swap_swapon(), since
# BlockDev.swap is an ErrorProxy instance and BlockDev.swap_swapon() is
# the function it calls
# test that exceptions are correctly transformed
try:
# no need to specify priority since we are using overrides that
# define the default value for the parameter (-1)
BlockDev.swap.swapon("/non/existing", -1)
except BlockDev.BlockDevError as e:
# we caught the generic error, now let's test that it is also the
# fine-grained one
self.assertTrue(isinstance(e, BlockDev.SwapError))
# test that a second call like that works the same (should go from the cache)
try:
BlockDev.swap.swapon("/non/existing", -1)
except BlockDev.BlockDevError as e:
self.assertTrue(isinstance(e, BlockDev.SwapError))
# test that successful calls propagate the results
self.assertTrue(BlockDev.lvm.is_supported_pe_size(4 * 1024))
self.assertEqual(BlockDev.lvm.round_size_to_pe(11 * 1024**2, 4 * 1024**2, True), 12 * 1024**2)
# test that utils functions are available via a proxy too
try:
BlockDev.utils.version_cmp("1.1", "malformed")
except BlockDev.BlockDevError as e:
self.assertTrue(isinstance(e, BlockDev.UtilsError))
self.assertEqual(BlockDev.utils.version_cmp("1.1", "1.2"), -1)
# test that overrides are used over the proxy
expected_padding = BlockDev.lvm_round_size_to_pe(int(math.ceil(11 * 1024**2 * 0.2)),
4 * 1024**2, True)
# the original lvm_get_thpool_padding takes 3 arguments, but one is enough for the overridden version
self.assertEqual(BlockDev.lvm_get_thpool_padding(11 * 1024**2),
expected_padding)
class OverridesUnloadTestCase(unittest.TestCase):
# all plugins except for 'btrfs', 'fs' and 'mpath' -- these don't have all
# the dependencies on CentOS/Debian and we don't need them for this test
requested_plugins = BlockDev.plugin_specs_from_names(("crypto", "dm",
"loop", "mdraid",
"part", "swap"))
@classmethod
def setUpClass(cls):
if not BlockDev.is_initialized():
BlockDev.init(cls.requested_plugins, None)
else:
BlockDev.reinit(cls.requested_plugins, True, None)
def tearDown(self):
# make sure the library is initialized with all plugins loaded for other
# tests
self.assertTrue(BlockDev.reinit(self.requested_plugins, True, None))
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_xrules(self):
"""Verify that regexp-based transformation rules work as expected"""
# unload all plugins first
self.assertTrue(BlockDev.reinit([], True, None))
# no longer loaded
with self.assertRaises(BlockDev.BlockDevNotImplementedError):
BlockDev.md.canonicalize_uuid("3386ff85:f5012621:4a435f06:1eb47236")
# load the plugins back
self.assertTrue(BlockDev.reinit(self.requested_plugins, True, None))
@tag_test(TestTags.NOSTORAGE, TestTags.CORE)
def test_exception_inheritance(self):
# unload all plugins first
self.assertTrue(BlockDev.reinit([], True, None))
# the exception should be properly inherited from two classes
with self.assertRaises(NotImplementedError):
BlockDev.md.canonicalize_uuid("3386ff85:f5012621:4a435f06:1eb47236")
with self.assertRaises(BlockDev.BlockDevError):
BlockDev.md.canonicalize_uuid("3386ff85:f5012621:4a435f06:1eb47236")
# load the plugins back
self.assertTrue(BlockDev.reinit(self.requested_plugins, True, None))
|
