# Path: elastalert/kibana.py
# def add_filter(dashboard, es_filter):
#     next_id = max(dashboard['services']['filter']['ids']) + 1
# 
#     kibana_filter = {'active': True,
#                      'alias': '',
#                      'id': next_id,
#                      'mandate': 'must'}
# 
#     if 'not' in es_filter:
#         es_filter = es_filter['not']
#         kibana_filter['mandate'] = 'mustNot'
# 
#     if 'query' in es_filter:
#         es_filter = es_filter['query']
#         if 'query_string' in es_filter:
#             kibana_filter['type'] = 'querystring'
#             kibana_filter['query'] = es_filter['query_string']['query']
#     elif 'term' in es_filter:
#         kibana_filter['type'] = 'field'
#         f_field, f_query = list(es_filter['term'].items())[0]
#         # Wrap query in quotes, otherwise certain characters cause Kibana to throw errors
#         if isinstance(f_query, str):
#             f_query = '"%s"' % (f_query.replace('"', '\\"'))
#         if isinstance(f_query, list):
#             # Escape quotes
#             f_query = [item.replace('"', '\\"') for item in f_query]
#             # Wrap in quotes
#             f_query = ['"%s"' % (item) for item in f_query]
#             # Convert into joined query
#             f_query = '(%s)' % (' AND '.join(f_query))
#         kibana_filter['field'] = f_field
#         kibana_filter['query'] = f_query
#     elif 'range' in es_filter:
#         kibana_filter['type'] = 'range'
#         f_field, f_range = list(es_filter['range'].items())[0]
#         kibana_filter['field'] = f_field
#         kibana_filter.update(f_range)
#     else:
#         raise EAException("Could not parse filter %s for Kibana" % (es_filter))
# 
#     dashboard['services']['filter']['ids'].append(next_id)
#     dashboard['services']['filter']['list'][str(next_id)] = kibana_filter
# 
# Path: elastalert/kibana.py
# def set_time(dashboard, start, end):
# def set_index_name(dashboard, name):
# def set_timestamp_field(dashboard, field):
# def add_filter(dashboard, es_filter):
# def set_name(dashboard, name):
# def set_included_fields(dashboard, fields):
# def filters_from_dashboard(db):
# def kibana4_dashboard_link(dashboard, starttime, endtime):
# 
# Path: elastalert/kibana.py
# def filters_from_dashboard(db):
#     filters = db['services']['filter']['list']
#     config_filters = []
#     or_filters = []
#     for filter in list(filters.values()):
#         filter_type = filter['type']
#         if filter_type == 'time':
#             continue
# 
#         if filter_type == 'querystring':
#             config_filter = {'query': {'query_string': {'query': filter['query']}}}
# 
#         if filter_type == 'field':
#             config_filter = {'term': {filter['field']: filter['query']}}
# 
#         if filter_type == 'range':
#             config_filter = {'range': {filter['field']: {'from': filter['from'], 'to': filter['to']}}}
# 
#         if filter['mandate'] == 'mustNot':
#             config_filter = {'not': config_filter}
# 
#         if filter['mandate'] == 'either':
#             or_filters.append(config_filter)
#         else:
#             config_filters.append(config_filter)
# 
#     if or_filters:
#         config_filters.append({'or': or_filters})
# 
#     return config_filters
# 
# Path: elastalert/kibana.py
# def kibana4_dashboard_link(dashboard, starttime, endtime):
#     dashboard = os.path.expandvars(dashboard)
#     time_settings = kibana4_time_temp % (starttime, endtime)
#     time_settings = urllib.parse.quote(time_settings)
#     return "%s?_g=%s" % (dashboard, time_settings)


