cdf518d82145b4715ddc19e96745e76488194d45
[yardstick.git] / tests / unit / benchmark / scenarios / compute / test_memload.py
1 #!/usr/bin/env python
2
3 ##############################################################################
4 # Copyright (c) 2016 Huawei Technologies Co.,Ltd and others.
5 #
6 # All rights reserved. This program and the accompanying materials
7 # are made available under the terms of the Apache License, Version 2.0
8 # which accompanies this distribution, and is available at
9 # http://www.apache.org/licenses/LICENSE-2.0
10 ##############################################################################
11
12 # Unittest for yardstick.benchmark.scenarios.compute.memload.MEMLoad
13
14 import mock
15 import unittest
16 import os
17
18 from yardstick.benchmark.scenarios.compute import memload
19
20
21 @mock.patch('yardstick.benchmark.scenarios.compute.memload.ssh')
22 class MEMLoadTestCase(unittest.TestCase):
23
24     def setUp(self):
25         self.ctx = {
26             'host': {
27                 'ip': '172.16.0.137',
28                 'user': 'root',
29                 'key_filename': "mykey.key"
30             }
31         }
32
33         self.result = {}
34
35     def test_memload_successful_setup(self, mock_ssh):
36         m = memload.MEMLoad({}, self.ctx)
37         mock_ssh.SSH().execute.return_value = (0, '', '')
38
39         m.setup()
40         self.assertIsNotNone(m.client)
41         self.assertTrue(m.setup_done)
42
43     def test_execute_command_success(self, mock_ssh):
44         m = memload.MEMLoad({}, self.ctx)
45         mock_ssh.SSH().execute.return_value = (0, '', '')
46         m.setup()
47
48         expected_result = 'abcdefg'
49         mock_ssh.SSH().execute.return_value = (0, expected_result, '')
50         result = m._execute_command("foo")
51         self.assertEqual(result, expected_result)
52
53     def test_execute_command_failed(self, mock_ssh):
54         m = memload.MEMLoad({}, self.ctx)
55         mock_ssh.SSH().execute.return_value = (0, '', '')
56         m.setup()
57
58         mock_ssh.SSH().execute.return_value = (127, '', 'Failed executing \
59                                                command')
60         self.assertRaises(RuntimeError, m._execute_command,
61                           "cat /proc/meminfo")
62
63     def test_get_mem_usage_successful(self, mock_ssh):
64         options = {
65             "interval": 1,
66             "count": 1
67         }
68         args = {"options": options}
69         m = memload.MEMLoad(args, self.ctx)
70         mock_ssh.SSH().execute.return_value = (0, '', '')
71         m.setup()
72
73         output = self._read_file("memload_sample_output.txt")
74         mock_ssh.SSH().execute.return_value = (0, output, '')
75         result = m._get_mem_usage()
76         expected_result = {"max": {"used": 76737332, "cached": 67252400,
77                            "free": 187016644, "shared": 2844,
78                            "total": 263753976, "buffers": 853528},
79                            "average": {"used": 76737332, "cached": 67252400,
80                            "free": 187016644, "shared": 2844,
81                            "total": 263753976, "buffers": 853528},
82                            "free": {"memory0": {"used": "76737332",
83                            "cached": "67252400", "free": "187016644",
84                            "shared": "2844", "total": "263753976",
85                            "buffers": "853528"}}}
86         self.assertEqual(result, expected_result)
87
88     def _read_file(self, filename):
89         curr_path = os.path.dirname(os.path.abspath(__file__))
90         output = os.path.join(curr_path, filename)
91         with open(output) as f:
92             sample_output = f.read()
93         return sample_output
94