Jason Thistlethwaite
88 lines
3.1 KiB
Python
88 lines
3.1 KiB
Python
import json
|
|
import unittest
|
|
from unittest.mock import patch
|
|
|
|
from semantic_index.client import SemanticIndexClient
|
|
from semantic_index.models import SearchResult
|
|
|
|
|
|
class FakeSearchService:
|
|
def __init__(self):
|
|
self.queries = []
|
|
|
|
def search(self, query):
|
|
self.queries.append(query)
|
|
return [
|
|
SearchResult(
|
|
id="redmine:issue:1:chunk:0",
|
|
score=0.7,
|
|
text="Candidate follow up",
|
|
payload={
|
|
"source": "redmine",
|
|
"project_identifier": "hiring",
|
|
"doc_type": "issue",
|
|
"issue_id": 1,
|
|
"redmine_url": "http://redmine/issues/1",
|
|
"source_record_id": "issue:1",
|
|
},
|
|
)
|
|
]
|
|
|
|
def get_document(self, document_id):
|
|
return {"id": document_id, "text": "Full text", "payload": {"project_identifier": "hiring"}}
|
|
|
|
|
|
class SemanticIndexClientTest(unittest.TestCase):
|
|
def test_in_process_client_returns_normalized_search_response(self):
|
|
search = FakeSearchService()
|
|
client = SemanticIndexClient(search_service=search)
|
|
|
|
response = client.search("candidate follow up", project_identifier="hiring", limit=3)
|
|
|
|
self.assertEqual("candidate follow up", response["query"])
|
|
self.assertEqual({"project_identifier": "hiring", "limit": 3}, response["filters"])
|
|
self.assertEqual("redmine:issue:1:chunk:0", response["results"][0]["id"])
|
|
self.assertEqual("hiring", response["results"][0]["citation"]["project_identifier"])
|
|
self.assertEqual("hiring", search.queries[0].project_identifier)
|
|
|
|
def test_in_process_client_get_document(self):
|
|
client = SemanticIndexClient(search_service=FakeSearchService())
|
|
|
|
document = client.get_document("redmine:issue:1:chunk:0")
|
|
|
|
self.assertEqual("Full text", document["text"])
|
|
|
|
def test_http_client_sends_auth_header_and_parses_search_response(self):
|
|
body = json.dumps({"query": "printer", "filters": {}, "results": []}).encode()
|
|
|
|
class FakeResponse:
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
def read(self):
|
|
return body
|
|
|
|
captured = {}
|
|
|
|
def fake_urlopen(request, timeout):
|
|
captured["url"] = request.full_url
|
|
captured["authorization"] = request.headers.get("Authorization")
|
|
captured["body"] = json.loads(request.data.decode())
|
|
return FakeResponse()
|
|
|
|
with patch("urllib.request.urlopen", fake_urlopen):
|
|
client = SemanticIndexClient(base_url="http://semantic.local", api_key="secret")
|
|
response = client.search("printer", project_identifier="customer-service")
|
|
|
|
self.assertEqual("http://semantic.local/search", captured["url"])
|
|
self.assertEqual("Bearer secret", captured["authorization"])
|
|
self.assertEqual("customer-service", captured["body"]["project_identifier"])
|
|
self.assertEqual("printer", response["query"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|