tests/unit/benchmark/scenarios/compute/test_cyclictest.py

   1 #!/usr/bin/env python
   2
   3 ##############################################################################
   4 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and other.
   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.cyclictest.Cyclictest
  13
  14 import mock
  15 import unittest
  16 import json
  17
  18 from yardstick.benchmark.scenarios.compute import cyclictest
  19
  20
  21 @mock.patch('yardstick.benchmark.scenarios.compute.cyclictest.ssh')
  22 class CyclictestTestCase(unittest.TestCase):
  23
  24     def setUp(self):
  25         self.ctx = {
  26             "host": "192.168.50.28",
  27             "user": "root",
  28             "key_filename": "mykey.key"
  29         }
  30
  31     def test_cyclictest_successful_setup(self, mock_ssh):
  32
  33         c = cyclictest.Cyclictest(self.ctx)
  34         c.setup()
  35
  36         mock_ssh.SSH().execute.return_value = (0, '', '')
  37         self.assertIsNotNone(c.client)
  38         self.assertEqual(c.setup_done, True)
  39
  40     def test_cyclictest_successful_no_sla(self, mock_ssh):
  41
  42         c = cyclictest.Cyclictest(self.ctx)
  43         options = {
  44             "affinity": 2,
  45             "interval": 100,
  46             "priority": 88,
  47             "loops": 10000,
  48             "threads": 2,
  49             "histogram": 80
  50         }
  51         args = {
  52             "options": options,
  53         }
  54         result = {}
  55
  56         c.server = mock_ssh.SSH()
  57
  58         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
  59         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  60
  61         c.run(args, result)
  62         expected_result = json.loads(sample_output)
  63         self.assertEqual(result, expected_result)
  64
  65     def test_cyclictest_successful_sla(self, mock_ssh):
  66
  67         c = cyclictest.Cyclictest(self.ctx)
  68         options = {
  69             "affinity": 2,
  70             "interval": 100,
  71             "priority": 88,
  72             "loops": 10000,
  73             "threads": 2,
  74             "histogram": 80
  75         }
  76         sla = {
  77             "max_min_latency": 100,
  78             "max_avg_latency": 500,
  79             "max_max_latency": 1000,
  80         }
  81         args = {
  82             "options": options,
  83             "sla": sla
  84         }
  85         result = {}
  86
  87         c.server = mock_ssh.SSH()
  88
  89         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
  90         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  91
  92         c.run(args, result)
  93         expected_result = json.loads(sample_output)
  94         self.assertEqual(result, expected_result)
  95
  96     def test_cyclictest_unsuccessful_sla_min_latency(self, mock_ssh):
  97
  98         c = cyclictest.Cyclictest(self.ctx)
  99         args = {
 100             "options": {},
 101             "sla": {"max_min_latency": 10}
 102         }
 103         result = {}
 104
 105         c.server = mock_ssh.SSH()
 106         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
 107
 108         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 109         self.assertRaises(AssertionError, c.run, args, result)
 110
 111     def test_cyclictest_unsuccessful_sla_avg_latency(self, mock_ssh):
 112
 113         c = cyclictest.Cyclictest(self.ctx)
 114         args = {
 115             "options": {},
 116             "sla": {"max_avg_latency": 10}
 117         }
 118         result = {}
 119
 120         c.server = mock_ssh.SSH()
 121         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
 122
 123         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 124         self.assertRaises(AssertionError, c.run, args, result)
 125
 126     def test_cyclictest_unsuccessful_sla_max_latency(self, mock_ssh):
 127
 128         c = cyclictest.Cyclictest(self.ctx)
 129         args = {
 130             "options": {},
 131             "sla": {"max_max_latency": 10}
 132         }
 133         result = {}
 134
 135         c.server = mock_ssh.SSH()
 136         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
 137
 138         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 139         self.assertRaises(AssertionError, c.run, args, result)
 140
 141     def test_cyclictest_unsuccessful_script_error(self, mock_ssh):
 142
 143         c = cyclictest.Cyclictest(self.ctx)
 144         options = {
 145             "affinity": 2,
 146             "interval": 100,
 147             "priority": 88,
 148             "loops": 10000,
 149             "threads": 2,
 150             "histogram": 80
 151         }
 152         sla = {
 153             "max_min_latency": 100,
 154             "max_avg_latency": 500,
 155             "max_max_latency": 1000,
 156         }
 157         args = {
 158             "options": options,
 159             "sla": sla
 160         }
 161         result = {}
 162
 163         c.server = mock_ssh.SSH()
 164
 165         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 166         self.assertRaises(RuntimeError, c.run, args, result)
 167
 168
 169 def main():
 170     unittest.main()
 171
 172 if __name__ == '__main__':
 173     main()