wehub-resource-sync
fed8b2eed7
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Backend release / release (push) Has been cancelled
Bandit Security Scan / bandit_scan (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / manifest (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / manifest (push) Has been cancelled
Python linting / ruff (push) Has been cancelled
Run python tests with pytest / Run tests and count coverage (3.12) (push) Has been cancelled
React Widget Build / build (push) Has been cancelled
1037 lines
41 KiB
Python
1037 lines
41 KiB
Python
"""Tests for SharePointLoader."""
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
import requests as real_requests
|
|
|
|
from application.parser.schema.base import Document
|
|
|
|
|
|
def _make_loader(access_token="at", refresh_token="rt", allows_shared=False):
|
|
"""Create a SharePointLoader with mocked dependencies."""
|
|
with patch("application.parser.connectors.share_point.loader.SharePointAuth") as MockAuth:
|
|
mock_auth = MagicMock()
|
|
mock_auth.get_token_info_from_session.return_value = {
|
|
"access_token": access_token,
|
|
"refresh_token": refresh_token,
|
|
"allows_shared_content": allows_shared,
|
|
}
|
|
mock_auth.is_token_expired.return_value = False
|
|
MockAuth.return_value = mock_auth
|
|
|
|
from application.parser.connectors.share_point.loader import SharePointLoader
|
|
loader = SharePointLoader("session_tok")
|
|
return loader
|
|
|
|
|
|
@pytest.fixture
|
|
def loader():
|
|
return _make_loader()
|
|
|
|
|
|
@pytest.fixture
|
|
def loader_shared():
|
|
return _make_loader(allows_shared=True)
|
|
|
|
|
|
class TestSharePointLoaderInit:
|
|
|
|
@pytest.mark.unit
|
|
def test_init_sets_attributes(self, loader):
|
|
assert loader.session_token == "session_tok"
|
|
assert loader.access_token == "at"
|
|
assert loader.refresh_token == "rt"
|
|
assert loader.allows_shared_content is False
|
|
assert loader.next_page_token is None
|
|
|
|
@pytest.mark.unit
|
|
def test_no_access_token_raises(self):
|
|
with patch("application.parser.connectors.share_point.loader.SharePointAuth") as MockAuth:
|
|
mock_auth = MagicMock()
|
|
mock_auth.get_token_info_from_session.return_value = {
|
|
"access_token": None,
|
|
"refresh_token": "rt",
|
|
}
|
|
MockAuth.return_value = mock_auth
|
|
|
|
from application.parser.connectors.share_point.loader import SharePointLoader
|
|
with pytest.raises(ValueError, match="No access token"):
|
|
SharePointLoader("st")
|
|
|
|
|
|
class TestGetHeaders:
|
|
|
|
@pytest.mark.unit
|
|
def test_returns_bearer_token(self, loader):
|
|
headers = loader._get_headers()
|
|
assert headers["Authorization"] == "Bearer at"
|
|
assert headers["Accept"] == "application/json"
|
|
|
|
|
|
class TestEnsureValidToken:
|
|
|
|
@pytest.mark.unit
|
|
def test_valid_token_no_op(self, loader):
|
|
loader.auth.is_token_expired.return_value = False
|
|
loader._ensure_valid_token()
|
|
|
|
@pytest.mark.unit
|
|
def test_no_access_token_raises(self, loader):
|
|
loader.access_token = None
|
|
with pytest.raises(ValueError, match="No access token"):
|
|
loader._ensure_valid_token()
|
|
|
|
@pytest.mark.unit
|
|
def test_expired_token_refreshes(self, loader):
|
|
loader.auth.is_token_expired.return_value = True
|
|
loader.auth.refresh_access_token.return_value = {"access_token": "new_at"}
|
|
loader._ensure_valid_token()
|
|
assert loader.access_token == "new_at"
|
|
|
|
@pytest.mark.unit
|
|
def test_refresh_failure_raises(self, loader):
|
|
loader.auth.is_token_expired.return_value = True
|
|
loader.auth.refresh_access_token.side_effect = Exception("fail")
|
|
with pytest.raises(ValueError, match="Failed to refresh"):
|
|
loader._ensure_valid_token()
|
|
|
|
|
|
class TestGetItemUrl:
|
|
|
|
@pytest.mark.unit
|
|
def test_simple_id(self, loader):
|
|
url = loader._get_item_url("item123")
|
|
assert url == f"{loader.GRAPH_API_BASE}/me/drive/items/item123"
|
|
|
|
@pytest.mark.unit
|
|
def test_drive_colon_item(self, loader):
|
|
url = loader._get_item_url("driveABC:itemXYZ")
|
|
assert url == f"{loader.GRAPH_API_BASE}/drives/driveABC/items/itemXYZ"
|
|
|
|
|
|
class TestResolveMimeType:
|
|
|
|
@pytest.mark.unit
|
|
def test_supported_mime_from_file_data(self, loader):
|
|
resource = {"file": {"mimeType": "application/pdf"}, "name": "test.pdf"}
|
|
mime, supported = loader._resolve_mime_type(resource)
|
|
assert mime == "application/pdf"
|
|
assert supported is True
|
|
|
|
@pytest.mark.unit
|
|
def test_fallback_to_extension(self, loader):
|
|
resource = {"file": {"mimeType": "application/octet-stream"}, "name": "test.docx"}
|
|
mime, supported = loader._resolve_mime_type(resource)
|
|
assert mime == "application/vnd.openxmlformats-officedocument.wordprocessingml.document"
|
|
assert supported is True
|
|
|
|
@pytest.mark.unit
|
|
def test_unsupported_returns_false(self, loader):
|
|
resource = {"file": {"mimeType": "application/zip"}, "name": "test.zip"}
|
|
mime, supported = loader._resolve_mime_type(resource)
|
|
assert supported is False
|
|
|
|
@pytest.mark.unit
|
|
def test_no_file_data_fallback_to_extension(self, loader):
|
|
resource = {"name": "readme.txt"}
|
|
mime, supported = loader._resolve_mime_type(resource)
|
|
assert mime == "text/plain"
|
|
assert supported is True
|
|
|
|
@pytest.mark.unit
|
|
def test_no_file_data_no_extension(self, loader):
|
|
resource = {"name": "noext"}
|
|
mime, supported = loader._resolve_mime_type(resource)
|
|
assert mime == "application/octet-stream"
|
|
assert supported is False
|
|
|
|
|
|
class TestProcessFile:
|
|
|
|
@pytest.mark.unit
|
|
def test_supported_file_with_content(self, loader):
|
|
loader._download_file_content = MagicMock(return_value="content")
|
|
metadata = {
|
|
"id": "f1",
|
|
"name": "test.pdf",
|
|
"file": {"mimeType": "application/pdf"},
|
|
"size": 1024,
|
|
"createdDateTime": "2025-01-01",
|
|
"lastModifiedDateTime": "2025-01-02",
|
|
}
|
|
doc = loader._process_file(metadata, load_content=True)
|
|
assert doc is not None
|
|
assert doc.text == "content"
|
|
assert doc.extra_info["source"] == "share_point"
|
|
|
|
@pytest.mark.unit
|
|
def test_supported_file_no_content(self, loader):
|
|
metadata = {
|
|
"id": "f1",
|
|
"name": "test.pdf",
|
|
"file": {"mimeType": "application/pdf"},
|
|
}
|
|
doc = loader._process_file(metadata, load_content=False)
|
|
assert doc.text == ""
|
|
|
|
@pytest.mark.unit
|
|
def test_unsupported_returns_none(self, loader):
|
|
metadata = {
|
|
"id": "f1",
|
|
"name": "test.zip",
|
|
"file": {"mimeType": "application/zip"},
|
|
}
|
|
doc = loader._process_file(metadata)
|
|
assert doc is None
|
|
|
|
@pytest.mark.unit
|
|
def test_download_failure_returns_none(self, loader):
|
|
loader._download_file_content = MagicMock(return_value=None)
|
|
metadata = {
|
|
"id": "f1",
|
|
"name": "test.txt",
|
|
"file": {"mimeType": "text/plain"},
|
|
}
|
|
doc = loader._process_file(metadata, load_content=True)
|
|
assert doc is None
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_none(self, loader):
|
|
loader._download_file_content = MagicMock(side_effect=Exception("fail"))
|
|
metadata = {"id": "f1", "name": "test.txt", "file": {"mimeType": "text/plain"}}
|
|
doc = loader._process_file(metadata, load_content=True)
|
|
assert doc is None
|
|
|
|
|
|
class TestLoadData:
|
|
|
|
@pytest.mark.unit
|
|
def test_load_specific_files(self, loader):
|
|
doc = Document(text="c", doc_id="f1", extra_info={"file_name": "test.pdf"})
|
|
loader._load_file_by_id = MagicMock(return_value=doc)
|
|
result = loader.load_data({"file_ids": ["f1"]})
|
|
assert len(result) == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_load_files_with_search_filter(self, loader):
|
|
doc = Document(text="c", doc_id="f1", extra_info={"file_name": "report.pdf"})
|
|
loader._load_file_by_id = MagicMock(return_value=doc)
|
|
result = loader.load_data({"file_ids": ["f1"], "search_query": "report"})
|
|
assert len(result) == 1
|
|
result = loader.load_data({"file_ids": ["f1"], "search_query": "other"})
|
|
assert len(result) == 0
|
|
|
|
@pytest.mark.unit
|
|
def test_load_files_error_continues(self, loader):
|
|
loader._load_file_by_id = MagicMock(side_effect=Exception("fail"))
|
|
result = loader.load_data({"file_ids": ["f1"]})
|
|
assert len(result) == 0
|
|
|
|
@pytest.mark.unit
|
|
def test_shared_mode(self, loader_shared):
|
|
loader_shared._list_shared_items = MagicMock(return_value=[
|
|
Document(text="", doc_id="s1", extra_info={})
|
|
])
|
|
result = loader_shared.load_data({"shared": True})
|
|
assert len(result) == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_shared_mode_personal_account_returns_empty(self, loader):
|
|
result = loader.load_data({"shared": True})
|
|
assert len(result) == 0
|
|
|
|
@pytest.mark.unit
|
|
def test_browse_mode_uses_list_items(self, loader):
|
|
loader._list_items_in_parent = MagicMock(return_value=[])
|
|
loader.load_data({"folder_id": "folder1"})
|
|
loader._list_items_in_parent.assert_called_once()
|
|
|
|
@pytest.mark.unit
|
|
def test_browse_mode_defaults_to_root(self, loader):
|
|
loader._list_items_in_parent = MagicMock(return_value=[])
|
|
loader.load_data({})
|
|
args = loader._list_items_in_parent.call_args
|
|
assert args[0][0] == "root"
|
|
|
|
@pytest.mark.unit
|
|
def test_list_only_mode(self, loader):
|
|
loader._list_items_in_parent = MagicMock(return_value=[])
|
|
loader.load_data({"list_only": True})
|
|
args = loader._list_items_in_parent.call_args
|
|
assert args[1]["load_content"] is False
|
|
|
|
@pytest.mark.unit
|
|
def test_outer_exception_raises(self, loader):
|
|
"""The outer try/except in load_data re-raises unexpected errors."""
|
|
loader._list_items_in_parent = MagicMock(side_effect=RuntimeError("unexpected"))
|
|
with pytest.raises(RuntimeError, match="unexpected"):
|
|
loader.load_data({})
|
|
|
|
|
|
class TestLoadFileById:
|
|
|
|
@pytest.mark.unit
|
|
def test_loads_and_processes(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"id": "f1", "name": "test.pdf", "file": {"mimeType": "application/pdf"}
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._process_file = MagicMock(return_value=Document(text="", doc_id="f1", extra_info={}))
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
doc = loader._load_file_by_id("f1")
|
|
assert doc is not None
|
|
|
|
@pytest.mark.unit
|
|
def test_http_error_retries_with_refresh(self, loader):
|
|
mock_resp_401 = MagicMock()
|
|
mock_resp_401.status_code = 401
|
|
|
|
http_error = real_requests.exceptions.HTTPError(response=mock_resp_401)
|
|
|
|
mock_resp_ok = MagicMock()
|
|
mock_resp_ok.json.return_value = {"id": "f1", "name": "t.pdf", "file": {"mimeType": "application/pdf"}}
|
|
mock_resp_ok.raise_for_status = MagicMock()
|
|
|
|
call_count = [0]
|
|
def get_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise http_error
|
|
return mock_resp_ok
|
|
|
|
loader._process_file = MagicMock(return_value=Document(text="", doc_id="f1", extra_info={}))
|
|
loader.auth.refresh_access_token.return_value = {"access_token": "new_at"}
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=get_side_effect):
|
|
doc = loader._load_file_by_id("f1")
|
|
assert doc is not None
|
|
|
|
@pytest.mark.unit
|
|
def test_general_exception_returns_none(self, loader):
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=Exception("fail")):
|
|
doc = loader._load_file_by_id("f1")
|
|
assert doc is None
|
|
|
|
|
|
class TestListItemsInParent:
|
|
|
|
@pytest.mark.unit
|
|
def test_lists_files_and_folders(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [
|
|
{"id": "folder1", "name": "Docs", "folder": {}, "size": 0, "createdDateTime": "d1", "lastModifiedDateTime": "d2"},
|
|
{"id": "file1", "name": "test.pdf", "file": {"mimeType": "application/pdf"}, "size": 100},
|
|
],
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._process_file = MagicMock(return_value=Document(text="", doc_id="file1", extra_info={}))
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
docs = loader._list_items_in_parent("root")
|
|
assert len(docs) == 2
|
|
assert docs[0].extra_info["is_folder"] is True
|
|
|
|
@pytest.mark.unit
|
|
def test_search_query_uses_search_url(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"value": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response) as mock_get:
|
|
loader._list_items_in_parent("root", search_query="report")
|
|
call_url = mock_get.call_args[0][0]
|
|
assert "search" in call_url
|
|
|
|
@pytest.mark.unit
|
|
def test_search_with_drive_id(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"value": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response) as mock_get:
|
|
loader._list_items_in_parent("drive1:folder1", search_query="test")
|
|
call_url = mock_get.call_args[0][0]
|
|
assert "drives/drive1" in call_url
|
|
|
|
@pytest.mark.unit
|
|
def test_pagination_token_stored(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [
|
|
{"id": "f1", "name": "test.txt", "file": {"mimeType": "text/plain"}},
|
|
],
|
|
"@odata.nextLink": "https://graph.microsoft.com/v1.0/items?$skiptoken=abc123",
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
loader._process_file = MagicMock(return_value=Document(text="", doc_id="f1", extra_info={}))
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
loader._list_items_in_parent("root")
|
|
assert loader.next_page_token == "abc123"
|
|
|
|
@pytest.mark.unit
|
|
def test_no_next_link_clears_token(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"value": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
loader._list_items_in_parent("root")
|
|
assert loader.next_page_token is None
|
|
|
|
@pytest.mark.unit
|
|
def test_limit_stops_early(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [
|
|
{"id": f"f{i}", "name": f"f{i}.txt", "file": {"mimeType": "text/plain"}} for i in range(10)
|
|
],
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
loader._process_file = MagicMock(side_effect=lambda m, **kw: Document(text="", doc_id=m["id"], extra_info={}))
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
docs = loader._list_items_in_parent("root", limit=3)
|
|
assert len(docs) == 3
|
|
|
|
@pytest.mark.unit
|
|
def test_page_token_passed_as_skip_token(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"value": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response) as mock_get:
|
|
loader._list_items_in_parent("root", page_token="tok123")
|
|
call_kwargs = mock_get.call_args[1]
|
|
assert call_kwargs["params"]["$skipToken"] == "tok123"
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_partial(self, loader):
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=Exception("fail")):
|
|
docs = loader._list_items_in_parent("root")
|
|
assert docs == []
|
|
|
|
@pytest.mark.unit
|
|
def test_next_link_without_skiptoken(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [],
|
|
"@odata.nextLink": "https://graph.microsoft.com/v1.0/items?other=param",
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
loader._list_items_in_parent("root")
|
|
assert loader.next_page_token is None
|
|
|
|
|
|
class TestDownloadFileContent:
|
|
|
|
@pytest.mark.unit
|
|
def test_downloads_and_decodes(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.content = b"file content"
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
content = loader._download_file_content("f1")
|
|
assert content == "file content"
|
|
|
|
@pytest.mark.unit
|
|
def test_unicode_decode_error_returns_none(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.content = MagicMock()
|
|
mock_response.content.decode.side_effect = UnicodeDecodeError("utf-8", b"", 0, 1, "bad")
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
result = loader._download_file_content("f1")
|
|
assert result is None
|
|
|
|
@pytest.mark.unit
|
|
def test_general_exception_returns_none(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=Exception("fail")):
|
|
result = loader._download_file_content("f1")
|
|
assert result is None
|
|
|
|
@pytest.mark.unit
|
|
def test_http_401_retries_with_refresh(self, loader):
|
|
mock_resp_401 = MagicMock()
|
|
mock_resp_401.status_code = 401
|
|
http_error = real_requests.exceptions.HTTPError(response=mock_resp_401)
|
|
|
|
mock_resp_ok = MagicMock()
|
|
mock_resp_ok.content = b"content after refresh"
|
|
mock_resp_ok.raise_for_status = MagicMock()
|
|
|
|
call_count = [0]
|
|
def get_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise http_error
|
|
return mock_resp_ok
|
|
|
|
loader.auth.refresh_access_token.return_value = {"access_token": "new_at"}
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=get_side_effect):
|
|
content = loader._download_file_content("f1")
|
|
assert content == "content after refresh"
|
|
|
|
|
|
class TestListSharedItems:
|
|
|
|
@pytest.mark.unit
|
|
def test_lists_shared_files(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value="https://user.sharepoint.com/drive")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{
|
|
"hitsContainers": [{
|
|
"total": 2,
|
|
"hits": [
|
|
{
|
|
"resource": {
|
|
"id": "item1",
|
|
"name": "shared.pdf",
|
|
"file": {"mimeType": "application/pdf"},
|
|
"size": 100,
|
|
"parentReference": {"driveId": "d1"},
|
|
}
|
|
},
|
|
{
|
|
"resource": {
|
|
"id": "item2",
|
|
"name": "folder1",
|
|
"folder": {},
|
|
"parentReference": {"driveId": "d1"},
|
|
}
|
|
},
|
|
],
|
|
}]
|
|
}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(limit=100, load_content=False)
|
|
assert len(docs) == 2
|
|
|
|
@pytest.mark.unit
|
|
def test_deduplicates_hits(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{
|
|
"hitsContainers": [{
|
|
"total": 2,
|
|
"hits": [
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}},
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}},
|
|
],
|
|
}]
|
|
}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(limit=100, load_content=False)
|
|
assert len(docs) == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_pagination_with_offset(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
hits = [
|
|
{"resource": {"id": f"item{i}", "name": f"f{i}.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}}
|
|
for i in range(5)
|
|
]
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 5, "hits": hits}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(limit=2, page_token="2", load_content=False)
|
|
assert len(docs) == 2
|
|
assert loader_shared.next_page_token == "4"
|
|
|
|
@pytest.mark.unit
|
|
def test_offset_beyond_results_returns_empty(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 1, "hits": [
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}}
|
|
]}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(page_token="999")
|
|
assert len(docs) == 0
|
|
assert loader_shared.next_page_token is None
|
|
|
|
@pytest.mark.unit
|
|
def test_invalid_page_token_defaults_to_zero(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 1, "hits": [
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}}
|
|
]}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(page_token="invalid", load_content=False)
|
|
assert len(docs) == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_negative_page_token_defaults_to_zero(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 1, "hits": [
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}}
|
|
]}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(page_token="-5", load_content=False)
|
|
assert len(docs) == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_empty_search_response(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"value": []}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items()
|
|
assert len(docs) == 0
|
|
|
|
@pytest.mark.unit
|
|
def test_empty_hits_containers(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": []}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items()
|
|
assert len(docs) == 0
|
|
|
|
@pytest.mark.unit
|
|
def test_unsupported_shared_file_skipped(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 1, "hits": [
|
|
{"resource": {"id": "item1", "name": "f.zip", "file": {"mimeType": "application/zip"}, "parentReference": {"driveId": "d1"}}}
|
|
]}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(load_content=False)
|
|
assert len(docs) == 0
|
|
|
|
@pytest.mark.unit
|
|
def test_shared_item_with_content_download(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
loader_shared._download_file_content = MagicMock(return_value="content")
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 1, "hits": [
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}}
|
|
]}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
docs = loader_shared._list_shared_items(load_content=True)
|
|
assert len(docs) == 1
|
|
assert docs[0].text == "content"
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_empty(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", side_effect=Exception("fail")):
|
|
docs = loader_shared._list_shared_items()
|
|
assert docs == []
|
|
|
|
@pytest.mark.unit
|
|
def test_last_page_clears_next_token(self, loader_shared):
|
|
loader_shared._get_user_drive_web_url = MagicMock(return_value=None)
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [{"hitsContainers": [{"total": 1, "hits": [
|
|
{"resource": {"id": "item1", "name": "f.pdf", "file": {"mimeType": "application/pdf"}, "parentReference": {"driveId": "d1"}}}
|
|
]}]}]
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.post", return_value=mock_response):
|
|
loader_shared._list_shared_items(limit=100, load_content=False)
|
|
assert loader_shared.next_page_token is None
|
|
|
|
|
|
class TestGetUserDriveWebUrl:
|
|
|
|
@pytest.mark.unit
|
|
def test_returns_web_url(self, loader):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"webUrl": "https://user.sharepoint.com/drive"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
url = loader._get_user_drive_web_url()
|
|
assert url == "https://user.sharepoint.com/drive"
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_none(self, loader):
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=Exception("fail")):
|
|
url = loader._get_user_drive_web_url()
|
|
assert url is None
|
|
|
|
|
|
class TestBuildSharedKqlQuery:
|
|
|
|
@pytest.mark.unit
|
|
def test_with_search_and_drive_url(self, loader):
|
|
query = loader._build_shared_kql_query("test", "https://drive.url")
|
|
assert 'test AND -path:"https://drive.url"' == query
|
|
|
|
@pytest.mark.unit
|
|
def test_with_search_no_drive_url(self, loader):
|
|
query = loader._build_shared_kql_query("test", None)
|
|
assert query == "test"
|
|
|
|
@pytest.mark.unit
|
|
def test_no_search_with_drive_url(self, loader):
|
|
query = loader._build_shared_kql_query(None, "https://drive.url")
|
|
assert '* AND -path:"https://drive.url"' == query
|
|
|
|
@pytest.mark.unit
|
|
def test_no_search_no_drive_url(self, loader):
|
|
query = loader._build_shared_kql_query(None, None)
|
|
assert query == "*"
|
|
|
|
|
|
class TestDownloadToDirectory:
|
|
|
|
@pytest.mark.unit
|
|
def test_download_files(self, loader, tmp_path):
|
|
loader._download_file_to_directory = MagicMock(return_value=True)
|
|
result = loader.download_to_directory(str(tmp_path), {"file_ids": ["f1", "f2"]})
|
|
assert result["files_downloaded"] == 2
|
|
assert result["source_type"] == "share_point"
|
|
|
|
@pytest.mark.unit
|
|
def test_download_files_string_id(self, loader, tmp_path):
|
|
loader._download_file_to_directory = MagicMock(return_value=True)
|
|
result = loader.download_to_directory(str(tmp_path), {"file_ids": "single"})
|
|
assert result["files_downloaded"] == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_download_folders(self, loader, tmp_path):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"name": "MyFolder"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._download_folder_recursive = MagicMock(return_value=3)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
result = loader.download_to_directory(str(tmp_path), {"folder_ids": ["folder1"]})
|
|
assert result["files_downloaded"] == 3
|
|
|
|
@pytest.mark.unit
|
|
def test_download_folders_string_id(self, loader, tmp_path):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {"name": "F"}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._download_folder_recursive = MagicMock(return_value=1)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
result = loader.download_to_directory(str(tmp_path), {"folder_ids": "single_folder"})
|
|
assert result["files_downloaded"] == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_no_ids_returns_error(self, loader, tmp_path):
|
|
result = loader.download_to_directory(str(tmp_path), {})
|
|
assert "error" in result
|
|
|
|
@pytest.mark.unit
|
|
def test_none_config(self, loader, tmp_path):
|
|
result = loader.download_to_directory(str(tmp_path))
|
|
assert "error" in result
|
|
|
|
@pytest.mark.unit
|
|
def test_folder_error_continues(self, loader, tmp_path):
|
|
mock_response = MagicMock()
|
|
mock_response.raise_for_status.side_effect = Exception("fail")
|
|
|
|
loader._download_file_to_directory = MagicMock(return_value=True)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
result = loader.download_to_directory(str(tmp_path), {"file_ids": ["f1"], "folder_ids": ["bad"]})
|
|
assert result["files_downloaded"] == 1
|
|
|
|
|
|
class TestDownloadSingleFile:
|
|
|
|
@pytest.mark.unit
|
|
def test_downloads_supported_file(self, loader, tmp_path):
|
|
mock_meta_resp = MagicMock()
|
|
mock_meta_resp.json.return_value = {
|
|
"name": "test.pdf", "file": {"mimeType": "application/pdf"}
|
|
}
|
|
mock_meta_resp.raise_for_status = MagicMock()
|
|
|
|
mock_dl_resp = MagicMock()
|
|
mock_dl_resp.content = b"pdf content"
|
|
mock_dl_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=[mock_meta_resp, mock_dl_resp]):
|
|
result = loader._download_single_file("f1", str(tmp_path))
|
|
assert result is True
|
|
assert os.path.exists(os.path.join(str(tmp_path), "test.pdf"))
|
|
|
|
@pytest.mark.unit
|
|
def test_unsupported_returns_false(self, loader, tmp_path):
|
|
mock_resp = MagicMock()
|
|
mock_resp.json.return_value = {
|
|
"name": "test.zip", "file": {"mimeType": "application/zip"}
|
|
}
|
|
mock_resp.raise_for_status = MagicMock()
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_resp):
|
|
result = loader._download_single_file("f1", str(tmp_path))
|
|
assert result is False
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_false(self, loader, tmp_path):
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=Exception("fail")):
|
|
result = loader._download_single_file("f1", str(tmp_path))
|
|
assert result is False
|
|
|
|
|
|
class TestDownloadFolderRecursive:
|
|
|
|
@pytest.mark.unit
|
|
def test_downloads_files_in_folder(self, loader, tmp_path):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [
|
|
{"id": "f1", "name": "file1.txt"},
|
|
],
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._download_single_file = MagicMock(return_value=True)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
count = loader._download_folder_recursive("folder1", str(tmp_path))
|
|
assert count == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_recurses_into_subfolders(self, loader, tmp_path):
|
|
call_count = [0]
|
|
def get_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
resp = MagicMock()
|
|
resp.raise_for_status = MagicMock()
|
|
if call_count[0] == 1:
|
|
resp.json.return_value = {
|
|
"value": [
|
|
{"id": "sub1", "name": "subfolder", "folder": {}},
|
|
{"id": "f1", "name": "file1.txt"},
|
|
],
|
|
}
|
|
else:
|
|
resp.json.return_value = {
|
|
"value": [
|
|
{"id": "f2", "name": "file2.txt"},
|
|
],
|
|
}
|
|
return resp
|
|
|
|
loader._download_single_file = MagicMock(return_value=True)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=get_side_effect):
|
|
count = loader._download_folder_recursive("folder1", str(tmp_path), recursive=True)
|
|
assert count == 2
|
|
|
|
@pytest.mark.unit
|
|
def test_non_recursive_skips_subfolders(self, loader, tmp_path):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [
|
|
{"id": "sub1", "name": "subfolder", "folder": {}},
|
|
{"id": "f1", "name": "file1.txt"},
|
|
],
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._download_single_file = MagicMock(return_value=True)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
count = loader._download_folder_recursive("folder1", str(tmp_path), recursive=False)
|
|
assert count == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_download_failure_continues(self, loader, tmp_path):
|
|
mock_response = MagicMock()
|
|
mock_response.json.return_value = {
|
|
"value": [
|
|
{"id": "f1", "name": "fail.txt"},
|
|
{"id": "f2", "name": "ok.txt"},
|
|
],
|
|
}
|
|
mock_response.raise_for_status = MagicMock()
|
|
|
|
loader._download_single_file = MagicMock(side_effect=[False, True])
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", return_value=mock_response):
|
|
count = loader._download_folder_recursive("folder1", str(tmp_path))
|
|
assert count == 1
|
|
|
|
@pytest.mark.unit
|
|
def test_pagination_follows_next_link(self, loader, tmp_path):
|
|
call_count = [0]
|
|
def get_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
resp = MagicMock()
|
|
resp.raise_for_status = MagicMock()
|
|
if call_count[0] == 1:
|
|
resp.json.return_value = {
|
|
"value": [{"id": "f1", "name": "file1.txt"}],
|
|
"@odata.nextLink": "https://graph.microsoft.com/next",
|
|
}
|
|
else:
|
|
resp.json.return_value = {
|
|
"value": [{"id": "f2", "name": "file2.txt"}],
|
|
}
|
|
return resp
|
|
|
|
loader._download_single_file = MagicMock(return_value=True)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=get_side_effect):
|
|
count = loader._download_folder_recursive("folder1", str(tmp_path))
|
|
assert count == 2
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_partial(self, loader, tmp_path):
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=Exception("fail")):
|
|
count = loader._download_folder_recursive("folder1", str(tmp_path))
|
|
assert count == 0
|
|
|
|
|
|
class TestDownloadFileToDirectory:
|
|
|
|
@pytest.mark.unit
|
|
def test_delegates_to_download_single(self, loader, tmp_path):
|
|
loader._download_single_file = MagicMock(return_value=True)
|
|
assert loader._download_file_to_directory("f1", str(tmp_path)) is True
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_false(self, loader, tmp_path):
|
|
loader._download_single_file = MagicMock(side_effect=Exception("fail"))
|
|
# _download_file_to_directory calls _ensure_valid_token first, which may raise
|
|
# but the method wraps everything in try/except
|
|
assert loader._download_file_to_directory("f1", str(tmp_path)) is False
|
|
|
|
|
|
class TestDownloadFolderContents:
|
|
|
|
@pytest.mark.unit
|
|
def test_delegates_to_recursive(self, loader, tmp_path):
|
|
loader._download_folder_recursive = MagicMock(return_value=5)
|
|
count = loader._download_folder_contents("folder1", str(tmp_path))
|
|
assert count == 5
|
|
|
|
@pytest.mark.unit
|
|
def test_exception_returns_zero(self, loader, tmp_path):
|
|
loader.auth.is_token_expired.side_effect = Exception("fail")
|
|
count = loader._download_folder_contents("folder1", str(tmp_path))
|
|
assert count == 0
|
|
|
|
|
|
class TestRetryOnAuthFailureDecorator:
|
|
|
|
@pytest.mark.unit
|
|
def test_retries_on_401(self, loader):
|
|
"""The _retry_on_auth_failure decorator should retry after refreshing on 401."""
|
|
mock_resp_401 = MagicMock()
|
|
mock_resp_401.status_code = 401
|
|
http_error = real_requests.exceptions.HTTPError(response=mock_resp_401)
|
|
|
|
mock_resp_ok = MagicMock()
|
|
mock_resp_ok.content = b"ok"
|
|
mock_resp_ok.raise_for_status = MagicMock()
|
|
|
|
call_count = [0]
|
|
def get_side_effect(*args, **kwargs):
|
|
call_count[0] += 1
|
|
if call_count[0] == 1:
|
|
raise http_error
|
|
return mock_resp_ok
|
|
|
|
loader.auth.refresh_access_token.return_value = {"access_token": "new_at"}
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=get_side_effect):
|
|
content = loader._download_file_content("f1")
|
|
assert content == "ok"
|
|
assert loader.access_token == "new_at"
|
|
|
|
@pytest.mark.unit
|
|
def test_refresh_failure_raises_valueerror(self, loader):
|
|
"""If token refresh fails during retry, should raise ValueError."""
|
|
mock_resp_403 = MagicMock()
|
|
mock_resp_403.status_code = 403
|
|
http_error = real_requests.exceptions.HTTPError(response=mock_resp_403)
|
|
|
|
loader.auth.refresh_access_token.side_effect = Exception("refresh fail")
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=http_error):
|
|
with pytest.raises(ValueError, match="could not be refreshed"):
|
|
loader._download_file_content("f1")
|
|
|
|
@pytest.mark.unit
|
|
def test_non_auth_http_error_not_retried(self, loader):
|
|
"""Non-401/403 HTTP errors should not trigger retry."""
|
|
mock_resp_500 = MagicMock()
|
|
mock_resp_500.status_code = 500
|
|
http_error = real_requests.exceptions.HTTPError(response=mock_resp_500)
|
|
|
|
with patch("application.parser.connectors.share_point.loader.requests.get", side_effect=http_error):
|
|
with pytest.raises(real_requests.exceptions.HTTPError):
|
|
loader._download_file_content("f1")
|