f8da9dab78
It was introduced the concept of start and end periods in Cloudkitty rating rules. Therefore to make Cloudkitty CLI compatible with the Cloudkitty REST API, we need to add those new available attributes in the CLI as well. Change-Id: I0cd9b61fa81232d235c959da551a7840465fae88 Signed-off-by: Pedro Henrique <phpm13@gmail.com>
346 lines
14 KiB
Python
346 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
# Copyright 2018 Objectif Libre
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
|
|
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
|
|
# License for the specific language governing permissions and limitations
|
|
# under the License.
|
|
#
|
|
from unittest import mock
|
|
|
|
from cloudkittyclient import exc
|
|
from cloudkittyclient.tests.unit.v1 import base
|
|
from cloudkittyclient.tests import utils
|
|
|
|
|
|
class TestHashmap(base.BaseAPIEndpointTestCase):
|
|
|
|
def test_get_mapping_types(self):
|
|
self.hashmap.get_mapping_types()
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/types/')
|
|
|
|
def test_get_service(self):
|
|
self.hashmap.get_service()
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/services/')
|
|
|
|
def test_get_service_service_id(self):
|
|
self.hashmap.get_service(service_id='service_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/services/service_id')
|
|
|
|
def test_create_service(self):
|
|
kwargs = dict(name='service')
|
|
self.hashmap.create_service(**kwargs)
|
|
self.api_client.post.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/services/', json=kwargs)
|
|
|
|
def test_create_service_no_name(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_service)
|
|
|
|
def test_delete_service(self):
|
|
self.hashmap.delete_service(service_id='service_id')
|
|
self.api_client.delete.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/services/',
|
|
json={'service_id': 'service_id'})
|
|
|
|
def test_delete_service_no_id(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.delete_service)
|
|
|
|
def test_get_fields_of_service(self):
|
|
self.hashmap.get_field(service_id='service_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/fields/?service_id=service_id')
|
|
|
|
def test_get_field(self):
|
|
self.hashmap.get_field(field_id='field_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/fields/field_id')
|
|
|
|
def test_get_field_no_args(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.get_field)
|
|
|
|
def test_get_field_with_service_id_and_field_id(self):
|
|
self.assertRaises(exc.InvalidArgumentError, self.hashmap.get_field,
|
|
service_id='service_id', field_id='field_id')
|
|
|
|
def test_create_field(self):
|
|
kwargs = dict(name='name', service_id='service_id')
|
|
self.hashmap.create_field(**kwargs)
|
|
self.api_client.post.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/fields/', json=kwargs)
|
|
|
|
def test_create_field_no_name(self):
|
|
self.assertRaises(exc.ArgumentRequired,
|
|
self.hashmap.create_field,
|
|
service_id='service_id')
|
|
|
|
def test_create_field_no_service_id(self):
|
|
self.assertRaises(
|
|
exc.ArgumentRequired, self.hashmap.create_field, name='name')
|
|
|
|
def test_delete_field(self):
|
|
kwargs = dict(field_id='field_id')
|
|
self.hashmap.delete_field(**kwargs)
|
|
self.api_client.delete.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/fields/', json=kwargs)
|
|
|
|
def test_delete_field_no_arg(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.delete_field)
|
|
|
|
def test_get_mapping_with_id(self):
|
|
self.hashmap.get_mapping(mapping_id='mapping_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/mappings/mapping_id')
|
|
|
|
def test_get_mapping_service_id(self):
|
|
self.hashmap.get_mapping(service_id='service_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/mappings/?service_id=service_id')
|
|
|
|
def test_get_mapping_no_args(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.get_mapping)
|
|
|
|
def test_create_mapping(self):
|
|
kwargs = dict(cost=2, value='value', field_id='field_id',
|
|
name='name', start="2024-01-01",
|
|
end="2024-01-01",
|
|
description="description")
|
|
body = dict(
|
|
cost=kwargs.get('cost'),
|
|
value=kwargs.get('value'),
|
|
service_id=kwargs.get('service_id'),
|
|
field_id=kwargs.get('field_id'),
|
|
group_id=kwargs.get('group_id'),
|
|
tenant_id=kwargs.get('tenant_id'),
|
|
type=kwargs.get('type') or 'flat',
|
|
start="2024-01-01",
|
|
end="2024-01-01",
|
|
description="description",
|
|
name='name'
|
|
)
|
|
self.hashmap.create_mapping(**kwargs)
|
|
self.api_client.post.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/mappings/', json=body)
|
|
|
|
def test_create_mapping_no_cost(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_mapping,
|
|
value='value', field_id='field_id')
|
|
|
|
def test_create_mapping_no_id(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_mapping,
|
|
value='value', cost=12)
|
|
|
|
def test_create_mapping_field_and_service_id(self):
|
|
self.assertRaises(
|
|
exc.InvalidArgumentError, self.hashmap.create_mapping, cost=12,
|
|
field_id='field_id', service_id='service_id')
|
|
|
|
def test_create_mapping_value_and_service_id(self):
|
|
self.assertRaises(
|
|
exc.InvalidArgumentError, self.hashmap.create_mapping,
|
|
value='value', service_id='service_id', cost=0.8)
|
|
|
|
def test_update_mapping(self):
|
|
kwargs = dict(
|
|
cost=12,
|
|
value='value',
|
|
service_id='service_id',
|
|
field_id='field_id',
|
|
tenant_id='tenant_id',
|
|
type='type',
|
|
mapping_id='mapping_id',
|
|
)
|
|
fake_get = mock.Mock(return_value=utils.FakeRequest(
|
|
cost='Bad value',
|
|
value='Bad value',
|
|
service_id='Bad value',
|
|
field_id='Bad value',
|
|
tenant_id='Bad value',
|
|
type='Bad value',
|
|
mapping_id='mapping_id',
|
|
))
|
|
self.api_client.get = fake_get
|
|
self.hashmap.update_mapping(**kwargs)
|
|
self.api_client.get.assert_called_with(
|
|
'/v1/rating/module_config/hashmap/mappings/mapping_id')
|
|
self.api_client.put.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/mappings/', json=kwargs)
|
|
|
|
def test_update_mapping_no_arg(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.update_mapping)
|
|
|
|
def test_get_mapping_group(self):
|
|
self.hashmap.get_mapping_group(mapping_id='mapping_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/'
|
|
'hashmap/mappings/group?mapping_id=mapping_id')
|
|
|
|
def test_delete_mapping(self):
|
|
kwargs = dict(mapping_id='mapping_id')
|
|
self.hashmap.delete_mapping(**kwargs)
|
|
self.api_client.delete.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/mappings/', json=kwargs)
|
|
|
|
def test_delete_mapping_no_arg(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.delete_mapping)
|
|
|
|
def test_get_mapping_group_no_arg(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.get_mapping_group)
|
|
|
|
def test_get_group_no_arg(self):
|
|
self.hashmap.get_group()
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/')
|
|
|
|
def test_get_group(self):
|
|
self.hashmap.get_group(group_id='group_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/group_id')
|
|
|
|
def test_create_group(self):
|
|
kwargs = dict(name='group')
|
|
self.hashmap.create_group(**kwargs)
|
|
self.api_client.post.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/',
|
|
json=kwargs)
|
|
|
|
def test_create_group_no_name(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_group)
|
|
|
|
def test_delete_group(self):
|
|
kwargs = dict(group_id='group_id')
|
|
self.hashmap.delete_group(**kwargs)
|
|
kwargs['recursive'] = False
|
|
self.api_client.delete.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/',
|
|
json=kwargs)
|
|
|
|
def test_delete_group_recursive(self):
|
|
kwargs = dict(group_id='group_id', recursive=True)
|
|
self.hashmap.delete_group(**kwargs)
|
|
self.api_client.delete.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/',
|
|
json=kwargs)
|
|
|
|
def test_delete_group_no_id(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_group)
|
|
|
|
def test_get_group_mappings(self):
|
|
self.hashmap.get_group_mappings(group_id='group_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/mappings'
|
|
'?group_id=group_id')
|
|
|
|
def test_get_group_mappings_no_args(self):
|
|
self.assertRaises(
|
|
exc.ArgumentRequired, self.hashmap.get_group_mappings)
|
|
|
|
def test_get_group_thresholds(self):
|
|
self.hashmap.get_group_thresholds(group_id='group_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/groups/thresholds'
|
|
'?group_id=group_id')
|
|
|
|
def test_get_group_thresholds_no_args(self):
|
|
self.assertRaises(
|
|
exc.ArgumentRequired, self.hashmap.get_group_thresholds)
|
|
|
|
def test_get_threshold_with_id(self):
|
|
self.hashmap.get_threshold(threshold_id='threshold_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/thresholds/threshold_id')
|
|
|
|
def test_get_threshold_service_id(self):
|
|
self.hashmap.get_threshold(service_id='service_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/thresholds/'
|
|
'?service_id=service_id')
|
|
|
|
def test_get_threshold_no_args(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.get_threshold)
|
|
|
|
def test_create_threshold(self):
|
|
kwargs = dict(cost=2, level=123, field_id='field_id')
|
|
body = dict(
|
|
cost=kwargs.get('cost'),
|
|
level=kwargs.get('level'),
|
|
service_id=kwargs.get('service_id'),
|
|
field_id=kwargs.get('field_id'),
|
|
group_id=kwargs.get('group_id'),
|
|
tenant_id=kwargs.get('tenant_id'),
|
|
type=kwargs.get('type') or 'flat',
|
|
)
|
|
self.hashmap.create_threshold(**kwargs)
|
|
self.api_client.post.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/thresholds/', json=body)
|
|
|
|
def test_create_threshold_no_cost(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_threshold,
|
|
level=123, field_id='field_id')
|
|
|
|
def test_create_threshold_no_id(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.create_threshold,
|
|
level=123, cost=12)
|
|
|
|
def test_create_threshold_field_and_service_id(self):
|
|
self.assertRaises(
|
|
exc.ArgumentRequired, self.hashmap.create_threshold, cost=12,
|
|
field_id='field_id', service_id='service_id')
|
|
|
|
def test_delete_threshold(self):
|
|
kwargs = dict(threshold_id='threshold_id')
|
|
self.hashmap.delete_threshold(**kwargs)
|
|
self.api_client.delete.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/thresholds/', json=kwargs)
|
|
|
|
def test_delete_threshold_no_arg(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.delete_threshold)
|
|
|
|
def test_update_threshold(self):
|
|
kwargs = dict(
|
|
cost=12,
|
|
level=123,
|
|
service_id='service_id',
|
|
field_id='field_id',
|
|
tenant_id='tenant_id',
|
|
type='type',
|
|
threshold_id='threshold_id'
|
|
)
|
|
fake_get = mock.Mock(return_value=utils.FakeRequest(
|
|
cost='Bad value',
|
|
level='Bad value',
|
|
service_id='Bad value',
|
|
field_id='Bad value',
|
|
tenant_id='Bad value',
|
|
type='Bad value',
|
|
threshold_id='threshold_id'
|
|
))
|
|
self.api_client.get = fake_get
|
|
self.hashmap.update_threshold(**kwargs)
|
|
self.api_client.get.assert_called_with(
|
|
'/v1/rating/module_config/hashmap/thresholds/threshold_id')
|
|
self.api_client.put.assert_called_once_with(
|
|
'/v1/rating/module_config/hashmap/thresholds/', json=kwargs)
|
|
|
|
def test_update_threshold_no_arg(self):
|
|
self.assertRaises(exc.ArgumentRequired, self.hashmap.update_threshold)
|
|
|
|
def test_get_threshold_group(self):
|
|
self.hashmap.get_threshold_group(threshold_id='threshold_id')
|
|
self.api_client.get.assert_called_once_with(
|
|
'/v1/rating/module_config/'
|
|
'hashmap/thresholds/group?threshold_id=threshold_id')
|
|
|
|
def test_get_threshold_group_no_arg(self):
|
|
self.assertRaises(
|
|
exc.ArgumentRequired, self.hashmap.get_threshold_group)
|