--- /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('create_alarm_request')
+obj = Connection()
+connections = obj.setEnvironment()
+connections_res = obj.connection_instance()
+cloudwatch_conn = connections_res['cloudwatch_connection']
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+
+'''Test E2E Flow : Test cases has been tested one at a time.
+1) Commom Request is generated using request function in test_producer.py(/test/core)
+2) The request is then consumed by the comsumer (plugin)
+3) The response is sent back on the message bus in plugin_alarm.py using
+ response functions in producer.py(/core/message-bus)
+4) The response is then again consumed by the unit_tests_alarms.py
+ and the test cases has been applied on the response.
+'''
+
+class config_alarm_name_test(unittest.TestCase):
+
+
+ def setUp(self):
+ pass
+ #To generate a request of testing new alarm name and new instance id in create alarm request
+ def test_differentName_differentInstance(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/create_alarm_differentName_differentInstance.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info)
+ time.sleep(1)
+ self.assertTrue(info['alarm_create_response']['status'])
+ return
+
+ #To generate a request of testing new alarm name and existing instance id in create alarm request
+ def test_differentName_sameInstance(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/create_alarm_differentName_sameInstance.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid_delete1.json",'delete_alarm_request','','alarm_request')
+ self.assertTrue(info['alarm_create_response']['status'])
+ return
+
+ #To generate a request of testing existing alarm name and new instance id in create alarm request
+ def test_sameName_differentInstance(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/create_alarm_sameName_differentInstance.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid_delete2.json",'delete_alarm_request', '','alarm_request')
+ self.assertTrue(info['alarm_create_response']['status'])
+ return
+
+ #To generate a request of testing existing alarm name and existing instance id in create alarm request
+ def test_sameName_sameInstance(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/create_alarm_sameName_sameInstance.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info, "---")
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid.json",'delete_alarm_request', '','alarm_request')
+ self.assertEqual(info, None)
+ return
+
+ #To generate a request of testing valid statistics in create alarm request
+ def test_statisticValid(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/statistic_valid.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid_delete3.json",'delete_alarm_request', '','alarm_request')
+ self.assertTrue(info['alarm_create_response']['status'])
+ return
+
+ #To generate a request of testing Invalid statistics in create alarm request
+ def test_statisticValidNot(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/statistic_invalid.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info, "---")
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid_delete3.json",'delete_alarm_request', '','alarm_request')
+ self.assertEqual(info, None)
+ return
+
+ #To generate a request of testing valid operation in create alarm request
+ def test_operationValid(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/operation_valid.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid_delete3.json",'delete_alarm_request', '','alarm_request')
+ self.assertTrue(info['alarm_create_response']['status'])
+ return
+
+ #To generate a request of testing Invalid operation in create alarm request
+ def test_operationValidNot(self):
+ time.sleep(2)
+ producer.request("test_schemas/create_alarm/operation_invalid.json",'create_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "create_alarm_response":
+ info = json.loads(json.loads(message.value))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(info,None)
+ return
+
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+class update_alarm_name_test(unittest.TestCase):
+
+ #To generate a request of testing valid alarm_id in update alarm request
+ def test_nameValid(self):
+ producer.request("test_schemas/update_alarm/update_alarm_new_alarm.json",'create_alarm_request', '','alarm_request')
+ time.sleep(2)
+ producer.request("test_schemas/update_alarm/name_valid.json",'update_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "update_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid_delete4.json",'delete_alarm_request', '','alarm_request')
+ self.assertTrue(info['alarm_update_response']['status'])
+ return
+
+ #To generate a request of testing invalid alarm_id in update alarm request
+ def test_nameInvalid(self):
+ time.sleep(2)
+ producer.request("test_schemas/update_alarm/name_invalid.json",'update_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "update_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(info,None)
+ return
+
+ #To generate a request of testing valid statistics in update alarm request
+ def test_statisticValid(self):
+ producer.request("test_schemas/create_alarm/create_alarm_differentName_differentInstance.json",'create_alarm_request', '','alarm_request')
+ time.sleep(2)
+ producer.request("test_schemas/update_alarm/statistic_valid.json",'update_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "update_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid.json",'delete_alarm_request', '','alarm_request')
+ self.assertTrue(info['alarm_update_response']['status'])
+ return
+
+ #To generate a request of testing Invalid statistics in update alarm request
+ def test_statisticInvalid(self):
+ time.sleep(2)
+ producer.request("test_schemas/update_alarm/statistic_invalid.json",'update_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "update_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(info,None)
+ return
+
+ #To generate a request of testing valid operation in update alarm request
+ def test_operationValid(self):
+ producer.request("test_schemas/create_alarm/create_alarm_differentName_differentInstance.json",'create_alarm_request', '','alarm_request')
+ time.sleep(2)
+ producer.request("test_schemas/update_alarm/operation_valid.json",'update_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "update_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ producer.request("test_schemas/delete_alarm/name_valid.json",'delete_alarm_request', '','alarm_request')
+ self.assertTrue(info['alarm_update_response']['status'])
+ return
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+class delete_alarm_test(unittest.TestCase):
+
+ #To generate a request of testing valid alarm_id in delete alarm request
+ def test_nameValid(self):
+ producer.request("test_schemas/create_alarm/create_alarm_differentName_differentInstance.json",'create_alarm_request', '','alarm_request')
+ time.sleep(2)
+ producer.request("test_schemas/delete_alarm/name_valid.json",'delete_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "delete_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertTrue(info['alarm_deletion_response']['status'])
+ return
+
+ #To generate a request of testing Invalid alarm_id in delete alarm request
+ def test_nameInvalid(self):
+ time.sleep(2)
+ producer.request("test_schemas/delete_alarm/name_invalid.json",'delete_alarm_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "delete_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(info,None)
+ return
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+class list_alarm_test(unittest.TestCase):
+
+ #To generate a request of testing valid input fields in alarm list request
+ def test_valid_no_arguments(self):
+ time.sleep(2)
+ producer.request("test_schemas/list_alarm/list_alarm_valid_no_arguments.json",'alarm_list_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "list_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(type(info),dict)
+ return
+
+ #To generate a request of testing valid input fields in alarm list request
+ def test_valid_one_arguments(self):
+ time.sleep(2)
+ producer.request("test_schemas/list_alarm/list_alarm_valid_one_arguments.json",'alarm_list_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "list_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(type(info),dict)
+ return
+
+ #To generate a request of testing valid input fields in alarm list request
+ def test_valid_two_arguments(self):
+ time.sleep(2)
+ producer.request("test_schemas/list_alarm/list_alarm_valid_two_arguments.json",'alarm_list_request', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "list_alarm_response":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(type(info),dict)
+ return
+
+
+#--------------------------------------------------------------------------------------------------------------------------------------
+class alarm_details_test(unittest.TestCase):
+
+ #To generate a request of testing valid input fields in acknowledge alarm
+ def test_Valid(self):
+ time.sleep(2)
+ producer.request("test_schemas/alarm_details/acknowledge_alarm.json",'acknowledge_alarm', '','alarm_request')
+ server = {'server': 'localhost:9092', 'topic': 'alarm_request'}
+
+ _consumer = KafkaConsumer(bootstrap_servers=server['server'])
+ _consumer.subscribe(['alarm_response'])
+
+ for message in _consumer:
+ if message.key == "notify_alarm":
+ info = json.loads(json.loads(json.loads(message.value)))
+ print(info)
+ time.sleep(1)
+ self.assertEqual(type(info),dict)
+ 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