--- /dev/null
+#
+# Copyright (C) 2015 SUSE LINUX GmbH
+# Copyright (C) 2015 <contact@redhat.com>
+#
+# Author: Owen Synge <osynge@suse.com>
+# Author: Loic Dachary <loic@dachary.org>
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Affero General Public License as
+# published by the Free Software Foundation, either version 3 of the
+# License, or (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU Affero General Public License for more details.
+#
+# You should have received a copy of the GNU Affero General Public License
+# along with this program. If not, see `<http://www.gnu.org/licenses/>`.
+#
+import logging
+import mock
+import testtools
+
+import ceph_detect_init
+from ceph_detect_init import alpine
+from ceph_detect_init import arch
+from ceph_detect_init import centos
+from ceph_detect_init import debian
+from ceph_detect_init import exc
+from ceph_detect_init import fedora
+from ceph_detect_init import main
+from ceph_detect_init import rhel
+from ceph_detect_init import suse
+from ceph_detect_init import gentoo
+from ceph_detect_init import freebsd
+from ceph_detect_init import docker
+from ceph_detect_init import oraclevms
+
+logging.basicConfig(format='%(asctime)s %(levelname)s %(message)s',
+ level=logging.DEBUG)
+
+
+class TestCephDetectInit(testtools.TestCase):
+
+ def test_alpine(self):
+ self.assertEqual('openrc', alpine.choose_init())
+
+ def test_arch(self):
+ self.assertEqual('systemd', arch.choose_init())
+
+ def test_freebsd(self):
+ self.assertEqual('bsdrc', freebsd.choose_init())
+
+ def test_docker(self):
+ self.assertEqual('none', docker.choose_init())
+
+ def test_oraclevms(self):
+ self.assertEqual('sysvinit', oraclevms.choose_init())
+
+ def test_centos(self):
+ with mock.patch('ceph_detect_init.centos.release',
+ '7.0'):
+ self.assertEqual('systemd', centos.choose_init())
+ self.assertEqual('sysvinit', centos.choose_init())
+
+ def test_debian(self):
+ with mock.patch.multiple('os.path',
+ isdir=lambda x: x == '/run/systemd/system'):
+ self.assertEqual('systemd', debian.choose_init())
+
+ def mock_call_with_upstart(*args, **kwargs):
+ if args[0] == '. /lib/lsb/init-functions ; init_is_upstart' and \
+ kwargs['shell']:
+ return 0
+ else:
+ return 1
+ with mock.patch.multiple('os.path',
+ isdir=lambda x: False,
+ isfile=lambda x: False):
+ with mock.patch.multiple('subprocess',
+ call=mock_call_with_upstart):
+ self.assertEqual('upstart', debian.choose_init())
+ with mock.patch.multiple('os.path',
+ isdir=lambda x: False,
+ isfile=lambda x: x == '/sbin/init',
+ islink=lambda x: x != '/sbin/init'):
+ with mock.patch.multiple('subprocess',
+ call=lambda *args, **kwargs: 1):
+ self.assertEqual('sysvinit', debian.choose_init())
+ with mock.patch.multiple('os.path',
+ isdir=lambda x: False,
+ isfile=lambda x: False):
+ with mock.patch.multiple('subprocess',
+ call=lambda *args, **kwargs: 1):
+ self.assertIs(None, debian.choose_init())
+
+ def test_fedora(self):
+ with mock.patch('ceph_detect_init.fedora.release',
+ '22'):
+ self.assertEqual('systemd', fedora.choose_init())
+ self.assertEqual('sysvinit', fedora.choose_init())
+
+ def test_rhel(self):
+ with mock.patch('ceph_detect_init.rhel.release',
+ '7.0'):
+ self.assertEqual('systemd', rhel.choose_init())
+ self.assertEqual('sysvinit', rhel.choose_init())
+
+ def test_suse(self):
+ with mock.patch('ceph_detect_init.suse.release',
+ '11'):
+ self.assertEqual('sysvinit', suse.choose_init())
+ with mock.patch('ceph_detect_init.suse.release',
+ '12'):
+ self.assertEqual('systemd', suse.choose_init())
+ with mock.patch('ceph_detect_init.suse.release',
+ '13.1'):
+ self.assertEqual('systemd', suse.choose_init())
+ with mock.patch('ceph_detect_init.suse.release',
+ '13.2'):
+ self.assertEqual('systemd', suse.choose_init())
+
+ def test_gentoo_is_openrc(self):
+ with mock.patch('os.path.isdir', return_value=True):
+ self.assertEqual(gentoo.is_openrc(), True)
+ with mock.patch('os.path.isdir', return_value=False):
+ self.assertEqual(gentoo.is_openrc(), False)
+
+ def test_gentoo_is_systemd(self):
+ import sys
+ if sys.version_info >= (3, 0):
+ mocked_fn = 'builtins.open'
+ else:
+ mocked_fn = '__builtin__.open'
+
+ f = mock.mock_open(read_data='systemd')
+ with mock.patch(mocked_fn, f, create=True) as m:
+ self.assertEqual(gentoo.is_systemd(), True)
+ m.assert_called_once_with('/proc/1/comm')
+ f = mock.mock_open(read_data='init')
+ with mock.patch(mocked_fn, f, create=True) as m:
+ self.assertEqual(gentoo.is_systemd(), False)
+ m.assert_called_once_with('/proc/1/comm')
+ f = mock.mock_open(read_data='upstart')
+ with mock.patch(mocked_fn, f, create=True) as m:
+ self.assertEqual(gentoo.is_systemd(), False)
+ m.assert_called_once_with('/proc/1/comm')
+
+ def test_gentoo(self):
+ with mock.patch.multiple('ceph_detect_init.gentoo',
+ is_systemd=(lambda: True),
+ is_openrc=(lambda: True)):
+ self.assertEqual('openrc', gentoo.choose_init())
+ with mock.patch.multiple('ceph_detect_init.gentoo',
+ is_systemd=(lambda: True),
+ is_openrc=(lambda: False)):
+ self.assertEqual('systemd', gentoo.choose_init())
+ with mock.patch.multiple('ceph_detect_init.gentoo',
+ is_systemd=(lambda: False),
+ is_openrc=(lambda: True)):
+ self.assertEqual('openrc', gentoo.choose_init())
+ with mock.patch.multiple('ceph_detect_init.gentoo',
+ is_systemd=(lambda: False),
+ is_openrc=(lambda: False)):
+ self.assertEqual('unknown', gentoo.choose_init())
+
+ def test_get(self):
+ with mock.patch.multiple(
+ 'platform',
+ system=lambda: 'Linux',
+ linux_distribution=lambda **kwargs: (('unknown', '', ''))):
+ g = ceph_detect_init.get
+ self.assertRaises(exc.UnsupportedPlatform, g)
+ try:
+ g()
+ except exc.UnsupportedPlatform as e:
+ self.assertIn('Platform is not supported', str(e))
+
+ with mock.patch.multiple(
+ 'platform',
+ system=lambda: 'Linux',
+ linux_distribution=lambda **kwargs: (('debian', '6.0', ''))):
+ distro = ceph_detect_init.get()
+ self.assertEqual(debian, distro)
+ self.assertEqual('debian', distro.name)
+ self.assertEqual('debian', distro.normalized_name)
+ self.assertEqual('debian', distro.distro)
+ self.assertEqual(False, distro.is_el)
+ self.assertEqual('6.0', distro.release)
+
+ with mock.patch.multiple('platform',
+ system=lambda: 'FreeBSD',
+ release=lambda: '12.0-CURRENT',
+ version=lambda: 'FreeBSD 12.0 #1 r306554M:'):
+ distro = ceph_detect_init.get()
+ self.assertEqual(freebsd, distro)
+ self.assertEqual('freebsd', distro.name)
+ self.assertEqual('freebsd', distro.normalized_name)
+ self.assertEqual('freebsd', distro.distro)
+ self.assertFalse(distro.is_el)
+ self.assertEqual('12.0-CURRENT', distro.release)
+ self.assertEqual('r306554M', distro.codename)
+ self.assertEqual('bsdrc', distro.init)
+
+ with mock.patch('platform.system',
+ lambda: 'cephix'):
+ self.assertRaises(exc.UnsupportedPlatform, ceph_detect_init.get)
+
+ def test_get_distro(self):
+ g = ceph_detect_init._get_distro
+ self.assertEqual(None, g(None))
+ self.assertEqual(debian, g('debian'))
+ self.assertEqual(debian, g('ubuntu'))
+ self.assertEqual(centos, g('centos'))
+ self.assertEqual(centos, g('scientific'))
+ self.assertEqual(centos, g('Oracle Linux Server'))
+ self.assertEqual(oraclevms, g('Oracle VM server'))
+ self.assertEqual(fedora, g('fedora'))
+ self.assertEqual(suse, g('suse'))
+ self.assertEqual(rhel, g('redhat', use_rhceph=True))
+ self.assertEqual(gentoo, g('gentoo'))
+ self.assertEqual(centos, g('virtuozzo'))
+
+ def test_normalized_distro_name(self):
+ n = ceph_detect_init._normalized_distro_name
+ self.assertEqual('redhat', n('RedHat'))
+ self.assertEqual('redhat', n('redhat'))
+ self.assertEqual('redhat', n('Red Hat'))
+ self.assertEqual('redhat', n('red hat'))
+ self.assertEqual('scientific', n('scientific'))
+ self.assertEqual('scientific', n('Scientific'))
+ self.assertEqual('scientific', n('Scientific Linux'))
+ self.assertEqual('scientific', n('scientific linux'))
+ self.assertEqual('oraclel', n('Oracle Linux Server'))
+ self.assertEqual('oraclevms', n('Oracle VM server'))
+ self.assertEqual('suse', n('SUSE'))
+ self.assertEqual('suse', n('suse'))
+ self.assertEqual('suse', n('openSUSE'))
+ self.assertEqual('suse', n('opensuse'))
+ self.assertEqual('centos', n('CentOS'))
+ self.assertEqual('centos', n('centos'))
+ self.assertEqual('debian', n('Debian'))
+ self.assertEqual('debian', n('debian'))
+ self.assertEqual('ubuntu', n('Ubuntu'))
+ self.assertEqual('ubuntu', n('ubuntu'))
+ self.assertEqual('gentoo', n('Gentoo'))
+ self.assertEqual('gentoo', n('gentoo'))
+ self.assertEqual('gentoo', n('Funtoo'))
+ self.assertEqual('gentoo', n('funtoo'))
+ self.assertEqual('gentoo', n('Exherbo'))
+ self.assertEqual('gentoo', n('exherbo'))
+ self.assertEqual('virtuozzo', n('Virtuozzo Linux'))
+
+ @mock.patch('platform.system', lambda: 'Linux')
+ def test_platform_information_linux(self):
+ with mock.patch('platform.linux_distribution',
+ lambda **kwargs: (('debian', '8.0', ''))):
+ self.assertEqual(('debian', '8.0'),
+ ceph_detect_init.platform_information()[:-1])
+
+ with mock.patch('platform.linux_distribution',
+ lambda **kwargs: (('Oracle Linux Server', '7.3', ''))):
+ self.assertEqual(('Oracle Linux Server', '7.3', 'OL7.3'),
+ ceph_detect_init.platform_information())
+
+ with mock.patch('platform.linux_distribution',
+ lambda **kwargs: (('Oracle VM server', '3.4.2', ''))):
+ self.assertEqual(('Oracle VM server', '3.4.2', 'OVS3.4.2'),
+ ceph_detect_init.platform_information())
+
+ with mock.patch('platform.linux_distribution',
+ lambda **kwargs: (('Virtuozzo Linux', '7.3', ''))):
+ self.assertEqual(('Virtuozzo Linux', '7.3', 'virtuozzo'),
+ ceph_detect_init.platform_information())
+
+ @mock.patch('platform.linux_distribution')
+ def test_platform_information_container(self, mock_linux_dist):
+ import sys
+ if sys.version_info >= (3, 0):
+ mocked_fn = 'builtins.open'
+ else:
+ mocked_fn = '__builtin__.open'
+
+ with mock.patch(mocked_fn,
+ mock.mock_open(read_data="""1:name=systemd:/system.slice \
+ /docker-39cc1fb.scope"""),
+ create=True) as m:
+ self.assertEqual(('docker',
+ 'docker',
+ 'docker'),
+ ceph_detect_init.platform_information(),)
+ m.assert_called_once_with('/proc/self/cgroup', 'r')
+
+ with mock.patch(mocked_fn, mock.mock_open(), create=True) as m:
+ m.side_effect = IOError()
+ mock_linux_dist.return_value = ('Red Hat Enterprise Linux Server',
+ '7.3', 'Maipo')
+ # Just run the code to validate the code won't raise IOError
+ ceph_detect_init.platform_information()
+
+ with mock.patch('os.path.isfile', mock.MagicMock()) as m:
+ m.return_value = True
+ self.assertEqual(('docker',
+ 'docker',
+ 'docker'),
+ ceph_detect_init.platform_information(),)
+ m.assert_called_once_with('/.dockerenv')
+
+ @mock.patch('platform.system', lambda: 'FreeBSD')
+ def test_platform_information_freebsd(self):
+ with mock.patch.multiple('platform',
+ release=lambda: '12.0-CURRENT',
+ version=lambda: 'FreeBSD 12.0 #1 r306554M:'):
+ self.assertEqual(('freebsd', '12.0-CURRENT', 'r306554M'),
+ ceph_detect_init.platform_information())
+
+ def test_run(self):
+ argv = ['--use-rhceph', '--verbose']
+ self.assertEqual(0, main.run(argv))
+
+ with mock.patch.multiple(
+ 'platform',
+ system=lambda: 'Linux',
+ linux_distribution=lambda **kwargs: (('unknown', '', ''))):
+ self.assertRaises(exc.UnsupportedPlatform, main.run, argv)
+ self.assertEqual(0, main.run(argv + ['--default=sysvinit']))
+
+# Local Variables:
+# compile-command: "cd .. ; .tox/py27/bin/py.test tests/test_all.py"
+# End:
Code Review / stor4nfv.git / blobdiff