Jason Thistlethwaite
116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
import unittest
|
|
from pathlib import Path
|
|
|
|
from semantic_index.app import create_app
|
|
from semantic_index.config import Settings
|
|
from semantic_index.models import SearchResult
|
|
|
|
|
|
class FakeSearchService:
|
|
def search(self, query):
|
|
return [
|
|
SearchResult(
|
|
id="redmine:issue:1:chunk:0",
|
|
score=0.8,
|
|
text="Snippet text",
|
|
payload={
|
|
"source": "redmine",
|
|
"project_identifier": "customer-service",
|
|
"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": {}}
|
|
|
|
|
|
class FakeStore:
|
|
def list_projects(self, source=None, limit=1000):
|
|
return [{"project_identifier": "customer-service", "document_count": 10}]
|
|
|
|
|
|
class FakeRefreshService:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def refresh_redmine_project_limits(self, project_limits, dry_run=False, force_rebuild=False, overlap_minutes=15):
|
|
self.calls.append((project_limits, dry_run, force_rebuild, overlap_minutes))
|
|
return {"source": "redmine", "projects": len(project_limits), "dry_run": dry_run}
|
|
|
|
|
|
def fake_services():
|
|
refresh = FakeRefreshService()
|
|
return {
|
|
"settings": Settings(
|
|
openai_api_key="",
|
|
qdrant_url="http://qdrant",
|
|
qdrant_api_key=None,
|
|
qdrant_collection="semantic",
|
|
redmine_url="http://redmine",
|
|
redmine_api_key="",
|
|
redmine_project_identifier=None,
|
|
sample_limit=50,
|
|
bind_host="127.0.0.1",
|
|
bind_port=8787,
|
|
service_api_key=None,
|
|
refresh_state_path=Path(".cache/semantic_index/refresh_state.json"),
|
|
),
|
|
"search": FakeSearchService(),
|
|
"store": FakeStore(),
|
|
"refresh": refresh,
|
|
}
|
|
|
|
|
|
class SemanticIndexAppTest(unittest.TestCase):
|
|
def test_health_does_not_build_live_services(self):
|
|
def broken_builder():
|
|
raise AssertionError("health should not build live clients")
|
|
|
|
app = create_app(service_builder=broken_builder)
|
|
routes = {route.path: route.endpoint for route in app.routes}
|
|
|
|
self.assertEqual({"status": "ok"}, routes["/health"]())
|
|
|
|
def test_search_endpoint_returns_normalized_agent_response(self):
|
|
app = create_app(service_builder=fake_services)
|
|
routes = {route.path: route.endpoint for route in app.routes}
|
|
|
|
response = routes["/search"]({"query": "printer", "project_identifier": "customer-service", "limit": 3})
|
|
|
|
self.assertEqual("printer", response["query"])
|
|
self.assertEqual("customer-service", response["filters"]["project_identifier"])
|
|
self.assertEqual("customer-service", response["results"][0]["citation"]["project_identifier"])
|
|
|
|
def test_projects_endpoint_lists_indexed_projects(self):
|
|
app = create_app(service_builder=fake_services)
|
|
routes = {route.path: route.endpoint for route in app.routes}
|
|
|
|
response = routes["/projects"]()
|
|
|
|
self.assertEqual("customer-service", response["projects"][0]["project_identifier"])
|
|
|
|
def test_refresh_endpoint_passes_project_limits_and_cost_flags(self):
|
|
services = fake_services()
|
|
app = create_app(service_builder=lambda: services)
|
|
routes = {route.path: route.endpoint for route in app.routes}
|
|
|
|
response = routes["/sources/redmine/refresh"](
|
|
{
|
|
"project_limits": {"customer-service": 5},
|
|
"dry_run": True,
|
|
"force_rebuild": False,
|
|
"overlap_minutes": 30,
|
|
}
|
|
)
|
|
|
|
self.assertTrue(response["dry_run"])
|
|
self.assertEqual(({"customer-service": 5}, True, False, 30), services["refresh"].calls[0])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|