mirror of
https://github.com/tiennm99/DocsGPT.git
synced 2026-10-12 22:14:05 +00:00
chore: connector tests
This commit is contained in:
Alex committed
1 parent
6492852f7d
commit
e06debad5f
8 files changed
+2866
No files matched your search
Whitespace-only changes.
Whitespace-only changes.
@@ -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
|
||||
@@ -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)
|
||||
@@ -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
|
||||
@@ -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"
|
||||
@@ -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")
|
||||
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user