diff --git a/tests/parser/__init__.py b/tests/parser/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/parser/connectors/__init__.py b/tests/parser/connectors/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/parser/connectors/test_base.py b/tests/parser/connectors/test_base.py new file mode 100644 index 00000000..2cd6fb75 --- /dev/null +++ b/tests/parser/connectors/test_base.py @@ -0,0 +1,110 @@ +"""Tests for connector base classes.""" + + +import pytest + +from application.parser.connectors.base import BaseConnectorAuth, BaseConnectorLoader +from application.parser.schema.base import Document + + +class ConcreteAuth(BaseConnectorAuth): + """Minimal concrete implementation for testing the ABC.""" + + def get_authorization_url(self, state=None): + return f"https://example.com/auth?state={state}" + + def exchange_code_for_tokens(self, authorization_code): + return {"access_token": "tok", "code": authorization_code} + + def refresh_access_token(self, refresh_token): + return {"access_token": "new_tok", "refresh_token": refresh_token} + + def is_token_expired(self, token_info): + return token_info.get("expired", False) + + +class ConcreteLoader(BaseConnectorLoader): + """Minimal concrete implementation for testing the ABC.""" + + def __init__(self, session_token): + self.session_token = session_token + + def load_data(self, inputs): + return [Document(text="test", doc_id="1", extra_info={})] + + def download_to_directory(self, local_dir, source_config=None): + return {"files_downloaded": 0, "directory_path": local_dir} + + +class TestBaseConnectorAuth: + + @pytest.mark.unit + def test_sanitize_token_info_extracts_standard_fields(self): + auth = ConcreteAuth() + token_info = { + "access_token": "at", + "refresh_token": "rt", + "token_uri": "https://token.uri", + "expiry": 12345, + "extra_field": "should_not_appear", + } + result = auth.sanitize_token_info(token_info) + assert result == { + "access_token": "at", + "refresh_token": "rt", + "token_uri": "https://token.uri", + "expiry": 12345, + } + + @pytest.mark.unit + def test_sanitize_token_info_with_extra_kwargs(self): + auth = ConcreteAuth() + token_info = { + "access_token": "at", + "refresh_token": "rt", + "token_uri": "https://token.uri", + "expiry": 100, + } + result = auth.sanitize_token_info(token_info, custom_field="custom_val") + assert result["custom_field"] == "custom_val" + assert result["access_token"] == "at" + + @pytest.mark.unit + def test_sanitize_token_info_missing_fields_returns_none(self): + auth = ConcreteAuth() + result = auth.sanitize_token_info({}) + assert result["access_token"] is None + assert result["refresh_token"] is None + assert result["token_uri"] is None + assert result["expiry"] is None + + @pytest.mark.unit + def test_abstract_methods_invocable_on_concrete(self): + auth = ConcreteAuth() + assert "example.com" in auth.get_authorization_url("s1") + assert auth.exchange_code_for_tokens("code1")["access_token"] == "tok" + assert auth.refresh_access_token("rt")["access_token"] == "new_tok" + assert auth.is_token_expired({"expired": True}) is True + assert auth.is_token_expired({"expired": False}) is False + + +class TestBaseConnectorLoader: + + @pytest.mark.unit + def test_concrete_loader_init(self): + loader = ConcreteLoader("session123") + assert loader.session_token == "session123" + + @pytest.mark.unit + def test_concrete_loader_load_data(self): + loader = ConcreteLoader("s") + docs = loader.load_data({}) + assert len(docs) == 1 + assert docs[0].text == "test" + + @pytest.mark.unit + def test_concrete_loader_download_to_directory(self): + loader = ConcreteLoader("s") + result = loader.download_to_directory("/tmp/test") + assert result["directory_path"] == "/tmp/test" + assert result["files_downloaded"] == 0 diff --git a/tests/parser/connectors/test_connector_creator.py b/tests/parser/connectors/test_connector_creator.py new file mode 100644 index 00000000..509b87b6 --- /dev/null +++ b/tests/parser/connectors/test_connector_creator.py @@ -0,0 +1,102 @@ +"""Tests for ConnectorCreator factory class.""" + +from unittest.mock import patch, MagicMock + +import pytest + + +class TestConnectorCreator: + + @pytest.fixture(autouse=True) + def _patch_settings(self): + """Patch settings so connector imports don't fail on missing credentials.""" + mock_settings = MagicMock() + mock_settings.GOOGLE_CLIENT_ID = "gid" + mock_settings.GOOGLE_CLIENT_SECRET = "gsecret" + mock_settings.CONNECTOR_REDIRECT_BASE_URI = "https://redirect" + mock_settings.MICROSOFT_CLIENT_ID = "mid" + mock_settings.MICROSOFT_CLIENT_SECRET = "msecret" + mock_settings.MICROSOFT_TENANT_ID = "tid" + mock_settings.MONGO_DB_NAME = "test_db" + + with patch("application.core.settings.settings", mock_settings), \ + patch("application.parser.connectors.share_point.auth.settings", mock_settings), \ + patch("application.parser.connectors.google_drive.auth.settings", mock_settings), \ + patch("application.parser.connectors.share_point.auth.ConfidentialClientApplication"): + from application.parser.connectors.connector_creator import ConnectorCreator + self.ConnectorCreator = ConnectorCreator + yield + + @pytest.mark.unit + def test_get_supported_connectors(self): + supported = self.ConnectorCreator.get_supported_connectors() + assert "google_drive" in supported + assert "share_point" in supported + + @pytest.mark.unit + def test_is_supported_valid(self): + assert self.ConnectorCreator.is_supported("google_drive") is True + assert self.ConnectorCreator.is_supported("share_point") is True + + @pytest.mark.unit + def test_is_supported_case_insensitive(self): + assert self.ConnectorCreator.is_supported("Google_Drive") is True + assert self.ConnectorCreator.is_supported("SHARE_POINT") is True + + @pytest.mark.unit + def test_is_supported_invalid(self): + assert self.ConnectorCreator.is_supported("dropbox") is False + assert self.ConnectorCreator.is_supported("") is False + + @pytest.mark.unit + def test_create_auth_google_drive(self): + auth = self.ConnectorCreator.create_auth("google_drive") + from application.parser.connectors.google_drive.auth import GoogleDriveAuth + assert isinstance(auth, GoogleDriveAuth) + + @pytest.mark.unit + def test_create_auth_share_point(self): + auth = self.ConnectorCreator.create_auth("share_point") + from application.parser.connectors.share_point.auth import SharePointAuth + assert isinstance(auth, SharePointAuth) + + @pytest.mark.unit + def test_create_auth_invalid_raises(self): + with pytest.raises(ValueError, match="No auth class found"): + self.ConnectorCreator.create_auth("invalid") + + @pytest.mark.unit + def test_create_connector_invalid_raises(self): + with pytest.raises(ValueError, match="No connector class found"): + self.ConnectorCreator.create_connector("invalid", "session_tok") + + @pytest.mark.unit + def test_create_connector_google_drive(self): + with patch("application.parser.connectors.google_drive.loader.GoogleDriveAuth") as MockAuth: + mock_auth_instance = MagicMock() + mock_auth_instance.get_token_info_from_session.return_value = { + "access_token": "at", "refresh_token": "rt" + } + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.expired = False + mock_auth_instance.create_credentials_from_token_info.return_value = mock_creds + mock_auth_instance.build_drive_service.return_value = MagicMock() + MockAuth.return_value = mock_auth_instance + + loader = self.ConnectorCreator.create_connector("google_drive", "session_tok") + from application.parser.connectors.google_drive.loader import GoogleDriveLoader + assert isinstance(loader, GoogleDriveLoader) + + @pytest.mark.unit + def test_create_connector_share_point(self): + with patch("application.parser.connectors.share_point.loader.SharePointAuth") as MockAuth: + mock_auth_instance = MagicMock() + mock_auth_instance.get_token_info_from_session.return_value = { + "access_token": "at", "refresh_token": "rt" + } + MockAuth.return_value = mock_auth_instance + + loader = self.ConnectorCreator.create_connector("share_point", "session_tok") + from application.parser.connectors.share_point.loader import SharePointLoader + assert isinstance(loader, SharePointLoader) diff --git a/tests/parser/connectors/test_google_drive_auth.py b/tests/parser/connectors/test_google_drive_auth.py new file mode 100644 index 00000000..ec4e86c9 --- /dev/null +++ b/tests/parser/connectors/test_google_drive_auth.py @@ -0,0 +1,441 @@ +"""Tests for GoogleDriveAuth.""" + +import datetime +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def mock_settings(): + s = MagicMock() + s.GOOGLE_CLIENT_ID = "test-client-id" + s.GOOGLE_CLIENT_SECRET = "test-client-secret" + s.CONNECTOR_REDIRECT_BASE_URI = "https://redirect.example.com/callback" + s.MONGO_DB_NAME = "test_db" + return s + + +@pytest.fixture +def auth(mock_settings): + with patch("application.parser.connectors.google_drive.auth.settings", mock_settings): + from application.parser.connectors.google_drive.auth import GoogleDriveAuth + return GoogleDriveAuth() + + +class TestGoogleDriveAuthInit: + + @pytest.mark.unit + def test_init_sets_credentials(self, auth, mock_settings): + assert auth.client_id == "test-client-id" + assert auth.client_secret == "test-client-secret" + assert auth.redirect_uri == "https://redirect.example.com/callback" + + @pytest.mark.unit + def test_init_missing_client_id_raises(self, mock_settings): + mock_settings.GOOGLE_CLIENT_ID = None + with patch("application.parser.connectors.google_drive.auth.settings", mock_settings): + from application.parser.connectors.google_drive.auth import GoogleDriveAuth + with pytest.raises(ValueError, match="Google OAuth credentials not configured"): + GoogleDriveAuth() + + @pytest.mark.unit + def test_init_missing_client_secret_raises(self, mock_settings): + mock_settings.GOOGLE_CLIENT_SECRET = None + with patch("application.parser.connectors.google_drive.auth.settings", mock_settings): + from application.parser.connectors.google_drive.auth import GoogleDriveAuth + with pytest.raises(ValueError, match="Google OAuth credentials not configured"): + GoogleDriveAuth() + + +class TestGetAuthorizationUrl: + + @pytest.mark.unit + def test_returns_authorization_url(self, auth): + mock_flow = MagicMock() + mock_flow.authorization_url.return_value = ("https://accounts.google.com/auth?state=s1", "s1") + + with patch("application.parser.connectors.google_drive.auth.Flow") as MockFlow: + MockFlow.from_client_config.return_value = mock_flow + url = auth.get_authorization_url(state="s1") + + assert url == "https://accounts.google.com/auth?state=s1" + mock_flow.authorization_url.assert_called_once_with( + access_type='offline', + prompt='consent', + include_granted_scopes='false', + state="s1" + ) + + @pytest.mark.unit + def test_raises_on_flow_error(self, auth): + with patch("application.parser.connectors.google_drive.auth.Flow") as MockFlow: + MockFlow.from_client_config.side_effect = Exception("flow error") + with pytest.raises(Exception, match="flow error"): + auth.get_authorization_url() + + +class TestExchangeCodeForTokens: + + @pytest.mark.unit + def test_successful_exchange(self, auth): + mock_creds = MagicMock() + mock_creds.token = "access_tok" + mock_creds.refresh_token = "refresh_tok" + mock_creds.token_uri = "https://oauth2.googleapis.com/token" + mock_creds.client_id = "test-client-id" + mock_creds.client_secret = "test-client-secret" + mock_creds.scopes = ["https://www.googleapis.com/auth/drive.file"] + mock_creds.expiry = datetime.datetime(2025, 1, 1, 12, 0, 0) + + mock_flow = MagicMock() + mock_flow.credentials = mock_creds + + with patch("application.parser.connectors.google_drive.auth.Flow") as MockFlow: + MockFlow.from_client_config.return_value = mock_flow + result = auth.exchange_code_for_tokens("auth_code_123") + + assert result["access_token"] == "access_tok" + assert result["refresh_token"] == "refresh_tok" + assert result["token_uri"] == "https://oauth2.googleapis.com/token" + assert result["client_id"] == "test-client-id" + assert result["client_secret"] == "test-client-secret" + assert result["expiry"] == "2025-01-01T12:00:00" + + @pytest.mark.unit + def test_empty_code_raises(self, auth): + with pytest.raises(ValueError, match="Authorization code is required"): + auth.exchange_code_for_tokens("") + + @pytest.mark.unit + def test_no_access_token_raises(self, auth): + mock_creds = MagicMock() + mock_creds.token = None + mock_creds.refresh_token = "rt" + mock_flow = MagicMock() + mock_flow.credentials = mock_creds + + with patch("application.parser.connectors.google_drive.auth.Flow") as MockFlow: + MockFlow.from_client_config.return_value = mock_flow + with pytest.raises(ValueError, match="did not return an access token"): + auth.exchange_code_for_tokens("code") + + @pytest.mark.unit + def test_no_refresh_token_raises(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = None + mock_flow = MagicMock() + mock_flow.credentials = mock_creds + + with patch("application.parser.connectors.google_drive.auth.Flow") as MockFlow: + MockFlow.from_client_config.return_value = mock_flow + with pytest.raises(ValueError, match="No refresh token received"): + auth.exchange_code_for_tokens("code") + + @pytest.mark.unit + def test_fills_in_missing_token_uri(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = "rt" + mock_creds.token_uri = None + mock_creds.client_id = None + mock_creds.client_secret = None + mock_creds.scopes = [] + mock_creds.expiry = None + mock_flow = MagicMock() + mock_flow.credentials = mock_creds + + with patch("application.parser.connectors.google_drive.auth.Flow") as MockFlow: + MockFlow.from_client_config.return_value = mock_flow + result = auth.exchange_code_for_tokens("code") + + assert result["token_uri"] == "https://oauth2.googleapis.com/token" + assert result["client_id"] == "test-client-id" + assert result["client_secret"] == "test-client-secret" + + +class TestRefreshAccessToken: + + @pytest.mark.unit + def test_successful_refresh(self, auth): + mock_request_cls = MagicMock() + with patch("application.parser.connectors.google_drive.auth.Credentials") as MockCreds, \ + patch("google.auth.transport.requests.Request", mock_request_cls): + mock_cred_instance = MagicMock() + mock_cred_instance.token = "new_access" + mock_cred_instance.token_uri = "https://oauth2.googleapis.com/token" + mock_cred_instance.client_id = "cid" + mock_cred_instance.client_secret = "cs" + mock_cred_instance.scopes = [] + mock_cred_instance.expiry = datetime.datetime(2025, 6, 1, 0, 0, 0) + MockCreds.return_value = mock_cred_instance + + result = auth.refresh_access_token("old_refresh") + + assert result["access_token"] == "new_access" + assert result["refresh_token"] == "old_refresh" + mock_cred_instance.refresh.assert_called_once() + + @pytest.mark.unit + def test_empty_refresh_token_raises(self, auth): + with pytest.raises(ValueError, match="Refresh token is required"): + auth.refresh_access_token("") + + @pytest.mark.unit + def test_refresh_failure_raises(self, auth): + with patch("application.parser.connectors.google_drive.auth.Credentials") as MockCreds, \ + patch("google.auth.transport.requests.Request"): + mock_cred_instance = MagicMock() + mock_cred_instance.refresh.side_effect = Exception("refresh failed") + MockCreds.return_value = mock_cred_instance + + with pytest.raises(Exception, match="refresh failed"): + auth.refresh_access_token("rt") + + +class TestCreateCredentialsFromTokenInfo: + + @pytest.mark.unit + def test_creates_credentials(self, auth, mock_settings): + with patch("application.parser.connectors.google_drive.auth.Credentials") as MockCreds, \ + patch("application.parser.connectors.google_drive.auth.settings", mock_settings): + mock_cred = MagicMock() + mock_cred.token = "at" + MockCreds.return_value = mock_cred + + creds = auth.create_credentials_from_token_info({ + "access_token": "at", + "refresh_token": "rt", + "scopes": ["scope1"], + }) + assert creds.token == "at" + + @pytest.mark.unit + def test_missing_access_token_raises(self, auth, mock_settings): + with patch("application.parser.connectors.google_drive.auth.settings", mock_settings): + with pytest.raises(ValueError, match="No access token found"): + auth.create_credentials_from_token_info({}) + + @pytest.mark.unit + def test_credentials_without_valid_token_raises(self, auth, mock_settings): + with patch("application.parser.connectors.google_drive.auth.Credentials") as MockCreds, \ + patch("application.parser.connectors.google_drive.auth.settings", mock_settings): + mock_cred = MagicMock() + mock_cred.token = None + MockCreds.return_value = mock_cred + + with pytest.raises(ValueError, match="Credentials created without valid access token"): + auth.create_credentials_from_token_info({"access_token": "at"}) + + +class TestBuildDriveService: + + @pytest.mark.unit + def test_builds_service(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = "rt" + mock_creds.expired = False + + with patch("application.parser.connectors.google_drive.auth.build") as mock_build: + mock_build.return_value = MagicMock() + service = auth.build_drive_service(mock_creds) + mock_build.assert_called_once_with('drive', 'v3', credentials=mock_creds) + assert service is not None + + @pytest.mark.unit + def test_no_credentials_raises(self, auth): + with pytest.raises(ValueError, match="No credentials provided"): + auth.build_drive_service(None) + + @pytest.mark.unit + def test_no_token_no_refresh_raises(self, auth): + mock_creds = MagicMock() + mock_creds.token = None + mock_creds.refresh_token = None + with pytest.raises(ValueError, match="No access token or refresh token"): + auth.build_drive_service(mock_creds) + + @pytest.mark.unit + def test_expired_token_refreshes(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = "rt" + mock_creds.expired = True + + with patch("application.parser.connectors.google_drive.auth.build") as mock_build, \ + patch("google.auth.transport.requests.Request"): + mock_build.return_value = MagicMock() + auth.build_drive_service(mock_creds) + mock_creds.refresh.assert_called_once() + + @pytest.mark.unit + def test_expired_no_refresh_token_raises(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = None + mock_creds.expired = True + with pytest.raises(ValueError, match="No access token or refresh token"): + auth.build_drive_service(mock_creds) + + @pytest.mark.unit + def test_refresh_failure_raises(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = "rt" + mock_creds.expired = True + + with patch("google.auth.transport.requests.Request"): + mock_creds.refresh.side_effect = Exception("Cannot refresh") + with pytest.raises(ValueError, match="Failed to refresh credentials"): + auth.build_drive_service(mock_creds) + + @pytest.mark.unit + def test_http_error_raises(self, auth): + from googleapiclient.errors import HttpError + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = "rt" + mock_creds.expired = False + + mock_resp = MagicMock() + mock_resp.status = 500 + + with patch("application.parser.connectors.google_drive.auth.build") as mock_build: + mock_build.side_effect = HttpError(mock_resp, b"error") + with pytest.raises(ValueError, match="HTTP 500"): + auth.build_drive_service(mock_creds) + + +class TestIsTokenExpired: + + @pytest.mark.unit + def test_expired_token(self, auth): + past = (datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(hours=1)).isoformat() + assert auth.is_token_expired({"expiry": past}) is True + + @pytest.mark.unit + def test_valid_token(self, auth): + future = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(hours=1)).isoformat() + assert auth.is_token_expired({"expiry": future}) is False + + @pytest.mark.unit + def test_token_within_buffer(self, auth): + almost_expired = (datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(seconds=30)).isoformat() + assert auth.is_token_expired({"expiry": almost_expired}) is True + + @pytest.mark.unit + def test_no_expiry_with_access_token(self, auth): + assert auth.is_token_expired({"access_token": "at"}) is False + + @pytest.mark.unit + def test_no_expiry_no_access_token(self, auth): + assert auth.is_token_expired({}) is True + + @pytest.mark.unit + def test_invalid_expiry_format_returns_true(self, auth): + assert auth.is_token_expired({"expiry": "not-a-date"}) is True + + @pytest.mark.unit + def test_none_expiry_with_access_token(self, auth): + assert auth.is_token_expired({"expiry": None, "access_token": "at"}) is False + + +class TestGetTokenInfoFromSession: + + def _mock_mongo(self, mock_settings, find_one_return): + mock_collection = MagicMock() + mock_collection.find_one.return_value = find_one_return + mock_db = MagicMock() + mock_db.__getitem__ = MagicMock(return_value=mock_collection) + return {mock_settings.MONGO_DB_NAME: mock_db} + + @pytest.mark.unit + def test_valid_session(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, { + "session_token": "st", + "token_info": {"access_token": "at", "refresh_token": "rt"}, + }) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + result = auth.get_token_info_from_session("st") + assert result["access_token"] == "at" + assert result["token_uri"] == "https://oauth2.googleapis.com/token" + + @pytest.mark.unit + def test_session_not_found_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, None) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve Google Drive token"): + auth.get_token_info_from_session("bad_token") + + @pytest.mark.unit + def test_session_missing_token_info_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, {"session_token": "st"}) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve Google Drive token"): + auth.get_token_info_from_session("st") + + @pytest.mark.unit + def test_missing_required_fields_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, { + "session_token": "st", + "token_info": {"access_token": "at"}, + }) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve Google Drive token"): + auth.get_token_info_from_session("st") + + @pytest.mark.unit + def test_empty_token_info_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, { + "session_token": "st", + "token_info": None, + }) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve Google Drive token"): + auth.get_token_info_from_session("st") + + +class TestValidateCredentials: + + @pytest.mark.unit + def test_valid_credentials(self, auth): + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.refresh_token = "rt" + mock_creds.expired = False + + mock_service = MagicMock() + mock_service.about.return_value.get.return_value.execute.return_value = {"user": {}} + + with patch.object(auth, 'build_drive_service', return_value=mock_service): + assert auth.validate_credentials(mock_creds) is True + + @pytest.mark.unit + def test_http_error_returns_false(self, auth): + from googleapiclient.errors import HttpError + mock_creds = MagicMock() + mock_resp = MagicMock() + mock_resp.status = 401 + mock_service = MagicMock() + mock_service.about.return_value.get.return_value.execute.side_effect = HttpError(mock_resp, b"unauth") + + with patch.object(auth, 'build_drive_service', return_value=mock_service): + assert auth.validate_credentials(mock_creds) is False + + @pytest.mark.unit + def test_general_error_returns_false(self, auth): + mock_creds = MagicMock() + with patch.object(auth, 'build_drive_service', side_effect=Exception("fail")): + assert auth.validate_credentials(mock_creds) is False diff --git a/tests/parser/connectors/test_google_drive_loader.py b/tests/parser/connectors/test_google_drive_loader.py new file mode 100644 index 00000000..e7897cc6 --- /dev/null +++ b/tests/parser/connectors/test_google_drive_loader.py @@ -0,0 +1,851 @@ +"""Tests for GoogleDriveLoader.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from application.parser.schema.base import Document + + +def _make_loader(service=None): + """Create a GoogleDriveLoader with mocked dependencies.""" + with patch("application.parser.connectors.google_drive.loader.GoogleDriveAuth") as MockAuth: + mock_auth = MagicMock() + mock_auth.get_token_info_from_session.return_value = { + "access_token": "at", + "refresh_token": "rt", + } + mock_creds = MagicMock() + mock_creds.token = "at" + mock_creds.expired = False + mock_creds.refresh_token = "rt" + mock_auth.create_credentials_from_token_info.return_value = mock_creds + mock_auth.build_drive_service.return_value = service or MagicMock() + MockAuth.return_value = mock_auth + + from application.parser.connectors.google_drive.loader import GoogleDriveLoader + loader = GoogleDriveLoader("session_tok") + return loader + + +@pytest.fixture +def mock_service(): + return MagicMock() + + +@pytest.fixture +def loader(mock_service): + return _make_loader(mock_service) + + +class TestGoogleDriveLoaderInit: + + @pytest.mark.unit + def test_init_sets_attributes(self, loader): + assert loader.session_token == "session_tok" + assert loader.credentials is not None + assert loader.service is not None + assert loader.next_page_token is None + + @pytest.mark.unit + def test_init_service_failure_sets_none(self): + with patch("application.parser.connectors.google_drive.loader.GoogleDriveAuth") as MockAuth: + mock_auth = MagicMock() + mock_auth.get_token_info_from_session.return_value = { + "access_token": "at", "refresh_token": "rt" + } + mock_creds = MagicMock() + mock_creds.token = "at" + mock_auth.create_credentials_from_token_info.return_value = mock_creds + mock_auth.build_drive_service.side_effect = Exception("service fail") + MockAuth.return_value = mock_auth + + from application.parser.connectors.google_drive.loader import GoogleDriveLoader + loader = GoogleDriveLoader("st") + assert loader.service is None + + +class TestProcessFile: + + @pytest.mark.unit + def test_supported_mime_type_with_content(self, loader): + loader._download_file_content = MagicMock(return_value="file content") + metadata = { + "id": "f1", + "name": "test.pdf", + "mimeType": "application/pdf", + "size": 1024, + "createdTime": "2025-01-01T00:00:00Z", + "modifiedTime": "2025-01-02T00:00:00Z", + "parents": ["root"], + } + doc = loader._process_file(metadata, load_content=True) + assert doc is not None + assert doc.text == "file content" + assert doc.doc_id == "f1" + assert doc.extra_info["file_name"] == "test.pdf" + assert doc.extra_info["source"] == "google_drive" + + @pytest.mark.unit + def test_supported_mime_type_no_content(self, loader): + metadata = { + "id": "f1", + "name": "test.pdf", + "mimeType": "application/pdf", + } + doc = loader._process_file(metadata, load_content=False) + assert doc is not None + assert doc.text == "" + assert doc.doc_id == "f1" + + @pytest.mark.unit + def test_unsupported_mime_type_returns_none(self, loader): + metadata = { + "id": "f1", + "name": "test.zip", + "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", + "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", + "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="content", 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 + assert result[0].doc_id == "f1" + + @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", "f2"]}) + assert len(result) == 0 + + @pytest.mark.unit + def test_browse_mode_uses_list_items(self, loader): + docs = [Document(text="", doc_id="f1", extra_info={})] + loader._list_items_in_parent = MagicMock(return_value=docs) + + result = loader.load_data({"folder_id": "folder1", "limit": 50}) + loader._list_items_in_parent.assert_called_once_with( + "folder1", limit=50, load_content=True, page_token=None, search_query=None + ) + assert len(result) == 1 + + @pytest.mark.unit + def test_browse_mode_defaults_to_root(self, loader): + loader._list_items_in_parent = MagicMock(return_value=[]) + loader.load_data({}) + loader._list_items_in_parent.assert_called_once_with( + "root", limit=100, load_content=True, page_token=None, search_query=None + ) + + @pytest.mark.unit + def test_session_token_mismatch_logs_warning(self, loader): + loader._list_items_in_parent = MagicMock(return_value=[]) + loader.load_data({"session_token": "different_token"}) + # Should not raise, just logs + + @pytest.mark.unit + def test_load_data_with_list_only(self, loader): + loader._list_items_in_parent = MagicMock(return_value=[]) + loader.load_data({"list_only": True}) + loader._list_items_in_parent.assert_called_once_with( + "root", limit=100, load_content=False, page_token=None, search_query=None + ) + + @pytest.mark.unit + def test_load_data_with_page_token(self, loader): + loader._list_items_in_parent = MagicMock(return_value=[]) + loader.load_data({"page_token": "next_page"}) + loader._list_items_in_parent.assert_called_once_with( + "root", limit=100, load_content=True, page_token="next_page", search_query=None + ) + + @pytest.mark.unit + def test_credential_refresh_retry(self, loader): + """When _load_file_by_id returns None and _credential_refreshed is set, retry.""" + loader._credential_refreshed = True + call_count = [0] + def side_effect(fid, load_content=True): + call_count[0] += 1 + if call_count[0] == 1: + return None + return Document(text="c", doc_id=fid, extra_info={"file_name": "test.pdf"}) + + loader._load_file_by_id = MagicMock(side_effect=side_effect) + result = loader.load_data({"file_ids": ["f1"]}) + assert len(result) == 1 + + @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_file_metadata_and_processes(self, loader, mock_service): + mock_service.files.return_value.get.return_value.execute.return_value = { + "id": "f1", "name": "test.txt", "mimeType": "text/plain" + } + loader._process_file = MagicMock(return_value=Document(text="t", doc_id="f1", extra_info={})) + doc = loader._load_file_by_id("f1") + assert doc is not None + + @pytest.mark.unit + def test_http_401_refreshes_credentials(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 401 + mock_service.files.return_value.get.return_value.execute.side_effect = HttpError(resp, b"unauth") + + with patch("google.auth.transport.requests.Request"): + result = loader._load_file_by_id("f1") + assert result is None + loader.credentials.refresh.assert_called_once() + + @pytest.mark.unit + def test_http_401_no_refresh_token_raises(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 401 + mock_service.files.return_value.get.return_value.execute.side_effect = HttpError(resp, b"unauth") + loader.credentials.refresh_token = None + + with pytest.raises(ValueError, match="missing refresh_token"): + loader._load_file_by_id("f1") + + @pytest.mark.unit + def test_http_500_returns_none(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 500 + mock_service.files.return_value.get.return_value.execute.side_effect = HttpError(resp, b"server error") + result = loader._load_file_by_id("f1") + assert result is None + + @pytest.mark.unit + def test_general_exception_returns_none(self, loader, mock_service): + mock_service.files.return_value.get.return_value.execute.side_effect = Exception("fail") + result = loader._load_file_by_id("f1") + assert result is None + + @pytest.mark.unit + def test_ensure_service_called(self, loader): + loader.service = None + loader.auth.build_drive_service.return_value = MagicMock() + loader.auth.build_drive_service.return_value.files.return_value.get.return_value.execute.return_value = { + "id": "f1", "name": "t.txt", "mimeType": "text/plain" + } + loader._process_file = MagicMock(return_value=None) + loader._load_file_by_id("f1") + loader.auth.build_drive_service.assert_called() + + @pytest.mark.unit + def test_http_401_refresh_failure_raises(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 401 + mock_service.files.return_value.get.return_value.execute.side_effect = HttpError(resp, b"unauth") + loader.credentials.refresh_token = "rt" + + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh.side_effect = Exception("refresh broke") + with pytest.raises(ValueError, match="could not be refreshed"): + loader._load_file_by_id("f1") + + +class TestListItemsInParent: + + @pytest.mark.unit + def test_lists_files_and_folders(self, loader, mock_service): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [ + {"id": "folder1", "name": "Docs", "mimeType": "application/vnd.google-apps.folder"}, + {"id": "file1", "name": "test.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": None, + } + loader._process_file = MagicMock(return_value=Document(text="", doc_id="file1", extra_info={})) + docs = loader._list_items_in_parent("root", limit=100) + assert len(docs) == 2 + assert docs[0].extra_info.get("is_folder") is True + + @pytest.mark.unit + def test_search_query_modifies_drive_query(self, loader, mock_service): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [], "nextPageToken": None + } + loader._list_items_in_parent("root", search_query="report") + call_args = mock_service.files.return_value.list.call_args + assert "name contains 'report'" in call_args.kwargs.get("q", "") + + @pytest.mark.unit + def test_limit_stops_early(self, loader, mock_service): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [ + {"id": f"f{i}", "name": f"file{i}.txt", "mimeType": "text/plain"} for i in range(10) + ], + "nextPageToken": "next", + } + loader._process_file = MagicMock(side_effect=lambda m, **kw: Document(text="", doc_id=m["id"], extra_info={})) + docs = loader._list_items_in_parent("root", limit=3) + assert len(docs) == 3 + + @pytest.mark.unit + def test_pagination_stores_next_page_token(self, loader, mock_service): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [{"id": "f1", "name": "f.txt", "mimeType": "text/plain"}], + "nextPageToken": "page2", + } + loader._process_file = MagicMock(return_value=Document(text="", doc_id="f1", extra_info={})) + loader._list_items_in_parent("root", limit=1) + assert loader.next_page_token == "page2" + + @pytest.mark.unit + def test_limit_breaks_loop_when_remaining_zero(self, loader, mock_service): + """When limit is exactly met after first page, the while loop breaks via remaining==0.""" + call_count = [0] + def list_side_effect(**kw): + call_count[0] += 1 + mock = MagicMock() + if call_count[0] == 1: + mock.execute.return_value = { + "files": [ + {"id": "f1", "name": "f1.txt", "mimeType": "text/plain"}, + {"id": "f2", "name": "f2.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": "page2", + } + else: + # Should not reach here if break works correctly + mock.execute.return_value = {"files": [], "nextPageToken": None} + return mock + + mock_service.files.return_value.list.side_effect = list_side_effect + loader._process_file = MagicMock(side_effect=lambda m, **kw: Document(text="", doc_id=m["id"], extra_info={})) + # limit=2, first page returns exactly 2 files with nextPageToken + # Since items don't hit the inner limit check (it checks after each item), + # it should loop back and break at remaining==0 + docs = loader._list_items_in_parent("root", limit=2) + assert len(docs) == 2 + + @pytest.mark.unit + def test_exception_returns_partial_results(self, loader, mock_service): + mock_service.files.return_value.list.return_value.execute.side_effect = Exception("api error") + docs = loader._list_items_in_parent("root") + assert docs == [] + + +class TestDownloadFileContent: + + @pytest.mark.unit + def test_download_regular_file(self, loader, mock_service): + mock_request = MagicMock() + mock_service.files.return_value.get_media.return_value = mock_request + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.side_effect = [(None, False), (None, True)] + MockDownload.return_value = mock_dl + + with patch("io.BytesIO") as MockBytesIO: + mock_bio = MagicMock() + mock_bio.getvalue.return_value = b"file content" + MockBytesIO.return_value = mock_bio + + content = loader._download_file_content("f1", "text/plain") + + assert content == "file content" + + @pytest.mark.unit + def test_download_google_workspace_file_uses_export(self, loader, mock_service): + mock_request = MagicMock() + mock_service.files.return_value.export_media.return_value = mock_request + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.return_value = (None, True) + MockDownload.return_value = mock_dl + + with patch("io.BytesIO") as MockBytesIO: + mock_bio = MagicMock() + mock_bio.getvalue.return_value = b"exported" + MockBytesIO.return_value = mock_bio + + content = loader._download_file_content("f1", "application/vnd.google-apps.document") + + mock_service.files.return_value.export_media.assert_called_once() + assert content == "exported" + + @pytest.mark.unit + def test_unicode_decode_error_returns_none(self, loader, mock_service): + mock_request = MagicMock() + mock_service.files.return_value.get_media.return_value = mock_request + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.return_value = (None, True) + MockDownload.return_value = mock_dl + + with patch("io.BytesIO") as MockBytesIO: + mock_bio = MagicMock() + mock_bio.getvalue.return_value = MagicMock() + mock_bio.getvalue.return_value.decode.side_effect = UnicodeDecodeError("utf-8", b"", 0, 1, "bad") + MockBytesIO.return_value = mock_bio + + result = loader._download_file_content("f1", "application/pdf") + assert result is None + + @pytest.mark.unit + def test_no_access_token_with_refresh(self, loader): + loader.credentials.token = None + loader.credentials.refresh_token = "rt" + loader.credentials.expired = False + + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh = MagicMock() + # After refresh, set token + def set_token(req): + loader.credentials.token = "new_at" + loader.credentials.refresh.side_effect = set_token + loader._ensure_service = MagicMock() + loader.service.files.return_value.get_media.return_value = MagicMock() + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.return_value = (None, True) + MockDownload.return_value = mock_dl + with patch("io.BytesIO") as MockBytesIO: + mock_bio = MagicMock() + mock_bio.getvalue.return_value = b"data" + MockBytesIO.return_value = mock_bio + content = loader._download_file_content("f1", "text/plain") + + assert content == "data" + + @pytest.mark.unit + def test_no_access_token_no_refresh_raises(self, loader): + loader.credentials.token = None + loader.credentials.refresh_token = None + with pytest.raises(ValueError, match="missing refresh_token"): + loader._download_file_content("f1", "text/plain") + + @pytest.mark.unit + def test_no_token_refresh_fails_raises(self, loader): + loader.credentials.token = None + loader.credentials.refresh_token = "rt" + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh.side_effect = Exception("fail") + with pytest.raises(ValueError, match="missing or invalid refresh_token"): + loader._download_file_content("f1", "text/plain") + + @pytest.mark.unit + def test_expired_credentials_refresh(self, loader): + loader.credentials.token = "at" + loader.credentials.expired = True + loader.credentials.refresh_token = "rt" + + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh = MagicMock() + def fix_expired(req): + loader.credentials.expired = False + loader.credentials.refresh.side_effect = fix_expired + loader._ensure_service = MagicMock() + loader.service.files.return_value.get_media.return_value = MagicMock() + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.return_value = (None, True) + MockDownload.return_value = mock_dl + with patch("io.BytesIO") as MockBytesIO: + mock_bio = MagicMock() + mock_bio.getvalue.return_value = b"ok" + MockBytesIO.return_value = mock_bio + content = loader._download_file_content("f1", "text/plain") + + assert content == "ok" + + @pytest.mark.unit + def test_expired_no_refresh_token_raises(self, loader): + loader.credentials.token = "at" + loader.credentials.expired = True + loader.credentials.refresh_token = None + with pytest.raises(ValueError, match="missing refresh_token"): + loader._download_file_content("f1", "text/plain") + + @pytest.mark.unit + def test_expired_refresh_fails_raises(self, loader): + loader.credentials.token = "at" + loader.credentials.expired = True + loader.credentials.refresh_token = "rt" + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh.side_effect = Exception("fail") + with pytest.raises(ValueError, match="expired credentials"): + loader._download_file_content("f1", "text/plain") + + @pytest.mark.unit + def test_http_error_during_download_chunk_returns_none(self, loader, mock_service): + from googleapiclient.errors import HttpError + mock_request = MagicMock() + mock_service.files.return_value.get_media.return_value = mock_request + + resp = MagicMock() + resp.status = 500 + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.side_effect = HttpError(resp, b"server error") + MockDownload.return_value = mock_dl + with patch("io.BytesIO") as MockBytesIO: + MockBytesIO.return_value = MagicMock() + result = loader._download_file_content("f1", "text/plain") + assert result is None + + @pytest.mark.unit + def test_general_error_during_download_chunk_returns_none(self, loader, mock_service): + mock_request = MagicMock() + mock_service.files.return_value.get_media.return_value = mock_request + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDownload: + mock_dl = MagicMock() + mock_dl.next_chunk.side_effect = RuntimeError("chunk fail") + MockDownload.return_value = mock_dl + with patch("io.BytesIO") as MockBytesIO: + MockBytesIO.return_value = MagicMock() + result = loader._download_file_content("f1", "text/plain") + assert result is None + + @pytest.mark.unit + def test_http_401_during_download_refreshes_and_returns_none(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 401 + + mock_service.files.return_value.get_media.side_effect = HttpError(resp, b"unauth") + loader.credentials.refresh_token = "rt" + + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh = MagicMock() + loader._ensure_service = MagicMock() + result = loader._download_file_content("f1", "text/plain") + + assert result is None + assert loader._credential_refreshed is True + + @pytest.mark.unit + def test_http_401_no_refresh_token_raises(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 401 + mock_service.files.return_value.get_media.side_effect = HttpError(resp, b"unauth") + loader.credentials.refresh_token = None + + with pytest.raises(ValueError, match="missing refresh_token"): + loader._download_file_content("f1", "text/plain") + + @pytest.mark.unit + def test_http_401_refresh_fails_raises(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 401 + mock_service.files.return_value.get_media.side_effect = HttpError(resp, b"unauth") + loader.credentials.refresh_token = "rt" + + with patch("google.auth.transport.requests.Request"): + loader.credentials.refresh.side_effect = Exception("refresh fail") + with pytest.raises(ValueError, match="could not be refreshed"): + loader._download_file_content("f1", "text/plain") + + @pytest.mark.unit + def test_http_500_returns_none(self, loader, mock_service): + from googleapiclient.errors import HttpError + resp = MagicMock() + resp.status = 500 + mock_service.files.return_value.get_media.side_effect = HttpError(resp, b"error") + result = loader._download_file_content("f1", "text/plain") + assert result is None + + @pytest.mark.unit + def test_general_exception_returns_none(self, loader, mock_service): + mock_service.files.return_value.get_media.side_effect = RuntimeError("fail") + result = loader._download_file_content("f1", "text/plain") + assert result is None + + +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"] == "google_drive" + assert result["empty_result"] is False + + @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_id"}) + assert result["files_downloaded"] == 1 + + @pytest.mark.unit + def test_download_folders(self, loader, tmp_path, mock_service): + mock_service.files.return_value.get.return_value.execute.return_value = {"name": "MyFolder"} + loader._download_folder_recursive = MagicMock(return_value=3) + 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_service): + mock_service.files.return_value.get.return_value.execute.return_value = {"name": "F"} + loader._download_folder_recursive = MagicMock(return_value=1) + 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 + assert result["empty_result"] is True + + @pytest.mark.unit + def test_none_config_uses_empty(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_service): + mock_service.files.return_value.get.return_value.execute.side_effect = Exception("fail") + loader._download_file_to_directory = MagicMock(return_value=True) + result = loader.download_to_directory(str(tmp_path), {"file_ids": ["f1"], "folder_ids": ["bad_folder"]}) + assert result["files_downloaded"] == 1 + + +class TestDownloadSingleFile: + + @pytest.mark.unit + def test_downloads_supported_file(self, loader, mock_service, tmp_path): + mock_service.files.return_value.get.return_value.execute.return_value = { + "name": "test.txt", "mimeType": "text/plain" + } + mock_request = MagicMock() + mock_service.files.return_value.get_media.return_value = mock_request + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDl: + mock_dl = MagicMock() + mock_dl.next_chunk.return_value = (None, True) + MockDl.return_value = mock_dl + + result = loader._download_single_file("f1", str(tmp_path)) + assert result is True + + @pytest.mark.unit + def test_unsupported_mime_returns_false(self, loader, mock_service, tmp_path): + mock_service.files.return_value.get.return_value.execute.return_value = { + "name": "test.zip", "mimeType": "application/zip" + } + result = loader._download_single_file("f1", str(tmp_path)) + assert result is False + + @pytest.mark.unit + def test_google_workspace_file_export(self, loader, mock_service, tmp_path): + mock_service.files.return_value.get.return_value.execute.return_value = { + "name": "doc", "mimeType": "application/vnd.google-apps.document" + } + mock_request = MagicMock() + mock_service.files.return_value.export_media.return_value = mock_request + + with patch("application.parser.connectors.google_drive.loader.MediaIoBaseDownload") as MockDl: + mock_dl = MagicMock() + mock_dl.next_chunk.return_value = (None, True) + MockDl.return_value = mock_dl + result = loader._download_single_file("f1", str(tmp_path)) + + assert result is True + mock_service.files.return_value.export_media.assert_called_once() + + +class TestDownloadFolderRecursive: + + @pytest.mark.unit + def test_downloads_files_in_folder(self, loader, mock_service, tmp_path): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [ + {"id": "f1", "name": "file1.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": None, + } + loader._download_single_file = MagicMock(return_value=True) + count = loader._download_folder_recursive("folder1", str(tmp_path)) + assert count == 1 + + @pytest.mark.unit + def test_recurses_into_subfolders(self, loader, mock_service, tmp_path): + # First call: folder with subfolder and file + # Second call: subfolder contents + call_count = [0] + def list_side_effect(): + mock = MagicMock() + if call_count[0] == 0: + call_count[0] += 1 + mock.execute.return_value = { + "files": [ + {"id": "sub1", "name": "subfolder", "mimeType": "application/vnd.google-apps.folder"}, + {"id": "f1", "name": "file1.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": None, + } + else: + mock.execute.return_value = { + "files": [ + {"id": "f2", "name": "file2.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": None, + } + return mock + + mock_service.files.return_value.list.side_effect = lambda **kw: list_side_effect() + loader._download_single_file = MagicMock(return_value=True) + 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, mock_service, tmp_path): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [ + {"id": "sub1", "name": "subfolder", "mimeType": "application/vnd.google-apps.folder"}, + {"id": "f1", "name": "file1.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": None, + } + loader._download_single_file = MagicMock(return_value=True) + count = loader._download_folder_recursive("folder1", str(tmp_path), recursive=False) + assert count == 1 + + @pytest.mark.unit + def test_download_failure_continues(self, loader, mock_service, tmp_path): + mock_service.files.return_value.list.return_value.execute.return_value = { + "files": [ + {"id": "f1", "name": "fail.txt", "mimeType": "text/plain"}, + {"id": "f2", "name": "ok.txt", "mimeType": "text/plain"}, + ], + "nextPageToken": None, + } + loader._download_single_file = MagicMock(side_effect=[False, True]) + count = loader._download_folder_recursive("folder1", str(tmp_path)) + assert count == 1 + + @pytest.mark.unit + def test_exception_returns_partial_count(self, loader, mock_service, tmp_path): + mock_service.files.return_value.list.return_value.execute.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")) + 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.service = None + loader.auth.build_drive_service.side_effect = Exception("fail") + count = loader._download_folder_contents("folder1", str(tmp_path)) + assert count == 0 + + +class TestEnsureService: + + @pytest.mark.unit + def test_builds_service_when_none(self, loader): + loader.service = None + mock_svc = MagicMock() + loader.auth.build_drive_service.return_value = mock_svc + loader._ensure_service() + assert loader.service == mock_svc + + @pytest.mark.unit + def test_noop_when_service_exists(self, loader, mock_service): + loader._ensure_service() + assert loader.service == mock_service + + @pytest.mark.unit + def test_build_failure_raises(self, loader): + loader.service = None + loader.auth.build_drive_service.side_effect = Exception("fail") + with pytest.raises(ValueError, match="Cannot access Google Drive"): + loader._ensure_service() + + +class TestGetExtensionForMimeType: + + @pytest.mark.unit + def test_known_mime_types(self, loader): + assert loader._get_extension_for_mime_type("application/pdf") == ".pdf" + assert loader._get_extension_for_mime_type("text/plain") == ".txt" + assert loader._get_extension_for_mime_type("text/html") == ".html" + assert loader._get_extension_for_mime_type("text/markdown") == ".md" + + @pytest.mark.unit + def test_unknown_returns_bin(self, loader): + assert loader._get_extension_for_mime_type("application/unknown") == ".bin" diff --git a/tests/parser/connectors/test_share_point_auth.py b/tests/parser/connectors/test_share_point_auth.py new file mode 100644 index 00000000..9cca1290 --- /dev/null +++ b/tests/parser/connectors/test_share_point_auth.py @@ -0,0 +1,326 @@ +"""Tests for SharePointAuth.""" + +import datetime +from unittest.mock import MagicMock, patch + +import pytest + + +@pytest.fixture +def mock_settings(): + s = MagicMock() + s.MICROSOFT_CLIENT_ID = "ms-client-id" + s.MICROSOFT_CLIENT_SECRET = "ms-client-secret" + s.MICROSOFT_TENANT_ID = "tenant-id-123" + s.CONNECTOR_REDIRECT_BASE_URI = "https://redirect.example.com/callback" + s.MONGO_DB_NAME = "test_db" + # Delete MICROSOFT_AUTHORITY so getattr falls back to default + del s.MICROSOFT_AUTHORITY + return s + + +@pytest.fixture +def mock_msal(): + with patch("application.parser.connectors.share_point.auth.ConfidentialClientApplication") as MockMSAL: + mock_app = MagicMock() + MockMSAL.return_value = mock_app + yield mock_app + + +@pytest.fixture +def auth(mock_settings, mock_msal): + with patch("application.parser.connectors.share_point.auth.settings", mock_settings): + from application.parser.connectors.share_point.auth import SharePointAuth + return SharePointAuth() + + +class TestSharePointAuthInit: + + @pytest.mark.unit + def test_init_sets_attributes(self, auth, mock_settings): + assert auth.client_id == "ms-client-id" + assert auth.client_secret == "ms-client-secret" + assert auth.redirect_uri == "https://redirect.example.com/callback" + assert auth.tenant_id == "tenant-id-123" + + @pytest.mark.unit + def test_missing_client_id_raises(self, mock_settings): + mock_settings.MICROSOFT_CLIENT_ID = None + with patch("application.parser.connectors.share_point.auth.settings", mock_settings), \ + patch("application.parser.connectors.share_point.auth.ConfidentialClientApplication"): + from application.parser.connectors.share_point.auth import SharePointAuth + with pytest.raises(ValueError, match="MICROSOFT_CLIENT_ID"): + SharePointAuth() + + @pytest.mark.unit + def test_missing_client_secret_raises(self, mock_settings): + mock_settings.MICROSOFT_CLIENT_SECRET = None + with patch("application.parser.connectors.share_point.auth.settings", mock_settings), \ + patch("application.parser.connectors.share_point.auth.ConfidentialClientApplication"): + from application.parser.connectors.share_point.auth import SharePointAuth + with pytest.raises(ValueError, match="MICROSOFT_CLIENT_SECRET"): + SharePointAuth() + + @pytest.mark.unit + def test_default_authority(self, auth): + assert "login.microsoftonline.com" in auth.authority + assert "tenant-id-123" in auth.authority + + +class TestGetAuthorizationUrl: + + @pytest.mark.unit + def test_returns_url(self, auth, mock_msal): + mock_msal.get_authorization_request_url.return_value = "https://login.microsoftonline.com/auth?state=s1" + url = auth.get_authorization_url(state="s1") + assert url == "https://login.microsoftonline.com/auth?state=s1" + mock_msal.get_authorization_request_url.assert_called_once() + + +class TestExchangeCodeForTokens: + + @pytest.mark.unit + def test_successful_exchange(self, auth, mock_msal): + mock_msal.acquire_token_by_authorization_code.return_value = { + "access_token": "at", + "refresh_token": "rt", + "scope": ["Files.Read"], + "id_token_claims": { + "iss": "https://login.microsoftonline.com/tid/v2.0", + "exp": 1700000000, + "tid": "work-tenant-id", + "name": "Test User", + "preferred_username": "test@example.com", + }, + } + result = auth.exchange_code_for_tokens("auth_code") + assert result["access_token"] == "at" + assert result["refresh_token"] == "rt" + assert result["user_info"]["name"] == "Test User" + assert result["user_info"]["email"] == "test@example.com" + assert result["allows_shared_content"] is True + + @pytest.mark.unit + def test_error_in_response_raises(self, auth, mock_msal): + mock_msal.acquire_token_by_authorization_code.return_value = { + "error": "invalid_grant", + "error_description": "Code expired", + } + with pytest.raises(ValueError, match="Code expired"): + auth.exchange_code_for_tokens("bad_code") + + +class TestRefreshAccessToken: + + @pytest.mark.unit + def test_successful_refresh(self, auth, mock_msal): + mock_msal.acquire_token_by_refresh_token.return_value = { + "access_token": "new_at", + "refresh_token": "new_rt", + "scope": ["Files.Read"], + "id_token_claims": { + "iss": "https://issuer", + "exp": 1700001000, + "tid": "work-tid", + "name": "User", + "preferred_username": "u@example.com", + }, + } + result = auth.refresh_access_token("old_rt") + assert result["access_token"] == "new_at" + assert result["refresh_token"] == "new_rt" + + @pytest.mark.unit + def test_error_in_response_raises(self, auth, mock_msal): + mock_msal.acquire_token_by_refresh_token.return_value = { + "error": "invalid_grant", + "error_description": "Token revoked", + } + with pytest.raises(ValueError, match="Token revoked"): + auth.refresh_access_token("bad_rt") + + +class TestIsTokenExpired: + + @pytest.mark.unit + def test_expired_token(self, auth): + past = int((datetime.datetime.now() - datetime.timedelta(hours=1)).timestamp()) + assert auth.is_token_expired({"expiry": past}) is True + + @pytest.mark.unit + def test_valid_token(self, auth): + future = int((datetime.datetime.now() + datetime.timedelta(hours=1)).timestamp()) + assert auth.is_token_expired({"expiry": future}) is False + + @pytest.mark.unit + def test_within_buffer(self, auth): + almost_expired = int((datetime.datetime.now() + datetime.timedelta(seconds=30)).timestamp()) + assert auth.is_token_expired({"expiry": almost_expired}) is True + + @pytest.mark.unit + def test_none_token_info(self, auth): + assert auth.is_token_expired(None) is True + + @pytest.mark.unit + def test_missing_expiry(self, auth): + assert auth.is_token_expired({}) is True + + @pytest.mark.unit + def test_none_expiry(self, auth): + assert auth.is_token_expired({"expiry": None}) is True + + +class TestSanitizeTokenInfo: + + @pytest.mark.unit + def test_includes_allows_shared_content(self, auth): + token_info = { + "access_token": "at", + "refresh_token": "rt", + "token_uri": "https://uri", + "expiry": 123, + "allows_shared_content": True, + } + result = auth.sanitize_token_info(token_info) + assert result["allows_shared_content"] is True + assert result["access_token"] == "at" + + @pytest.mark.unit + def test_defaults_allows_shared_content_to_false(self, auth): + token_info = { + "access_token": "at", + "refresh_token": "rt", + "token_uri": "https://uri", + "expiry": 123, + } + result = auth.sanitize_token_info(token_info) + assert result["allows_shared_content"] is False + + @pytest.mark.unit + def test_with_extra_fields(self, auth): + token_info = { + "access_token": "at", + "refresh_token": "rt", + "token_uri": "https://uri", + "expiry": 123, + "allows_shared_content": True, + } + result = auth.sanitize_token_info(token_info, custom="val") + assert result["custom"] == "val" + + +class TestAllowsSharedContent: + + @pytest.mark.unit + def test_work_account_returns_true(self, auth): + claims = {"tid": "some-work-tenant-id"} + assert auth._allows_shared_content(claims) is True + + @pytest.mark.unit + def test_personal_account_returns_false(self, auth): + claims = {"tid": "9188040d-6c67-4c5b-b112-36a304b66dad"} + assert auth._allows_shared_content(claims) is False + + @pytest.mark.unit + def test_empty_tid_returns_false(self, auth): + assert auth._allows_shared_content({"tid": ""}) is False + assert auth._allows_shared_content({}) is False + + +class TestMapTokenResponse: + + @pytest.mark.unit + def test_maps_all_fields(self, auth): + result = { + "access_token": "at", + "refresh_token": "rt", + "scope": ["Files.Read"], + "id_token_claims": { + "iss": "https://issuer", + "exp": 1700000000, + "tid": "work-tid", + "name": "User Name", + "preferred_username": "user@example.com", + }, + } + mapped = auth.map_token_response(result) + assert mapped["access_token"] == "at" + assert mapped["refresh_token"] == "rt" + assert mapped["token_uri"] == "https://issuer" + assert mapped["scopes"] == ["Files.Read"] + assert mapped["expiry"] == 1700000000 + assert mapped["user_info"]["name"] == "User Name" + assert mapped["user_info"]["email"] == "user@example.com" + assert mapped["allows_shared_content"] is True + + @pytest.mark.unit + def test_missing_claims_uses_defaults(self, auth): + result = {"access_token": "at", "refresh_token": "rt"} + mapped = auth.map_token_response(result) + assert mapped["token_uri"] is None + assert mapped["expiry"] is None + assert mapped["user_info"]["name"] is None + assert mapped["allows_shared_content"] is False + + +class TestGetTokenInfoFromSession: + + def _mock_mongo(self, mock_settings, find_one_return): + mock_collection = MagicMock() + mock_collection.find_one.return_value = find_one_return + mock_db = MagicMock() + mock_db.__getitem__ = MagicMock(return_value=mock_collection) + mock_client = {mock_settings.MONGO_DB_NAME: mock_db} + return mock_client + + @pytest.mark.unit + def test_valid_session(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, { + "session_token": "st", + "token_info": {"access_token": "at", "refresh_token": "rt"}, + }) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + result = auth.get_token_info_from_session("st") + assert result["access_token"] == "at" + assert "token_uri" in result + + @pytest.mark.unit + def test_session_not_found_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, None) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve SharePoint token"): + auth.get_token_info_from_session("bad") + + @pytest.mark.unit + def test_missing_token_info_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, {"session_token": "st"}) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve SharePoint token"): + auth.get_token_info_from_session("st") + + @pytest.mark.unit + def test_empty_token_info_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, {"session_token": "st", "token_info": None}) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve SharePoint token"): + auth.get_token_info_from_session("st") + + @pytest.mark.unit + def test_missing_required_fields_raises(self, auth, mock_settings): + mock_client = self._mock_mongo(mock_settings, { + "session_token": "st", + "token_info": {"access_token": "at"}, + }) + + with patch("application.core.mongo_db.MongoDB.get_client", return_value=mock_client), \ + patch("application.core.settings.settings", mock_settings): + with pytest.raises(ValueError, match="Failed to retrieve SharePoint token"): + auth.get_token_info_from_session("st") diff --git a/tests/parser/connectors/test_share_point_loader.py b/tests/parser/connectors/test_share_point_loader.py new file mode 100644 index 00000000..b8028356 --- /dev/null +++ b/tests/parser/connectors/test_share_point_loader.py @@ -0,0 +1,1036 @@ +"""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")