copilot-swe-agent[bot]
204000aabc
Resolve 3 merge conflicts and renumber the automation_hooks migration to follow main's migration chain (036_add_document_translation_fields). Conflicts resolved: - app/api/__init__.py: add automation_router alongside main's new routers - app/utils/settings_service.py: add automation_hooks_enabled alongside compliance_enabled - tests/conftest.py: add AutomationHook alongside AuditLog/ComplianceTemplate imports Migration renumbered: - 027_add_automation_hooks → 037_add_automation_hooks - down_revision: 026_add_scheduled_jobs → 036_add_document_translation_fields Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
359 lines
13 KiB
Python
359 lines
13 KiB
Python
"""Comprehensive unit tests for app/api/onedrive.py module."""
|
|
|
|
from datetime import timedelta
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestExchangeOneDriveToken:
|
|
"""Tests for POST /onedrive/exchange-token endpoint."""
|
|
|
|
@patch("app.api.onedrive.exchange_oauth_token")
|
|
def test_exchange_token_success(self, mock_exchange):
|
|
"""Test successful token exchange."""
|
|
mock_exchange.return_value = {
|
|
"refresh_token": "refresh_token_value",
|
|
"expires_in": 3600,
|
|
}
|
|
|
|
# Response should include tokens
|
|
pass
|
|
|
|
@patch("app.api.onedrive.exchange_oauth_token")
|
|
def test_exchange_token_with_tenant_id(self, mock_exchange):
|
|
"""Test token exchange with specific tenant ID."""
|
|
mock_exchange.return_value = {
|
|
"refresh_token": "refresh_token_value",
|
|
"expires_in": 3600,
|
|
}
|
|
|
|
# Should use provided tenant_id in token URL
|
|
pass
|
|
|
|
@patch("app.api.onedrive.exchange_oauth_token")
|
|
def test_exchange_token_calls_oauth_helper(self, mock_exchange):
|
|
"""Test that exchange_oauth_token is called correctly."""
|
|
mock_exchange.return_value = {
|
|
"refresh_token": "token",
|
|
"expires_in": 3600,
|
|
}
|
|
|
|
# Should call with provider_name="OneDrive"
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestTestOneDriveToken:
|
|
"""Tests for GET /onedrive/test-token endpoint."""
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
@patch("httpx.AsyncClient.get", new_callable=AsyncMock)
|
|
def test_test_token_success(self, mock_get, mock_post):
|
|
"""Test successful token validation."""
|
|
from app.config import settings
|
|
|
|
# Mock token refresh response
|
|
mock_post_response = MagicMock()
|
|
mock_post_response.status_code = 200
|
|
mock_post_response.json.return_value = {
|
|
"access_token": "new_access_token",
|
|
"expires_in": 3600,
|
|
}
|
|
mock_post.return_value = mock_post_response
|
|
|
|
# Mock user info response
|
|
mock_get_response = MagicMock()
|
|
mock_get_response.status_code = 200
|
|
mock_get_response.json.return_value = {
|
|
"displayName": "Test User",
|
|
"userPrincipalName": "test@example.com",
|
|
}
|
|
mock_get.return_value = mock_get_response
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", "token"):
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
with patch.object(settings, "onedrive_tenant_id", "common"):
|
|
# Should return success
|
|
pass
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
def test_test_token_not_configured(self, mock_post):
|
|
"""Test when credentials are not configured."""
|
|
from app.config import settings
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", None):
|
|
# Should return error
|
|
pass
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
def test_test_token_refresh_failed(self, mock_post):
|
|
"""Test when token refresh fails."""
|
|
from app.config import settings
|
|
|
|
mock_response = MagicMock()
|
|
mock_response.status_code = 400
|
|
mock_response.text = "Invalid refresh token"
|
|
mock_post.return_value = mock_response
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", "token"):
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
# Should return error with needs_reauth
|
|
pass
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
@patch("httpx.AsyncClient.get", new_callable=AsyncMock)
|
|
def test_test_token_user_info_failed(self, mock_get, mock_post):
|
|
"""Test when user info request fails."""
|
|
from app.config import settings
|
|
|
|
# Token refresh succeeds
|
|
mock_post_response = MagicMock()
|
|
mock_post_response.status_code = 200
|
|
mock_post_response.json.return_value = {"access_token": "token", "expires_in": 3600}
|
|
mock_post.return_value = mock_post_response
|
|
|
|
# User info fails
|
|
mock_get_response = MagicMock()
|
|
mock_get_response.status_code = 401
|
|
mock_get_response.text = "Unauthorized"
|
|
mock_get.return_value = mock_get_response
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", "token"):
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
# Should return error
|
|
pass
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
@patch("httpx.AsyncClient.get", new_callable=AsyncMock)
|
|
@patch("builtins.open", create=True)
|
|
@patch("os.path.exists")
|
|
def test_test_token_updates_refresh_token(self, mock_exists, mock_open, mock_get, mock_post):
|
|
"""Test that new refresh token is saved when received."""
|
|
from app.config import settings
|
|
|
|
# Mock token refresh with new refresh token
|
|
mock_post_response = MagicMock()
|
|
mock_post_response.status_code = 200
|
|
mock_post_response.json.return_value = {
|
|
"access_token": "new_access_token",
|
|
"refresh_token": "new_refresh_token",
|
|
"expires_in": 3600,
|
|
}
|
|
mock_post.return_value = mock_post_response
|
|
|
|
# Mock user info
|
|
mock_get_response = MagicMock()
|
|
mock_get_response.status_code = 200
|
|
mock_get_response.json.return_value = {
|
|
"displayName": "Test User",
|
|
"userPrincipalName": "test@example.com",
|
|
}
|
|
mock_get.return_value = mock_get_response
|
|
|
|
# Mock .env file
|
|
mock_exists.return_value = True
|
|
mock_file = MagicMock()
|
|
mock_file.readlines.return_value = ["ONEDRIVE_REFRESH_TOKEN=old_token\n"]
|
|
mock_open.return_value.__enter__.return_value = mock_file
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", "old_token"):
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
# Should update refresh token in memory and file
|
|
pass
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
@patch("httpx.AsyncClient.get", new_callable=AsyncMock)
|
|
def test_test_token_expiration_info(self, mock_get, mock_post):
|
|
"""Test that expiration info is included."""
|
|
from app.config import settings
|
|
|
|
mock_post_response = MagicMock()
|
|
mock_post_response.status_code = 200
|
|
mock_post_response.json.return_value = {
|
|
"access_token": "token",
|
|
"expires_in": 3600,
|
|
}
|
|
mock_post.return_value = mock_post_response
|
|
|
|
mock_get_response = MagicMock()
|
|
mock_get_response.status_code = 200
|
|
mock_get_response.json.return_value = {
|
|
"displayName": "Test User",
|
|
"userPrincipalName": "test@example.com",
|
|
}
|
|
mock_get.return_value = mock_get_response
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", "token"):
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
# token_info should include expiration details
|
|
pass
|
|
|
|
@patch("httpx.AsyncClient.post", new_callable=AsyncMock)
|
|
def test_test_token_exception_handling(self, mock_post):
|
|
"""Test handling of exceptions."""
|
|
from app.config import settings
|
|
|
|
mock_post.side_effect = Exception("Network error")
|
|
|
|
with patch.object(settings, "onedrive_refresh_token", "token"):
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
# Should return error
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestFormatTimeRemainingOneDrive:
|
|
"""Tests for format_time_remaining helper function."""
|
|
|
|
def test_format_expired(self):
|
|
"""Test formatting expired time."""
|
|
from app.api.onedrive import format_time_remaining
|
|
|
|
delta = timedelta(seconds=-100)
|
|
result = format_time_remaining(delta)
|
|
assert result == "Expired"
|
|
|
|
def test_format_days(self):
|
|
"""Test formatting with days."""
|
|
from app.api.onedrive import format_time_remaining
|
|
|
|
delta = timedelta(days=5, hours=3)
|
|
result = format_time_remaining(delta)
|
|
assert "5 days" in result
|
|
|
|
def test_format_hours(self):
|
|
"""Test formatting with hours."""
|
|
from app.api.onedrive import format_time_remaining
|
|
|
|
delta = timedelta(hours=3, minutes=30)
|
|
result = format_time_remaining(delta)
|
|
assert "3 hours" in result
|
|
|
|
def test_format_minutes(self):
|
|
"""Test formatting with minutes."""
|
|
from app.api.onedrive import format_time_remaining
|
|
|
|
delta = timedelta(minutes=45)
|
|
result = format_time_remaining(delta)
|
|
assert "45 minutes" in result
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestSaveOneDriveSettings:
|
|
"""Tests for POST /onedrive/save-settings endpoint."""
|
|
|
|
@patch("builtins.open", create=True)
|
|
@patch("os.path.exists")
|
|
def test_save_settings_success(self, mock_exists, mock_open):
|
|
"""Test successful saving to .env file."""
|
|
mock_exists.return_value = True
|
|
mock_file = MagicMock()
|
|
mock_file.readlines.return_value = []
|
|
mock_open.return_value.__enter__.return_value = mock_file
|
|
|
|
# Should save settings
|
|
pass
|
|
|
|
@patch("os.path.exists")
|
|
def test_save_settings_no_env_file(self, mock_exists):
|
|
"""Test when .env file doesn't exist."""
|
|
mock_exists.return_value = False
|
|
|
|
# Should raise HTTPException
|
|
pass
|
|
|
|
@patch("builtins.open", create=True)
|
|
@patch("os.path.exists")
|
|
def test_save_settings_all_fields(self, mock_exists, mock_open):
|
|
"""Test saving all OneDrive settings."""
|
|
mock_exists.return_value = True
|
|
mock_file = MagicMock()
|
|
mock_file.readlines.return_value = []
|
|
mock_open.return_value.__enter__.return_value = mock_file
|
|
|
|
# Should save all fields
|
|
pass
|
|
|
|
@patch("builtins.open", create=True)
|
|
@patch("os.path.exists")
|
|
def test_save_settings_updates_memory(self, mock_exists, mock_open):
|
|
"""Test that in-memory settings are updated."""
|
|
|
|
mock_exists.return_value = True
|
|
mock_file = MagicMock()
|
|
mock_file.readlines.return_value = []
|
|
mock_open.return_value.__enter__.return_value = mock_file
|
|
|
|
# Should update settings object
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestUpdateOneDriveSettings:
|
|
"""Tests for POST /onedrive/update-settings endpoint."""
|
|
|
|
@patch("app.tasks.upload_to_onedrive.get_onedrive_token")
|
|
def test_update_settings_success(self, mock_get_token):
|
|
"""Test successful settings update."""
|
|
|
|
mock_get_token.return_value = "access_token"
|
|
|
|
# Should update settings and test token
|
|
pass
|
|
|
|
@patch("app.tasks.upload_to_onedrive.get_onedrive_token")
|
|
def test_update_settings_token_test_failed(self, mock_get_token):
|
|
"""Test when token test fails after update."""
|
|
|
|
mock_get_token.side_effect = Exception("Token test failed")
|
|
|
|
# Should return warning
|
|
pass
|
|
|
|
def test_update_settings_exception_handling(self):
|
|
"""Test handling of exceptions."""
|
|
# Should raise HTTPException with 500 status
|
|
pass
|
|
|
|
|
|
@pytest.mark.unit
|
|
class TestGetOneDriveFullConfig:
|
|
"""Tests for GET /onedrive/get-full-config endpoint."""
|
|
|
|
def test_get_full_config_success(self):
|
|
"""Test successful config retrieval."""
|
|
from app.config import settings
|
|
|
|
with patch.object(settings, "onedrive_client_id", "client_id"):
|
|
with patch.object(settings, "onedrive_client_secret", "secret"):
|
|
with patch.object(settings, "onedrive_tenant_id", "tenant"):
|
|
with patch.object(settings, "onedrive_refresh_token", "token"):
|
|
# Should return config object
|
|
pass
|
|
|
|
def test_get_full_config_env_format(self):
|
|
"""Test that env_format is generated correctly."""
|
|
|
|
# env_format should contain all settings as KEY=value
|
|
pass
|
|
|
|
def test_get_full_config_default_values(self):
|
|
"""Test default values when settings not configured."""
|
|
from app.config import settings
|
|
|
|
with patch.object(settings, "onedrive_client_id", None):
|
|
# Should use empty string for missing values
|
|
pass
|
|
|
|
def test_get_full_config_exception_handling(self):
|
|
"""Test handling of exceptions."""
|
|
# Should return error status
|