--- /dev/null
+from connection import Connection
+import unittest
+import sys
+import jsmin
+import json
+import os
+import time
+from jsmin import jsmin
+sys.path.append("../../test/core/")
+from test_producer import KafkaProducer
+from kafka import KafkaConsumer
+try:
+ import boto
+ import boto.ec2
+ import boto.vpc
+ import boto.ec2.cloudwatch
+ import boto.ec2.connection
+except:
+ exit("Boto not avialable. Try activating your virtualenv OR `pip install boto`")
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+
+# Test Producer object to generate request
+
+producer = KafkaProducer('')
+obj = Connection()
+connections = obj.setEnvironment()
+connections_res = obj.connection_instance()
+cloudwatch_conn = connections_res['cloudwatch_connection']
+
+# Consumer Object to consume response from message bus
+server = {'server': 'localhost:9092', 'topic': 'metric_request'}
+_consumer = KafkaConsumer(bootstrap_servers=server['server'])
+_consumer.subscribe(['metric_response'])
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+
+'''Test E2E Flow : Test cases has been tested one at a time.
+1) Commom Request is generated using request function in test_producer.py(/core/message-bus)
+2) The request is then consumed by the comsumer (plugin)
+3) The response is sent back on the message bus in plugin_metrics.py using
+ response functions in producer.py(/core/message-bus)
+4) The response is then again consumed by the unit_tests_metrics.py
+ and the test cases has been applied on the response.
+'''
+class test_create_metrics(unittest.TestCase):
+
+ def test_status_positive(self):
+ time.sleep(2)
+ # To generate Request of testing valid meric_name in create metrics requests
+ producer.request("create_metrics/create_metric_req_valid.json",'create_metric_request', '','metric_request')
+
+ for message in _consumer:
+ if message.key == "create_metric_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertTrue(resp['metric_create_response']['status'])
+ self.assertEqual(resp['metric_create_response']['metric_uuid'],0)
+ return
+
+ def test_status_negative(self):
+ time.sleep(2)
+ # To generate Request of testing invalid meric_name in create metrics requests
+ producer.request("create_metrics/create_metric_req_invalid.json",'create_metric_request', '','metric_request')
+
+ for message in _consumer:
+ if message.key == "create_metric_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp['metric_create_response']['status'])
+ self.assertEqual(resp['metric_create_response']['metric_uuid'],None)
+ return
+
+class test_metrics_data(unittest.TestCase):
+
+ def test_met_name_positive(self):
+ time.sleep(2)
+ # To generate Request of testing valid meric_name in read_metric_data_request
+ producer.request("read_metrics_data/read_metric_name_req_valid.json",'read_metric_data_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "read_metric_data_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertEqual(type(resp['metrics_data']),dict)
+ return
+
+ def test_met_name_negative(self):
+ time.sleep(2)
+ # To generate Request of testing invalid meric_name in read_metric_data_request
+ producer.request("read_metrics_data/read_metric_name_req_invalid.json",'read_metric_data_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "read_metric_data_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp['metrics_data'])
+ return
+
+ def test_coll_period_positive(self):
+ # To generate Request of testing valid collection_period in read_metric_data_request
+ # For AWS metric_data_stats collection period should be a multiple of 60
+ time.sleep(2)
+ producer.request("read_metrics_data/read_coll_period_req_valid.json",'read_metric_data_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "read_metric_data_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertEqual(type(resp),dict)
+ return
+
+ def test_coll_period_negative(self):
+ time.sleep(2)
+ # To generate Request of testing invalid collection_period in read_metric_data_request
+ producer.request("read_metrics_data/read_coll_period_req_invalid.json",'read_metric_data_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "read_metric_data_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp['metrics_data'])
+ return
+
+class test_update_metrics(unittest.TestCase):
+
+ def test_upd_status_positive(self):
+ time.sleep(2)
+ # To generate Request of testing valid meric_name in update metrics requests
+ producer.request("update_metrics/update_metric_req_valid.json",'update_metric_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "update_metric_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertTrue(resp['metric_update_response']['status'])
+ self.assertEqual(resp['metric_update_response']['metric_uuid'],0)
+ return
+
+ def test_upd_status_negative(self):
+ time.sleep(2)
+ # To generate Request of testing invalid meric_name in update metrics requests
+ producer.request("update_metrics/update_metric_req_invalid.json",'update_metric_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "update_metric_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp['metric_update_response']['status'])
+ self.assertEqual(resp['metric_update_response']['metric_uuid'],None)
+ return
+
+class test_delete_metrics(unittest.TestCase):
+
+ def test_del_met_name_positive(self):
+ time.sleep(2)
+ # To generate Request of testing valid meric_name in delete metrics requests
+ producer.request("delete_metrics/delete_metric_req_valid.json",'delete_metric_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "delete_metric_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp['status'])
+ return
+
+ def test_del_met_name_negative(self):
+ time.sleep(2)
+ # To generate Request of testing invalid meric_name in delete metrics requests
+ producer.request("delete_metrics/delete_metric_req_invalid.json",'delete_metric_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "delete_metric_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp)
+ return
+
+class test_list_metrics(unittest.TestCase):
+
+ def test_list_met_name_positive(self):
+ time.sleep(2)
+ # To generate Request of testing valid meric_name in list metrics requests
+ producer.request("list_metrics/list_metric_req_valid.json",'list_metric_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "list_metrics_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertEqual(type(resp['metrics_list']),list)
+ return
+
+ def test_list_met_name_negitive(self):
+ time.sleep(2)
+ # To generate Request of testing invalid meric_name in list metrics requests
+ producer.request("list_metrics/list_metric_req_invalid.json",'list_metric_request', '','metric_request')
+ for message in _consumer:
+ if message.key == "list_metrics_response":
+ resp = json.loads(json.loads(json.loads(message.value)))
+ time.sleep(1)
+ self.assertFalse(resp['metrics_list'])
+ return
+
+
+if __name__ == '__main__':
+
+ # Saving test reults in Log file
+
+ log_file = 'log_file.txt'
+ f = open(log_file, "w")
+ runner = unittest.TextTestRunner(f)
+ unittest.main(testRunner=runner)
+ f.close()
+
+ # For printing results on Console
+ # unittest.main()
+
projects / osm / MON.git / blobdiff