test/ut_behave_tests/test_steps.py

   1 #!/usr/bin/env python
   2 # Copyright 2022 Orange
   3 #
   4 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 #    not use this file except in compliance with the License. You may obtain
   6 #    a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 #    Unless required by applicable law or agreed to in writing, software
  11 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 #    License for the specific language governing permissions and limitations
  14 #    under the License.
  15 #
  16
  17 """
  18 Unit tests for some of the functions found in behave_tests/features/steps/steps.py
  19 """
  20
  21 import logging
  22 import unittest
  23 from unittest.mock import call, Mock, patch
  24
  25 from behave_tests.features.steps.steps import get_last_result
  26 from .test_utils import setup_logging, stub_requests_get
  27
  28
  29 def setUpModule():
  30     setup_logging()
  31
  32
  33 class TestGetLastResult(unittest.TestCase):
  34     def setUp(self):
  35         # Mock requests.get() so that TestAPI results come from JSON files
  36         # found in test_data/ directory.
  37         patcher = patch('behave_tests.features.steps.testapi.requests')
  38         self._mock_requests = patcher.start()
  39         self._mock_requests.get.side_effect = stub_requests_get
  40         self.addCleanup(patcher.stop)
  41
  42         # Setup a mock for behave context
  43         self._context = Mock()
  44         self._context.data = {
  45             'PROJECT_NAME': "nfvbench",
  46             'TEST_DB_URL': "http://127.0.0.1:8000/api/v1/results"
  47         }
  48         self._context.logger = logging.getLogger("behave_tests")
  49
  50     def test_get_last_result_throughput_characterization(self):
  51         self._context.json = {
  52             "frame_sizes": ['64'],
  53             "flow_count": "100k",
  54             "duration_sec": '10',
  55             "rate": "ndr",
  56             "user_label": "amical_tc18_loopback"
  57         }
  58         self._context.tag = "throughput"
  59
  60         last_result = get_last_result(self._context, reference=True)
  61
  62         self.assertIsNotNone(last_result)
  63         self.assertEqual(16765582, last_result["synthesis"]["total_tx_rate"])
  64         self.assertEqual(25, round(last_result["synthesis"]["avg_delay_usec"]))
  65
  66         self._mock_requests.get.assert_called_once_with(
  67             "http://127.0.0.1:8000/api/v1/results?"
  68             "project=nfvbench&case=characterization&criteria=PASS&page=1")
  69
  70     def test_get_last_result_latency_characterization(self):
  71         self._context.json = {
  72             "frame_sizes": ['768'],
  73             "flow_count": "100k",
  74             "duration_sec": '10',
  75             "rate": "90%",
  76             "user_label": "amical_tc6_intensive"
  77         }
  78         self._context.tag = "latency"
  79
  80         last_result = get_last_result(self._context, reference=True)
  81
  82         self.assertIsNotNone(last_result)
  83         self.assertEqual(262275, last_result["synthesis"]["total_tx_rate"])
  84         self.assertEqual(353, round(last_result["synthesis"]["avg_delay_usec"]))
  85
  86         self._mock_requests.get.assert_has_calls([
  87             call("http://127.0.0.1:8000/api/v1/results?"
  88                  "project=nfvbench&case=characterization&criteria=PASS&page=1"),
  89             call("http://127.0.0.1:8000/api/v1/results?"
  90                  "project=nfvbench&case=characterization&criteria=PASS&page=2")])
  91
  92     def test_last_result_not_found(self):
  93         self._context.json = {
  94             "frame_sizes": ['64'],
  95             "flow_count": "100k",
  96             "duration_sec": '10',
  97             "rate": "ndr",
  98             "user_label": "toto_titi_tata"  # User label not in test data
  99         }
 100         self._context.tag = "throughput"
 101
 102         with self.assertRaises(AssertionError):
 103             get_last_result(self._context, reference=True)
 104
 105         self._mock_requests.get.assert_has_calls([
 106             call("http://127.0.0.1:8000/api/v1/results?"
 107                  "project=nfvbench&case=characterization&criteria=PASS&page=1"),
 108             call("http://127.0.0.1:8000/api/v1/results?"
 109                  "project=nfvbench&case=characterization&criteria=PASS&page=2")])
 110
 111     def test_get_last_result_throughput_non_regression(self):
 112         self._context.CASE_NAME = "non-regression"
 113         self._context.json = {
 114             "frame_sizes": ['1518'],
 115             "flow_count": "100k",
 116             "duration_sec": '10',
 117             "rate": "ndr",
 118             "user_label": "amical_tc12_basic"
 119         }
 120         self._context.tag = "throughput"
 121
 122         last_result = get_last_result(self._context)
 123
 124         self.assertIsNotNone(last_result)
 125         self.assertEqual(512701, last_result["synthesis"]["total_tx_rate"])
 126         self.assertEqual(148, round(last_result["synthesis"]["avg_delay_usec"]))
 127
 128         self._mock_requests.get.assert_called_once_with(
 129             "http://127.0.0.1:8000/api/v1/results?"
 130             "project=nfvbench&case=non-regression&criteria=PASS&page=1")
 131
 132     def test_get_last_result_latency_non_regression(self):
 133         self._context.CASE_NAME = "non-regression"
 134         self._context.json = {
 135             "frame_sizes": ['1518'],
 136             "flow_count": "100k",
 137             "duration_sec": '10',
 138             "rate": "70%",
 139             "user_label": "amical_tc12_basic"
 140         }
 141         self._context.tag = "latency"
 142
 143         last_result = get_last_result(self._context)
 144
 145         self.assertIsNotNone(last_result)
 146         self.assertEqual(352040, last_result["synthesis"]["total_tx_rate"])
 147         self.assertEqual(114, round(last_result["synthesis"]["avg_delay_usec"]))
 148
 149         self._mock_requests.get.assert_called_once_with(
 150             "http://127.0.0.1:8000/api/v1/results?"
 151             "project=nfvbench&case=non-regression&criteria=PASS&page=1")