7 logging.basicConfig(level=logging.DEBUG)
8 log = logging.getLogger()
10 def run_command(args, except_on_error=True):
11 log.debug('running command "%s"', ' '.join(args))
12 proc = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
13 out, err = proc.communicate()
15 log.debug('stdout: %s', out)
17 log.debug('stderr: %s', err)
19 log.debug('ret: %d', proc.returncode)
21 raise subprocess.CalledProcessError(proc.returncode, ' '.join(args))
22 return (proc.returncode, out, err)
24 def setup(image_name):
25 run_command(['rbd', 'create', '-s', '100', image_name])
26 run_command(['rbd', 'snap', 'create', image_name + '@snap'])
27 run_command(['rbd', 'map', image_name])
28 run_command(['rbd', 'map', image_name + '@snap'])
30 def teardown(image_name, fail_on_error=True):
31 run_command(['rbd', 'unmap', '/dev/rbd/rbd/' + image_name + '@snap'], fail_on_error)
32 run_command(['rbd', 'unmap', '/dev/rbd/rbd/' + image_name], fail_on_error)
33 run_command(['rbd', 'snap', 'rm', image_name + '@snap'], fail_on_error)
34 run_command(['rbd', 'rm', image_name], fail_on_error)
36 def write(target, expect_fail=False):
38 with open(target, 'w', 0) as f:
41 assert not expect_fail, 'writing should have failed'
43 assert expect_fail, 'writing should not have failed'
45 def test_ro(image_name):
46 dev = '/dev/rbd/rbd/' + image_name
47 snap_dev = dev + '@snap'
49 log.info('basic device is readable')
52 log.info('basic snapshot is read-only')
55 log.info('cannot set snapshot rw')
56 ret, _, _ = run_command(['blockdev', '--setrw', snap_dev], False)
57 assert ret != 0, 'snapshot was set read-write!'
58 run_command(['udevadm', 'settle'])
61 log.info('set device ro')
62 run_command(['blockdev', '--setro', dev])
63 run_command(['udevadm', 'settle'])
66 log.info('cannot set device rw when in-use')
67 with open(dev, 'r') as f:
68 ret, _, _ = run_command(['blockdev', '--setro', dev], False)
69 assert ret != 0, 'in-use device was set read-only!'
70 run_command(['udevadm', 'settle'])
73 run_command(['blockdev', '--setro', dev])
74 run_command(['udevadm', 'settle'])
77 run_command(['blockdev', '--setrw', dev])
78 run_command(['udevadm', 'settle'])
80 run_command(['udevadm', 'settle'])
81 run_command(['blockdev', '--setrw', dev])
82 run_command(['udevadm', 'settle'])
85 log.info('cannot set device ro when in-use')
86 with open(dev, 'r') as f:
87 ret, _, _ = run_command(['blockdev', '--setro', dev], False)
88 assert ret != 0, 'in-use device was set read-only!'
89 run_command(['udevadm', 'settle'])
91 run_command(['rbd', 'unmap', '/dev/rbd/rbd/' + image_name])
92 run_command(['rbd', 'map', '--read-only', image_name])
94 log.info('cannot write to newly mapped ro device')
97 log.info('can set ro mapped device rw')
98 run_command(['blockdev', '--setrw', dev])
99 run_command(['udevadm', 'settle'])
104 # clean up any state from previous test runs
105 teardown(image_name, False)
112 if __name__ == '__main__':