Skip to content
Snippets Groups Projects
Commit 3eb8f97d authored by Benjamin Becker's avatar Benjamin Becker
Browse files

adds unit tests for before_search

parent 1d9cca45
No related branches found
No related tags found
No related merge requests found
......@@ -119,7 +119,7 @@ def extend_search_convert_local_to_utc_timestamp(str_timestamp):
return None
if not re.match(r'\d\d\d\d-\d\d-\d\d', str_timestamp):
raise 'wrong format'
raise ValueError('wrong format')
dt = parser.parse(str_timestamp, dayfirst=False).isoformat()
......
......@@ -24,6 +24,7 @@ import ckanext.odsh.helper_pkg_dict as helper_pkg_dict
from helper_pkg_dict import HelperPgkDict
import ckanext.odsh.logic.action as action
import ckanext.odsh.validation as validation
import ckanext.odsh.search as search
from ckanext.odsh.odsh_logger import OdshLogger
......@@ -360,66 +361,7 @@ class OdshPlugin(plugins.SingletonPlugin, DefaultTranslation, DefaultDatasetForm
# use several daterange queries agains temporal_start and temporal_end field
# TODO: use field of type date_range in solr index instead
def before_search(self, search_params):
search_params['facet.mincount'] = 0
extras = search_params.get('extras')
if not extras:
# There are no extras in the search params, so do nothing.
return search_params
fq = search_params['fq']
start_date = None
end_date = None
try:
start_date = odsh_helpers.extend_search_convert_local_to_utc_timestamp(
extras.get('ext_startdate'))
end_date = odsh_helpers.extend_search_convert_local_to_utc_timestamp(
extras.get('ext_enddate'))
except:
return search_params
empty_range = start_date and end_date and start_date > end_date
if not start_date and not end_date:
return search_params
do_enclosing_query = True
if not start_date:
do_enclosing_query = False
start_date = '*'
if not end_date:
do_enclosing_query = False
end_date = '*'
start_query = '+extras_temporal_start:[{start_date} TO {end_date}]'.format(
start_date=start_date, end_date=end_date)
end_query = '+extras_temporal_end:[{start_date} TO {end_date}]'.format(
start_date=start_date, end_date=end_date)
enclosing_query = ''
if do_enclosing_query and not empty_range:
enclosing_query_start = 'extras_temporal_start:[* TO {start_date}]'.format(
start_date=start_date)
enclosing_query_end = 'extras_temporal_end:[{end_date} TO *]'.format(
end_date=end_date)
enclosing_query = ' OR ({enclosing_query_start} AND {enclosing_query_end})'.format(
enclosing_query_start=enclosing_query_start, enclosing_query_end=enclosing_query_end)
if end_date is '*':
open_end_query = '(*:* NOT extras_temporal_end:[* TO *])'
else:
open_end_query = '((*:* NOT extras_temporal_end:[* TO *]) AND extras_temporal_start:[* TO {end_date}])'.format(
end_date=end_date)
fq = u'{fq} ({start_query} OR {end_query} {enclosing_query} OR {open_end_query})'.format(
fq=fq, start_query=start_query, end_query=end_query, enclosing_query=enclosing_query, open_end_query=open_end_query)
search_params['fq'] = fq
return search_params
return search.before_search(search_params)
scores = [['0OL'], ['0OL', '1RE'], ['0OL', '1RE', '2OF'], [
'0OL', '1RE', '2OF', '3URI'], ['0OL', '1RE', '2OF', '3URI', '4LD']]
......
import ckanext.odsh.helpers as odsh_helpers
def before_search(search_params):
search_params['facet.mincount'] = 0
extras = search_params.get('extras')
if not extras:
# There are no extras in the search params, so do nothing.
return search_params
start_date = None
end_date = None
try:
start_date = odsh_helpers.extend_search_convert_local_to_utc_timestamp(
extras.get('ext_startdate'))
end_date = odsh_helpers.extend_search_convert_local_to_utc_timestamp(
extras.get('ext_enddate'))
except ValueError:
return search_params
empty_range = start_date and end_date and start_date > end_date
if not start_date and not end_date:
return search_params
do_enclosing_query = True
if not start_date:
do_enclosing_query = False
start_date = '*'
if not end_date:
do_enclosing_query = False
end_date = '*'
start_query = '+extras_temporal_start:[{start_date} TO {end_date}]'.format(
start_date=start_date, end_date=end_date)
end_query = '+extras_temporal_end:[{start_date} TO {end_date}]'.format(
start_date=start_date, end_date=end_date)
enclosing_query = ''
if do_enclosing_query and not empty_range:
enclosing_query_start = 'extras_temporal_start:[* TO {start_date}]'.format(
start_date=start_date)
enclosing_query_end = 'extras_temporal_end:[{end_date} TO *]'.format(
end_date=end_date)
enclosing_query = ' OR ({enclosing_query_start} AND {enclosing_query_end})'.format(
enclosing_query_start=enclosing_query_start, enclosing_query_end=enclosing_query_end)
if end_date is '*':
open_end_query = '(*:* NOT extras_temporal_end:[* TO *])'
else:
open_end_query = '((*:* NOT extras_temporal_end:[* TO *]) AND extras_temporal_start:[* TO {end_date}])'.format(
end_date=end_date)
fq_preset = search_params.get('fq')
fq = u'{fq_preset} ({start_query} OR {end_query} {enclosing_query} OR {open_end_query})'.format(
fq_preset=fq_preset,
start_query=start_query,
end_query=end_query,
enclosing_query=enclosing_query,
open_end_query=open_end_query
)
search_params['fq'] = fq
return search_params
\ No newline at end of file
import nose.tools as nt
import ckanext.odsh.search as search
class Test_before_search(object):
def setUp(self):
self.search_params_before_test = {
'extras': {},
'facet.field': ['organization', 'subject_text', 'groups'],
'fq': u'organization:"test-organisation" groups:"gove" subject_text:"T\xe4tigkeitsbericht" +dataset_type:dataset',
'include_private': True,
'q': u'',
'rows': 20,
'sort': u'score desc, metadata_modified desc',
'start': 0
}
self.search_params_with_facet_mincount = self.search_params_before_test.copy()
self.search_params_with_facet_mincount.update({'facet.mincount': 0})
def test_it_solely_adds_facet_mincount_to_dict_if_no_extras(self):
# arange
search_params = self.search_params_before_test.copy()
# act
search.before_search(search_params)
# assert
search_params_expected = self.search_params_with_facet_mincount.copy()
nt.assert_equal(search_params, search_params_expected)
def test_it_adds_fq_if_empty_range(self):
# arange
search_params = self.search_params_before_test.copy()
extras = {'ext_enddate': u'2019-08-01', 'ext_startdate': u'2019-08-02'}
search_params.update({'extras': extras})
search_params_expected = self.search_params_with_facet_mincount.copy()
search_params_expected.update({'extras': extras})
search_params_expected.update({
'fq': (
u'organization:"test-organisation" groups:"gove" subject_text:"T\xe4tigkeitsbericht" '
u'+dataset_type:dataset (+extras_temporal_start:[2019-08-02T00:00:00Z TO 2019-08-01T00:00:00Z] '
u'OR +extras_temporal_end:[2019-08-02T00:00:00Z TO 2019-08-01T00:00:00Z] OR '
u'((*:* NOT extras_temporal_end:[* TO *]) AND extras_temporal_start:[* TO 2019-08-01T00:00:00Z]))'
)
})
# act
search.before_search(search_params)
# assert
nt.assert_equal(search_params, search_params_expected)
def test_it_solely_adds_facet_mincount_to_dict_if_wrong_date_format_in_extras(self):
# arange
search_params = self.search_params_before_test.copy()
extras = {'ext_enddate': u'some_date', 'ext_startdate': u'some_date'}
search_params.update({'extras': extras})
search_params_expected = self.search_params_with_facet_mincount.copy()
search_params_expected.update({'extras': extras})
# act
search.before_search(search_params)
# assert
nt.assert_equal(search_params, search_params_expected)
def test_it_adds_fq_if_enclosing_range(self):
# arange
search_params = self.search_params_before_test.copy()
extras = {'ext_enddate': u'2019-08-02', 'ext_startdate': u'2019-08-01'}
search_params.update({'extras': extras})
search_params_expected = self.search_params_with_facet_mincount.copy()
search_params_expected.update({'extras': extras})
search_params_expected.update({
'fq': (
u'organization:"test-organisation" groups:"gove" '
u'subject_text:"T\xe4tigkeitsbericht" +dataset_type:dataset '
u'(+extras_temporal_start:[2019-08-01T00:00:00Z TO 2019-08-02T00:00:00Z] '
u'OR +extras_temporal_end:[2019-08-01T00:00:00Z TO 2019-08-02T00:00:00Z] '
u'OR (extras_temporal_start:[* TO 2019-08-01T00:00:00Z] AND '
u'extras_temporal_end:[2019-08-02T00:00:00Z TO *]) OR '
u'((*:* NOT extras_temporal_end:[* TO *]) AND '
u'extras_temporal_start:[* TO 2019-08-02T00:00:00Z]))'
)
})
# act
search.before_search(search_params)
# assert
nt.assert_equal(search_params, search_params_expected)
def test_it_adds_fq_if_start_only(self):
# arange
search_params = self.search_params_before_test.copy()
extras = {'ext_startdate': u'2019-08-01'}
search_params.update({'extras': extras})
search_params_expected = self.search_params_with_facet_mincount.copy()
search_params_expected.update({'extras': extras})
search_params_expected.update({
'fq': (
u'organization:"test-organisation" groups:"gove" '
u'subject_text:"T\xe4tigkeitsbericht" +dataset_type:dataset '
u'(+extras_temporal_start:[2019-08-01T00:00:00Z TO *] '
u'OR +extras_temporal_end:[2019-08-01T00:00:00Z TO *] '
u'OR (*:* NOT extras_temporal_end:[* TO *]))'
)
})
# act
search.before_search(search_params)
# assert
nt.assert_equal(search_params, search_params_expected)
def test_it_adds_fq_if_end_only(self):
# arange
search_params = self.search_params_before_test.copy()
extras = {'ext_enddate': u'2019-08-02'}
search_params.update({'extras': extras})
search_params_expected = self.search_params_with_facet_mincount.copy()
search_params_expected.update({'extras': extras})
search_params_expected.update({
'fq': (
u'organization:"test-organisation" groups:"gove" '
u'subject_text:"T\xe4tigkeitsbericht" +dataset_type:dataset '
u'(+extras_temporal_start:[* TO 2019-08-02T00:00:00Z] OR '
u'+extras_temporal_end:[* TO 2019-08-02T00:00:00Z] '
u'OR ((*:* NOT extras_temporal_end:[* TO *]) '
u'AND extras_temporal_start:[* TO 2019-08-02T00:00:00Z]))'
)
})
# act
search.before_search(search_params)
# assert
nt.assert_equal(search_params, search_params_expected)
def test_it_returns_search_params(self):
# arange
search_params = self.search_params_before_test.copy()
# act
return_value = search.before_search(search_params)
# assert
nt.assert_equal(return_value, search_params)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment