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