+class ReadMeminfoTestCase(ut_base.BaseUnitTestCase):
+
+ MEMINFO = (b'MemTotal: 65860500 kB\n'
+ b'MemFree: 28690900 kB\n'
+ b'MemAvailable: 52873764 kB\n'
+ b'Active(anon): 3015676 kB\n'
+ b'HugePages_Total: 8\n'
+ b'Hugepagesize: 1048576 kB')
+ MEMINFO_DICT = {'MemTotal': '65860500',
+ 'MemFree': '28690900',
+ 'MemAvailable': '52873764',
+ 'Active(anon)': '3015676',
+ 'HugePages_Total': '8',
+ 'Hugepagesize': '1048576'}
+
+ def test_read_meminfo(self):
+ ssh_client = ssh.SSH('user', 'host')
+ with mock.patch.object(ssh_client, 'get_file_obj') as \
+ mock_get_client, \
+ mock.patch.object(six, 'BytesIO',
+ return_value=six.BytesIO(self.MEMINFO)):
+ output = utils.read_meminfo(ssh_client)
+ mock_get_client.assert_called_once_with('/proc/meminfo', mock.ANY)
+ self.assertEqual(self.MEMINFO_DICT, output)
+
+
+class TimerTestCase(ut_base.BaseUnitTestCase):
+
+ def test__getattr(self):
+ with utils.Timer() as timer:
+ time.sleep(1)
+ self.assertEqual(1, round(timer.total_seconds(), 0))
+ self.assertEqual(1, timer.delta.seconds)
+
+ def test__enter_with_timeout(self):
+ with utils.Timer(timeout=10) as timer:
+ time.sleep(1)
+ self.assertEqual(1, round(timer.total_seconds(), 0))
+
+ def test__enter_with_timeout_exception(self):
+ with self.assertRaises(exceptions.TimerTimeout):
+ with utils.Timer(timeout=1):
+ time.sleep(2)
+
+ def test__enter_with_timeout_no_exception(self):
+ with utils.Timer(timeout=1, raise_exception=False):
+ time.sleep(2)
+
+ def test__iter(self):
+ iterations = []
+ for i in utils.Timer(timeout=2):
+ iterations.append(i)
+ time.sleep(1.1)
+ self.assertEqual(2, len(iterations))
+
+
+class WaitUntilTrueTestCase(ut_base.BaseUnitTestCase):
+
+ def test_no_timeout(self):
+ self.assertIsNone(utils.wait_until_true(lambda: True,
+ timeout=1, sleep=1))
+
+ def test_timeout_generic_exception(self):
+ with self.assertRaises(exceptions.WaitTimeout):
+ self.assertIsNone(utils.wait_until_true(lambda: False,
+ timeout=1, sleep=1))
+
+ def test_timeout_given_exception(self):
+ class MyTimeoutException(exceptions.YardstickException):
+ message = 'My timeout exception'
+
+ with self.assertRaises(MyTimeoutException):
+ self.assertIsNone(
+ utils.wait_until_true(lambda: False, timeout=1, sleep=1,
+ exception=MyTimeoutException))