import mock
+from osm_mon.core.message_bus.producer import KafkaProducer
from osm_mon.plugins.OpenStack.Gnocchi import metrics as metric_req
from osm_mon.plugins.OpenStack.common import Common
"""Initialize a mocked message instance."""
self.topic = "metric_request"
self.key = None
- self.value = json.dumps({"vim_uuid": "test_id", "mock_message": "message_details"})
+ self.value = json.dumps({"mock_message": "message_details"})
+@mock.patch.object(KafkaProducer, 'publish', mock.Mock())
class TestMetricReq(unittest.TestCase):
"""Integration test for metric request keys."""
super(TestMetricReq, self).setUp()
self.metrics = metric_req.Metrics()
- @mock.patch.object(Common, 'get_endpoint')
- @mock.patch.object(Common, "get_auth_token")
- def test_access_cred_metric_auth(self, get_token, get_endpoint):
- """Test authentication with access credentials."""
- message = Message()
-
- self.metrics.metric_calls(message)
-
- get_token.assert_called_with('test_id')
- get_endpoint.assert_any_call('metric', 'test_id')
-
@mock.patch.object(Common, "get_auth_token", mock.Mock())
@mock.patch.object(Common, 'get_endpoint', mock.Mock())
@mock.patch.object(metric_req.Metrics, "delete_metric")
# Mock a message value and key
message = Message()
message.key = "delete_metric_request"
- message.value = json.dumps({"vim_uuid": "test_id", "metric_name": "disk_write_ops", "resource_uuid": "my_r_id"})
+ message.value = json.dumps({"metric_name": "disk_write_ops", "resource_uuid": "my_r_id", "correlation_id": 1})
+
+ del_metric.return_value = True
# Call the metric functionality and check delete request
get_metric_id.return_value = "my_metric_id"
- self.metrics.metric_calls(message)
+ self.metrics.metric_calls(message, 'test_id')
del_metric.assert_called_with(mock.ANY, mock.ANY, "my_metric_id")
@mock.patch.object(Common, "get_auth_token", mock.Mock())
# Mock a message with list metric key and value
message = Message()
message.key = "list_metric_request"
- message.value = json.dumps({"vim_uuid": "test_id", "metrics_list_request": "metric_details"})
+ message.value = json.dumps({"metrics_list_request": {"correlation_id": 1}})
+
+ list_metrics.return_value = []
# Call the metric functionality and check list functionality
- self.metrics.metric_calls(message)
- list_metrics.assert_called_with(mock.ANY, mock.ANY, "metric_details")
+ self.metrics.metric_calls(message, 'test_id')
+ list_metrics.assert_called_with(mock.ANY, mock.ANY, {"correlation_id": 1})
@mock.patch.object(Common, "get_auth_token", mock.Mock())
@mock.patch.object(Common, 'get_endpoint', mock.Mock())
# Mock a message with update metric key and value
message = Message()
message.key = "update_metric_request"
- message.value = json.dumps({"vim_uuid": "test_id",
- "metric_create":
- {"metric_name": "my_metric",
+ message.value = json.dumps({"metric_create_request":
+ {"correlation_id": 1,
+ "metric_name": "my_metric",
"resource_uuid": "my_r_id"}})
# Call metric functionality and confirm no function is called
# Gnocchi does not support updating a metric configuration
- self.metrics.metric_calls(message)
+ self.metrics.metric_calls(message, 'test_id')
config_metric.assert_not_called()
list_metrics.assert_not_called()
delete_metric.assert_not_called()
# Mock a message with create metric key and value
message = Message()
message.key = "create_metric_request"
- message.value = json.dumps({"vim_uuid": "test_id", "metric_create": "metric_details"})
+ message.value = json.dumps({"metric_create_request": "metric_details"})
# Call metric functionality and check config metric
config_metric.return_value = "metric_id", "resource_id", True
- self.metrics.metric_calls(message)
+ self.metrics.metric_calls(message, 'test_id')
config_metric.assert_called_with(mock.ANY, mock.ANY, "metric_details")
@mock.patch.object(Common, "get_auth_token", mock.Mock())
# Mock a message with a read data key and value
message = Message()
message.key = "read_metric_data_request"
- message.value = json.dumps({"vim_uuid": "test_id", "alarm_uuid": "alarm_id"})
+ message.value = json.dumps({"alarm_uuid": "alarm_id"})
# Call metric functionality and check read data metrics
read_data.return_value = "time_stamps", "data_values"
- self.metrics.metric_calls(message)
+ self.metrics.metric_calls(message, 'test_id')
read_data.assert_called_with(
mock.ANY, mock.ANY, json.loads(message.value))