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.thread_data.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.index.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.thread_data.current_es.search.return_value = {'hits': {'total': 0, 'hits': []}}
    ea.add_aggregated_alert = mock.Mock()

    with mock.patch.object(ea, 'run_query'):
        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]) 

def test_trace_error_not_found(self, request_mock):
        msg = "record not found"
        exc = elasticsearch.exceptions.NotFoundError(404, msg)
        request_mock.return_value = (1, {}, {})
        request_mock.side_effect = exc
        self._test_trace_error(StatusCanonicalCode.NOT_FOUND, exc) 

def test_search_arg_named_index(elasticsearch_client, tracked_request):
    with pytest.raises(elasticsearch.exceptions.NotFoundError):
        # body, index
        elasticsearch_client.search(None, "myindex")

    assert len(tracked_request.complete_spans) == 1
    span = tracked_request.complete_spans[0]
    assert span.operation == "Elasticsearch/Myindex/Search" 

def test_search_kwarg_named_index(elasticsearch_client, tracked_request):
    with pytest.raises(elasticsearch.exceptions.NotFoundError):
        elasticsearch_client.search(index="myindex")

    assert len(tracked_request.complete_spans) == 1
    span = tracked_request.complete_spans[0]
    assert span.operation == "Elasticsearch/Myindex/Search" 

def test_search_kwarg_index_list(elasticsearch_client, tracked_request):
    with pytest.raises(elasticsearch.exceptions.NotFoundError):
        elasticsearch_client.search(index=["myindex", "myindex2"])

    assert len(tracked_request.complete_spans) == 1
    span = tracked_request.complete_spans[0]
    assert span.operation == "Elasticsearch/Myindex,Myindex2/Search" 

def _wrap_perform_request(tracer, span_name_prefix):
    def wrapper(wrapped, _, args, kwargs):
        method = url = None
        try:
            method, url, *_ = args
        except IndexError:
            logger.warning(
                "expected perform_request to receive two positional arguments. "
                "Got %d",
                len(args),
            )

        op_name = span_name_prefix + (url or method or _DEFALT_OP_NAME)
        params = kwargs.get("params", {})
        body = kwargs.get("body", None)

        attributes = {
            "component": "elasticsearch-py",
            "db.type": "elasticsearch",
        }

        if url:
            attributes["elasticsearch.url"] = url
        if method:
            attributes["elasticsearch.method"] = method
        if body:
            attributes["db.statement"] = str(body)
        if params:
            attributes["elasticsearch.params"] = str(params)

        with tracer.start_as_current_span(
            op_name, kind=SpanKind.CLIENT, attributes=attributes
        ) as span:
            try:
                rv = wrapped(*args, **kwargs)
                if isinstance(rv, dict):
                    for member in _ATTRIBUTES_FROM_RESULT:
                        if member in rv:
                            span.set_attribute(
                                "elasticsearch.{0}".format(member),
                                str(rv[member]),
                            )
                return rv
            except Exception as ex:  # pylint: disable=broad-except
                if isinstance(ex, elasticsearch.exceptions.NotFoundError):
                    status = StatusCanonicalCode.NOT_FOUND
                else:
                    status = StatusCanonicalCode.UNKNOWN
                span.set_status(Status(status, str(ex)))
                raise ex

    return wrapper