gitea-admin/gh-christianlouis-docuelevate
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(coverage): add coverage tests for app/api/onedrive.py and app/api/google_drive.py

Browse Source 
Add targeted tests covering uncovered lines:
- OneDrive: token refresh failure, token rotation with .env update,
  DB persist failure, user info failure, exception handlers
- Google Drive: service account test-token paths, update settings
  exception handler, get-token-info outer exception

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot]
2026-02-25 15:41:01 +00:00
parent 4449671c88
commit e3cd687757
2 changed files with 496 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_google_drive_coverage2.py
+151
View File
@@ -0,0 +1,151 @@
"""
Coverage-targeted tests for app/api/google_drive.py
Focuses on uncovered lines: 125-127, 214-255, 339-341.
"""
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
@pytest.mark.unit
class TestUpdateSettingsExceptionHandler:
"""Cover lines 125-127: update_google_drive_settings outer exception."""
def test_update_settings_outer_exception(self, client: TestClient):
"""Trigger the outer exception handler in update_google_drive_settings."""
with patch("app.api.google_drive.save_setting_to_db", side_effect=Exception("DB crash")):
response = client.post(
"/api/google-drive/update-settings",
data={
"refresh_token": "test_token",
},
)
assert response.status_code == 500
assert "Failed to update Google Drive settings" in response.json()["detail"]
@pytest.mark.unit
class TestTestTokenServiceAccount:
"""Cover lines 214-255: service account test-token paths."""
@patch("app.api.google_drive.get_google_drive_service")
def test_test_token_service_account_success_no_delegation(self, mock_get_service, client: TestClient):
"""Test successful service account connection without delegation."""
from app.config import settings
mock_service = MagicMock()
mock_service.about().get().execute.return_value = {
"user": {"emailAddress": "sa@project.iam.gserviceaccount.com"}
}
mock_get_service.return_value = mock_service
with (
patch.object(settings, "google_drive_use_oauth", False),
patch.object(settings, "google_drive_credentials_json", '{"type": "service_account"}'),
patch.object(settings, "google_drive_delegate_to", None),
):
response = client.get("/api/google-drive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert data["auth_type"] == "service_account"
assert "sa@project.iam.gserviceaccount.com" in data["message"]
@patch("app.api.google_drive.get_google_drive_service")
def test_test_token_service_account_with_delegation(self, mock_get_service, client: TestClient):
"""Test service account with delegation shows delegated user info."""
from app.config import settings
mock_service = MagicMock()
mock_service.about().get().execute.return_value = {
"user": {"emailAddress": "sa@project.iam.gserviceaccount.com"}
}
mock_get_service.return_value = mock_service
with (
patch.object(settings, "google_drive_use_oauth", False),
patch.object(settings, "google_drive_credentials_json", '{"type": "service_account"}'),
patch.object(settings, "google_drive_delegate_to", "user@domain.com"),
):
response = client.get("/api/google-drive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
assert "delegating as user@domain.com" in data["message"]
def test_test_token_service_account_not_configured(self, client: TestClient):
"""Test service account not configured returns error."""
from app.config import settings
with (
patch.object(settings, "google_drive_use_oauth", False),
patch.object(settings, "google_drive_credentials_json", ""),
):
response = client.get("/api/google-drive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"
assert "not configured" in data["message"]
@patch("app.api.google_drive.get_google_drive_service")
def test_test_token_service_account_connection_error(self, mock_get_service, client: TestClient):
"""Test service account connection error (lines 245-251)."""
from app.config import settings
mock_get_service.side_effect = Exception("Service account credentials invalid")
with (
patch.object(settings, "google_drive_use_oauth", False),
patch.object(settings, "google_drive_credentials_json", '{"type": "service_account"}'),
):
response = client.get("/api/google-drive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"
assert "Service account validation failed" in data["message"]
@pytest.mark.unit
class TestGetTokenInfoOuterException:
"""Cover lines 339-341: get_google_drive_token_info outer exception."""
def test_get_token_info_outer_exception(self, client: TestClient):
"""Trigger the outer exception handler in get_google_drive_token_info."""
from app.config import settings
with patch.object(
type(settings),
"google_drive_use_oauth",
property(fget=lambda self: (_ for _ in ()).throw(Exception("boom"))),
):
response = client.get("/api/google-drive/get-token-info")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"
assert "Unexpected error" in data["message"]
@pytest.mark.unit
class TestTestTokenOuterException:
"""Cover lines 253-255: test_google_drive_token outer exception handler."""
def test_test_token_outer_exception(self, client: TestClient):
"""Trigger the outermost exception handler in test_google_drive_token."""
from app.config import settings
# Patch getattr on settings to raise on the first call within the try block
with patch("app.api.google_drive.getattr", side_effect=Exception("unexpected error")):
response = client.get("/api/google-drive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"
assert "Unexpected error" in data["message"]
tests/test_api_onedrive_coverage.py
+345
View File
@@ -0,0 +1,345 @@
"""
Coverage-targeted tests for app/api/onedrive.py
Focuses on uncovered lines: 98-99, 121-143, 160-161, 170-171,
324-326, 400-402, 436-438.
"""
import os
from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient
@pytest.mark.unit
class TestTestTokenRefreshFailed:
"""Cover lines 98-99: token refresh returns non-200."""
@patch("app.api.onedrive.requests.post")
def test_test_token_refresh_returns_non_200(self, mock_post, client: TestClient):
"""Test token refresh returning a failure status hits the error branch."""
from app.config import settings
mock_resp = MagicMock()
mock_resp.status_code = 400
mock_resp.text = "invalid_grant"
mock_post.return_value = mock_resp
with (
patch.object(settings, "onedrive_refresh_token", "tok"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
):
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"
assert data["needs_reauth"] is True
@pytest.mark.unit
class TestTestTokenRotation:
"""Cover lines 121-143, 160-161: token rotation with .env and DB persist."""
@patch("app.api.onedrive.requests.get")
@patch("app.api.onedrive.requests.post")
def test_token_rotation_env_file_exists(self, mock_post, mock_get, client: TestClient, tmp_path):
"""When a new refresh token is received and .env file exists, it should be updated."""
from app.config import settings
# Create a temporary .env file
env_file = tmp_path / ".env"
env_file.write_text("ONEDRIVE_REFRESH_TOKEN=old_token\nOTHER=value\n")
# Mock token refresh returning a new refresh token
mock_post_resp = MagicMock()
mock_post_resp.status_code = 200
mock_post_resp.json.return_value = {
"access_token": "new_access",
"refresh_token": "brand_new_token",
"expires_in": 3600,
}
mock_post.return_value = mock_post_resp
# Mock user info
mock_get_resp = MagicMock()
mock_get_resp.status_code = 200
mock_get_resp.json.return_value = {
"displayName": "Test User",
"userPrincipalName": "test@example.com",
}
mock_get.return_value = mock_get_resp
with (
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.join", return_value=str(env_file)),
patch("app.api.onedrive.os.path.exists", return_value=True),
patch("app.database.SessionLocal") as mock_session_local,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
):
mock_db = MagicMock()
mock_session_local.return_value = mock_db
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "success"
@patch("app.api.onedrive.requests.get")
@patch("app.api.onedrive.requests.post")
def test_token_rotation_env_not_existing(self, mock_post, mock_get, client: TestClient):
"""Token rotation when .env doesn't exist still succeeds."""
from app.config import settings
mock_post_resp = MagicMock()
mock_post_resp.status_code = 200
mock_post_resp.json.return_value = {
"access_token": "new_access",
"refresh_token": "brand_new_token",
"expires_in": 3600,
}
mock_post.return_value = mock_post_resp
mock_get_resp = MagicMock()
mock_get_resp.status_code = 200
mock_get_resp.json.return_value = {
"displayName": "Test User",
"userPrincipalName": "test@example.com",
}
mock_get.return_value = mock_get_resp
with (
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.exists", return_value=False),
patch("app.database.SessionLocal") as mock_session_local,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
):
mock_db = MagicMock()
mock_session_local.return_value = mock_db
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
assert response.json()["status"] == "success"
@patch("app.api.onedrive.requests.get")
@patch("app.api.onedrive.requests.post")
def test_token_rotation_env_write_failure(self, mock_post, mock_get, client: TestClient):
"""Token rotation when .env write fails (lines 142-143) still continues."""
from app.config import settings
mock_post_resp = MagicMock()
mock_post_resp.status_code = 200
mock_post_resp.json.return_value = {
"access_token": "new_access",
"refresh_token": "brand_new_token",
"expires_in": 3600,
}
mock_post.return_value = mock_post_resp
mock_get_resp = MagicMock()
mock_get_resp.status_code = 200
mock_get_resp.json.return_value = {
"displayName": "Test User",
"userPrincipalName": "test@example.com",
}
mock_get.return_value = mock_get_resp
with (
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.exists", return_value=True),
patch("builtins.open", side_effect=PermissionError("Permission denied")),
patch("app.database.SessionLocal") as mock_session_local,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
):
mock_db = MagicMock()
mock_session_local.return_value = mock_db
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
assert response.json()["status"] == "success"
@patch("app.api.onedrive.requests.get")
@patch("app.api.onedrive.requests.post")
def test_token_rotation_db_persist_failure(self, mock_post, mock_get, client: TestClient):
"""Token rotation when DB persist fails (lines 160-161) still continues."""
from app.config import settings
mock_post_resp = MagicMock()
mock_post_resp.status_code = 200
mock_post_resp.json.return_value = {
"access_token": "new_access",
"refresh_token": "brand_new_token",
"expires_in": 3600,
}
mock_post.return_value = mock_post_resp
mock_get_resp = MagicMock()
mock_get_resp.status_code = 200
mock_get_resp.json.return_value = {
"displayName": "Test User",
"userPrincipalName": "test@example.com",
}
mock_get.return_value = mock_get_resp
with (
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.exists", return_value=False),
patch("app.database.SessionLocal", side_effect=Exception("DB error")),
):
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
assert response.json()["status"] == "success"
@pytest.mark.unit
class TestTestTokenUserInfoFailed:
"""Cover lines 170-171: user info request fails."""
@patch("app.api.onedrive.requests.get")
@patch("app.api.onedrive.requests.post")
def test_user_info_returns_non_200(self, mock_post, mock_get, client: TestClient):
"""Test when user info request fails after successful token refresh."""
from app.config import settings
mock_post_resp = MagicMock()
mock_post_resp.status_code = 200
mock_post_resp.json.return_value = {
"access_token": "tok",
"expires_in": 3600,
}
mock_post.return_value = mock_post_resp
mock_get_resp = MagicMock()
mock_get_resp.status_code = 401
mock_get_resp.text = "Unauthorized"
mock_get.return_value = mock_get_resp
with (
patch.object(settings, "onedrive_refresh_token", "tok"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
):
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"
assert "401" in data["message"]
@pytest.mark.unit
class TestTokenRotationEnvAppendLine:
"""Cover the branch at line 134 where token line is not found in .env and must be appended."""
@patch("app.api.onedrive.requests.get")
@patch("app.api.onedrive.requests.post")
def test_token_rotation_appends_to_env(self, mock_post, mock_get, client: TestClient, tmp_path):
"""When .env exists but doesn't have ONEDRIVE_REFRESH_TOKEN, it should append."""
from app.config import settings
env_file = tmp_path / ".env"
env_file.write_text("OTHER_KEY=value\n")
mock_post_resp = MagicMock()
mock_post_resp.status_code = 200
mock_post_resp.json.return_value = {
"access_token": "new_access",
"refresh_token": "brand_new_token",
"expires_in": 3600,
}
mock_post.return_value = mock_post_resp
mock_get_resp = MagicMock()
mock_get_resp.status_code = 200
mock_get_resp.json.return_value = {
"displayName": "Test User",
"userPrincipalName": "test@example.com",
}
mock_get.return_value = mock_get_resp
with (
patch.object(settings, "onedrive_refresh_token", "old_token"),
patch.object(settings, "onedrive_client_id", "cid"),
patch.object(settings, "onedrive_client_secret", "sec"),
patch("app.api.onedrive.os.path.join", return_value=str(env_file)),
patch("app.api.onedrive.os.path.exists", return_value=True),
patch("app.database.SessionLocal") as mock_sl,
patch("app.api.onedrive.save_setting_to_db"),
patch("app.api.onedrive.notify_settings_updated"),
):
mock_sl.return_value = MagicMock()
response = client.get("/api/onedrive/test-token")
assert response.status_code == 200
assert response.json()["status"] == "success"
@pytest.mark.unit
class TestSaveSettingsException:
"""Cover lines 324-326: save_onedrive_settings outer exception handler."""
def test_save_settings_outer_exception(self, client: TestClient):
"""Trigger the outer exception handler in save_onedrive_settings."""
with patch("app.api.onedrive.os.path.join", side_effect=Exception("Unexpected boom")):
response = client.post(
"/api/onedrive/save-settings",
data={
"refresh_token": "test_token",
"client_id": "cid",
"client_secret": "sec",
},
)
assert response.status_code == 500
assert "Failed to save OneDrive settings" in response.json()["detail"]
@pytest.mark.unit
class TestUpdateSettingsException:
"""Cover lines 400-402: update_onedrive_settings outer exception handler."""
def test_update_settings_outer_exception(self, client: TestClient):
"""Trigger the outer exception handler in update_onedrive_settings."""
with patch("app.api.onedrive.save_setting_to_db", side_effect=Exception("DB crash")):
response = client.post(
"/api/onedrive/update-settings",
data={
"refresh_token": "test_token",
},
)
assert response.status_code == 500
assert "Failed to update OneDrive settings" in response.json()["detail"]
@pytest.mark.unit
class TestGetFullConfigException:
"""Cover lines 436-438: get_onedrive_full_config exception handler."""
def test_get_full_config_exception(self, client: TestClient):
"""Trigger the exception handler in get_onedrive_full_config."""
from app.config import settings
with patch.object(
type(settings), "onedrive_client_id", property(fget=lambda self: (_ for _ in ()).throw(Exception("boom")))
):
response = client.get("/api/onedrive/get-full-config")
assert response.status_code == 200
data = response.json()
assert data["status"] == "error"