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