python类exceptions()的实例源码

def tokenize(
        string, es_client,
        field_name=None, index_name=None, analyzer_name=None):
    """Tokenize a string based on analyzer of the provided field

    Args:
        string (string): the string to tokenize
        es_client (EsClient): elasticsearch client.
        field_name (string): the field whose analyzer is used to
            tokenize
        index_name (string): name of the index
    Returns:
        tokens (list): a list of tokens

    Raises:
        ElasticsearchClientError: if no field name or index name
            are available.
    """
    if field_name is None:
        field_name = es_client.field_name

    if index_name is None:
        index_name = es_client.index_name

    req = {'body': string, 'index': index_name}
    if analyzer_name is None:
        req['field'] = field_name
    else:
        req['analyzer'] = analyzer_name

    try:
        response = es_client.indices.analyze(**req)
        tokens = [d['token'] for d in response['tokens']]
    except elasticsearch.exceptions.RequestError:
        tokens = []

    return tokens

def test_recreate_index(self):
        """
        django.core.exceptions.ImproperlyConfigured:
            Model '<class 'tests.test_app.models.MockModel'>' has more than one 'SearchIndex`` handling it.
            Please exclude either '<tests.test_app.search_indexes.ElasticsearchAutocompleteMockModelSearchIndex object at 0x10b7881c8>'
                or
            '<tests.test_app.search_indexes.ElasticsearchComplexFacetsMockSearchIndex object at 0x10b788228>'
            using the 'EXCLUDED_INDEXES' setting defined in 'settings.HAYSTACK_CONNECTIONS'.
        """

        clear_elasticsearch_index()

        search_backend = connections['default'].get_backend()
        search_backend.silently_fail = True
        search_backend.setup()

        original_mapping = self.raw_es.indices.get_mapping(index=search_backend.index_name)

        search_backend.clear()
        search_backend.setup()

        try:
            updated_mapping = self.raw_es.indices.get_mapping(search_backend.index_name)
        except elasticsearch.NotFoundError:
            self.fail("There is no mapping after recreating the index")

        self.assertEqual(original_mapping, updated_mapping,
                         "Mapping after recreating the index differs from the original one")

def test_agg_no_writeback_connectivity(ea):
    """ Tests that if writeback_es throws an exception, the matches will be added to 'agg_matches' and when
    run again, that they will be passed again to add_aggregated_alert """
    hit1, hit2, hit3 = '2014-09-26T12:34:45', '2014-09-26T12:40:45', '2014-09-26T12:47:45'
    hits = generate_hits([hit1, hit2, hit3])
    ea.current_es.search.return_value = hits
    ea.rules[0]['aggregation'] = datetime.timedelta(minutes=10)
    ea.rules[0]['type'].matches = [{'@timestamp': hit1},
                                   {'@timestamp': hit2},
                                   {'@timestamp': hit3}]
    ea.writeback_es.create.side_effect = elasticsearch.exceptions.ElasticsearchException('Nope')
    with mock.patch('elastalert.elastalert.elasticsearch_client'):
        with mock.patch.object(ea, 'find_pending_aggregate_alert', return_value=None):
            ea.run_rule(ea.rules[0], END, START)

    assert ea.rules[0]['agg_matches'] == [{'@timestamp': hit1, 'num_hits': 0, 'num_matches': 3},
                                          {'@timestamp': hit2, 'num_hits': 0, 'num_matches': 3},
                                          {'@timestamp': hit3, 'num_hits': 0, 'num_matches': 3}]

    ea.current_es.search.return_value = {'hits': {'total': 0, 'hits': []}}
    ea.add_aggregated_alert = mock.Mock()

    with mock.patch('elastalert.elastalert.elasticsearch_client'):
        ea.run_rule(ea.rules[0], END, START)

    ea.add_aggregated_alert.assert_any_call({'@timestamp': hit1, 'num_hits': 0, 'num_matches': 3}, ea.rules[0])
    ea.add_aggregated_alert.assert_any_call({'@timestamp': hit2, 'num_hits': 0, 'num_matches': 3}, ea.rules[0])
    ea.add_aggregated_alert.assert_any_call({'@timestamp': hit3, 'num_hits': 0, 'num_matches': 3}, ea.rules[0])