3 # Copyright (C) 2015, 2016 Red Hat <contact@redhat.com>
5 # This program is free software; you can redistribute it and/or modify
6 # it under the terms of the GNU Library Public License as published by
7 # the Free Software Foundation; either version 2, or (at your option)
10 # This program is distributed in the hope that it will be useful,
11 # but WITHOUT ANY WARRANTY; without even the implied warranty of
12 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 # GNU Library Public License for more details.
24 from ceph_disk
import main
29 def setup_class(self
):
30 main
.setup_logging(True, False)
31 os
.environ
['PATH'] = "..:" + os
.environ
['PATH']
34 _
, self
.conf_file
= tempfile
.mkstemp()
35 os
.environ
['CEPH_CONF'] = self
.conf_file
36 self
.conf
= configobj
.ConfigObj(self
.conf_file
)
37 self
.conf
['global'] = {}
40 os
.unlink(self
.conf_file
)
43 self
.conf
.write(open(self
.conf_file
, 'wb'))
46 class TestPrepare(Base
):
48 @mock.patch('ceph_disk.main.get_fsid')
49 def test_init_filestore_dir(self
, m_get_fsid
):
50 parser
= argparse
.ArgumentParser('ceph-disk')
51 subparsers
= parser
.add_subparsers()
52 main
.Prepare
.set_subparser(subparsers
)
54 data
= tempfile
.mkdtemp()
55 main
.setup_statedir(data
)
56 args
= parser
.parse_args([
64 m_get_fsid
.return_value
= '571bb920-6d85-44d7-9eca-1bc114d1cd75'
65 with mock
.patch
.multiple(main
.PrepareData
,
67 prepare
= main
.Prepare
.factory(args
)
68 prepare
= main
.Prepare
.factory(args
)
69 assert isinstance(prepare
.data
, main
.PrepareFilestoreData
)
70 assert prepare
.data
.is_file()
71 assert isinstance(prepare
.journal
, main
.PrepareJournal
)
72 assert prepare
.journal
.is_none()
74 assert os
.path
.exists(os
.path
.join(data
, 'fsid'))
77 @mock.patch('stat.S_ISBLK')
78 @mock.patch('ceph_disk.main.is_partition')
79 @mock.patch('ceph_disk.main.get_fsid')
80 def test_init_filestore_dev(self
,
84 m_s_isblk
.return_value
= True
86 parser
= argparse
.ArgumentParser('ceph-disk')
87 subparsers
= parser
.add_subparsers()
88 main
.Prepare
.set_subparser(subparsers
)
90 m_is_partition
.return_value
= False
91 _
, data
= tempfile
.mkstemp()
93 args
= parser
.parse_args([
98 m_get_fsid
.return_value
= '571bb920-6d85-44d7-9eca-1bc114d1cd75'
99 prepare
= main
.Prepare
.factory(args
)
100 assert isinstance(prepare
.data
, main
.PrepareData
)
101 assert prepare
.data
.is_device()
102 assert isinstance(prepare
.journal
, main
.PrepareJournal
)
103 assert prepare
.journal
.is_device()
105 @mock.patch('ceph_disk.main.get_fsid')
106 def test_init_default_dir(self
, m_get_fsid
):
107 parser
= argparse
.ArgumentParser('ceph-disk')
108 subparsers
= parser
.add_subparsers()
109 main
.Prepare
.set_subparser(subparsers
)
111 data
= tempfile
.mkdtemp()
112 main
.setup_statedir(data
)
113 args
= parser
.parse_args([
119 self
.type = self
.FILE
120 m_get_fsid
.return_value
= '571bb920-6d85-44d7-9eca-1bc114d1cd75'
121 with mock
.patch
.multiple(main
.PrepareData
,
123 prepare
= main
.Prepare
.factory(args
)
124 assert isinstance(prepare
.data
, main
.PrepareBluestoreData
)
125 assert prepare
.data
.is_file()
127 assert os
.path
.exists(os
.path
.join(data
, 'fsid'))
130 def test_set_subparser(self
):
131 parser
= argparse
.ArgumentParser('ceph-disk')
132 subparsers
= parser
.add_subparsers()
133 main
.Prepare
.set_subparser(subparsers
)
134 osd_uuid
= 'OSD_UUID'
135 journal_uuid
= 'JOURNAL_UUID'
138 args
= parser
.parse_args([
140 '--osd-uuid', osd_uuid
,
141 '--journal-uuid', journal_uuid
,
142 '--fs-type', fs_type
,
145 assert args
.journal_uuid
== journal_uuid
146 assert args
.osd_uuid
== osd_uuid
147 assert args
.fs_type
== fs_type
148 assert args
.data
== data
151 class TestDevice(Base
):
153 @mock.patch('ceph_disk.main.is_partition')
154 def test_init(self
, m_is_partition
):
155 m_is_partition
.return_value
= False
156 device
= main
.Device('/dev/wholedisk', argparse
.Namespace())
157 assert device
.dev_size
is None
159 @mock.patch('ceph_disk.main.is_partition')
160 @mock.patch('ceph_disk.main.get_free_partition_index')
161 @mock.patch('ceph_disk.main.update_partition')
162 @mock.patch('ceph_disk.main.get_dm_uuid')
163 @mock.patch('ceph_disk.main.get_dev_size')
164 @mock.patch('ceph_disk.main.command_check_call')
165 def test_create_partition(self
,
166 m_command_check_call
,
170 m_get_free_partition_index
,
172 if platform
.system() == 'FreeBSD':
174 m_is_partition
.return_value
= False
176 m_get_free_partition_index
.return_value
= partition_number
177 path
= '/dev/wholedisk'
178 device
= main
.Device(path
, argparse
.Namespace(dmcrypt
=False))
180 m_get_dm_uuid
.return_value
= uuid
182 m_get_dev_size
.return_value
= size
+ 100
184 actual_partition_number
= device
.create_partition(
185 uuid
=uuid
, name
=name
, size
=size
)
186 assert actual_partition_number
== partition_number
188 '--new=%d:0:+%dM' % (partition_number
, size
),
189 '--change-name=%d:ceph %s' % (partition_number
, name
),
190 '--partition-guid=%d:%s' % (partition_number
, uuid
),
191 '--typecode=%d:%s' % (
193 main
.PTYPE
['regular']['journal']['ready']),
194 '--mbrtogpt', '--', path
]
195 m_command_check_call
.assert_called_with(command
, exit
=True)
196 m_update_partition
.assert_called_with(path
, 'created')
198 actual_partition_number
= device
.create_partition(
199 uuid
=uuid
, name
=name
)
201 '--largest-new=%d' % partition_number
,
202 '--change-name=%d:ceph %s' % (partition_number
, name
),
203 '--partition-guid=%d:%s' % (partition_number
, uuid
),
204 '--typecode=%d:%s' % (
206 main
.PTYPE
['regular']['journal']['ready']),
207 '--mbrtogpt', '--', path
]
208 m_command_check_call
.assert_called_with(command
, exit
=True)
211 class TestDevicePartition(Base
):
214 partition
= main
.DevicePartition(argparse
.Namespace())
215 for name
in ('osd', 'journal'):
216 assert (main
.PTYPE
['regular'][name
]['ready'] ==
217 partition
.ptype_for_name(name
))
219 def test_get_uuid(self
):
220 partition
= main
.DevicePartition(argparse
.Namespace())
222 with mock
.patch
.multiple(main
,
223 get_partition_uuid
=lambda path
: uuid
):
224 assert uuid
== partition
.get_uuid()
225 assert uuid
== partition
.get_uuid()
227 def test_get_ptype(self
):
228 partition
= main
.DevicePartition(argparse
.Namespace())
229 ptype
= main
.PTYPE
['regular']['osd']['tobe']
230 with mock
.patch
.multiple(main
,
231 get_partition_type
=lambda path
: ptype
):
232 assert ptype
== partition
.get_ptype()
233 assert ptype
== partition
.get_ptype()
235 def test_partition_number(self
):
236 partition
= main
.DevicePartition(argparse
.Namespace())
238 assert num
!= partition
.get_partition_number()
239 partition
.set_partition_number(num
)
240 assert num
== partition
.get_partition_number()
243 partition
= main
.DevicePartition(argparse
.Namespace())
245 assert dev
!= partition
.get_dev()
246 assert dev
!= partition
.get_rawdev()
247 partition
.set_dev(dev
)
248 assert dev
== partition
.get_dev()
249 assert dev
== partition
.get_rawdev()
251 @mock.patch('ceph_disk.main.is_mpath')
252 def test_factory(self
, m_is_mpath
):
253 parser
= argparse
.ArgumentParser('ceph-disk')
254 subparsers
= parser
.add_subparsers()
255 main
.Prepare
.set_subparser(subparsers
)
258 m_is_mpath
.return_value
= False
263 args
= parser
.parse_args([
267 partition
= main
.DevicePartition
.factory(
268 path
=path
, dev
=None, args
=args
)
269 assert isinstance(partition
, main
.DevicePartition
)
272 # Multipath device partition
274 m_is_mpath
.return_value
= True
275 args
= parser
.parse_args([
279 partition
= main
.DevicePartition
.factory(
280 path
=path
, dev
=None, args
=args
)
281 assert isinstance(partition
, main
.DevicePartitionMultipath
)
282 m_is_mpath
.return_value
= False
285 # Device partition encrypted via dmcrypt luks
287 args
= parser
.parse_args([
292 partition
= main
.DevicePartition
.factory(
293 path
=path
, dev
=None, args
=args
)
294 assert isinstance(partition
, main
.DevicePartitionCryptLuks
)
297 # Device partition encrypted via dmcrypt plain
299 self
.conf
['global']['osd dmcrypt type'] = 'plain'
301 args
= parser
.parse_args([
306 partition
= main
.DevicePartition
.factory(
307 path
=path
, dev
=None, args
=args
)
308 assert isinstance(partition
, main
.DevicePartitionCryptPlain
)
311 class TestDevicePartitionMultipath(Base
):
314 partition
= main
.DevicePartitionMultipath(argparse
.Namespace())
315 for name
in ('osd', 'journal'):
316 assert (main
.PTYPE
['mpath'][name
]['ready'] ==
317 partition
.ptype_for_name(name
))
320 class TestDevicePartitionCrypt(Base
):
322 @mock.patch('ceph_disk.main.get_conf')
323 def test_luks(self
, m_get_conf
):
324 parser
= argparse
.ArgumentParser('ceph-disk')
325 subparsers
= parser
.add_subparsers()
326 main
.Prepare
.set_subparser(subparsers
)
329 def get_conf(**kwargs
):
330 if kwargs
['variable'] == 'osd_dmcrypt_key_size':
332 elif kwargs
['variable'] == 'osd_dmcrypt_type':
334 elif kwargs
['variable'] == 'osd_cryptsetup_parameters':
339 m_get_conf
.side_effect
= get_conf
341 args
= parser
.parse_args([
346 partition
= main
.DevicePartitionCryptLuks(args
)
347 assert partition
.luks()
348 assert partition
.osd_dm_key
is None
350 with mock
.patch
.multiple(main
,
351 _dmcrypt_map
=mock
.DEFAULT
,
352 get_dmcrypt_key
=mock
.DEFAULT
,
353 get_partition_uuid
=lambda path
: uuid
) as m
:
355 assert m
['_dmcrypt_map'].called
356 m
['get_dmcrypt_key'].assert_called_with(
357 uuid
, '/etc/ceph/dmcrypt-keys', True)
360 class TestCryptHelpers(Base
):
362 @mock.patch('ceph_disk.main.get_conf')
363 def test_get_dmcrypt_type(self
, m_get_conf
):
364 args
= argparse
.Namespace(dmcrypt
=False)
365 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is None
367 m_get_conf
.return_value
= 'luks'
368 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
369 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is 'luks'
371 m_get_conf
.return_value
= None
372 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
373 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is 'luks'
375 m_get_conf
.return_value
= 'plain'
376 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
377 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is 'plain'
380 m_get_conf
.return_value
= invalid
381 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
382 with pytest
.raises(main
.Error
) as err
:
383 main
.CryptHelpers
.get_dmcrypt_type(args
)
384 assert invalid
in str(err
)
387 class TestPrepareData(Base
):
389 @mock.patch('ceph_disk.main.get_fsid')
390 def test_set_variables(self
, m_get_fsid
):
391 parser
= argparse
.ArgumentParser('ceph-disk')
392 subparsers
= parser
.add_subparsers()
393 main
.Prepare
.set_subparser(subparsers
)
395 osd_uuid
= 'OSD_UUID'
396 cluster_uuid
= '571bb920-6d85-44d7-9eca-1bc114d1cd75'
398 args
= parser
.parse_args([
400 '--osd-uuid', osd_uuid
,
401 '--cluster-uuid', cluster_uuid
,
406 self
.type = self
.FILE
407 m_get_fsid
.return_value
= cluster_uuid
408 with mock
.patch
.multiple(main
.PrepareData
,
410 data
= main
.PrepareData(args
)
411 assert data
.args
.osd_uuid
== osd_uuid
412 assert data
.args
.cluster_uuid
== cluster_uuid
415 args
= parser
.parse_args([
420 with mock
.patch
.multiple(main
.PrepareData
,
422 data
= main
.PrepareData(args
)
423 assert 36 == len(data
.args
.osd_uuid
)
424 assert 36 == len(data
.args
.cluster_uuid
)
426 self
.conf
['global']['fsid'] = cluster_uuid
429 args
= parser
.parse_args([
434 with mock
.patch
.multiple(main
.PrepareData
,
436 data
= main
.PrepareData(args
)
437 assert data
.args
.cluster_uuid
== cluster_uuid
440 class TestSecrets(Base
):
442 @mock.patch('ceph_disk.main.command')
443 def test_secrets(self
, m_command
):
445 m_command
.side_effect
= lambda cmd
: (key
+ "\n", '', 0)
447 assert {"cephx_secret": key
} == s
.keys
448 assert '{"cephx_secret": "' + key
+ '"}' == s
.get_json()
450 @mock.patch('ceph_disk.main.open')
451 @mock.patch('ceph_disk.main.CryptHelpers.get_dmcrypt_keysize')
452 @mock.patch('ceph_disk.main.command')
453 def test_lockbox_secrets(self
,
455 m_get_dmcrypt_keysize
,
458 m_command
.side_effect
= lambda cmd
: (key
+ "\n", '', 0)
459 m_get_dmcrypt_keysize
.side_effect
= lambda args
: 32
462 def read(self
, size
):
465 m_open
.side_effect
= lambda path
, mode
: File()
466 s
= main
.LockboxSecrets({})
468 "dmcrypt_key": 'T09PTw==',
470 "cephx_lockbox_secret": key
,