import os
from cloudinit.analyze.__main__ import (analyze_boot, get_parser)
from cloudinit.tests.helpers import CiTestCase, mock
from cloudinit.analyze.show import dist_check_timestamp, SystemctlReader, \
FAIL_CODE, CONTAINER_CODE
err_code = (FAIL_CODE, -1, -1, -1)
class TestDistroChecker(CiTestCase):
@mock.patch('cloudinit.util.system_info', return_value={'dist': ('', '',
''),
'system': ''})
@mock.patch('cloudinit.util.get_linux_distro', return_value=('', '', ''))
@mock.patch('cloudinit.util.is_FreeBSD', return_value=False)
def test_blank_distro(self, m_sys_info, m_get_linux_distro, m_free_bsd):
self.assertEqual(err_code, dist_check_timestamp())
@mock.patch('cloudinit.util.system_info', return_value={'dist': ('', '',
'')})
@mock.patch('cloudinit.util.get_linux_distro', return_value=('', '', ''))
@mock.patch('cloudinit.util.is_FreeBSD', return_value=True)
def test_freebsd_gentoo_cant_find(self, m_sys_info,
m_get_linux_distro, m_is_FreeBSD):
self.assertEqual(err_code, dist_check_timestamp())
@mock.patch('cloudinit.subp.subp', return_value=(0, 1))
def test_subp_fails(self, m_subp):
self.assertEqual(err_code, dist_check_timestamp())
class TestSystemCtlReader(CiTestCase):
def test_systemctl_invalid_property(self):
reader = SystemctlReader('dummyProperty')
with self.assertRaises(RuntimeError):
reader.parse_epoch_as_float()
def test_systemctl_invalid_parameter(self):
reader = SystemctlReader('dummyProperty', 'dummyParameter')
with self.assertRaises(RuntimeError):
reader.parse_epoch_as_float()
@mock.patch('cloudinit.subp.subp', return_value=('U=1000000', None))
def test_systemctl_works_correctly_threshold(self, m_subp):
reader = SystemctlReader('dummyProperty', 'dummyParameter')
self.assertEqual(1.0, reader.parse_epoch_as_float())
thresh = 1.0 - reader.parse_epoch_as_float()
self.assertTrue(thresh < 1e-6)
self.assertTrue(thresh > (-1 * 1e-6))
@mock.patch('cloudinit.subp.subp', return_value=('U=0', None))
def test_systemctl_succeed_zero(self, m_subp):
reader = SystemctlReader('dummyProperty', 'dummyParameter')
self.assertEqual(0.0, reader.parse_epoch_as_float())
@mock.patch('cloudinit.subp.subp', return_value=('U=1', None))
def test_systemctl_succeed_distinct(self, m_subp):
reader = SystemctlReader('dummyProperty', 'dummyParameter')
val1 = reader.parse_epoch_as_float()
m_subp.return_value = ('U=2', None)
reader2 = SystemctlReader('dummyProperty', 'dummyParameter')
val2 = reader2.parse_epoch_as_float()
self.assertNotEqual(val1, val2)
@mock.patch('cloudinit.subp.subp', return_value=('100', None))
def test_systemctl_epoch_not_splittable(self, m_subp):
reader = SystemctlReader('dummyProperty', 'dummyParameter')
with self.assertRaises(IndexError):
reader.parse_epoch_as_float()
@mock.patch('cloudinit.subp.subp', return_value=('U=foobar', None))
def test_systemctl_cannot_convert_epoch_to_float(self, m_subp):
reader = SystemctlReader('dummyProperty', 'dummyParameter')
with self.assertRaises(ValueError):
reader.parse_epoch_as_float()
class TestAnalyzeBoot(CiTestCase):
def set_up_dummy_file_ci(self, path, log_path):
infh = open(path, 'w+')
infh.write('2019-07-08 17:40:49,601 - util.py[DEBUG]: Cloud-init v. '
'19.1-1-gbaa47854-0ubuntu1~18.04.1 running \'init-local\' '
'at Mon, 08 Jul 2019 17:40:49 +0000. Up 18.84 seconds.')
infh.close()
outfh = open(log_path, 'w+')
outfh.close()
def set_up_dummy_file(self, path, log_path):
infh = open(path, 'w+')
infh.write('dummy data')
infh.close()
outfh = open(log_path, 'w+')
outfh.close()
def remove_dummy_file(self, path, log_path):
if os.path.isfile(path):
os.remove(path)
if os.path.isfile(log_path):
os.remove(log_path)
@mock.patch('cloudinit.analyze.show.dist_check_timestamp',
return_value=err_code)
def test_boot_invalid_distro(self, m_dist_check_timestamp):
path = os.path.dirname(os.path.abspath(__file__))
log_path = path + '/boot-test.log'
path += '/dummy.log'
self.set_up_dummy_file(path, log_path)
parser = get_parser()
args = parser.parse_args(args=['boot', '-i', path, '-o',
log_path])
name_default = ''
analyze_boot(name_default, args)
# now args have been tested, go into outfile and make sure error
# message is in the outfile
outfh = open(args.outfile, 'r')
data = outfh.read()
err_string = 'Your Linux distro or container does not support this ' \
'functionality.\nYou must be running a Kernel ' \
'Telemetry supported distro.\nPlease check ' \
'https://cloudinit.readthedocs.io/en/latest/topics' \
'/analyze.html for more information on supported ' \
'distros.\n'
self.remove_dummy_file(path, log_path)
self.assertEqual(err_string, data)
@mock.patch("cloudinit.util.is_container", return_value=True)
@mock.patch('cloudinit.subp.subp', return_value=('U=1000000', None))
def test_container_no_ci_log_line(self, m_is_container, m_subp):
path = os.path.dirname(os.path.abspath(__file__))
log_path = path + '/boot-test.log'
path += '/dummy.log'
self.set_up_dummy_file(path, log_path)
parser = get_parser()
args = parser.parse_args(args=['boot', '-i', path, '-o',
log_path])
name_default = ''
finish_code = analyze_boot(name_default, args)
self.remove_dummy_file(path, log_path)
self.assertEqual(FAIL_CODE, finish_code)
@mock.patch("cloudinit.util.is_container", return_value=True)
@mock.patch('cloudinit.subp.subp', return_value=('U=1000000', None))
@mock.patch('cloudinit.analyze.__main__._get_events', return_value=[{
'name': 'init-local', 'description': 'starting search', 'timestamp':
100000}])
@mock.patch('cloudinit.analyze.show.dist_check_timestamp',
return_value=(CONTAINER_CODE, 1, 1, 1))
def test_container_ci_log_line(self, m_is_container, m_subp, m_get, m_g):
path = os.path.dirname(os.path.abspath(__file__))
log_path = path + '/boot-test.log'
path += '/dummy.log'
self.set_up_dummy_file_ci(path, log_path)
parser = get_parser()
args = parser.parse_args(args=['boot', '-i', path, '-o',
log_path])
name_default = ''
finish_code = analyze_boot(name_default, args)
self.remove_dummy_file(path, log_path)
self.assertEqual(CONTAINER_CODE, finish_code)