- def _set_headers(self):
- """Mock getting the request headers."""
- pass
-
- def do_GET(self):
- """Mock functionality for GET request."""
- self._set_headers()
- pass
-
- def do_POST(self):
- """Mock functionality for a POST request."""
- self._set_headers()
- self.notify_alarm(json.loads(post_data))
-
- def notify_alarm(self, values):
- """Mock the notify_alarm functionality to generate a valid response."""
- config = Config.instance()
- config.read_environ()
- self._alarming = Alarming()
- self._common = Common()
- self._response = OpenStack_Response()
- self._producer = KafkaProducer('alarm_response')
- alarm_id = values['alarm_id']
-
- vim_uuid = 'test_id'
-
- auth_token = Common.get_auth_token(vim_uuid)
- endpoint = Common.get_endpoint("alarming", vim_uuid)
-
- # If authenticated generate and send response message
- if auth_token is not None and endpoint is not None:
- url = "{}/v2/alarms/%s".format(endpoint) % alarm_id
-
- # Get the resource_id of the triggered alarm and the date
- result = Common.perform_request(
- url, auth_token, req_type="get")
- alarm_details = json.loads(result.text)
- gnocchi_rule = alarm_details['gnocchi_resources_threshold_rule']
- resource_id = gnocchi_rule['resource_id']
- a_date = "dd-mm-yyyy 00:00"
-
- # Process an alarm notification if resource_id is valid
- if resource_id is not None:
- # Try generate and send response
- try:
- resp_message = self._response.generate_response(
- 'notify_alarm', a_id=alarm_id,
- r_id=resource_id,
- sev=values['severity'], date=a_date,
- state=values['current'], vim_type="OpenStack")
- self._producer.notify_alarm(
- 'notify_alarm', resp_message)
- except Exception:
- pass
-