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         c.server = mock_ssh.SSH()
  55
  56         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
  57         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  58
  59         result = c.run(args)
  60         expected_result = json.loads(sample_output)
  61         self.assertEqual(result, expected_result)
  62
  63     def test_cyclictest_successful_sla(self, mock_ssh):
  64
  65         c = cyclictest.Cyclictest(self.ctx)
  66         options = {
  67             "affinity": 2,
  68             "interval": 100,
  69             "priority": 88,
  70             "loops": 10000,
  71             "threads": 2,
  72             "histogram": 80
  73         }
  74         sla = {
  75             "max_min_latency": 100,
  76             "max_avg_latency": 500,
  77             "max_max_latency": 1000,
  78         }
  79         args = {
  80             "options": options,
  81             "sla": sla
  82         }
  83         c.server = mock_ssh.SSH()
  84
  85         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
  86         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
  87
  88         result = c.run(args)
  89         expected_result = json.loads(sample_output)
  90         self.assertEqual(result, expected_result)
  91
  92     def test_cyclictest_unsuccessful_sla_min_latency(self, mock_ssh):
  93
  94         c = cyclictest.Cyclictest(self.ctx)
  95         args = {
  96             "options": {},
  97             "sla": {"max_min_latency": 10}
  98         }
  99         c.server = mock_ssh.SSH()
 100         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
 101
 102         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 103         self.assertRaises(AssertionError, c.run, args)
 104
 105     def test_cyclictest_unsuccessful_sla_avg_latency(self, mock_ssh):
 106
 107         c = cyclictest.Cyclictest(self.ctx)
 108         args = {
 109             "options": {},
 110             "sla": {"max_avg_latency": 10}
 111         }
 112         c.server = mock_ssh.SSH()
 113         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
 114
 115         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 116         self.assertRaises(AssertionError, c.run, args)
 117
 118     def test_cyclictest_unsuccessful_sla_max_latency(self, mock_ssh):
 119
 120         c = cyclictest.Cyclictest(self.ctx)
 121         args = {
 122             "options": {},
 123             "sla": {"max_max_latency": 10}
 124         }
 125         c.server = mock_ssh.SSH()
 126         sample_output = '{"min": 100, "avg": 500, "max": 1000}'
 127
 128         mock_ssh.SSH().execute.return_value = (0, sample_output, '')
 129         self.assertRaises(AssertionError, c.run, args)
 130
 131     def test_cyclictest_unsuccessful_script_error(self, mock_ssh):
 132
 133         c = cyclictest.Cyclictest(self.ctx)
 134         options = {
 135             "affinity": 2,
 136             "interval": 100,
 137             "priority": 88,
 138             "loops": 10000,
 139             "threads": 2,
 140             "histogram": 80
 141         }
 142         sla = {
 143             "max_min_latency": 100,
 144             "max_avg_latency": 500,
 145             "max_max_latency": 1000,
 146         }
 147         args = {
 148             "options": options,
 149             "sla": sla
 150         }
 151         c.server = mock_ssh.SSH()
 152
 153         mock_ssh.SSH().execute.return_value = (1, '', 'FOOBAR')
 154         self.assertRaises(RuntimeError, c.run, args)
 155
 156
 157 def main():
 158     unittest.main()
 159
 160 if __name__ == '__main__':
 161     main()