Files
akyldash/backend/test_search_api_opensearch.py

42 lines
3.7 KiB
Python

import json
import os
import unittest
import uuid
from pathlib import Path
from search.api import Api
from search.minjust_opensearch import DEFAULT_MAPPING, request_json
@unittest.skipUnless(os.getenv("AKYLDASH_OPENSEARCH_URL"), "set AKYLDASH_OPENSEARCH_URL to run against local OpenSearch")
class SearchApiOpenSearchTest(unittest.TestCase):
def test_current_editions_filters_and_catalogs(self):
base_url = os.environ["AKYLDASH_OPENSEARCH_URL"].rstrip("/")
index = f"akyldash-api-test-{uuid.uuid4().hex}"
request_json(f"{base_url}/{index}", "PUT", DEFAULT_MAPPING.read_bytes(), "application/json")
try:
documents = [
{"document_code": "1", "edition_code": "1", "is_current_edition": False, "language": "ru", "position": 1, "fragment_type": "paragraph", "text_ru": "historic", "document_name_ru": "Old law", "document_type_ru": "Закон", "document_type_ky": "Мыйзам", "document_type_code": "law", "legal_force_code": "legislative", "status_ru": "Утратил силу", "status_ky": "Күчүн жоготту", "status_code": "repealed", "date_adopted": "2020-01-01", "authority_paths_ru": ["Парламент"], "authority_paths_ky": ["Парламент"], "authority_codes": ["parliament"]},
{"document_code": "1", "edition_code": "2", "is_current_edition": True, "language": "ru", "position": 1, "fragment_type": "paragraph", "text_ru": "needle", "document_name_ru": "Current law", "document_type_ru": "Закон", "document_type_ky": "Мыйзам", "document_type_code": "law", "legal_force_code": "legislative", "status_ru": "Действует", "status_ky": "Күчүндө", "status_code": "active", "date_adopted": "2021-01-01", "authority_paths_ru": ["Парламент"], "authority_paths_ky": ["Парламент"], "authority_codes": ["parliament"]},
{"document_code": "2", "edition_code": "1", "is_current_edition": True, "language": "ru", "position": 1, "fragment_type": "paragraph", "text_ru": "needle", "document_name_ru": "Current decree", "document_type_ru": "Указ", "document_type_ky": "Жарлык", "document_type_code": "decree", "legal_force_code": "subordinate", "status_ru": "Действует", "status_ky": "Күчүндө", "status_code": "active", "date_adopted": "2022-01-01", "authority_paths_ru": ["Президент"], "authority_paths_ky": ["Президент"], "authority_codes": ["president"]},
]
for number, document in enumerate(documents):
request_json(f"{base_url}/{index}/_doc/{number}", "PUT", json.dumps(document).encode(), "application/json")
request_json(f"{base_url}/{index}/_refresh", "POST", None, "application/json")
api = Api(base_url, index, Path("."))
self.assertEqual(api.handle("GET", "/search?q=historic")[1]["results"], [])
filtered = api.handle("GET", "/search?q=needle&document_type=law")[1]
self.assertEqual([result["code"] for result in filtered["results"]], ["1"])
force_filtered = api.handle("GET", "/search?q=needle&legal_force=subordinate")[1]
self.assertEqual([result["code"] for result in force_filtered["results"]], ["2"])
self.assertEqual(force_filtered["results"][0]["legal_force"], "Подзаконный уровень")
filters = api.handle("GET", "/search/filters")[1]
self.assertEqual({item["count"] for item in filters["statuses"]}, {2})
self.assertEqual({item["code"] for item in filters["legal_forces"]}, {"legislative", "subordinate"})
finally:
request_json(f"{base_url}/{index}", "DELETE", None, "application/json")
if __name__ == "__main__":
unittest.main()