diff --git a/.quality/exceptions.json b/.quality/exceptions.json index 8350e132..b5b9d361 100644 --- a/.quality/exceptions.json +++ b/.quality/exceptions.json @@ -313,23 +313,6 @@ "review": {"status": "active", "reference": "https://github.com/Azure/GPT-RAG/issues/681#issuecomment-5601804634"}, "review_by_stage": "initial" }, - { - "id": "sharepoint-run-failure-observation", - "module_id": "jobs.sharepoint_indexer", - "symbol": "SharePointIndexer._run_pipeline", - "handler_fingerprint": "db49e21755ec02b963f00bf183190e826369bec3e186f2c2bad75ae90269d19a", - "caught_types": ["Exception"], - "boundary": "SharePoint run failed-summary observation followed by exception propagation", - "reason": "A failed collection must mark the run failed and propagate to the scheduler/audit wrapper, not become a finished run with zero failed items. The handler preserves the existing summary schema with safe error text, then re-raises. Its outer resource guard closes clients on configuration, collection and cancellation paths.", - "failure_outcome": "explicit-failure-translation", - "diagnostic_path": "Failed terminal summary, safe run error and original exception propagation", - "evidence_tests": [ - "tests/test_sharepoint_failure_boundaries.py::test_sharepoint_run_propagates_failures_and_always_closes[configuration]", - "tests/test_sharepoint_failure_boundaries.py::test_sharepoint_run_propagates_failures_and_always_closes[collection]" - ], - "review": {"status": "active", "reference": "https://github.com/Azure/GPT-RAG/issues/681#issuecomment-5601804634"}, - "review_by_stage": "initial" - }, { "id": "sharepoint-summary-best-effort", "module_id": "jobs.sharepoint_indexer", @@ -344,25 +327,6 @@ "review": {"status": "active", "reference": "https://github.com/Azure/GPT-RAG/issues/681#issuecomment-5601804634"}, "review_by_stage": "initial" }, - { - "id": "sharepoint-item-failure", - "module_id": "jobs.sharepoint_indexer", - "symbol": "SharePointIndexer._process_collection.worker", - "handler_fingerprint": "6a5e6b398d3dcba804b8825b7bbfe04003fe9e2c1e99d1bd05b0284bf8828013", - "caught_types": ["Exception"], - "boundary": "SharePoint per-item processing failure, retry/block state and failed-count translation", - "reason": "The existing per-item boundary must record explicit failed items while siblings proceed. Permission acquisition now occurs within this bounded operation, so its failure records an error and cannot create empty ACLs or unobserved task exceptions. Failures before this boundary propagate through observed collection tasks; cancellation is not converted into a file success.", - "failure_outcome": "explicit-failure-translation", - "diagnostic_path": "Safe per-file error, failed item counter and no Search operation after ACL failure", - "evidence_tests": [ - "tests/test_sharepoint_failure_boundaries.py::test_collection_observes_item_failures_instead_of_finishing_empty[acl]", - "tests/test_sharepoint_failure_boundaries.py::test_collection_observes_item_failures_instead_of_finishing_empty[lookup]", - "tests/test_sharepoint_failure_boundaries.py::test_collection_failure_cancels_and_observes_its_children[late-page]", - "tests/test_sharepoint_failure_boundaries.py::test_collection_failure_cancels_and_observes_its_children[timeout]" - ], - "review": {"status": "active", "reference": "https://github.com/Azure/GPT-RAG/issues/681#issuecomment-5601804634"}, - "review_by_stage": "initial" - }, { "id": "blob-run-failure-observation", "module_id": "jobs.blob_storage_indexer", diff --git a/CHANGELOG.md b/CHANGELOG.md index a810654c..f2893785 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,18 @@ # Changelog +## [v3.1.3] - 2026-10-27 + +### Fixed + +- **SharePoint permission-only synchronization.** Refresh the supported Entra + user/group ACLs on every indexed item document and document-library chunk, + independently of content timestamps, without re-downloading or embedding + unchanged content. Document libraries read Graph's effective drive sharing + permissions, including inherited grants. Reconcile all permission pages, + replace revoked principals, skip equal ACLs, and confirm every Search merge. + Lookup and partial-update failures fail the job and remain retryable; content + retry blocking no longer prevents ACL refresh (Azure/agent-landing-zone#720). + ## [v3.1.2] - 2026-10-27 ### Changed diff --git a/VERSION b/VERSION index 6ebad148..711ee4f5 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -3.1.2 \ No newline at end of file +3.1.3 \ No newline at end of file diff --git a/jobs/sharepoint_graph_client.py b/jobs/sharepoint_graph_client.py index 8f2a4bb6..c41ba0cb 100644 --- a/jobs/sharepoint_graph_client.py +++ b/jobs/sharepoint_graph_client.py @@ -1,7 +1,7 @@ import asyncio import logging from typing import Any, AsyncIterator, Dict, List, Optional, Tuple -from urllib.parse import quote, urlencode +from urllib.parse import quote, urlencode, urlsplit from uuid import UUID import aiohttp @@ -13,7 +13,7 @@ class GraphRequestError(RuntimeError): - """A Graph transport operation exhausted its bounded retry budget.""" + """A Graph operation failed or returned an incomplete response.""" class SharePointGraphClient: @@ -340,48 +340,62 @@ async def get_item_permission_principal_ids( site_id: str, collection_id: str, item_id: str, + drive_item: Optional[Dict[str, Any]] = None, ) -> Tuple[List[str], List[str]]: - """Return (user_object_ids, group_object_ids) that have explicit permissions on an item.""" - # The permissions resource for list items currently exists only in the beta Graph surface. - url = ( - f"{self._graph_beta_base}/sites/{site_id}/lists/{collection_id}/items/{item_id}/permissions" - "?$select=id,grantedToIdentitiesV2,grantedTo" - ) - try: - data = await self._gget(session, url) - except ClientResponseError as exc: - if exc.status in (400, 401, 403, 404, 501): - logging.debug( - f"{LOG_SCOPE} permissions lookup failed ({exc.status}) for list {collection_id} item {item_id}" - ) - return ([], []) - raise + """Read all Graph-visible Entra grants; drive permissions include inherited sharing.""" + if drive_item is not None: + drive_id = drive_item.get("parentReference", {}).get("driveId") + drive_item_id = drive_item.get("id") + if not drive_id or not drive_item_id: + raise GraphRequestError("Drive permission lookup requires drive and item IDs.") + resource = f"{self._graph_base}/drives/{quote(str(drive_id), safe='')}/items/{quote(str(drive_item_id), safe='')}/permissions" + else: + resource = f"{self._graph_beta_base}/sites/{site_id}/lists/{collection_id}/items/{item_id}/permissions" + url = resource + "?$select=id,grantedToIdentitiesV2,grantedToV2,grantedTo" user_ids: List[str] = [] group_ids: List[str] = [] seen_users: set[str] = set() seen_groups: set[str] = set() + visited: set[str] = set() - for perm in data.get("value", []): - identities = perm.get("grantedToIdentitiesV2") or perm.get("grantedToIdentities") or [] - for identity in identities: - kind, obj_id = self._extract_identity_principal(identity) - if kind == "user" and obj_id and obj_id not in seen_users: - seen_users.add(obj_id) - user_ids.append(obj_id) - elif kind == "group" and obj_id and obj_id not in seen_groups: - seen_groups.add(obj_id) - group_ids.append(obj_id) - - granted_to = perm.get("grantedToV2") or perm.get("grantedTo") - if isinstance(granted_to, dict): - kind, obj_id = self._extract_identity_principal(granted_to) - if kind == "user" and obj_id and obj_id not in seen_users: - seen_users.add(obj_id) - user_ids.append(obj_id) - elif kind == "group" and obj_id and obj_id not in seen_groups: - seen_groups.add(obj_id) - group_ids.append(obj_id) + while url: + if url in visited: + raise GraphRequestError("Graph permission pagination repeated a page.") + visited.add(url) + data = await self._gget(session, url) + if not isinstance(data, dict) or not isinstance(data.get("value"), list): + raise GraphRequestError("Graph permission response has no permission collection.") + for perm in data["value"]: + if not isinstance(perm, dict): + raise GraphRequestError("Graph permission response contains an invalid entry.") + identities = perm.get("grantedToIdentitiesV2", perm.get("grantedToIdentities", [])) + if identities is None: + identities = [] + if not isinstance(identities, list): + raise GraphRequestError("Graph permission response contains invalid identities.") + granted_to = perm.get("grantedToV2", perm.get("grantedTo")) + if granted_to is not None: + identities = [*identities, granted_to] + for identity in identities: + if not isinstance(identity, dict): + raise GraphRequestError("Graph permission response contains an invalid identity.") + kind, obj_id = self._extract_identity_principal(identity) + if kind == "user" and obj_id and obj_id not in seen_users: + seen_users.add(obj_id) + user_ids.append(obj_id) + elif kind == "group" and obj_id and obj_id not in seen_groups: + seen_groups.add(obj_id) + group_ids.append(obj_id) + next_url = data.get("@odata.nextLink") + if next_url is not None: + if not isinstance(next_url, str): + raise GraphRequestError("Graph permission pagination link is invalid.") + target = urlsplit(next_url) + expected = urlsplit(resource) + if (target.scheme, target.netloc, target.path) != (expected.scheme, expected.netloc, expected.path): + raise GraphRequestError("Graph permission pagination left the permission resource.") + url = next_url return (user_ids, group_ids) diff --git a/jobs/sharepoint_indexer.py b/jobs/sharepoint_indexer.py index 2d940669..ca68ecef 100644 --- a/jobs/sharepoint_indexer.py +++ b/jobs/sharepoint_indexer.py @@ -14,7 +14,7 @@ from azure.search.documents.aio import SearchClient as AsyncSearchClient from azure.storage.blob import ContentSettings from azure.storage.blob.aio import BlobServiceClient -from openai import APIConnectionError, InternalServerError, RateLimitError +from openai import APIConnectionError, APIError, InternalServerError, RateLimitError from chunking import DocumentChunker from dependencies import get_config @@ -663,7 +663,7 @@ async def _with_backoff(self, func, **kw): result=result, source_type=self.cfg.indexer_name, ) - if getattr(func, "__name__", "") in {"upload_documents", "delete_documents"}: + if getattr(func, "__name__", "") in {"upload_documents", "delete_documents", "merge_documents"}: require_confirmed_batch(kw["documents"], result) return result @@ -729,6 +729,44 @@ async def _upload_docs(self, docs: List[Dict[str, Any]]): logging.debug(f"[{self.cfg.indexer_name}][INDEX-UPLOAD] Upload completed | totalDocumentsUploaded={uploaded_count}") + async def _refresh_item_acls( + self, + parent_id: str, + user_ids: List[str], + group_ids: List[str], + *, + include_files: bool, + ) -> int: + """Replace only changed ACLs, streaming every indexed document for this item.""" + escaped = parent_id.replace("'", "''") + parent_filter = f"parent_id eq '{escaped}'" + if include_files: + # File parents start with item-parent + '/'; '0' is its exclusive upper bound. + parent_filter = f"({parent_filter} or (parent_id ge '{escaped}/' and parent_id lt '{escaped}0'))" + fields = ["metadata_security_user_ids", "metadata_security_group_ids"] + expected = [set(user_ids), set(group_ids)] + results = await self._search_client.search( + search_text="*", filter=parent_filter, select=["id", *fields], + order_by=["id asc"], headers=_ELEVATED_HEADERS, + ) + batch: List[Dict[str, Any]] = [] + batch_size = min(self.cfg.batch_size, 1000) + updated = 0 + async for doc in results: + if all(set(doc.get(field) or []) == ids for field, ids in zip(fields, expected)): + continue + if not doc.get("id"): + raise ValueError("An indexed SharePoint document is missing its key.") + batch.append({"id": doc["id"], fields[0]: user_ids, fields[1]: group_ids}) + if len(batch) >= batch_size: + await self._with_backoff(self._search_client.merge_documents, documents=batch) + updated += len(batch) + batch = [] + if batch: + await self._with_backoff(self._search_client.merge_documents, documents=batch) + updated += len(batch) + return updated + # ---------- content builders ---------- async def _get_security_principals_for_item( self, @@ -737,15 +775,17 @@ async def _get_security_principals_for_item( site_id: str, collection_id: str, item_id: str, + drive_item: Optional[Dict[str, Any]] = None, ) -> tuple[List[str], List[str]]: if not self._graph_client: - return ([], []) + raise RuntimeError("SharePoint permission client is unavailable.") user_ids, group_ids = await self._graph_client.get_item_permission_principal_ids( session=session, site_id=site_id, collection_id=collection_id, item_id=item_id, + drive_item=drive_item, ) before_users = len(user_ids) @@ -968,10 +1008,12 @@ async def _process_document_library_file( security_group_ids: Optional[List[str]], stats: RunStats, stats_lock: asyncio.Lock, + drive_item: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: result = {"chunks": 0, "had_candidate": False} - drive_item = await self._graph_client.get_drive_item(session, site_id, collection_id, item_id) + if drive_item is None: + drive_item = await self._graph_client.get_drive_item(session, site_id, collection_id, item_id) if not drive_item: logging.debug( f"[{self.cfg.indexer_name}][DOC-LIB] Drive item missing | itemId={item_id} collection={collection_name}" @@ -1285,12 +1327,16 @@ async def _run_pipeline(self) -> None: raise r else: raise TypeError("Collection worker returned an invalid result.") + if summary["failed"] or stats.items_failed: + raise RuntimeError("SharePoint synchronization failed for one or more items.") # Add richer stats and flush a "finishing" snapshot summary.update({ "itemsDiscovered": stats.items_discovered, "candidateItems": stats.items_candidates, "indexedItems": stats.items_indexed, + "aclUpdatedItems": stats.items_acl_updated, + "aclUpdatedDocuments": stats.acl_documents_updated, "skippedNoChange": stats.items_skipped_nochange, "failed": max(summary["failed"], stats.items_failed), "documentLibraryStats": { @@ -1310,19 +1356,25 @@ async def _run_pipeline(self) -> None: self._log_event(logging.WARNING, "RUN-CANCELLED", runId=run_id) logging.info("[%s] Run cancelled: runId=%s", self.cfg.indexer_name, run_id) raise - except Exception as exc: + except (AzureError, aiohttp.ClientError, RuntimeError, ValueError) as exc: logging.error("[%s] Run failed (%s)", self.cfg.indexer_name, type(exc).__name__) summary["error"] = f"Run failed ({type(exc).__name__})." summary["status"] = "failed" self._log_event(logging.ERROR, "RUN-ERROR", runId=run_id, error=summary["error"]) raise finally: + if summary["status"] == "started": + summary["status"] = "failed" + summary["error"] = "SharePoint synchronization did not complete." + self._log_event(logging.ERROR, "RUN-ERROR", runId=run_id, error=summary["error"]) # Final, authoritative "finished" (or failed/cancelled) snapshot summary.setdefault("documentLibraryStats", {}) summary.update({ "itemsDiscovered": stats.items_discovered, "candidateItems": stats.items_candidates, "indexedItems": stats.items_indexed, + "aclUpdatedItems": stats.items_acl_updated, + "aclUpdatedDocuments": stats.acl_documents_updated, "skippedNoChange": stats.items_skipped_nochange, "skippedBlocked": stats.items_skipped_blocked, "failed": max(summary.get("failed", 0), stats.items_failed), @@ -1361,6 +1413,8 @@ async def _run_pipeline(self) -> None: collectionsSeen=summary.get("collections", 0), itemsDiscovered=stats.items_discovered, itemsIndexed=stats.items_indexed, + aclUpdatedItems=stats.items_acl_updated, + aclUpdatedDocuments=stats.acl_documents_updated, itemsFailed=stats.items_failed, skippedNoChange=stats.items_skipped_nochange, skippedBlocked=stats.items_skipped_blocked, @@ -1513,9 +1567,8 @@ async def worker(item: Dict[str, Any]): ) async with stats_lock: stats.items_skipped_blocked += 1 - return - - processing_attempts += 1 + else: + processing_attempts += 1 file_log = { **existing_log, @@ -1528,7 +1581,7 @@ async def worker(item: Dict[str, Any]): "chunksIds": _make_chunk_key_prefix(parent_item_id), "processingAttempts": processing_attempts, "lastAttemptAt": _utc_now(), - "blocked": False, + "blocked": is_blocked, "blockedAt": existing_log.get("blockedAt"), "blockedReason": existing_log.get("blockedReason"), } @@ -1540,11 +1593,21 @@ async def worker(item: Dict[str, Any]): try: async def _do() -> Dict[str, Any]: + drive_item = None + if is_document_library: + drive_item = await self._graph_client.get_drive_item(session, site_id, collection_id, item_id) + if drive_item is None: + raise GraphRequestError("Document-library permissions require a readable drive item.") security_user_ids, security_group_ids = await self._get_security_principals_for_item( session=session, site_id=site_id, collection_id=collection_id, item_id=item_id, + drive_item=drive_item, + ) + acl_updated = await self._refresh_item_acls( + parent_item_id, security_user_ids, security_group_ids, + include_files=is_document_library, ) local_chunks_for_item = 0 local_body_uploaded = False @@ -1553,7 +1616,7 @@ async def _do() -> Dict[str, Any]: logging.debug(f"[{self.cfg.indexer_name}][BODY] Checking body freshness | itemId={item_id} parentId={parent_item_id}") existing_body_mod = await self._get_body_lastmod_by_id(parent_item_id) - need_body = _is_strictly_newer(last_mod, existing_body_mod) + need_body = not is_blocked and _is_strictly_newer(last_mod, existing_body_mod) if need_body: local_item_had_candidate = True @@ -1602,7 +1665,7 @@ async def _do() -> Dict[str, Any]: f"itemId={item_id} parentId={parent_item_id} hasAttachmentsFlag={att_present}" ) - if is_document_library: + if is_document_library and not is_blocked: doclib_result = await self._process_document_library_file( session=session, site_id=site_id, @@ -1617,6 +1680,7 @@ async def _do() -> Dict[str, Any]: security_group_ids=security_group_ids, stats=stats, stats_lock=stats_lock, + drive_item=drive_item, ) local_chunks_for_item += int(doclib_result.get("chunks", 0)) if doclib_result.get("had_candidate"): @@ -1632,14 +1696,19 @@ async def _do() -> Dict[str, Any]: # ---- LOG + COUNTERS ---- did_upload_any = (local_body_uploaded or local_chunks_for_item > 0) + status = ( + "success" if did_upload_any else "acl-updated" if acl_updated else + "skipped-blocked" if is_blocked else "skipped-no-change" + ) file_log.update({ - "status": "success" if did_upload_any else "skipped-no-change", + "status": status, "finishedAt": _utc_now(), "chunks": local_chunks_for_item, - "processingAttempts": 0 if did_upload_any else file_log.get("processingAttempts", 0), - "blocked": False, - "blockedAt": None, - "blockedReason": None, + "aclDocumentsUpdated": acl_updated, + "processingAttempts": processing_attempts if is_blocked else 0, + "blocked": is_blocked, + "blockedAt": existing_log.get("blockedAt") if is_blocked else None, + "blockedReason": existing_log.get("blockedReason") if is_blocked else None, }) await self._write_file_log(f"{_sanitize_key_part(parent_item_id)}.json", file_log) @@ -1654,7 +1723,7 @@ async def _do() -> Dict[str, Any]: f"attachmentChunks={attach_chunks} " f"totalChunks={local_chunks_for_item} " f"hadCandidate={local_item_had_candidate} " - f"status={'uploaded' if did_upload_any else 'skipped-no-change'}" + f"status={status}" ) return { @@ -1662,6 +1731,8 @@ async def _do() -> Dict[str, Any]: "uploaded": did_upload_any, "had_candidate": local_item_had_candidate, "body_uploaded": local_body_uploaded, + "acl_updated": acl_updated, + "status": status, } result = await asyncio.wait_for(_do(), timeout=self._item_timeout_s) @@ -1671,11 +1742,16 @@ async def _do() -> Dict[str, Any]: if result.get("had_candidate"): async with stats_lock: stats.items_candidates += 1 - if result.get("uploaded"): + if result.get("uploaded") or result.get("acl_updated"): success += 1 + if result.get("uploaded"): async with stats_lock: stats.items_indexed += 1 - else: + if result.get("acl_updated"): + async with stats_lock: + stats.items_acl_updated += 1 + stats.acl_documents_updated += int(result["acl_updated"]) + elif not result.get("uploaded") and not is_blocked: async with stats_lock: stats.items_skipped_nochange += 1 if result.get("body_uploaded"): @@ -1693,7 +1769,8 @@ async def _do() -> Dict[str, Any]: listType=list_type, itemId=item_id, parentId=parent_item_id, - status="uploaded" if result.get("uploaded") else "skipped-no-change", + status=result["status"], + aclDocumentsUpdated=int(result.get("acl_updated", 0)), bodyUploaded=bool(result.get("body_uploaded")), attachmentChunks=total_attach_chunks, totalChunks=int(result.get("chunks", 0)), @@ -1734,7 +1811,7 @@ async def _do() -> Dict[str, Any]: timeoutSeconds=self._item_timeout_s, ) - except Exception as e: + except (AzureError, aiohttp.ClientError, APIError, RuntimeError, ValueError, OSError) as e: error = f"Item processing failed ({type(e).__name__})." logging.error("[%s] %s", self.cfg.indexer_name, error) processed += 1 diff --git a/jobs/sharepoint_ingestion_config.py b/jobs/sharepoint_ingestion_config.py index dfc614c8..e3e18d10 100644 --- a/jobs/sharepoint_ingestion_config.py +++ b/jobs/sharepoint_ingestion_config.py @@ -149,6 +149,8 @@ class RunStats: items_discovered: int = 0 items_candidates: int = 0 items_indexed: int = 0 + items_acl_updated: int = 0 + acl_documents_updated: int = 0 items_skipped_nochange: int = 0 items_failed: int = 0 items_skipped_blocked: int = 0 diff --git a/tests/test_sharepoint_acl_refresh.py b/tests/test_sharepoint_acl_refresh.py new file mode 100644 index 00000000..d153bad3 --- /dev/null +++ b/tests/test_sharepoint_acl_refresh.py @@ -0,0 +1,442 @@ +"""Permission-only synchronization exercises the real collection and Graph paths.""" + +import asyncio +import copy +from datetime import datetime, timezone +import re +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import aiohttp +from azure.core.exceptions import AzureError +from azure.search.documents.models import IndexingResult +import pytest + +from tests.test_worker_failure_boundaries import worker_module + + +USER = "11111111-1111-4111-8111-111111111111" +OLD_USER = "22222222-2222-4222-8222-222222222222" +GROUP = "33333333-3333-4333-8333-333333333333" +OLD_GROUP = "44444444-4444-4444-8444-444444444444" +STAMP = "2026-01-01T00:00:00Z" +ACL_FIELDS = ("metadata_security_user_ids", "metadata_security_group_ids") + + +def permissions(users=(USER,), groups=(GROUP,), *, inherited=False): + grants = [ + {"grantedToV2": {kind: {"id": principal}}} + for kind, principals in (("user", users), ("group", groups)) + for principal in principals + ] + if inherited: + for grant in grants: + grant["inheritedFrom"] = {"driveId": "drive", "id": "parent"} + return {"value": grants} + + +class Search: + def __init__(self, documents): + self.documents = {doc["id"]: copy.deepcopy(doc) for doc in documents} + self.queries = [] + self.batches = [] + self.reject = set() + + async def get_document(self, *, key, **kwargs): + return copy.deepcopy(self.documents[key]) + + async def search(self, **kwargs): + self.queries.append(kwargs) + parent = re.search(r"parent_id eq '((?:[^']|'')*)'", kwargs["filter"])[1].replace("''", "'") + bounds = re.search(r"parent_id ge '((?:[^']|'')*)' and parent_id lt '((?:[^']|'')*)'", kwargs["filter"]) + lower, upper = (value.replace("''", "'") for value in bounds.groups()) if bounds else ("", "") + + async def values(): + for doc in sorted(self.documents.values(), key=lambda value: value["id"]): + if doc["parent_id"] == parent or (bounds and lower <= doc["parent_id"] < upper): + yield {field: copy.deepcopy(doc.get(field)) for field in kwargs["select"]} + + return values() + + async def merge_documents(self, *, documents): + self.batches.append(copy.deepcopy(documents)) + results = [] + for doc in documents: + succeeded = doc["id"] not in self.reject + if succeeded: + self.documents[doc["id"]].update(copy.deepcopy(doc)) + results.append(IndexingResult.deserialize({ + "key": doc["id"], "status": succeeded, "statusCode": 200 if succeeded else 503, + })) + return results + + +@pytest.fixture +def connector(worker_module): + module = worker_module("sharepoint_indexer") + worker = object.__new__(module.SharePointIndexer) + worker.cfg = SimpleNamespace( + indexer_name="sharepoint-indexer", batch_size=2, max_file_processing_attempts=3, + max_concurrency=1, + ) + worker._item_timeout_s = 5 + worker._collection_gather_timeout_s = 5 + worker._http_total_timeout_s = 5 + worker._allowed_exts = {"pdf"} + worker._get_list_navigation_base_url = AsyncMock(return_value="") + worker._get_lookup_columns_map = AsyncMock(return_value={}) + worker._resolve_lookup_fields_for_item = AsyncMock(return_value={}) + worker._read_file_log = AsyncMock(return_value=None) + worker._write_file_log = AsyncMock() + worker._embed = AsyncMock(side_effect=AssertionError("Unchanged content must not be embedded")) + worker._delete_parent_docs = AsyncMock(side_effect=AssertionError("ACL refresh must not delete content")) + worker._upload_docs = AsyncMock(side_effect=AssertionError("ACL refresh must not upload content")) + graph = object.__new__(module.SharePointGraphClient) + graph._graph_base = "https://graph.microsoft.com/v1.0" + graph._graph_beta_base = "https://graph.microsoft.com/beta" + graph._gget = AsyncMock(return_value=permissions()) + graph.get_site_id = AsyncMock(return_value="site") + graph.get_drive_item = AsyncMock(return_value={ + "id": "file", "name": "document.pdf", "file": {"mimeType": "application/pdf"}, + "parentReference": {"driveId": "drive"}, + "fileSystemInfo": {"lastModifiedDateTime": STAMP}, + "lastModifiedDateTime": "2026-01-02T00:00:00Z", + }) + graph.download_drive_item = AsyncMock(side_effect=AssertionError("Unchanged content must not be downloaded")) + + async def items(**kwargs): + yield {"id": "item", "lastModifiedDateTime": STAMP, "fields": {"Title": "Synthetic item"}} + + graph.iter_items = items + worker._graph_client = graph + body = module._make_parent_key("example.org", "site", "list", "item") + file_parent = module._make_parent_key("example.org", "site", "list", "item", "document.pdf") + + def document(parent, number): + return { + "id": module._make_chunk_key(parent, number), "parent_id": parent, "source": "sharepoint", + "metadata_storage_last_modified": datetime(2026, 1, 1, tzinfo=timezone.utc), + "content": "Existing synthetic content", "contentVector": [0.1], + ACL_FIELDS[0]: [OLD_USER], ACL_FIELDS[1]: [OLD_GROUP], + } + + worker._search_client = Search([ + document(body, 0), *[document(file_parent, n) for n in range(3)], + document("unrelated", 0), + ]) + spec = {"siteDomain": "example.org", "siteName": "site", "listId": "list"} + return module, worker, graph, spec, document, body, file_parent + + +async def run_collection(connector, *, library=False): + module, worker, _, spec, *_ = connector + stats = module.RunStats() + result = await worker._process_collection( + object(), {**spec, "listType": "documentLibrary" if library else "genericList"}, + "run", asyncio.Semaphore(1), stats, asyncio.Lock(), + ) + return result, stats + + +@pytest.mark.asyncio +@pytest.mark.parametrize("library", [False, True]) +@pytest.mark.parametrize("change", ["grant", "revoke", "empty", "inherited", "break-inheritance", "restore-inheritance"]) +async def test_unchanged_content_refreshes_every_affected_acl(connector, library, change): + _, worker, graph, _, _, body, file_parent = connector + users = [OLD_USER, USER] if change == "grant" else ([] if change == "empty" else [USER]) + groups = [] if change == "empty" else [GROUP] + graph._gget.return_value = permissions( + users, groups, inherited=change in {"inherited", "restore-inheritance"}, + ) + before = copy.deepcopy(worker._search_client.documents) + result, stats = await run_collection(connector, library=library) + parents = {body, file_parent} if library else {body} + for key, doc in worker._search_client.documents.items(): + if doc["parent_id"] in parents: + assert set(doc[ACL_FIELDS[0]]) == set(users) + assert doc[ACL_FIELDS[1]] == groups + assert {k: v for k, v in doc.items() if k not in ACL_FIELDS} == { + k: v for k, v in before[key].items() if k not in ACL_FIELDS + } + else: + assert doc == before[key] + assert result["failed"] == 0 and result["success"] == 1 and result["chunks"] == 0 + assert stats.items_indexed == 0 and stats.items_skipped_nochange == 0 + assert worker._write_file_log.await_args.args[1]["status"] == "acl-updated" + assert all(set(doc) == {"id", *ACL_FIELDS} for batch in worker._search_client.batches for doc in batch) + assert all(len(batch) <= worker.cfg.batch_size for batch in worker._search_client.batches) + worker._embed.assert_not_awaited() + worker._upload_docs.assert_not_awaited() + worker._delete_parent_docs.assert_not_awaited() + graph.download_drive_item.assert_not_awaited() + if library: + assert "/drives/drive/items/file/permissions" in graph._gget.await_args.args[1] + + +@pytest.mark.asyncio +async def test_equal_acl_order_or_duplicates_do_not_write_or_reprocess(connector): + _, worker, graph, *_ = connector + for doc in worker._search_client.documents.values(): + doc[ACL_FIELDS[0]] = [USER, OLD_USER] + doc[ACL_FIELDS[1]] = [GROUP] + graph._gget.return_value = permissions([OLD_USER, USER, USER], [GROUP]) + result, stats = await run_collection(connector, library=True) + assert result["success"] == 0 and result["failed"] == 0 + assert stats.items_skipped_nochange == 1 + assert worker._search_client.batches == [] + worker._embed.assert_not_awaited() + graph.download_drive_item.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_all_chunks_are_scanned_without_a_top_limit(connector): + _, worker, _, _, make_document, _, file_parent = connector + worker.cfg.batch_size = 1500 + for n in range(1005): + doc = make_document(file_parent, n) + worker._search_client.documents[doc["id"]] = doc + result, _ = await run_collection(connector, library=True) + assert result["failed"] == 0 + assert sum(len(batch) for batch in worker._search_client.batches) == 1006 + assert all("top" not in query and "skip" not in query for query in worker._search_client.queries) + assert all(query["headers"] == {"x-ms-enable-elevated-read": "true"} for query in worker._search_client.queries) + assert all(query["order_by"] == ["id asc"] for query in worker._search_client.queries) + assert max(map(len, worker._search_client.batches)) <= 1000 + + +@pytest.mark.asyncio +async def test_item_range_updates_old_file_names_but_not_sibling_items(connector): + _, worker, _, _, make_document, body, _ = connector + before = {} + for parent in [body + "0", body + "2/document.pdf", body + "-sibling", body + "/previous-name.pdf"]: + doc = make_document(parent, 0) + worker._search_client.documents[doc["id"]] = doc + before[doc["id"]] = copy.deepcopy(doc) + result, _ = await run_collection(connector, library=True) + assert result["failed"] == 0 + for key, old in before.items(): + doc = worker._search_client.documents[key] + if old["parent_id"] == body + "/previous-name.pdf": + assert doc[ACL_FIELDS[0]] == [USER] and doc[ACL_FIELDS[1]] == [GROUP] + else: + assert doc == old + + +@pytest.mark.asyncio +async def test_inheritance_changes_replace_previous_effective_grants_without_parent_union(connector): + _, worker, graph, _, _, body, file_parent = connector + for users, groups, inherited in [ + ([OLD_USER], [OLD_GROUP], True), + ([USER], [GROUP], True), + ([USER], [], False), + ([OLD_USER], [GROUP], True), + ]: + graph._gget.return_value = permissions(users, groups, inherited=inherited) + result, _ = await run_collection(connector, library=True) + assert result["failed"] == 0 + for doc in worker._search_client.documents.values(): + if doc["parent_id"] in {body, file_parent}: + assert doc[ACL_FIELDS[0]] == users and doc[ACL_FIELDS[1]] == groups + assert graph._gget.await_count == 4 + assert all("/drives/drive/items/file/permissions?" in call.args[1] for call in graph._gget.await_args_list) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("status", [400, 401, 403, 404, 501]) +async def test_failed_permission_lookup_does_not_clear_acl_or_claim_success(connector, status, caplog): + _, worker, graph, *_ = connector + graph._gget.side_effect = aiohttp.ClientResponseError( + SimpleNamespace(real_url="https://graph.microsoft.com"), (), status=status, + message="private-permission-canary", + ) + before = copy.deepcopy(worker._search_client.documents) + result, stats = await run_collection(connector, library=True) + assert result["failed"] == stats.items_failed == 1 and result["success"] == 0 + assert worker._search_client.documents == before and not worker._search_client.batches + assert worker._write_file_log.await_args.args[1]["status"] == "error" + assert "private-permission-canary" not in caplog.text + + +@pytest.mark.asyncio +@pytest.mark.parametrize("payload", [{}, {"value": None}, {"value": [None]}, {"value": [{"grantedToV2": []}]}]) +async def test_invalid_permission_response_is_not_a_successful_empty_acl(connector, payload): + _, worker, graph, *_ = connector + graph._gget.return_value = payload + result, _ = await run_collection(connector) + assert result["failed"] == 1 and result["success"] == 0 + assert worker._search_client.batches == [] + + +@pytest.mark.asyncio +async def test_permission_pages_are_complete_before_any_acl_write(connector): + _, worker, graph, *_ = connector + next_url = graph._graph_beta_base + "/sites/site/lists/list/items/item/permissions?$skiptoken=next" + graph._gget.side_effect = [ + {**permissions([USER], []), "@odata.nextLink": next_url}, permissions([], [GROUP]), + ] + result, _ = await run_collection(connector) + assert result["failed"] == 0 + doc = worker._search_client.batches[0][0] + assert doc[ACL_FIELDS[0]] == [USER] and doc[ACL_FIELDS[1]] == [GROUP] + assert graph._gget.await_count == 2 + + +@pytest.mark.asyncio +async def test_late_permission_page_failure_never_commits_partial_acl(connector): + _, worker, graph, *_ = connector + next_url = graph._graph_beta_base + "/sites/site/lists/list/items/item/permissions?$skiptoken=next" + graph._gget.side_effect = [ + {**permissions([USER], []), "@odata.nextLink": next_url}, AzureError("private-late-page"), + ] + result, _ = await run_collection(connector) + assert result["failed"] == 1 and worker._search_client.batches == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("next_link", [ + "https://untrusted.example/permissions", "http://graph.microsoft.com/beta/sites/site/lists/list/items/item/permissions", + "https://graph.microsoft.com/beta/sites/other/lists/list/items/item/permissions", "", 7, +]) +async def test_invalid_permission_pagination_never_forwards_credentials_or_commits_acl(connector, next_link): + _, worker, graph, *_ = connector + graph._gget.return_value = {**permissions(), "@odata.nextLink": next_link} + result, _ = await run_collection(connector) + assert result["failed"] == 1 and worker._search_client.batches == [] + assert graph._gget.await_count == 1 + + +@pytest.mark.asyncio +async def test_permission_pagination_cycle_fails_instead_of_accepting_partial_acl(connector): + _, worker, graph, *_ = connector + next_url = graph._graph_beta_base + "/sites/site/lists/list/items/item/permissions?$skiptoken=cycle" + graph._gget.return_value = {**permissions(), "@odata.nextLink": next_url} + result, _ = await run_collection(connector) + assert result["failed"] == 1 and graph._gget.await_count == 2 + assert worker._search_client.batches == [] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("case", ["missing", "failed", "duplicate", "unrelated"]) +async def test_acl_merges_require_confirmation_for_every_requested_key(connector, case): + _, worker, *_ = connector + + def result(key, success=True): + return IndexingResult.deserialize({"key": key, "status": success, "statusCode": 200}) + + async def merge_documents(**kwargs): + return { + "missing": [], "failed": [result("key", False)], + "duplicate": [result("key"), result("key")], "unrelated": [result("other")], + }[case] + + with pytest.raises(AzureError, match="confirm"): + await worker._with_backoff(merge_documents, documents=[{"id": "key"}]) + + +@pytest.mark.asyncio +async def test_absent_permission_client_is_not_a_successful_empty_acl(connector): + _, worker, *_ = connector + worker._graph_client = None + with pytest.raises(RuntimeError, match="unavailable"): + await worker._get_security_principals_for_item( + session=object(), site_id="site", collection_id="list", item_id="item", + ) + + +@pytest.mark.asyncio +async def test_changed_content_retains_full_indexing_with_current_acl(connector): + _, worker, _, _, _, body, _ = connector + worker._search_client.documents[next( + key for key, doc in worker._search_client.documents.items() if doc["parent_id"] == body + )]["metadata_storage_last_modified"] = datetime(2025, 1, 1, tzinfo=timezone.utc) + worker._embed.side_effect = None + worker._embed.return_value = [0.2] + worker._delete_parent_docs.side_effect = None + worker._upload_docs.side_effect = None + result, stats = await run_collection(connector) + assert result["failed"] == 0 and result["success"] == 1 and stats.items_indexed == 1 + worker._embed.assert_awaited_once() + worker._delete_parent_docs.assert_awaited_once_with(body) + doc = worker._upload_docs.await_args.args[0][0] + assert doc[ACL_FIELDS[0]] == [USER] and doc[ACL_FIELDS[1]] == [GROUP] + assert doc["contentVector"] == [0.2] + + +@pytest.mark.asyncio +async def test_partial_search_merge_is_visible_and_retry_updates_only_stale_chunks(connector): + _, worker, _, _, _, _, file_parent = connector + failed_id = next( + key for key, doc in worker._search_client.documents.items() if doc["parent_id"] == file_parent + ) + worker._search_client.reject = {failed_id} + result, stats = await run_collection(connector, library=True) + assert result["failed"] == stats.items_failed == 1 and result["success"] == 0 + assert worker._write_file_log.await_args.args[1]["status"] == "error" + assert worker._search_client.documents[failed_id][ACL_FIELDS[0]] == [OLD_USER] + worker._search_client.reject.clear() + worker._search_client.batches.clear() + result, _ = await run_collection(connector, library=True) + assert result["failed"] == 0 and result["success"] == 1 + updated = [doc["id"] for batch in worker._search_client.batches for doc in batch] + assert failed_id in updated + assert len(updated) < 4 + assert all(doc[ACL_FIELDS[0]] == [USER] for doc in worker._search_client.documents.values() + if doc["parent_id"] != "unrelated") + + +@pytest.mark.asyncio +async def test_content_retry_block_does_not_block_permission_revocations(connector): + _, worker, *_ = connector + worker._read_file_log.return_value = { + "blocked": True, "processingAttempts": 3, "blockedReason": "max_attempts_exceeded", + } + result, stats = await run_collection(connector, library=True) + assert result["failed"] == 0 and result["success"] == 1 + assert stats.items_skipped_blocked == 1 + log = worker._write_file_log.await_args.args[1] + assert log["blocked"] is True and log["processingAttempts"] == 3 + worker._embed.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_run_with_partial_item_failure_is_not_reported_finished(connector): + _, worker, _, spec, *_ = connector + worker._ensure_clients = AsyncMock() + worker._close_clients = AsyncMock() + worker._hydrate_site_configs_from_cosmos = AsyncMock() + worker._parse_collections = lambda: [spec] + worker._process_collection = AsyncMock(return_value={"items": 1, "success": 0, "failed": 1, "chunks": 0}) + summaries = [] + + async def write(run, summary): + summaries.append(copy.deepcopy(summary)) + + worker._write_run_summary_safely = write + with pytest.raises(RuntimeError, match="failed"): + await worker.run() + assert summaries[-1]["status"] == "failed" + worker._close_clients.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_unexpected_item_defect_propagates_and_never_finishes_the_job(connector, caplog): + _, worker, graph, spec, *_ = connector + defect = TypeError("private-defect-canary") + graph._gget.side_effect = defect + worker._ensure_clients = AsyncMock() + worker._close_clients = AsyncMock() + worker._hydrate_site_configs_from_cosmos = AsyncMock() + worker._parse_collections = lambda: [spec] + summaries = [] + + async def write(run, summary): + summaries.append(copy.deepcopy(summary)) + + worker._write_run_summary_safely = write + with pytest.raises(TypeError) as raised: + await worker.run() + assert raised.value is defect + assert summaries[-1]["status"] == "failed" + assert not worker._search_client.batches + worker._close_clients.assert_awaited_once() + assert "private-defect-canary" not in caplog.text + str(summaries)