"Fossies" - the Fresh Open Source Software Archive

Member "cloudkitty-13.0.0/cloudkitty/tests/common/test_prometheus_client.py" (14 Oct 2020, 5395 Bytes) of package /linux/misc/openstack/cloudkitty-13.0.0.tar.gz:


As a special service "Fossies" has tried to format the requested source page into HTML format using (guessed) Python source code syntax highlighting (style: standard) with prefixed line numbers. Alternatively you can here view or download the uninterpreted source code file. See also the latest Fossies "Diffs" side-by-side code changes report for "test_prometheus_client.py": 12.1.0_vs_13.0.0.

    1 # -*- coding: utf-8 -*-
    2 # Copyright 2019 Objectif Libre
    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 from unittest import mock
   17 
   18 from cloudkitty.collector import prometheus
   19 from cloudkitty import tests
   20 from cloudkitty.tests import samples
   21 from cloudkitty.utils import json
   22 
   23 
   24 class PrometheusClientTest(tests.TestCase):
   25 
   26     class FakeResponse(object):
   27         """Mimics an HTTP ``requests`` response"""
   28 
   29         def __init__(self, url, text, status_code):
   30             self.url = url
   31             self.text = text
   32             self.status_code = status_code
   33 
   34         def json(self):
   35             return json.loads(self.text)
   36 
   37     @staticmethod
   38     def _mock_requests_get(text):
   39         """Factory to build FakeResponse with desired response body text"""
   40         return lambda *args, **kwargs: PrometheusClientTest.FakeResponse(
   41             args[0], text, 200,
   42         )
   43 
   44     def setUp(self):
   45         super(PrometheusClientTest, self).setUp()
   46         self.client = prometheus.PrometheusClient(
   47             'http://localhost:9090/api/v1',
   48         )
   49 
   50     def test_get_with_no_options(self):
   51         with mock.patch('requests.get') as mock_get:
   52             self.client._get(
   53                 'query_range',
   54                 params={
   55                     'query': 'max(http_requests_total) by (project_id)',
   56                     'start': samples.FIRST_PERIOD_BEGIN,
   57                     'end': samples.FIRST_PERIOD_END,
   58                     'step': 10,
   59                 },
   60             )
   61             mock_get.assert_called_once_with(
   62                 'http://localhost:9090/api/v1/query_range',
   63                 params={
   64                     'query': 'max(http_requests_total) by (project_id)',
   65                     'start': samples.FIRST_PERIOD_BEGIN,
   66                     'end': samples.FIRST_PERIOD_END,
   67                     'step': 10,
   68                 },
   69                 auth=None,
   70                 verify=True,
   71             )
   72 
   73     def test_get_with_options(self):
   74         client = prometheus.PrometheusClient(
   75             'http://localhost:9090/api/v1',
   76             auth=('foo', 'bar'),
   77             verify='/some/random/path',
   78         )
   79         with mock.patch('requests.get') as mock_get:
   80             client._get(
   81                 'query_range',
   82                 params={
   83                     'query': 'max(http_requests_total) by (project_id)',
   84                     'start': samples.FIRST_PERIOD_BEGIN,
   85                     'end': samples.FIRST_PERIOD_END,
   86                     'step': 10,
   87                 },
   88             )
   89             mock_get.assert_called_once_with(
   90                 'http://localhost:9090/api/v1/query_range',
   91                 params={
   92                     'query': 'max(http_requests_total) by (project_id)',
   93                     'start': samples.FIRST_PERIOD_BEGIN,
   94                     'end': samples.FIRST_PERIOD_END,
   95                     'step': 10,
   96                 },
   97                 auth=('foo', 'bar'),
   98                 verify='/some/random/path',
   99             )
  100 
  101     def test_get_instant(self):
  102         mock_get = mock.patch(
  103             'requests.get',
  104             side_effect=self._mock_requests_get('{"foo": "bar"}'),
  105         )
  106 
  107         with mock_get:
  108             res = self.client.get_instant(
  109                 'max(http_requests_total) by (project_id)',
  110             )
  111             self.assertEqual(res, {'foo': 'bar'})
  112 
  113     def test_get_range(self):
  114         mock_get = mock.patch(
  115             'requests.get',
  116             side_effect=self._mock_requests_get('{"foo": "bar"}'),
  117         )
  118 
  119         with mock_get:
  120             res = self.client.get_range(
  121                 'max(http_requests_total) by (project_id)',
  122                 samples.FIRST_PERIOD_BEGIN,
  123                 samples.FIRST_PERIOD_END,
  124                 10,
  125             )
  126             self.assertEqual(res, {'foo': 'bar'})
  127 
  128     def test_get_instant_raises_error_on_bad_json(self):
  129         # Simulating malformed JSON response from HTTP+PromQL instant request
  130         mock_get = mock.patch(
  131             'requests.get',
  132             side_effect=self._mock_requests_get('{"foo": "bar"'),
  133         )
  134         with mock_get:
  135             self.assertRaises(
  136                 prometheus.PrometheusResponseError,
  137                 self.client.get_instant,
  138                 'max(http_requests_total) by (project_id)',
  139             )
  140 
  141     def test_get_range_raises_error_on_bad_json(self):
  142         # Simulating malformed JSON response from HTTP+PromQL range request
  143         mock_get = mock.patch(
  144             'requests.get',
  145             side_effect=self._mock_requests_get('{"foo": "bar"'),
  146         )
  147         with mock_get:
  148             self.assertRaises(
  149                 prometheus.PrometheusResponseError,
  150                 self.client.get_range,
  151                 'max(http_requests_total) by (project_id)',
  152                 samples.FIRST_PERIOD_BEGIN,
  153                 samples.FIRST_PERIOD_END,
  154                 10,
  155             )