(self, request, tenant_id, entity_id)
return json.dumps(response_payload)
Test an alarm.

        This API can be driven using the control API to set an error
        or canned success response. If no error or success response is set,
        it will return success with a random state and status. Users should
        not expect this API to consistently return either OK, WARNING or
        CRITICAL without first setting the response in the control API.
Test an alarm.
Test an alarm .
def test_alarm(self, request, tenant_id, entity_id):
        """
        Test an alarm.

        This API can be driven using the control API to set an error
        or canned success response. If no error or success response is set,
        it will return success with a random state and status. Users should
        not expect this API to consistently return either OK, WARNING or
        CRITICAL without first setting the response in the control API.
        """
        content = request.content.read()
        payload = json.loads(content.decode("utf-8"))
        n_tests = len(payload['check_data'])
        current_time_milliseconds = int(
            1000 * self._session_store.clock.seconds())
        status = 200
        response_payload = []

        test_responses = self._entity_cache_for_tenant(
            tenant_id).test_alarm_responses
        test_errors = self._entity_cache_for_tenant(
            tenant_id).test_alarm_errors

        if entity_id in test_errors and len(test_errors[entity_id]) > 0:
            error_response = test_errors[entity_id].popleft()
            status = error_response['code']
            response_payload = error_response['response']
        elif entity_id in test_responses:
            n_responses = len(test_responses[entity_id])
            for i in range(n_tests):
                test_response = test_responses[entity_id][i % n_responses]
                response_payload.append({'state': test_response['state'],
                                         'status': test_response.get(
                                             'status', 'Matched default return statement'),
                                         'timestamp': current_time_milliseconds})
        else:
            for _ in range(n_tests):
                response_payload.append({'state': random.choice(['OK', 'WARNING', 'CRITICAL']),
                                         'status': random_hipsum(12),
                                         'timestamp': current_time_milliseconds})

        request.setResponseCode(status)
        self._audit('alarms', request, tenant_id, status, content)
        return json.dumps(response_payload)
def test_alarm ( self , request , tenant_id , entity_id ) : content = request . content . read ( ) payload = json . loads ( content . decode ( "utf-8" ) ) n_tests = len ( payload [ 'check_data' ] ) current_time_milliseconds = int ( 1000 * self . _session_store . clock . seconds ( ) ) status = 200 response_payload = [ ] test_responses = self . _entity_cache_for_tenant ( tenant_id ) . test_alarm_responses test_errors = self . _entity_cache_for_tenant ( tenant_id ) . test_alarm_errors if entity_id in test_errors and len ( test_errors [ entity_id ] ) > 0 : error_response = test_errors [ entity_id ] . popleft ( ) status = error_response [ 'code' ] response_payload = error_response [ 'response' ] elif entity_id in test_responses : n_responses = len ( test_responses [ entity_id ] ) for i in range ( n_tests ) : test_response = test_responses [ entity_id ] [ i % n_responses ] response_payload . append ( { 'state' : test_response [ 'state' ] , 'status' : test_response . get ( 'status' , 'Matched default return statement' ) , 'timestamp' : current_time_milliseconds } ) else : for _ in range ( n_tests ) : response_payload . append ( { 'state' : random . choice ( [ 'OK' , 'WARNING' , 'CRITICAL' ] ) , 'status' : random_hipsum ( 12 ) , 'timestamp' : current_time_milliseconds } ) request . setResponseCode ( status ) self . _audit ( 'alarms' , request , tenant_id , status , content ) return json . dumps ( response_payload )