# #
from osm_mon.collector.vnf_collectors.vrops.vrops_helper import vROPS_Helper
-from osm_mon.tests.unit.collector.vnf_collectors.vmware.mock_http import mock_http_response
+from osm_mon.tests.unit.collector.vnf_collectors.vmware.mock_http import (
+ mock_http_response,
+)
from unittest import TestCase
import json
class vROPS_Helper_Resource_List_Test(TestCase):
-
def setUp(self):
super().setUp()
self.vrops = vROPS_Helper()
def test_get_vm_resource_list_from_vrops(self):
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources\\?resourceKind=VirtualMachine',
- response_file='vrops_resources.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources\\?resourceKind=VirtualMachine",
+ response_file="vrops_resources.json",
+ )
resource_list = self.vrops.get_vm_resource_list_from_vrops()
- self.assertEqual(len(resource_list), 3, "List of resources from canned vrops_resources.json")
+ self.assertEqual(
+ len(resource_list),
+ 3,
+ "List of resources from canned vrops_resources.json",
+ )
def test_get_vm_resource_list_from_vrops_http_404(self):
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources\\?resourceKind=VirtualMachine',
- response_file='404.txt', status_code=404)
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources\\?resourceKind=VirtualMachine",
+ response_file="404.txt",
+ status_code=404,
+ )
resource_list = self.vrops.get_vm_resource_list_from_vrops()
self.assertEqual(len(resource_list), 0, "Should return an empty list")
def test_get_vm_resource_list_from_vrops_bad_json(self):
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources\\?resourceKind=VirtualMachine',
- response_file='malformed.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources\\?resourceKind=VirtualMachine",
+ response_file="malformed.json",
+ )
resource_list = self.vrops.get_vm_resource_list_from_vrops()
self.assertEqual(len(resource_list), 0, "Should return an empty list")
class vROPS_Helper_Get_Metrics_Test(TestCase):
-
def setUp(self):
super().setUp()
self.vrops = vROPS_Helper()
- with open(os.path.join(os.path.dirname(__file__), 'osm_mocks', 'VNFR.json'), 'r') as f:
+ with open(
+ os.path.join(os.path.dirname(__file__), "osm_mocks", "VNFR.json"), "r"
+ ) as f:
self.vnfr = json.load(f)
def tearDown(self):
super().tearDown()
def test_collect_one_metric_only(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
{"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
- ]
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 1, "Number of metrics returned")
self.assertEqual(metrics[0].value, 100.0, "CPU metric value")
def test_collect_cpu_and_memory(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
{"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
- {"id": "ubuntu_vnf_average_memory_utilization", "performance-metric": "average_memory_utilization"}
- ]
+ {
+ "id": "ubuntu_vnf_average_memory_utilization",
+ "performance-metric": "average_memory_utilization",
+ },
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 2, "Number of metrics returned")
self.assertEqual(metrics[0].name, "cpu_utilization", "First metric name")
self.assertEqual(metrics[0].value, 100.0, "CPU metric value")
- self.assertEqual(metrics[1].name, "average_memory_utilization", "Second metric name")
+ self.assertEqual(
+ metrics[1].name, "average_memory_utilization", "Second metric name"
+ )
self.assertEqual(metrics[1].value, 20.515941619873047, "Memory metric value")
def test_collect_adjusted_metric(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
- {'id': 'ubuntu_vnf_cpu_util', 'performance-metric': 'disk_read_bytes'}
- ]
+ {"id": "ubuntu_vnf_cpu_util", "performance-metric": "disk_read_bytes"}
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(metrics[0].value, 10240.0, "Disk read bytes (not KB/s)")
def test_collect_not_provided_metric(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
- {'id': 'cirros_vnf_packets_sent', 'performance-metric': 'packets_in_dropped'},
- ]
+ {
+ "id": "cirros_vnf_packets_sent",
+ "performance-metric": "packets_in_dropped",
+ },
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 0, "Number of metrics returned")
def test_collect_unkown_metric(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
- {'id': 'cirros_vnf-Unknown_Metric', 'performance-metric': 'unknown'},
- ]
+ {"id": "cirros_vnf-Unknown_Metric", "performance-metric": "unknown"},
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 0, "Number of metrics returned")
def test_collect_vrops_no_data(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
{"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
- ]
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='OK.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="OK.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 0, "Number of metrics returned")
def test_collect_vrops_unknown_vim_id(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {"name": "vmware-scaling-1-ubuntu_vnfd-VM-2"}
+ }
monitoring_params = [
{"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
- ]
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 0, "Number of metrics returned")
def test_collect_vrops_http_error(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
{"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
- ]
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='404.txt', status_code=404)
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="404.txt",
+ status_code=404,
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 0, "Number of metrics returned")
def test_collect_vrops_json_parse_error(self):
- vdu_mappings = {'VMWARE-OID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2',
- 'vrops_id': 'VROPS-UUID-1'}}
+ vdu_mappings = {
+ "VMWARE-OID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-1",
+ }
+ }
monitoring_params = [
{"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
- ]
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='404.txt')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="404.txt",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
self.assertEqual(len(metrics), 0, "Number of metrics returned")
def test_collect_multi_vdu(self):
- vdu_mappings = {'VMWARE-UUID-VM-1':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-1', 'vrops_id': 'VROPS-UUID-1'},
- 'VMWARE-UUID-VM-2':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2', 'vrops_id': 'VROPS-UUID-2'},
- 'VMWARE-UUID-VM-3':
- {'name': 'vmware-scaling-1-ubuntu_vnfd-VM-2', 'vrops_id': 'VROPS-UUID-3'}
- }
+ vdu_mappings = {
+ "VMWARE-UUID-VM-1": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-1",
+ "vrops_id": "VROPS-UUID-1",
+ },
+ "VMWARE-UUID-VM-2": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-2",
+ },
+ "VMWARE-UUID-VM-3": {
+ "name": "vmware-scaling-1-ubuntu_vnfd-VM-2",
+ "vrops_id": "VROPS-UUID-3",
+ },
+ }
monitoring_params = [
- {'id': 'ubuntu_vnf_cpu_util', 'performance-metric': 'cpu_utilization'},
- {'id': 'ubuntu_vnf_average_memory_utilization', 'performance-metric': 'average_memory_utilization'},
- {'id': 'ubuntu_vnf_disk_read_ops', 'performance-metric': 'disk_read_ops'},
- {'id': 'ubuntu_vnf_disk_write_ops', 'performance-metric': 'disk_write_ops'},
- {'id': 'ubuntu_vnf_disk_read_bytes', 'performance-metric': 'disk_read_bytes'},
- {'id': 'ubuntu_vnf_disk_write_bytes', 'performance-metric': 'disk_write_bytes'},
- {'id': 'ubuntu_vnf_packets_out_dropped', 'performance-metric': 'packets_out_dropped'},
- {'id': 'ubuntu_vnf_packets_received', 'performance-metric': 'packets_received'},
- {'id': 'ubuntu_vnf_packets_sent', 'performance-metric': 'packets_sent'}
- ]
+ {"id": "ubuntu_vnf_cpu_util", "performance-metric": "cpu_utilization"},
+ {
+ "id": "ubuntu_vnf_average_memory_utilization",
+ "performance-metric": "average_memory_utilization",
+ },
+ {"id": "ubuntu_vnf_disk_read_ops", "performance-metric": "disk_read_ops"},
+ {"id": "ubuntu_vnf_disk_write_ops", "performance-metric": "disk_write_ops"},
+ {
+ "id": "ubuntu_vnf_disk_read_bytes",
+ "performance-metric": "disk_read_bytes",
+ },
+ {
+ "id": "ubuntu_vnf_disk_write_bytes",
+ "performance-metric": "disk_write_bytes",
+ },
+ {
+ "id": "ubuntu_vnf_packets_out_dropped",
+ "performance-metric": "packets_out_dropped",
+ },
+ {
+ "id": "ubuntu_vnf_packets_received",
+ "performance-metric": "packets_received",
+ },
+ {"id": "ubuntu_vnf_packets_sent", "performance-metric": "packets_sent"},
+ ]
with requests_mock.Mocker() as mock_requests:
- mock_http_response(mock_requests, method='POST',
- url_pattern='/suite-api/api/auth/token/acquire',
- response_file='vrops_token.json')
- mock_http_response(mock_requests,
- url_pattern='/suite-api/api/resources/stats.*',
- response_file='vrops_multi.json')
+ mock_http_response(
+ mock_requests,
+ method="POST",
+ url_pattern="/suite-api/api/auth/token/acquire",
+ response_file="vrops_token.json",
+ )
+ mock_http_response(
+ mock_requests,
+ url_pattern="/suite-api/api/resources/stats.*",
+ response_file="vrops_multi.json",
+ )
metrics = self.vrops.get_metrics(vdu_mappings, monitoring_params, self.vnfr)
- self.assertEqual(len(metrics), len(monitoring_params) * len(vdu_mappings), "Number of metrics returned")
+ self.assertEqual(
+ len(metrics),
+ len(monitoring_params) * len(vdu_mappings),
+ "Number of metrics returned",
+ )
projects / osm / MON.git / blobdiff