""" Offline unit tests for the implemented /ingest and /deduplicate endpoints. Covers (Phase C item 1): - /ingest/check-updates: content-hash comparison classification - /ingest/status/{job_id}: Redis job-manager backing + tenant scoping - /ingest/repo-status/{repository}: wiki vs graph counts + job stats - /deduplicate/check: tenant-scoped similarity scan grouping All external clients are mocked - no shared services are contacted. """ import pytest from unittest.mock import AsyncMock from fastapi import HTTPException from src.core.hashing import compute_content_hash from src.main import ( check_updates, check_duplicates, get_ingestion_status, get_repo_status, CheckUpdatesRequest, DeduplicateCheckRequest, ) from src.services.vector_service import VectorService TEST_USER = "llm_tester" # --------------------------------------------------------------------------- # /ingest/check-updates # --------------------------------------------------------------------------- class TestCheckUpdates: @pytest.mark.asyncio async def test_classifies_changed_new_and_deleted(self): """Hash mismatch -> changed, no node -> new, stale node -> deleted.""" wikijs = AsyncMock() wikijs.list_all_pages = AsyncMock(return_value=[ {"id": 1, "path": f"users/{TEST_USER}/a", "title": "A", "tags": []}, {"id": 2, "path": f"users/{TEST_USER}/b", "title": "B", "tags": []}, ]) wikijs.get_page = AsyncMock(side_effect=[ {"id": 1, "content": "content-a"}, {"id": 2, "content": "content-b"}, ]) neo4j = AsyncMock() async def fake_query(cypher, params): # The endpoint sends one UNWIND query with page hashes pages = params["pages"] assert pages[0]["hash"] == compute_content_hash("content-a") return [{ "checked": [ { # page 1: node exists, hash matches -> up to date "page_id": 1, "path": pages[0]["path"], "title": "A", "is_new": False, "changed": False, "stored_hash_missing": False, }, { # page 2: node exists, hash differs -> changed "page_id": 2, "path": pages[1]["path"], "title": "B", "is_new": False, "changed": True, "stored_hash_missing": False, }, ], "deleted": [ {"page_id": 99, "path": f"users/{TEST_USER}/gone", "title": "Gone"}, ], }] neo4j.execute_query = AsyncMock(side_effect=fake_query) result = await check_updates( request=CheckUpdatesRequest(user=TEST_USER), neo4j=neo4j, wikijs=wikijs, api_key="", ) assert result["counts"] == { "changed": 1, "new": 0, "deleted": 1, "up_to_date": 1 } assert result["changed"][0]["page_id"] == 2 assert result["deleted"][0]["page_id"] == 99 assert result["duration_ms"] >= 0 @pytest.mark.asyncio async def test_new_page_detected_and_stub_pages_excluded(self): """Pages without Document nodes are new; entity stubs are skipped.""" wikijs = AsyncMock() wikijs.list_all_pages = AsyncMock(return_value=[ {"id": 5, "path": f"users/{TEST_USER}/fresh", "title": "Fresh", "tags": []}, {"id": 6, "path": f"users/{TEST_USER}/stub", "title": "Stub", "tags": ["entity-stub"]}, ]) wikijs.get_page = AsyncMock(return_value={"id": 5, "content": "x"}) neo4j = AsyncMock() neo4j.execute_query = AsyncMock(return_value=[{ "checked": [{ "page_id": 5, "path": f"users/{TEST_USER}/fresh", "title": "Fresh", "is_new": True, "changed": False, "stored_hash_missing": False, }], "deleted": [], }]) result = await check_updates( request=CheckUpdatesRequest(user=TEST_USER), neo4j=neo4j, wikijs=wikijs, api_key="", ) # Only page 5 was hashed (stub excluded -> get_page called once) assert wikijs.get_page.await_count == 1 assert result["counts"]["new"] == 1 assert result["new"][0]["page_id"] == 5 @pytest.mark.asyncio async def test_empty_wiki_reports_all_documents_deleted(self): """With no wiki pages, every Document node is reported deleted.""" wikijs = AsyncMock() wikijs.list_all_pages = AsyncMock(return_value=[]) neo4j = AsyncMock() neo4j.execute_query = AsyncMock(return_value=[{ "deleted": [{"page_id": 7, "path": f"users/{TEST_USER}/x", "title": "X"}], }]) result = await check_updates( request=CheckUpdatesRequest(user=TEST_USER), neo4j=neo4j, wikijs=wikijs, api_key="", ) assert result["counts"] == { "changed": 0, "new": 0, "deleted": 1, "up_to_date": 0 } def test_requires_user(self): """Empty user is rejected by the request model (Phase B rule).""" with pytest.raises(Exception): CheckUpdatesRequest(user=" ") # --------------------------------------------------------------------------- # /ingest/status/{job_id} # --------------------------------------------------------------------------- class TestIngestionStatus: @pytest.mark.asyncio async def test_returns_job_from_job_manager(self): job_manager = AsyncMock() job_manager.get_job = AsyncMock(return_value={ "job_id": "abc", "user": TEST_USER, "status": "completed", }) job = await get_ingestion_status( job_id="abc", user=TEST_USER, job_manager=job_manager, api_key="" ) assert job["status"] == "completed" job_manager.get_job.assert_awaited_once_with("abc") @pytest.mark.asyncio async def test_unknown_job_404(self): job_manager = AsyncMock() job_manager.get_job = AsyncMock(return_value=None) with pytest.raises(HTTPException) as exc: await get_ingestion_status( job_id="missing", user=TEST_USER, job_manager=job_manager, api_key="" ) assert exc.value.status_code == 404 @pytest.mark.asyncio async def test_other_tenants_job_is_hidden(self): """Tenant scoping: another user's job looks like a 404.""" job_manager = AsyncMock() job_manager.get_job = AsyncMock(return_value={ "job_id": "abc", "user": "someone_else", "status": "completed", }) with pytest.raises(HTTPException) as exc: await get_ingestion_status( job_id="abc", user=TEST_USER, job_manager=job_manager, api_key="" ) assert exc.value.status_code == 404 # --------------------------------------------------------------------------- # /ingest/repo-status/{repository} # --------------------------------------------------------------------------- class TestRepoStatus: @pytest.mark.asyncio async def test_counts_indexed_vs_total(self): wikijs = AsyncMock() wikijs.list_all_pages = AsyncMock(return_value=[ {"id": 1, "path": f"users/{TEST_USER}/tech/a"}, {"id": 2, "path": f"users/{TEST_USER}/tech/b"}, {"id": 3, "path": f"users/{TEST_USER}/tech/c"}, ]) neo4j = AsyncMock() neo4j.execute_query = AsyncMock(return_value=[{"indexed": 2}]) job_manager = AsyncMock() job_manager.get_job_stats = AsyncMock(return_value={"total": 4, "completed": 4}) result = await get_repo_status( repository="tech", user=TEST_USER, neo4j=neo4j, wikijs=wikijs, job_manager=job_manager, api_key="" ) assert result["total_documents"] == 3 assert result["indexed_documents"] == 2 assert result["unindexed_documents"] == 1 assert result["jobs"]["total"] == 4 assert result["path_prefix"] == f"users/{TEST_USER}/tech" # Wiki listing was scoped to the tenant namespace wikijs.list_all_pages.assert_awaited_once_with( path_prefix=f"users/{TEST_USER}/tech" ) # --------------------------------------------------------------------------- # /deduplicate/check + VectorService.find_duplicate_pairs # --------------------------------------------------------------------------- def _make_vector_service(points, search_results_by_id): qdrant = AsyncMock() qdrant.collection_exists = AsyncMock(return_value=True) qdrant.scroll_all_points = AsyncMock(return_value=points) async def fake_search(collection_name, query_vector, limit, score_threshold): # Route on the probe vector's marker value return search_results_by_id.get(query_vector[0], []) qdrant.search_vectors = AsyncMock(side_effect=fake_search) return VectorService(qdrant, AsyncMock(), AsyncMock()), qdrant class TestDeduplicateCheck: @pytest.mark.asyncio async def test_groups_pairs_by_page_and_dedupes_directions(self): points = [ {"id": "c1", "vector": [1.0], "payload": {"page_id": 10, "page_path": "users/llm_tester/a", "page_title": "A", "doc_type": "wiki"}}, {"id": "c2", "vector": [2.0], "payload": {"page_id": 20, "page_path": "users/llm_tester/b", "page_title": "B", "doc_type": "wiki"}}, ] search_results = { 1.0: [ # c1 finds c2 (cross-page) and itself (same page - ignored) {"id": "c1", "score": 1.0, "payload": points[0]["payload"]}, {"id": "c2", "score": 0.95, "payload": points[1]["payload"]}, ], 2.0: [ # c2 finds c1 - reverse direction must NOT double count {"id": "c1", "score": 0.95, "payload": points[0]["payload"]}, ], } service, _ = _make_vector_service(points, search_results) scan = await service.find_duplicate_pairs(TEST_USER, similarity_threshold=0.9) assert scan["chunks_scanned"] == 2 assert len(scan["duplicate_groups"]) == 1 group = scan["duplicate_groups"][0] assert group["max_similarity"] == 0.95 assert group["matching_chunk_pairs"] == 1 assert {p["page_id"] for p in group["pages"]} == {10, 20} @pytest.mark.asyncio async def test_missing_collection_returns_empty(self): qdrant = AsyncMock() qdrant.collection_exists = AsyncMock(return_value=False) service = VectorService(qdrant, AsyncMock(), AsyncMock()) scan = await service.find_duplicate_pairs(TEST_USER) assert scan == {"chunks_scanned": 0, "duplicate_groups": []} qdrant.scroll_all_points.assert_not_awaited() @pytest.mark.asyncio async def test_endpoint_shape(self): service_qdrant = AsyncMock() service_qdrant.collection_exists = AsyncMock(return_value=False) result = await check_duplicates( request=DeduplicateCheckRequest(user=TEST_USER), qdrant_client=service_qdrant, wiki_client=AsyncMock(), ollama_client=AsyncMock(), api_key="", ) assert result["user"] == TEST_USER assert result["similarity_threshold"] == 0.9 assert result["duplicate_groups"] == [] assert result["duplicate_group_count"] == 0 def test_requires_user(self): with pytest.raises(Exception): DeduplicateCheckRequest(user="") # --------------------------------------------------------------------------- # content hash canonicality # --------------------------------------------------------------------------- def test_content_hash_is_stable_and_none_safe(): assert compute_content_hash("abc") == compute_content_hash("abc") assert compute_content_hash("abc") != compute_content_hash("abd") assert compute_content_hash(None) == compute_content_hash("")