import copy
import json
from elastalert.kibana import add_filter
from elastalert.kibana import dashboard_temp
from elastalert.kibana import filters_from_dashboard
from elastalert.kibana import kibana4_dashboard_link
        2
      ]
    }
  }
}'''
test_dashboard = json.loads(test_dashboard)


def test_filters_from_dashboard():
    filters = filters_from_dashboard(test_dashboard)
    assert {'term': {'_log_type': '"active_directory"'}} in filters
    assert {'query': {'query_string': {'query': 'ad.security_auditing_code:4740'}}} in filters


def test_add_filter():
    basic_filter = {"term": {"this": "that"}}
    db = copy.deepcopy(dashboard_temp)
    add_filter(db, basic_filter)
    assert db['services']['filter']['list']['1'] == {
        'field': 'this',
        'alias': '',
        'mandate': 'must',
        'active': True,
        'query': '"that"',
        'type': 'field',
        'id': 1
    }

    list_filter = {"term": {"this": ["that", "those"]}}
    db = copy.deepcopy(dashboard_temp)

# Path: elastalert/kibana.py
# def add_filter(dashboard, es_filter):
#     next_id = max(dashboard['services']['filter']['ids']) + 1
# 
#     kibana_filter = {'active': True,
#                      'alias': '',
#                      'id': next_id,
#                      'mandate': 'must'}
# 
#     if 'not' in es_filter:
#         es_filter = es_filter['not']
#         kibana_filter['mandate'] = 'mustNot'
# 
#     if 'query' in es_filter:
#         es_filter = es_filter['query']
#         if 'query_string' in es_filter:
#             kibana_filter['type'] = 'querystring'
#             kibana_filter['query'] = es_filter['query_string']['query']
#     elif 'term' in es_filter:
#         kibana_filter['type'] = 'field'
#         f_field, f_query = list(es_filter['term'].items())[0]
#         # Wrap query in quotes, otherwise certain characters cause Kibana to throw errors
#         if isinstance(f_query, str):
#             f_query = '"%s"' % (f_query.replace('"', '\\"'))
#         if isinstance(f_query, list):
#             # Escape quotes
#             f_query = [item.replace('"', '\\"') for item in f_query]
#             # Wrap in quotes
#             f_query = ['"%s"' % (item) for item in f_query]
#             # Convert into joined query
#             f_query = '(%s)' % (' AND '.join(f_query))
#         kibana_filter['field'] = f_field
#         kibana_filter['query'] = f_query
#     elif 'range' in es_filter:
#         kibana_filter['type'] = 'range'
#         f_field, f_range = list(es_filter['range'].items())[0]
#         kibana_filter['field'] = f_field
#         kibana_filter.update(f_range)
#     else:
#         raise EAException("Could not parse filter %s for Kibana" % (es_filter))
# 
#     dashboard['services']['filter']['ids'].append(next_id)
#     dashboard['services']['filter']['list'][str(next_id)] = kibana_filter
# 
# Path: elastalert/kibana.py
# def set_time(dashboard, start, end):
# def set_index_name(dashboard, name):
# def set_timestamp_field(dashboard, field):
# def add_filter(dashboard, es_filter):
# def set_name(dashboard, name):
# def set_included_fields(dashboard, fields):
# def filters_from_dashboard(db):
# def kibana4_dashboard_link(dashboard, starttime, endtime):
# 
# Path: elastalert/kibana.py
# def filters_from_dashboard(db):
#     filters = db['services']['filter']['list']
#     config_filters = []
#     or_filters = []
#     for filter in list(filters.values()):
#         filter_type = filter['type']
#         if filter_type == 'time':
#             continue
# 
#         if filter_type == 'querystring':
#             config_filter = {'query': {'query_string': {'query': filter['query']}}}
# 
#         if filter_type == 'field':
#             config_filter = {'term': {filter['field']: filter['query']}}
# 
#         if filter_type == 'range':
#             config_filter = {'range': {filter['field']: {'from': filter['from'], 'to': filter['to']}}}
# 
#         if filter['mandate'] == 'mustNot':
#             config_filter = {'not': config_filter}
# 
#         if filter['mandate'] == 'either':
#             or_filters.append(config_filter)
#         else:
#             config_filters.append(config_filter)
# 
#     if or_filters:
#         config_filters.append({'or': or_filters})
# 
#     return config_filters
# 
# Path: elastalert/kibana.py
# def kibana4_dashboard_link(dashboard, starttime, endtime):
#     dashboard = os.path.expandvars(dashboard)
#     time_settings = kibana4_time_temp % (starttime, endtime)
#     time_settings = urllib.parse.quote(time_settings)
#     return "%s?_g=%s" % (dashboard, time_settings)


# Path: tests/kibana_test.py
import copy
import json
from elastalert.kibana import add_filter
from elastalert.kibana import dashboard_temp
from elastalert.kibana import filters_from_dashboard
from elastalert.kibana import kibana4_dashboard_link

        2
      ]
    }
  }
}'''
test_dashboard = json.loads(test_dashboard)


def test_filters_from_dashboard():
    filters = filters_from_dashboard(test_dashboard)
    assert {'term': {'_log_type': '"active_directory"'}} in filters
    assert {'query': {'query_string': {'query': 'ad.security_auditing_code:4740'}}} in filters


def test_add_filter():
    basic_filter = {"term": {"this": "that"}}
    db = copy.deepcopy(dashboard_temp)
    add_filter(db, basic_filter)
    assert db['services']['filter']['list']['1'] == {
        'field': 'this',
        'alias': '',
        'mandate': 'must',
        'active': True,
        'query': '"that"',
        'type': 'field',
        'id': 1
    }

    list_filter = {"term": {"this": ["that", "those"]}}
    db = copy.deepcopy(dashboard_temp)

    add_filter(db, list_filter)