src/ceph/src/ceph-detect-init/tests/test_all.py

   1 #
   2 # Copyright (C) 2015 SUSE LINUX GmbH
   3 # Copyright (C) 2015 <contact@redhat.com>
   4 #
   5 # Author: Owen Synge <osynge@suse.com>
   6 # Author: Loic Dachary <loic@dachary.org>
   7 #
   8 # This program is free software: you can redistribute it and/or modify
   9 # it under the terms of the GNU Affero General Public License as
  10 # published by the Free Software Foundation, either version 3 of the
  11 # License, or (at your option) any later version.
  12 #
  13 # This program is distributed in the hope that it will be useful,
  14 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  15 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16 # GNU Affero General Public License for more details.
  17 #
  18 # You should have received a copy of the GNU Affero General Public License
  19 # along with this program.  If not, see `<http://www.gnu.org/licenses/>`.
  20 #
  21 import logging
  22 import mock
  23 import testtools
  24
  25 import ceph_detect_init
  26 from ceph_detect_init import alpine
  27 from ceph_detect_init import arch
  28 from ceph_detect_init import centos
  29 from ceph_detect_init import debian
  30 from ceph_detect_init import exc
  31 from ceph_detect_init import fedora
  32 from ceph_detect_init import main
  33 from ceph_detect_init import rhel
  34 from ceph_detect_init import suse
  35 from ceph_detect_init import gentoo
  36 from ceph_detect_init import freebsd
  37 from ceph_detect_init import docker
  38 from ceph_detect_init import oraclevms
  39
  40 logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
  41                     level=logging.DEBUG)
  42
  43
  44 class TestCephDetectInit(testtools.TestCase):
  45
  46     def test_alpine(self):
  47         self.assertEqual('openrc', alpine.choose_init())
  48
  49     def test_arch(self):
  50         self.assertEqual('systemd', arch.choose_init())
  51
  52     def test_freebsd(self):
  53         self.assertEqual('bsdrc', freebsd.choose_init())
  54
  55     def test_docker(self):
  56         self.assertEqual('none', docker.choose_init())
  57
  58     def test_oraclevms(self):
  59         self.assertEqual('sysvinit', oraclevms.choose_init())
  60
  61     def test_centos(self):
  62         with mock.patch('ceph_detect_init.centos.release',
  63                         '7.0'):
  64             self.assertEqual('systemd', centos.choose_init())
  65         self.assertEqual('sysvinit', centos.choose_init())
  66
  67     def test_debian(self):
  68         with mock.patch.multiple('os.path',
  69                                  isdir=lambda x: x == '/run/systemd/system'):
  70             self.assertEqual('systemd', debian.choose_init())
  71
  72         def mock_call_with_upstart(*args, **kwargs):
  73             if args[0] == '. /lib/lsb/init-functions ; init_is_upstart' and \
  74                kwargs['shell']:
  75                 return 0
  76             else:
  77                 return 1
  78         with mock.patch.multiple('os.path',
  79                                  isdir=lambda x: False,
  80                                  isfile=lambda x: False):
  81             with mock.patch.multiple('subprocess',
  82                                      call=mock_call_with_upstart):
  83                 self.assertEqual('upstart', debian.choose_init())
  84         with mock.patch.multiple('os.path',
  85                                  isdir=lambda x: False,
  86                                  isfile=lambda x: x == '/sbin/init',
  87                                  islink=lambda x: x != '/sbin/init'):
  88             with mock.patch.multiple('subprocess',
  89                                      call=lambda *args, **kwargs: 1):
  90                 self.assertEqual('sysvinit', debian.choose_init())
  91         with mock.patch.multiple('os.path',
  92                                  isdir=lambda x: False,
  93                                  isfile=lambda x: False):
  94             with mock.patch.multiple('subprocess',
  95                                      call=lambda *args, **kwargs: 1):
  96                 self.assertIs(None, debian.choose_init())
  97
  98     def test_fedora(self):
  99         with mock.patch('ceph_detect_init.fedora.release',
 100                         '22'):
 101             self.assertEqual('systemd', fedora.choose_init())
 102         self.assertEqual('sysvinit', fedora.choose_init())
 103
 104     def test_rhel(self):
 105         with mock.patch('ceph_detect_init.rhel.release',
 106                         '7.0'):
 107             self.assertEqual('systemd', rhel.choose_init())
 108         self.assertEqual('sysvinit', rhel.choose_init())
 109
 110     def test_suse(self):
 111         with mock.patch('ceph_detect_init.suse.release',
 112                         '11'):
 113             self.assertEqual('sysvinit', suse.choose_init())
 114         with mock.patch('ceph_detect_init.suse.release',
 115                         '12'):
 116             self.assertEqual('systemd', suse.choose_init())
 117         with mock.patch('ceph_detect_init.suse.release',
 118                         '13.1'):
 119             self.assertEqual('systemd', suse.choose_init())
 120         with mock.patch('ceph_detect_init.suse.release',
 121                         '13.2'):
 122             self.assertEqual('systemd', suse.choose_init())
 123
 124     def test_gentoo_is_openrc(self):
 125         with mock.patch('os.path.isdir', return_value=True):
 126             self.assertEqual(gentoo.is_openrc(), True)
 127         with mock.patch('os.path.isdir', return_value=False):
 128             self.assertEqual(gentoo.is_openrc(), False)
 129
 130     def test_gentoo_is_systemd(self):
 131         import sys
 132         if sys.version_info >= (3, 0):
 133             mocked_fn = 'builtins.open'
 134         else:
 135             mocked_fn = '__builtin__.open'
 136
 137         f = mock.mock_open(read_data='systemd')
 138         with mock.patch(mocked_fn, f, create=True) as m:
 139             self.assertEqual(gentoo.is_systemd(), True)
 140             m.assert_called_once_with('/proc/1/comm')
 141         f = mock.mock_open(read_data='init')
 142         with mock.patch(mocked_fn, f, create=True) as m:
 143             self.assertEqual(gentoo.is_systemd(), False)
 144             m.assert_called_once_with('/proc/1/comm')
 145         f = mock.mock_open(read_data='upstart')
 146         with mock.patch(mocked_fn, f, create=True) as m:
 147             self.assertEqual(gentoo.is_systemd(), False)
 148             m.assert_called_once_with('/proc/1/comm')
 149
 150     def test_gentoo(self):
 151         with mock.patch.multiple('ceph_detect_init.gentoo',
 152                                  is_systemd=(lambda: True),
 153                                  is_openrc=(lambda: True)):
 154             self.assertEqual('openrc', gentoo.choose_init())
 155         with mock.patch.multiple('ceph_detect_init.gentoo',
 156                                  is_systemd=(lambda: True),
 157                                  is_openrc=(lambda: False)):
 158             self.assertEqual('systemd', gentoo.choose_init())
 159         with mock.patch.multiple('ceph_detect_init.gentoo',
 160                                  is_systemd=(lambda: False),
 161                                  is_openrc=(lambda: True)):
 162             self.assertEqual('openrc', gentoo.choose_init())
 163         with mock.patch.multiple('ceph_detect_init.gentoo',
 164                                  is_systemd=(lambda: False),
 165                                  is_openrc=(lambda: False)):
 166             self.assertEqual('unknown', gentoo.choose_init())
 167
 168     def test_get(self):
 169         with mock.patch.multiple(
 170                 'platform',
 171                 system=lambda: 'Linux',
 172                 linux_distribution=lambda **kwargs: (('unknown', '', ''))):
 173             g = ceph_detect_init.get
 174             self.assertRaises(exc.UnsupportedPlatform, g)
 175             try:
 176                 g()
 177             except exc.UnsupportedPlatform as e:
 178                 self.assertIn('Platform is not supported', str(e))
 179
 180         with mock.patch.multiple(
 181                 'platform',
 182                 system=lambda: 'Linux',
 183                 linux_distribution=lambda **kwargs: (('debian', '6.0', ''))):
 184             distro = ceph_detect_init.get()
 185             self.assertEqual(debian, distro)
 186             self.assertEqual('debian', distro.name)
 187             self.assertEqual('debian', distro.normalized_name)
 188             self.assertEqual('debian', distro.distro)
 189             self.assertEqual(False, distro.is_el)
 190             self.assertEqual('6.0', distro.release)
 191
 192         with mock.patch.multiple('platform',
 193                                  system=lambda: 'FreeBSD',
 194                                  release=lambda: '12.0-CURRENT',
 195                                  version=lambda: 'FreeBSD 12.0 #1 r306554M:'):
 196             distro = ceph_detect_init.get()
 197             self.assertEqual(freebsd, distro)
 198             self.assertEqual('freebsd', distro.name)
 199             self.assertEqual('freebsd', distro.normalized_name)
 200             self.assertEqual('freebsd', distro.distro)
 201             self.assertFalse(distro.is_el)
 202             self.assertEqual('12.0-CURRENT', distro.release)
 203             self.assertEqual('r306554M', distro.codename)
 204             self.assertEqual('bsdrc', distro.init)
 205
 206         with mock.patch('platform.system',
 207                         lambda: 'cephix'):
 208             self.assertRaises(exc.UnsupportedPlatform, ceph_detect_init.get)
 209
 210     def test_get_distro(self):
 211         g = ceph_detect_init._get_distro
 212         self.assertEqual(None, g(None))
 213         self.assertEqual(debian, g('debian'))
 214         self.assertEqual(debian, g('ubuntu'))
 215         self.assertEqual(centos, g('centos'))
 216         self.assertEqual(centos, g('scientific'))
 217         self.assertEqual(centos, g('Oracle Linux Server'))
 218         self.assertEqual(oraclevms, g('Oracle VM server'))
 219         self.assertEqual(fedora, g('fedora'))
 220         self.assertEqual(suse, g('suse'))
 221         self.assertEqual(rhel, g('redhat', use_rhceph=True))
 222         self.assertEqual(gentoo, g('gentoo'))
 223         self.assertEqual(centos, g('virtuozzo'))
 224
 225     def test_normalized_distro_name(self):
 226         n = ceph_detect_init._normalized_distro_name
 227         self.assertEqual('redhat', n('RedHat'))
 228         self.assertEqual('redhat', n('redhat'))
 229         self.assertEqual('redhat', n('Red Hat'))
 230         self.assertEqual('redhat', n('red hat'))
 231         self.assertEqual('scientific', n('scientific'))
 232         self.assertEqual('scientific', n('Scientific'))
 233         self.assertEqual('scientific', n('Scientific Linux'))
 234         self.assertEqual('scientific', n('scientific linux'))
 235         self.assertEqual('oraclel', n('Oracle Linux Server'))
 236         self.assertEqual('oraclevms', n('Oracle VM server'))
 237         self.assertEqual('suse', n('SUSE'))
 238         self.assertEqual('suse', n('suse'))
 239         self.assertEqual('suse', n('openSUSE'))
 240         self.assertEqual('suse', n('opensuse'))
 241         self.assertEqual('centos', n('CentOS'))
 242         self.assertEqual('centos', n('centos'))
 243         self.assertEqual('debian', n('Debian'))
 244         self.assertEqual('debian', n('debian'))
 245         self.assertEqual('ubuntu', n('Ubuntu'))
 246         self.assertEqual('ubuntu', n('ubuntu'))
 247         self.assertEqual('gentoo', n('Gentoo'))
 248         self.assertEqual('gentoo', n('gentoo'))
 249         self.assertEqual('gentoo', n('Funtoo'))
 250         self.assertEqual('gentoo', n('funtoo'))
 251         self.assertEqual('gentoo', n('Exherbo'))
 252         self.assertEqual('gentoo', n('exherbo'))
 253         self.assertEqual('virtuozzo', n('Virtuozzo Linux'))
 254
 255     @mock.patch('platform.system', lambda: 'Linux')
 256     def test_platform_information_linux(self):
 257         with mock.patch('platform.linux_distribution',
 258                         lambda **kwargs: (('debian', '8.0', ''))):
 259             self.assertEqual(('debian', '8.0'),
 260                              ceph_detect_init.platform_information()[:-1])
 261
 262         with mock.patch('platform.linux_distribution',
 263                         lambda **kwargs: (('Oracle Linux Server', '7.3', ''))):
 264             self.assertEqual(('Oracle Linux Server', '7.3', 'OL7.3'),
 265                              ceph_detect_init.platform_information())
 266
 267         with mock.patch('platform.linux_distribution',
 268                         lambda **kwargs: (('Oracle VM server', '3.4.2', ''))):
 269             self.assertEqual(('Oracle VM server', '3.4.2', 'OVS3.4.2'),
 270                              ceph_detect_init.platform_information())
 271
 272         with mock.patch('platform.linux_distribution',
 273                         lambda **kwargs: (('Virtuozzo Linux', '7.3', ''))):
 274             self.assertEqual(('Virtuozzo Linux', '7.3', 'virtuozzo'),
 275                              ceph_detect_init.platform_information())
 276
 277     @mock.patch('platform.linux_distribution')
 278     def test_platform_information_container(self, mock_linux_dist):
 279         import sys
 280         if sys.version_info >= (3, 0):
 281             mocked_fn = 'builtins.open'
 282         else:
 283             mocked_fn = '__builtin__.open'
 284
 285         with mock.patch(mocked_fn,
 286                         mock.mock_open(read_data="""1:name=systemd:/system.slice \
 287                                                  /docker-39cc1fb.scope"""),
 288                         create=True) as m:
 289             self.assertEqual(('docker',
 290                               'docker',
 291                               'docker'),
 292                              ceph_detect_init.platform_information(),)
 293             m.assert_called_once_with('/proc/self/cgroup', 'r')
 294
 295         with mock.patch(mocked_fn, mock.mock_open(), create=True) as m:
 296             m.side_effect = IOError()
 297             mock_linux_dist.return_value = ('Red Hat Enterprise Linux Server',
 298                                             '7.3', 'Maipo')
 299             # Just run the code to validate the code won't raise IOError
 300             ceph_detect_init.platform_information()
 301
 302         with mock.patch('os.path.isfile', mock.MagicMock()) as m:
 303             m.return_value = True
 304             self.assertEqual(('docker',
 305                               'docker',
 306                               'docker'),
 307                              ceph_detect_init.platform_information(),)
 308             m.assert_called_once_with('/.dockerenv')
 309
 310     @mock.patch('platform.system', lambda: 'FreeBSD')
 311     def test_platform_information_freebsd(self):
 312         with mock.patch.multiple('platform',
 313                                  release=lambda: '12.0-CURRENT',
 314                                  version=lambda: 'FreeBSD 12.0 #1 r306554M:'):
 315             self.assertEqual(('freebsd', '12.0-CURRENT', 'r306554M'),
 316                              ceph_detect_init.platform_information())
 317
 318     def test_run(self):
 319         argv = ['--use-rhceph', '--verbose']
 320         self.assertEqual(0, main.run(argv))
 321
 322         with mock.patch.multiple(
 323                 'platform',
 324                 system=lambda: 'Linux',
 325                 linux_distribution=lambda **kwargs: (('unknown', '', ''))):
 326             self.assertRaises(exc.UnsupportedPlatform, main.run, argv)
 327             self.assertEqual(0, main.run(argv + ['--default=sysvinit']))
 328
 329 # Local Variables:
 330 # compile-command: "cd .. ; .tox/py27/bin/py.test tests/test_all.py"
 331 # End: