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 def test_init_filestore_dir(self
):
49 parser
= argparse
.ArgumentParser('ceph-disk')
50 subparsers
= parser
.add_subparsers()
51 main
.Prepare
.set_subparser(subparsers
)
53 data
= tempfile
.mkdtemp()
54 main
.setup_statedir(data
)
55 args
= parser
.parse_args([
63 with mock
.patch
.multiple(main
.PrepareData
,
65 prepare
= main
.Prepare
.factory(args
)
66 assert isinstance(prepare
.data
, main
.PrepareFilestoreData
)
67 assert prepare
.data
.is_file()
68 assert isinstance(prepare
.journal
, main
.PrepareJournal
)
69 assert prepare
.journal
.is_none()
71 assert os
.path
.exists(os
.path
.join(data
, 'fsid'))
74 @mock.patch('stat.S_ISBLK')
75 @mock.patch('ceph_disk.main.is_partition')
76 def test_init_filestore_dev(self
, m_is_partition
, m_s_isblk
):
77 m_s_isblk
.return_value
= True
79 parser
= argparse
.ArgumentParser('ceph-disk')
80 subparsers
= parser
.add_subparsers()
81 main
.Prepare
.set_subparser(subparsers
)
83 m_is_partition
.return_value
= False
84 _
, data
= tempfile
.mkstemp()
86 args
= parser
.parse_args([
91 prepare
= main
.Prepare
.factory(args
)
92 assert isinstance(prepare
.data
, main
.PrepareData
)
93 assert prepare
.data
.is_device()
94 assert isinstance(prepare
.journal
, main
.PrepareJournal
)
95 assert prepare
.journal
.is_device()
97 def test_init_default_dir(self
):
98 parser
= argparse
.ArgumentParser('ceph-disk')
99 subparsers
= parser
.add_subparsers()
100 main
.Prepare
.set_subparser(subparsers
)
102 data
= tempfile
.mkdtemp()
103 main
.setup_statedir(data
)
104 args
= parser
.parse_args([
110 self
.type = self
.FILE
111 with mock
.patch
.multiple(main
.PrepareData
,
113 prepare
= main
.Prepare
.factory(args
)
114 assert isinstance(prepare
.data
, main
.PrepareBluestoreData
)
115 assert prepare
.data
.is_file()
117 assert os
.path
.exists(os
.path
.join(data
, 'fsid'))
120 def test_set_subparser(self
):
121 parser
= argparse
.ArgumentParser('ceph-disk')
122 subparsers
= parser
.add_subparsers()
123 main
.Prepare
.set_subparser(subparsers
)
124 osd_uuid
= 'OSD_UUID'
125 journal_uuid
= 'JOURNAL_UUID'
128 args
= parser
.parse_args([
130 '--osd-uuid', osd_uuid
,
131 '--journal-uuid', journal_uuid
,
132 '--fs-type', fs_type
,
135 assert args
.journal_uuid
== journal_uuid
136 assert args
.osd_uuid
== osd_uuid
137 assert args
.fs_type
== fs_type
138 assert args
.data
== data
141 class TestDevice(Base
):
143 @mock.patch('ceph_disk.main.is_partition')
144 def test_init(self
, m_is_partition
):
145 m_is_partition
.return_value
= False
146 device
= main
.Device('/dev/wholedisk', argparse
.Namespace())
147 assert device
.dev_size
is None
149 @mock.patch('ceph_disk.main.is_partition')
150 @mock.patch('ceph_disk.main.get_free_partition_index')
151 @mock.patch('ceph_disk.main.update_partition')
152 @mock.patch('ceph_disk.main.get_dm_uuid')
153 @mock.patch('ceph_disk.main.get_dev_size')
154 @mock.patch('ceph_disk.main.command_check_call')
155 def test_create_partition(self
,
156 m_command_check_call
,
160 m_get_free_partition_index
,
162 if platform
.system() == 'FreeBSD':
164 m_is_partition
.return_value
= False
166 m_get_free_partition_index
.return_value
= partition_number
167 path
= '/dev/wholedisk'
168 device
= main
.Device(path
, argparse
.Namespace(dmcrypt
=False))
170 m_get_dm_uuid
.return_value
= uuid
172 m_get_dev_size
.return_value
= size
+ 100
174 actual_partition_number
= device
.create_partition(
175 uuid
=uuid
, name
=name
, size
=size
)
176 assert actual_partition_number
== partition_number
178 '--new=%d:0:+%dM' % (partition_number
, size
),
179 '--change-name=%d:ceph %s' % (partition_number
, name
),
180 '--partition-guid=%d:%s' % (partition_number
, uuid
),
181 '--typecode=%d:%s' % (
183 main
.PTYPE
['regular']['journal']['ready']),
184 '--mbrtogpt', '--', path
]
185 m_command_check_call
.assert_called_with(command
, exit
=True)
186 m_update_partition
.assert_called_with(path
, 'created')
188 actual_partition_number
= device
.create_partition(
189 uuid
=uuid
, name
=name
)
191 '--largest-new=%d' % partition_number
,
192 '--change-name=%d:ceph %s' % (partition_number
, name
),
193 '--partition-guid=%d:%s' % (partition_number
, uuid
),
194 '--typecode=%d:%s' % (
196 main
.PTYPE
['regular']['journal']['ready']),
197 '--mbrtogpt', '--', path
]
198 m_command_check_call
.assert_called_with(command
, exit
=True)
201 class TestDevicePartition(Base
):
204 partition
= main
.DevicePartition(argparse
.Namespace())
205 for name
in ('osd', 'journal'):
206 assert (main
.PTYPE
['regular'][name
]['ready'] ==
207 partition
.ptype_for_name(name
))
209 def test_get_uuid(self
):
210 partition
= main
.DevicePartition(argparse
.Namespace())
212 with mock
.patch
.multiple(main
,
213 get_partition_uuid
=lambda path
: uuid
):
214 assert uuid
== partition
.get_uuid()
215 assert uuid
== partition
.get_uuid()
217 def test_get_ptype(self
):
218 partition
= main
.DevicePartition(argparse
.Namespace())
219 ptype
= main
.PTYPE
['regular']['osd']['tobe']
220 with mock
.patch
.multiple(main
,
221 get_partition_type
=lambda path
: ptype
):
222 assert ptype
== partition
.get_ptype()
223 assert ptype
== partition
.get_ptype()
225 def test_partition_number(self
):
226 partition
= main
.DevicePartition(argparse
.Namespace())
228 assert num
!= partition
.get_partition_number()
229 partition
.set_partition_number(num
)
230 assert num
== partition
.get_partition_number()
233 partition
= main
.DevicePartition(argparse
.Namespace())
235 assert dev
!= partition
.get_dev()
236 assert dev
!= partition
.get_rawdev()
237 partition
.set_dev(dev
)
238 assert dev
== partition
.get_dev()
239 assert dev
== partition
.get_rawdev()
241 @mock.patch('ceph_disk.main.is_mpath')
242 def test_factory(self
, m_is_mpath
):
243 parser
= argparse
.ArgumentParser('ceph-disk')
244 subparsers
= parser
.add_subparsers()
245 main
.Prepare
.set_subparser(subparsers
)
248 m_is_mpath
.return_value
= False
253 args
= parser
.parse_args([
257 partition
= main
.DevicePartition
.factory(
258 path
=path
, dev
=None, args
=args
)
259 assert isinstance(partition
, main
.DevicePartition
)
262 # Multipath device partition
264 m_is_mpath
.return_value
= True
265 args
= parser
.parse_args([
269 partition
= main
.DevicePartition
.factory(
270 path
=path
, dev
=None, args
=args
)
271 assert isinstance(partition
, main
.DevicePartitionMultipath
)
272 m_is_mpath
.return_value
= False
275 # Device partition encrypted via dmcrypt luks
277 args
= parser
.parse_args([
282 partition
= main
.DevicePartition
.factory(
283 path
=path
, dev
=None, args
=args
)
284 assert isinstance(partition
, main
.DevicePartitionCryptLuks
)
287 # Device partition encrypted via dmcrypt plain
289 self
.conf
['global']['osd dmcrypt type'] = 'plain'
291 args
= parser
.parse_args([
296 partition
= main
.DevicePartition
.factory(
297 path
=path
, dev
=None, args
=args
)
298 assert isinstance(partition
, main
.DevicePartitionCryptPlain
)
301 class TestDevicePartitionMultipath(Base
):
304 partition
= main
.DevicePartitionMultipath(argparse
.Namespace())
305 for name
in ('osd', 'journal'):
306 assert (main
.PTYPE
['mpath'][name
]['ready'] ==
307 partition
.ptype_for_name(name
))
310 class TestDevicePartitionCrypt(Base
):
312 @mock.patch('ceph_disk.main.get_conf')
313 def test_luks(self
, m_get_conf
):
314 parser
= argparse
.ArgumentParser('ceph-disk')
315 subparsers
= parser
.add_subparsers()
316 main
.Prepare
.set_subparser(subparsers
)
319 def get_conf(**kwargs
):
320 if kwargs
['variable'] == 'osd_dmcrypt_key_size':
322 elif kwargs
['variable'] == 'osd_dmcrypt_type':
324 elif kwargs
['variable'] == 'osd_cryptsetup_parameters':
329 m_get_conf
.side_effect
= get_conf
331 args
= parser
.parse_args([
336 partition
= main
.DevicePartitionCryptLuks(args
)
337 assert partition
.luks()
338 assert partition
.osd_dm_key
is None
340 with mock
.patch
.multiple(main
,
341 _dmcrypt_map
=mock
.DEFAULT
,
342 get_dmcrypt_key
=mock
.DEFAULT
,
343 get_partition_uuid
=lambda path
: uuid
) as m
:
345 assert m
['_dmcrypt_map'].called
346 m
['get_dmcrypt_key'].assert_called_with(
347 uuid
, '/etc/ceph/dmcrypt-keys', True)
350 class TestCryptHelpers(Base
):
352 @mock.patch('ceph_disk.main.get_conf')
353 def test_get_dmcrypt_type(self
, m_get_conf
):
354 args
= argparse
.Namespace(dmcrypt
=False)
355 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is None
357 m_get_conf
.return_value
= 'luks'
358 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
359 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is 'luks'
361 m_get_conf
.return_value
= None
362 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
363 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is 'luks'
365 m_get_conf
.return_value
= 'plain'
366 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
367 assert main
.CryptHelpers
.get_dmcrypt_type(args
) is 'plain'
370 m_get_conf
.return_value
= invalid
371 args
= argparse
.Namespace(dmcrypt
=True, cluster
='ceph')
372 with pytest
.raises(main
.Error
) as err
:
373 main
.CryptHelpers
.get_dmcrypt_type(args
)
374 assert invalid
in str(err
)
377 class TestPrepareData(Base
):
379 def test_set_variables(self
):
380 parser
= argparse
.ArgumentParser('ceph-disk')
381 subparsers
= parser
.add_subparsers()
382 main
.Prepare
.set_subparser(subparsers
)
384 osd_uuid
= 'OSD_UUID'
385 cluster_uuid
= '571bb920-6d85-44d7-9eca-1bc114d1cd75'
387 args
= parser
.parse_args([
389 '--osd-uuid', osd_uuid
,
390 '--cluster-uuid', cluster_uuid
,
395 self
.type = self
.FILE
396 with mock
.patch
.multiple(main
.PrepareData
,
398 data
= main
.PrepareData(args
)
399 assert data
.args
.osd_uuid
== osd_uuid
400 assert data
.args
.cluster_uuid
== cluster_uuid
403 args
= parser
.parse_args([
408 with mock
.patch
.multiple(main
.PrepareData
,
410 data
= main
.PrepareData(args
)
411 assert 36 == len(data
.args
.osd_uuid
)
412 assert 36 == len(data
.args
.cluster_uuid
)
414 self
.conf
['global']['fsid'] = cluster_uuid
417 args
= parser
.parse_args([
422 with mock
.patch
.multiple(main
.PrepareData
,
424 data
= main
.PrepareData(args
)
425 assert data
.args
.cluster_uuid
== cluster_uuid