@@ -0,0 +1,406 @@
1+# SPDX-License-Identifier: AGPL-3.0-or-later
2+
3+import typing
4+import types
5+import functools
6+import itertools
7+from time import time
8+from urllib.parse import urlparse
9+
10+import re
11+import cld3
12+import requests.exceptions
13+
14+from searx import poolrequests, logger
15+from searx.results import ResultContainer
16+from searx.search.models import SearchQuery, EngineRef
17+from searx.search.processors import EngineProcessor
18+
19+
20+logger = logger.getChild('searx.search.checker')
21+
22+HTML_TAGS = [
23+ 'embed', 'iframe', 'object', 'param', 'picture', 'source', 'svg', 'math', 'canvas', 'noscript', 'script',
24+ 'del', 'ins', 'area', 'audio', 'img', 'map', 'track', 'video', 'a', 'abbr', 'b', 'bdi', 'bdo', 'br', 'cite',
25+ 'code', 'data', 'dfn', 'em', 'i', 'kdb', 'mark', 'q', 'rb', 'rp', 'rt', 'rtc', 'ruby', 's', 'samp', 'small',
26+ 'span', 'strong', 'sub', 'sup', 'time', 'u', 'var', 'wbr', 'style', 'blockquote', 'dd', 'div', 'dl', 'dt',
27+ 'figcaption', 'figure', 'hr', 'li', 'ol', 'p', 'pre', 'ul', 'button', 'datalist', 'fieldset', 'form', 'input',
28+ 'label', 'legend', 'meter', 'optgroup', 'option', 'output', 'progress', 'select', 'textarea', 'applet',
29+ 'frame', 'frameset'
30+]
31+
32+
33+def get_check_no_html():
34+ rep = ['<' + tag + '[^\>]*>' for tag in HTML_TAGS]
35+ rep += ['</' + tag + '>' for tag in HTML_TAGS]
36+ pattern = re.compile('|'.join(rep))
37+
38+ def f(text):
39+ return pattern.search(text.lower()) is None
40+
41+ return f
42+
43+
44+_check_no_html = get_check_no_html()
45+
46+
47+def _is_url(url):
48+ try:
49+ result = urlparse(url)
50+ except ValueError:
51+ return False
52+ if result.scheme not in ('http', 'https'):
53+ return False
54+ return True
55+
56+
57+@functools.lru_cache(maxsize=8192)
58+def _is_url_image(image_url):
59+ if not isinstance(image_url, str):
60+ return False
61+
62+ if image_url.startswith('//'):
63+ image_url = 'https:' + image_url
64+
65+ if image_url.startswith('data:'):
66+ return image_url.startswith('data:image/')
67+
68+ if not _is_url(image_url):
69+ return False
70+
71+ retry = 2
72+
73+ while retry > 0:
74+ a = time()
75+ try:
76+ poolrequests.set_timeout_for_thread(10.0, time())
77+ r = poolrequests.get(image_url, timeout=10.0, allow_redirects=True, headers={
78+ 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:84.0) Gecko/20100101 Firefox/84.0',
79+ 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
80+ 'Accept-Language': 'en-US;q=0.5,en;q=0.3',
81+ 'Accept-Encoding': 'gzip, deflate, br',
82+ 'DNT': '1',
83+ 'Connection': 'keep-alive',
84+ 'Upgrade-Insecure-Requests': '1',
85+ 'Sec-GPC': '1',
86+ 'Cache-Control': 'max-age=0'
87+ })
88+ if r.headers["content-type"].startswith('image/'):
89+ return True
90+ return False
91+ except requests.exceptions.Timeout:
92+ logger.error('Timeout for %s: %i', image_url, int(time() - a))
93+ retry -= 1
94+ except requests.exceptions.RequestException:
95+ logger.exception('Exception for %s', image_url)
96+ return False
97+
98+
99+def _search_query_to_dict(search_query: SearchQuery) -> typing.Dict[str, typing.Any]:
100+ return {
101+ 'query': search_query.query,
102+ 'lang': search_query.lang,
103+ 'pageno': search_query.pageno,
104+ 'safesearch': search_query.safesearch,
105+ 'time_range': search_query.time_range,
106+ }
107+
108+
109+def _search_query_diff(sq1: SearchQuery, sq2: SearchQuery)\
110+ -> typing.Tuple[typing.Dict[str, typing.Any], typing.Dict[str, typing.Any]]:
111+ param1 = _search_query_to_dict(sq1)
112+ param2 = _search_query_to_dict(sq2)
113+ common = {}
114+ diff = {}
115+ for k, value1 in param1.items():
116+ value2 = param2[k]
117+ if value1 == value2:
118+ common[k] = value1
119+ else:
120+ diff[k] = (value1, value2)
121+ return (common, diff)
122+
123+
124+class TestResults:
125+
126+ __slots__ = 'errors', 'logs', 'languages'
127+
128+ def __init__(self):
129+ self.errors: typing.Dict[str, typing.List[str]] = {}
130+ self.logs: typing.Dict[str, typing.List[typing.Any]] = {}
131+ self.languages: typing.Set[str] = set()
132+
133+ def add_error(self, test, message, *args):
134+ # message to self.errors
135+ errors_for_test = self.errors.setdefault(test, [])
136+ if message not in errors_for_test:
137+ errors_for_test.append(message)
138+ # (message, *args) to self.logs
139+ logs_for_test = self.logs.setdefault(test, [])
140+ if (message, *args) not in logs_for_test:
141+ logs_for_test.append((message, *args))
142+
143+ def add_language(self, language):
144+ self.languages.add(language)
145+
146+ @property
147+ def succesfull(self):
148+ return len(self.errors) == 0
149+
150+ def __iter__(self):
151+ for test_name, errors in self.errors.items():
152+ for error in sorted(errors):
153+ yield (test_name, error)
154+
155+
156+class ResultContainerTests:
157+
158+ __slots__ = 'test_name', 'search_query', 'result_container', 'languages', 'stop_test', 'test_results'
159+
160+ def __init__(self,
161+ test_results: TestResults,
162+ test_name: str,
163+ search_query: SearchQuery,
164+ result_container: ResultContainer):
165+ self.test_name = test_name
166+ self.search_query = search_query
167+ self.result_container = result_container
168+ self.languages: typing.Set[str] = set()
169+ self.test_results = test_results
170+ self.stop_test = False
171+
172+ @property
173+ def result_urls(self):
174+ results = self.result_container.get_ordered_results()
175+ return [result['url'] for result in results]
176+
177+ def _record_error(self, message: str, *args) -> None:
178+ sq = _search_query_to_dict(self.search_query)
179+ sqstr = ' '.join(['{}={!r}'.format(k, v) for k, v in sq.items()])
180+ self.test_results.add_error(self.test_name, message, *args, '(' + sqstr + ')')
181+
182+ def _add_language(self, text: str) -> typing.Optional[str]:
183+ r = cld3.get_language(str(text)) # pylint: disable=E1101
184+ if r is not None and r.probability >= 0.98 and r.is_reliable:
185+ self.languages.add(r.language)
186+ self.test_results.add_language(r.language)
187+ return None
188+
189+ def _check_result(self, result):
190+ if not _check_no_html(result.get('title', '')):
191+ self._record_error('HTML in title', repr(result.get('title', '')))
192+ if not _check_no_html(result.get('content', '')):
193+ self._record_error('HTML in content', repr(result.get('content', '')))
194+
195+ self._add_language(result.get('title', ''))
196+ self._add_language(result.get('content', ''))
197+
198+ template = result.get('template', 'default.html')
199+ if template == 'default.html':
200+ return
201+ if template == 'code.html':
202+ return
203+ if template == 'torrent.html':
204+ return
205+ if template == 'map.html':
206+ return
207+ if template == 'images.html':
208+ thumbnail_src = result.get('thumbnail_src')
209+ if thumbnail_src is not None:
210+ if not _is_url_image(thumbnail_src):
211+ self._record_error('thumbnail_src URL is invalid', thumbnail_src)
212+ elif not _is_url_image(result.get('img_src')):
213+ self._record_error('img_src URL is invalid', result.get('img_src'))
214+ if template == 'videos.html' and not _is_url_image(result.get('thumbnail')):
215+ self._record_error('thumbnail URL is invalid', result.get('img_src'))
216+
217+ def _check_results(self, results: list):
218+ for result in results:
219+ self._check_result(result)
220+
221+ def _check_answers(self, answers):
222+ for answer in answers:
223+ if not _check_no_html(answer):
224+ self._record_error('HTML in answer', answer)
225+
226+ def _check_infoboxes(self, infoboxes):
227+ for infobox in infoboxes:
228+ if not _check_no_html(infobox.get('content', '')):
229+ self._record_error('HTML in infobox content', infobox.get('content', ''))
230+ self._add_language(infobox.get('content', ''))
231+ for attribute in infobox.get('attributes', {}):
232+ if not _check_no_html(attribute.get('value', '')):
233+ self._record_error('HTML in infobox attribute value', attribute.get('value', ''))
234+
235+ def check_basic(self):
236+ if len(self.result_container.unresponsive_engines) > 0:
237+ for message in self.result_container.unresponsive_engines:
238+ self._record_error(message[1] + ' ' + (message[2] or ''))
239+ self.stop_test = True
240+ return
241+
242+ results = self.result_container.get_ordered_results()
243+ if len(results) > 0:
244+ self._check_results(results)
245+
246+ if len(self.result_container.answers) > 0:
247+ self._check_answers(self.result_container.answers)
248+
249+ if len(self.result_container.infoboxes) > 0:
250+ self._check_infoboxes(self.result_container.infoboxes)
251+
252+ def has_infobox(self):
253+ """Check the ResultContainer has at least one infobox"""
254+ if len(self.result_container.infoboxes) == 0:
255+ self._record_error('No infobox')
256+
257+ def has_answer(self):
258+ """Check the ResultContainer has at least one answer"""
259+ if len(self.result_container.answers) == 0:
260+ self._record_error('No answer')
261+
262+ def has_language(self, lang):
263+ """Check at least one title or content of the results is written in the `lang`.
264+
265+ Detected using pycld3, may be not accurate"""
266+ if lang not in self.languages:
267+ self._record_error(lang + ' not found')
268+
269+ def not_empty(self):
270+ """Check the ResultContainer has at least one answer or infobox or result"""
271+ result_types = set()
272+ results = self.result_container.get_ordered_results()
273+ if len(results) > 0:
274+ result_types.add('results')
275+
276+ if len(self.result_container.answers) > 0:
277+ result_types.add('answers')
278+
279+ if len(self.result_container.infoboxes) > 0:
280+ result_types.add('infoboxes')
281+
282+ if len(result_types) == 0:
283+ self._record_error('No result')
284+
285+ def one_title_contains(self, title: str):
286+ """Check one of the title contains `title` (case insensitive comparaison)"""
287+ title = title.lower()
288+ for result in self.result_container.get_ordered_results():
289+ if title in result['title'].lower():
290+ return
291+ self._record_error(('{!r} not found in the title'.format(title)))
292+
293+
294+class CheckerTests:
295+
296+ __slots__ = 'test_results', 'test_name', 'result_container_tests_list'
297+
298+ def __init__(self,
299+ test_results: TestResults,
300+ test_name: str,
301+ result_container_tests_list: typing.List[ResultContainerTests]):
302+ self.test_results = test_results
303+ self.test_name = test_name
304+ self.result_container_tests_list = result_container_tests_list
305+
306+ def unique_results(self):
307+ """Check the results of each ResultContain is unique"""
308+ urls_list = [rct.result_urls for rct in self.result_container_tests_list]
309+ if len(urls_list[0]) > 0:
310+ # results on the first page
311+ for i, urls_i in enumerate(urls_list):
312+ for j, urls_j in enumerate(urls_list):
313+ if i < j and urls_i == urls_j:
314+ common, diff = _search_query_diff(self.result_container_tests_list[i].search_query,
315+ self.result_container_tests_list[j].search_query)
316+ common_str = ' '.join(['{}={!r}'.format(k, v) for k, v in common.items()])
317+ diff1_str = ', ' .join(['{}={!r}'.format(k, v1) for (k, (v1, v2)) in diff.items()])
318+ diff2_str = ', ' .join(['{}={!r}'.format(k, v2) for (k, (v1, v2)) in diff.items()])
319+ self.test_results.add_error(self.test_name,
320+ 'results are identitical for {} and {} ({})'
321+ .format(diff1_str, diff2_str, common_str))
322+
323+
324+class Checker:
325+
326+ __slots__ = 'processor', 'tests', 'test_results'
327+
328+ def __init__(self, processor: EngineProcessor):
329+ self.processor = processor
330+ self.tests = self.processor.get_tests()
331+ self.test_results = TestResults()
332+
333+ @property
334+ def engineref_list(self):
335+ engine_name = self.processor.engine_name
336+ engine_category = self.processor.engine.categories[0]
337+ return [EngineRef(engine_name, engine_category)]
338+
339+ @staticmethod
340+ def search_query_matrix_iterator(engineref_list, matrix):
341+ p = []
342+ for name, values in matrix.items():
343+ if isinstance(values, (tuple, list)):
344+ l = [(name, value) for value in values]
345+ else:
346+ l = [(name, values)]
347+ p.append(l)
348+
349+ for kwargs in itertools.product(*p):
350+ kwargs = {k: v for k, v in kwargs}
351+ query = kwargs['query']
352+ params = dict(kwargs)
353+ del params['query']
354+ yield SearchQuery(query, engineref_list, **params)
355+
356+ def call_test(self, obj, test_description):
357+ if isinstance(test_description, (tuple, list)):
358+ method, args = test_description[0], test_description[1:]
359+ else:
360+ method = test_description
361+ args = ()
362+ if isinstance(method, str) and hasattr(obj, method):
363+ getattr(obj, method)(*args)
364+ elif isinstance(method, types.FunctionType):
365+ method(*args)
366+ else:
367+ self.test_results.add_error(obj.test_name,
368+ 'method {!r} ({}) not found for {}'
369+ .format(method, method.__class__.__name__, obj.__class__.__name__))
370+
371+ def call_tests(self, obj, test_descriptions):
372+ for test_description in test_descriptions:
373+ self.call_test(obj, test_description)
374+
375+ def search(self, search_query: SearchQuery) -> ResultContainer:
376+ result_container = ResultContainer()
377+ engineref_category = search_query.engineref_list[0].category
378+ params = self.processor.get_params(search_query, engineref_category)
379+ if params is not None:
380+ self.processor.search(search_query.query, params, result_container, time(), 5)
381+ return result_container
382+
383+ def get_result_container_tests(self, test_name: str, search_query: SearchQuery) -> ResultContainerTests:
384+ result_container = self.search(search_query)
385+ result_container_check = ResultContainerTests(self.test_results, test_name, search_query, result_container)
386+ result_container_check.check_basic()
387+ return result_container_check
388+
389+ def run_test(self, test_name):
390+ test_parameters = self.tests[test_name]
391+ search_query_list = list(Checker.search_query_matrix_iterator(self.engineref_list, test_parameters['matrix']))
392+ rct_list = [self.get_result_container_tests(test_name, search_query) for search_query in search_query_list]
393+ stop_test = False
394+ if 'result_container' in test_parameters:
395+ for rct in rct_list:
396+ stop_test = stop_test or rct.stop_test
397+ if not rct.stop_test:
398+ self.call_tests(rct, test_parameters['result_container'])
399+ if not stop_test:
400+ if 'test' in test_parameters:
401+ checker_tests = CheckerTests(self.test_results, test_name, rct_list)
402+ self.call_tests(checker_tests, test_parameters['test'])
403+
404+ def run(self):
405+ for test_name in self.tests:
406+ self.run_test(test_name)