85 lines
5.3 KiB
Python
85 lines
5.3 KiB
Python
import json
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
from search.api import Api, ApiError
|
||
|
||
|
||
class SearchApiTest(unittest.TestCase):
|
||
def make_api(self, root: Path) -> Api:
|
||
document = root / "documents/7"
|
||
edition = document / "editions/10"
|
||
edition.mkdir(parents=True)
|
||
(document / "document.json").write_text(json.dumps({
|
||
"source_code": "7", "available_languages": ["ru"],
|
||
"editions": [{"source_code": "10", "available_languages": ["ru"]},],
|
||
}), encoding="utf-8")
|
||
(edition / "edition.json").write_text(json.dumps({"source_code": "10", "available_languages": ["ru"]}), encoding="utf-8")
|
||
(edition / "ru").mkdir()
|
||
(edition / "ru/content.html").write_text("<p>Текст</p>", encoding="utf-8")
|
||
(edition / "ru/content.txt").write_text("Текст\n", encoding="utf-8")
|
||
return Api("http://opensearch:9200", "current", root)
|
||
|
||
def test_search_pagination_filters_and_highlight(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
api = self.make_api(Path(temporary))
|
||
response = {"hits": {"hits": [{"_source": {"document_code": "7", "edition_code": "10", "document_name_ru": "Закон"}, "highlight": {"text_ru": ["<em>Закон</em>"]}}]}}
|
||
with patch("search.api.request_json", return_value=response) as request:
|
||
status, payload = api.handle("GET", "/search?q=%D0%B7%D0%B0%D0%BA%D0%BE%D0%BD&language=ru&page=2&page_size=5&status=active")
|
||
self.assertEqual(status, 200)
|
||
self.assertEqual(payload["results"][0]["snippet"], "Закон")
|
||
body = json.loads(request.call_args.args[2])
|
||
self.assertEqual((body["from"], body["size"]), (5, 6))
|
||
self.assertIn({"term": {"status_code": "active"}}, body["query"]["bool"]["filter"])
|
||
with self.assertRaisesRegex(ApiError, "within 10000 results"):
|
||
api.handle("GET", "/search?q=x&page=100&page_size=100")
|
||
|
||
def test_production_api_uses_registration_intent_boosts(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
api = self.make_api(Path(temporary))
|
||
response = {"hits": {"hits": []}}
|
||
with patch("search.api.request_json", return_value=response) as request:
|
||
api.handle("GET", "/search?q=%D0%BA%D0%B0%D0%BA+%D0%BE%D1%82%D0%BA%D1%80%D1%8B%D1%82%D1%8C+%D0%9E%D1%81%D0%9E%D0%9E&language=ru")
|
||
body = json.loads(request.call_args.args[2])
|
||
planned = body["query"]["bool"]["must"]["bool"]
|
||
self.assertEqual(planned["minimum_should_match"], 1)
|
||
self.assertNotIn("filter", planned)
|
||
self.assertTrue(any("constant_score" in clause for clause in planned["should"]))
|
||
self.assertTrue(any("общество с ограниченной ответственностью" in clause.get("multi_match", {}).get("query", "") for clause in planned["should"]))
|
||
|
||
def test_document_editions_openapi_and_validation(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
api = self.make_api(Path(temporary))
|
||
specification = api.handle("GET", "/openapi.json")[1]
|
||
self.assertEqual(specification["info"]["version"], "v1")
|
||
self.assertEqual(specification["paths"]["/documents/{code}"]["get"]["parameters"][0]["required"], True)
|
||
self.assertEqual(specification["paths"]["/documents/{code}/editions/{edition}"]["get"]["parameters"][1]["name"], "edition")
|
||
self.assertEqual(api.handle("GET", "/documents/7")[1]["current_edition"]["source_code"], "10")
|
||
self.assertEqual(api.handle("GET", "/documents/7/editions/10?language=ru")[1]["content"]["ru"]["text"], "Текст\n")
|
||
with self.assertRaisesRegex(ApiError, "q is required"):
|
||
api.handle("GET", "/search")
|
||
with self.assertRaisesRegex(ApiError, "date_from must be an ISO date"):
|
||
api.handle("GET", "/search?q=x&date_from=tomorrow")
|
||
with self.assertRaisesRegex(ApiError, "document not found"):
|
||
api.handle("GET", "/documents/%2E%2E")
|
||
|
||
def test_filters_count_documents_and_return_bilingual_labels(self):
|
||
with tempfile.TemporaryDirectory() as temporary:
|
||
api = self.make_api(Path(temporary))
|
||
keys = {"document_types": "law", "statuses": "active", "authorities": "parliament", "legal_forces": "legislative"}
|
||
response = {"aggregations": {name: {"buckets": [{"key": keys[name], "documents": {"value": 3}}]} for name in keys}}
|
||
with patch("search.api.request_json", return_value=response) as request:
|
||
payload = api.handle("GET", "/search/filters?language=ky")[1]
|
||
self.assertEqual(payload["document_types"][0], {"code": "law", "labels": {"ru": "Закон", "ky": "Мыйзам"}, "count": 3})
|
||
self.assertEqual(payload["legal_forces"][0]["labels"]["ky"], "Мыйзам деңгээли")
|
||
body = json.loads(request.call_args.args[2])
|
||
self.assertIn("terms", body["aggs"]["document_types"])
|
||
self.assertIn("legal_force_code", body["aggs"]["legal_forces"]["terms"]["field"])
|
||
self.assertEqual(body["query"], {"term": {"is_current_edition": True}})
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|