prepare.mount_osd('/dev/sda1', 1)
expected = [
'mount', '-t', 'xfs', '-o',
- 'rw', 'noatime', 'inode64', # default flags
+ 'rw,noatime,inode64', # default flags
'/dev/sda1', '/var/lib/ceph/osd/ceph-1']
assert expected == fake_run.calls[0]['args'][0]
'/dev/sda1', '/var/lib/ceph/osd/ceph-1']
assert expected == fake_run.calls[0]['args'][0]
- def test_multiple_options_are_used(self, conf_ceph_stub, fake_run):
+ def test_multiple_whitespace_options_are_used(self, conf_ceph_stub, fake_run):
conf_ceph_stub(dedent("""[global]
fsid = 1234lkjh1234
[osd]
prepare.mount_osd('/dev/sda1', 1)
expected = [
'mount', '-t', 'xfs', '-o',
- 'rw', 'auto', 'exec',
+ 'rw,auto,exec',
+ '/dev/sda1', '/var/lib/ceph/osd/ceph-1']
+ assert expected == fake_run.calls[0]['args'][0]
+
+ def test_multiple_comma_whitespace_options_are_used(self, conf_ceph_stub, fake_run):
+ conf_ceph_stub(dedent("""[global]
+ fsid = 1234lkjh1234
+ [osd]
+ osd mount options xfs = rw, auto, exec"""))
+ conf.cluster = 'ceph'
+ prepare.mount_osd('/dev/sda1', 1)
+ expected = [
+ 'mount', '-t', 'xfs', '-o',
+ 'rw,auto,exec',
'/dev/sda1', '/var/lib/ceph/osd/ceph-1']
assert expected == fake_run.calls[0]['args'][0]
'rw',
'/dev/sda1', '/var/lib/ceph/osd/ceph-1']
assert expected == fake_run.calls[0]['args'][0]
+
+
+ceph_conf_mount_values = [
+ ['rw,', 'auto,' 'exec'],
+ ['rw', 'auto', 'exec'],
+ [' rw ', ' auto ', ' exec '],
+ ['rw,', 'auto,', 'exec,'],
+ [',rw ', ',auto ', ',exec,'],
+ [',rw,', ',auto,', ',exec,'],
+]
+
+string_mount_values = [
+ 'rw, auto exec ',
+ 'rw auto exec',
+ ',rw, auto, exec,',
+ ' rw auto exec ',
+ ' rw,auto,exec ',
+ 'rw,auto,exec',
+ ',rw,auto,exec,',
+ 'rw,auto,exec ',
+ 'rw, auto, exec ',
+]
+
+
+class TestNormalizeFlags(object):
+ # a bit overkill since most of this is already tested in prepare.mount_osd
+ # tests
+
+ @pytest.mark.parametrize("flags", ceph_conf_mount_values)
+ def test_normalize_lists(self, flags):
+ result = prepare._normalize_mount_flags(flags)
+ assert result == 'rw,auto,exec'
+
+ @pytest.mark.parametrize("flags", string_mount_values)
+ def test_normalize_strings(self, flags):
+ result = prepare._normalize_mount_flags(flags)
+ assert result == 'rw,auto,exec'