235 lines
14 KiB
Python
235 lines
14 KiB
Python
import json
|
|
import contextlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
from search.api import Api, ApiError, handler
|
|
|
|
|
|
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_static_font_assets_are_allowlisted(self):
|
|
content_type, body = Api.static_asset("/static/fonts/noto-sans-cyrillic-ext.woff2")
|
|
self.assertEqual(content_type, "font/woff2")
|
|
self.assertTrue(body.startswith(b"wOF2"))
|
|
content_type, body = Api.static_asset("/static/favicon.svg")
|
|
self.assertEqual(content_type, "image/svg+xml")
|
|
self.assertIn(b"<svg", body)
|
|
content_type, body = Api.static_asset("/static/favicon-32.png")
|
|
self.assertEqual(content_type, "image/png")
|
|
self.assertTrue(body.startswith(b"\x89PNG\r\n\x1a\n"))
|
|
with self.assertRaisesRegex(ApiError, "static asset not found"):
|
|
Api.static_asset("/static/fonts/../../DESIGN.md")
|
|
|
|
def test_users_screen_is_served_without_exposing_admin_api(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
api = self.make_api(Path(temporary))
|
|
request_handler = handler(api)
|
|
|
|
def get(path):
|
|
request = request_handler.__new__(request_handler)
|
|
request.path = path
|
|
request.headers = {}
|
|
request.wfile = io.BytesIO()
|
|
request.send_response = Mock()
|
|
request.send_header = Mock()
|
|
request.end_headers = Mock()
|
|
request.do_GET()
|
|
return request
|
|
|
|
users_page = get("/users")
|
|
users_page.send_response.assert_called_once_with(200)
|
|
self.assertIn(b'id="users-page"', users_page.wfile.getvalue())
|
|
admin_api = get("/admin/users")
|
|
admin_api.send_response.assert_called_once_with(401)
|
|
|
|
def test_account_capability_matches_the_admin_users_api_guard(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
api = self.make_api(Path(temporary))
|
|
api.auth_store.create_admin("Fixture-only-admin-password-2026")
|
|
admin_token, _ = api.auth_store.login("admin", "Fixture-only-admin-password-2026")
|
|
api.auth_store.create_user("fixture-user", "Fixture-only-temporary-password-2026")
|
|
api.auth_store.change_password("fixture-user", "Fixture-only-temporary-password-2026", "Fixture-only-user-password-2026")
|
|
user_token, _ = api.auth_store.login("fixture-user", "Fixture-only-user-password-2026")
|
|
request_handler = handler(api)
|
|
|
|
def get(path, token):
|
|
request = request_handler.__new__(request_handler)
|
|
request.path = path
|
|
request.headers = {"Cookie": f"akyldash_session={token}"}
|
|
request.wfile = io.BytesIO()
|
|
request.send_response = Mock()
|
|
request.send_header = Mock()
|
|
request.end_headers = Mock()
|
|
request.do_GET()
|
|
return request
|
|
|
|
admin_account = json.loads(get("/auth/me", admin_token).wfile.getvalue())["account"]
|
|
user_account = json.loads(get("/auth/me", user_token).wfile.getvalue())["account"]
|
|
self.assertEqual((admin_account["role"], admin_account["can_manage_users"]), ("Администратор", True))
|
|
self.assertEqual((user_account["role"], user_account["can_manage_users"]), ("Пользователь", False))
|
|
get("/admin/users", admin_token).send_response.assert_called_once_with(200)
|
|
get("/admin/users", user_token).send_response.assert_called_once_with(403)
|
|
|
|
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_review_secret_file_is_read_without_command_line_value(self):
|
|
from search.api import main
|
|
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
secret_file = Path(temporary) / "secret"
|
|
secret_file.write_text("secret-value\n", encoding="utf-8")
|
|
with patch("sys.argv", ["search.api", "--review-secret-file", str(secret_file)]), patch("search.api.Api") as api, patch("search.api.ThreadingHTTPServer"):
|
|
main()
|
|
self.assertEqual(api.call_args.args[4], b"secret-value")
|
|
|
|
def test_review_secret_file_rejects_empty_content(self):
|
|
from search.api import main
|
|
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
secret_file = Path(temporary) / "secret"
|
|
secret_file.touch()
|
|
with patch("sys.argv", ["search.api", "--review-secret-file", str(secret_file)]), contextlib.redirect_stderr(io.StringIO()):
|
|
with self.assertRaises(SystemExit) as result:
|
|
main()
|
|
self.assertEqual(result.exception.code, 2)
|
|
|
|
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_production_api_boosts_tax_code_for_profit_tax_query(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
api = self.make_api(Path(temporary))
|
|
with patch("search.api.request_json", return_value={"hits": {"hits": []}}) as request:
|
|
api.handle("GET", "/search?q=%D0%BD%D0%B0%D0%BB%D0%BE%D0%B3+%D0%BD%D0%B0+%D0%BF%D1%80%D0%B8%D0%B1%D1%8B%D0%BB%D1%8C&language=ru")
|
|
body = json.loads(request.call_args.args[2])
|
|
planned = body["query"]["bool"]["must"]["bool"]
|
|
anchors = [clause["constant_score"]["filter"]["bool"]["filter"][0] for clause in planned["should"] if "constant_score" in clause]
|
|
self.assertIn({"term": {"document_code": "112340"}}, anchors)
|
|
|
|
def test_search_sorts_repealed_documents_after_other_statuses(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
api = self.make_api(Path(temporary))
|
|
with patch("search.api.request_json", return_value={"hits": {"hits": []}}) as request:
|
|
api.handle("GET", "/search?q=НДС+на+импорт&language=ru")
|
|
body = json.loads(request.call_args.args[2])
|
|
source = body["sort"][0]["_script"]["script"]["source"]
|
|
self.assertIn("'active' ? 0", source)
|
|
self.assertIn("'repealed' ? 2", source)
|
|
self.assertEqual(body["sort"][1], {"_score": "desc"})
|
|
|
|
api.handle("GET", "/search?q=НДС+на+импорт&language=ru&sort=date")
|
|
body = json.loads(request.call_args.args[2])
|
|
self.assertEqual(body["sort"][1], {"date_adopted": "desc"})
|
|
|
|
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}})
|
|
|
|
def test_global_dfs_is_only_sent_for_explicit_multi_index_searches(self):
|
|
with tempfile.TemporaryDirectory() as temporary:
|
|
multi = Api("http://opensearch:9200", "base-v2,supplement-v1", Path(temporary), review_secret=b"test-secret")
|
|
alias = Api("http://opensearch:9200", "current", Path(temporary), review_secret=b"test-secret")
|
|
single = Api("http://opensearch:9200", "single-v2", Path(temporary), review_secret=b"test-secret")
|
|
one_hit = {"hits": {"hits": [{"_index": "base-v2", "_source": {"document_code": "5", "document_name_ru": "Кодекс", "document_name_ky": "Кодекс"}}]}}
|
|
empty = {"hits": {"hits": []}}
|
|
alias_hit = {"hits": {"hits": [{"_index": "current-20261103", "_source": {"document_code": "7", "document_name_ru": "Закон"}}]}}
|
|
aggregations = {"aggregations": {name: {"buckets": []} for name in ("document_types", "statuses", "authorities", "legal_forces")}}
|
|
with patch("search.api.request_json", side_effect=[one_hit, empty, alias_hit, one_hit, aggregations]) as request:
|
|
multi_ru = multi.search({"q": ["test"], "language": ["ru"], "page_size": ["10"]})
|
|
multi_ky_empty = multi.search({"q": ["test"], "language": ["ky"], "page_size": ["10"]})
|
|
alias_ru = alias.search({"q": ["test"], "language": ["ru"], "page_size": ["10"]})
|
|
single_ky = single.search({"q": ["test"], "language": ["ky"], "page_size": ["10"]})
|
|
alias.filters({"language": ["ru"]})
|
|
|
|
calls = request.call_args_list
|
|
multi_ru_url = urllib.parse.urlsplit(calls[0].args[0])
|
|
multi_ky_url = urllib.parse.urlsplit(calls[1].args[0])
|
|
self.assertEqual(multi_ru_url.path, "/base-v2%2Csupplement-v1/_search")
|
|
self.assertEqual(urllib.parse.parse_qs(multi_ru_url.query), {"search_type": ["dfs_query_then_fetch"]})
|
|
self.assertEqual(urllib.parse.parse_qs(multi_ky_url.query), {"search_type": ["dfs_query_then_fetch"]})
|
|
self.assertEqual([item["code"] for item in multi_ru["results"]], ["5"])
|
|
self.assertEqual(multi_ky_empty["results"], [])
|
|
self.assertEqual(multi.review_snapshots.verify(multi_ru["review_token"])["index_name"], "base-v2,supplement-v1")
|
|
self.assertEqual(multi.review_snapshots.verify(multi_ky_empty["review_token"])["index_name"], "base-v2,supplement-v1")
|
|
|
|
for call in calls[2:]:
|
|
parsed = urllib.parse.urlsplit(call.args[0])
|
|
self.assertEqual(parsed.query, "")
|
|
self.assertEqual(alias.review_snapshots.verify(alias_ru["review_token"])["index_name"], "current-20261103")
|
|
self.assertEqual(single.review_snapshots.verify(single_ky["review_token"])["index_name"], "base-v2")
|
|
multi_body = json.loads(calls[0].args[2])
|
|
alias_body = json.loads(calls[2].args[2])
|
|
self.assertEqual(multi_body, alias_body)
|
|
self.assertNotIn("search_type", multi_body)
|
|
self.assertEqual(json.loads(calls[1].args[2])["query"]["bool"]["filter"][0], {"term": {"language": "ky"}})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|