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
|
#!/usr/bin/env python
import logging
import subprocess
import sys
logging.basicConfig(level=logging.DEBUG)
log = logging.getLogger()
def run_command(args, except_on_error=True):
log.debug('running command "%s"', ' '.join(args))
proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
out, err = proc.communicate()
if out:
log.debug('stdout: %s', out)
if err:
log.debug('stderr: %s', err)
if proc.returncode:
log.debug('ret: %d', proc.returncode)
if except_on_error:
raise subprocess.CalledProcessError(proc.returncode, ' '.join(args))
return (proc.returncode, out, err)
def setup(image_name):
run_command(['rbd', 'create', '-s', '100', image_name])
run_command(['rbd', 'snap', 'create', image_name + '@snap'])
run_command(['rbd', 'map', image_name])
run_command(['rbd', 'map', image_name + '@snap'])
def teardown(image_name, fail_on_error=True):
run_command(['rbd', 'unmap', '/dev/rbd/rbd/' + image_name + '@snap'], fail_on_error)
run_command(['rbd', 'unmap', '/dev/rbd/rbd/' + image_name], fail_on_error)
run_command(['rbd', 'snap', 'rm', image_name + '@snap'], fail_on_error)
run_command(['rbd', 'rm', image_name], fail_on_error)
def write(target, expect_fail=False):
try:
with open(target, 'w', 0) as f:
f.write('test')
f.flush()
assert not expect_fail, 'writing should have failed'
except IOError:
assert expect_fail, 'writing should not have failed'
def test_ro(image_name):
dev = '/dev/rbd/rbd/' + image_name
snap_dev = dev + '@snap'
log.info('basic device is readable')
write(dev)
log.info('basic snapshot is read-only')
write(snap_dev, True)
log.info('cannot set snapshot rw')
ret, _, _ = run_command(['blockdev', '--setrw', snap_dev], False)
assert ret != 0, 'snapshot was set read-write!'
run_command(['udevadm', 'settle'])
write(snap_dev, True)
log.info('set device ro')
run_command(['blockdev', '--setro', dev])
run_command(['udevadm', 'settle'])
write(dev, True)
log.info('cannot set device rw when in-use')
with open(dev, 'r') as f:
ret, _, _ = run_command(['blockdev', '--setro', dev], False)
assert ret != 0, 'in-use device was set read-only!'
run_command(['udevadm', 'settle'])
write(dev, True)
run_command(['blockdev', '--setro', dev])
run_command(['udevadm', 'settle'])
write(dev, True)
run_command(['blockdev', '--setrw', dev])
run_command(['udevadm', 'settle'])
write(dev)
run_command(['udevadm', 'settle'])
run_command(['blockdev', '--setrw', dev])
run_command(['udevadm', 'settle'])
write(dev)
log.info('cannot set device ro when in-use')
with open(dev, 'r') as f:
ret, _, _ = run_command(['blockdev', '--setro', dev], False)
assert ret != 0, 'in-use device was set read-only!'
run_command(['udevadm', 'settle'])
run_command(['rbd', 'unmap', '/dev/rbd/rbd/' + image_name])
run_command(['rbd', 'map', '--read-only', image_name])
log.info('cannot write to newly mapped ro device')
write(dev, True)
log.info('can set ro mapped device rw')
run_command(['blockdev', '--setrw', dev])
run_command(['udevadm', 'settle'])
write(dev)
def main():
image_name = 'test1'
# clean up any state from previous test runs
teardown(image_name, False)
setup(image_name)
test_ro(image_name)
teardown(image_name)
if __name__ == '__main__':
main()
|
