Files
gh-christianlouis-docuelevate/tests/test_connections.py
T
copilot-swe-agent[bot] c7d3ec57c3 fix: restore all code deleted/truncated by d2217531 Jules SSRF commit
Commit d2217531 (google-labs-jules SSRF fix) catastrophically deleted
11,500+ lines across 100+ files while fixing an unrelated IMAP issue.

Restored from d2217531^ (pre-bad-commit state):

Deleted files (fully restored):
- app/api/{automation,classification_rules,comments,sharing}.py
- app/middleware/upload_rate_limit.py
- app/tasks/{automation_tasks,classify_document}.py
- app/utils/{automation_hooks,classification_rules}.py
- docs/AppleAppStoreCompliance.md
- frontend/input.css, package.json, package-lock.json, tailwind.config.js
- frontend/static/js/{annotations,claim,comments,sharing}.js
- frontend/templates/{admin_connections,file_annotations,file_summary}.html
- tests/{test_api_files_comprehensive,test_auth_extended,test_sharing,
         test_comments,test_connections,test_imap_profiles,test_api_sessions,
         test_automation,test_classification_rules,test_api_advanced_filters,
         test_api_classification_rules,test_upload_rate_limit,test_api_dropbox,
         test_classify_document,test_comments_ui,test_upload_to_icloud,
         test_api_onedrive_comprehensive,test_frontend_build,test_sentry,
         test_diagnostic,test_database,test_views_dropbox,test_local_auth}.py

Truncated files (content restored):
- app/{auth,config,main,models,celery_worker,database}.py
- app/api/{__init__,api_tokens,diagnostic,dropbox,files,google_drive,
           integrations,local_auth,mobile,onedrive,pipelines,qr_auth,
           settings,url_upload}.py
- app/middleware/upload_rate_limit.py
- app/tasks/upload_to_nextcloud.py
- app/utils/{allowed_types,settings_service,settings_sync,user_scope,webhook}.py
- app/views/{base,dropbox,files,google_drive,onedrive,settings}.py
- docs/{API,AuthenticationSetup,ConfigurationGuide,DatabaseConfiguration,
        DeploymentGuide,DropboxSetup,GoogleDriveSetup,KubernetesDeployment,
        MobileApp,OneDriveSetup,ProductionReadiness,SentrySetup,
        SocialLoginSetup,UserGuide}.md
- frontend/static/{js/upload.js,styles.css}
- frontend/templates/{api_tokens,base,devices,dropbox,dropbox_callback,
                      file_view,files,google_drive,onedrive,onedrive_callback,
                      signup}.html
- frontend/translations/en.json
- migrations/env.py
- tests/{conftest,test_api_integrations,test_api_mobile,test_api_settings,
         test_api_tokens,test_audit_logs,test_duplicates,test_imap_tasks,
         test_setup_wizard,test_views_files_comprehensive}.py

Security fixes kept from post-d2217531 commits:
- app/utils/network.py: DNS SSRF fail-secure fix (06b0fced)
- app/utils/file_operations.py: path traversal fix (1018ea17)
- tests/test_imap_tasks.py: re-applied 4 is_private_ip mock patches

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
Agent-Logs-Url: https://github.com/christianlouis/DocuElevate/sessions/51133dd8-9bec-41ab-aa10-3de753634187
2026-03-23 23:52:39 +00:00

495 lines
20 KiB
Python

"""Tests for the Connections admin page and new authentication providers."""
from unittest.mock import MagicMock, patch
import pytest
from fastapi import status
@pytest.mark.unit
class TestDropboxComplianceFix:
"""Tests for the Dropbox userinfo compliance fix."""
def test_dropbox_compliance_fix_adds_sub(self):
"""Test that compliance fix adds 'sub' from account_id."""
from app.auth import _dropbox_userinfo_compliance_fix
data = {"account_id": "dbid:abc123", "email": "test@example.com"}
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
assert result["sub"] == "dbid:abc123"
def test_dropbox_compliance_fix_does_not_overwrite_sub(self):
"""Test that compliance fix preserves existing 'sub'."""
from app.auth import _dropbox_userinfo_compliance_fix
data = {"account_id": "dbid:abc123", "sub": "existing-sub"}
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
assert result["sub"] == "existing-sub"
def test_dropbox_compliance_fix_normalizes_name(self):
"""Test that compliance fix normalizes nested name object."""
from app.auth import _dropbox_userinfo_compliance_fix
data = {"name": {"display_name": "John Doe", "given_name": "John"}}
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
assert result["name"] == "John Doe"
def test_dropbox_compliance_fix_handles_no_name(self):
"""Test compliance fix works when no name is present."""
from app.auth import _dropbox_userinfo_compliance_fix
data = {"email": "test@example.com"}
result = _dropbox_userinfo_compliance_fix(None, None, None, data)
assert "email" in result
@pytest.mark.unit
class TestGitHubNormalization:
"""Tests for GitHub userinfo normalization."""
def test_github_normalize_standard(self):
"""Test GitHub userinfo normalization with standard response."""
from app.auth import _normalize_social_userinfo
raw = {
"id": 12345,
"login": "octocat",
"name": "The Octocat",
"email": "octocat@github.com",
"avatar_url": "https://avatars.githubusercontent.com/u/12345",
}
result = _normalize_social_userinfo("github", {}, raw)
assert result["sub"] == "12345"
assert result["email"] == "octocat@github.com"
assert result["name"] == "The Octocat"
assert result["preferred_username"] == "octocat"
assert result["picture"] == "https://avatars.githubusercontent.com/u/12345"
def test_github_normalize_no_name_uses_login(self):
"""Test GitHub normalization falls back to login when name is empty."""
from app.auth import _normalize_social_userinfo
raw = {"id": 12345, "login": "octocat", "name": "", "email": "octocat@github.com"}
result = _normalize_social_userinfo("github", {}, raw)
assert result["name"] == "octocat"
def test_github_normalize_missing_fields(self):
"""Test GitHub normalization handles missing fields gracefully."""
from app.auth import _normalize_social_userinfo
result = _normalize_social_userinfo("github", {}, {})
assert result["sub"] == ""
assert result["email"] == ""
assert result["name"] == ""
assert result["preferred_username"] == ""
@pytest.mark.unit
class TestSSOAutoLogin:
"""Tests for SSO Auto Login configuration."""
def test_sso_auto_login_default_false(self):
"""Test that SSO auto login defaults to False."""
from app.config import Settings
s = Settings(
_env_file=None,
auth_enabled=True,
)
assert s.sso_auto_login is False
def test_sso_auto_login_can_be_enabled(self):
"""Test that SSO auto login can be set to True."""
from app.config import Settings
s = Settings(
_env_file=None,
auth_enabled=True,
sso_auto_login=True,
)
assert s.sso_auto_login is True
@pytest.mark.unit
class TestNewConfigFields:
"""Tests for new configuration fields."""
def test_github_config_defaults(self):
"""Test GitHub social auth config defaults."""
from app.config import Settings
s = Settings(_env_file=None, auth_enabled=True)
assert s.social_auth_github_enabled is False
assert s.social_auth_github_client_id is None
assert s.social_auth_github_client_secret is None
def test_keycloak_config_defaults(self):
"""Test Keycloak social auth config defaults."""
from app.config import Settings
s = Settings(_env_file=None, auth_enabled=True)
assert s.social_auth_keycloak_enabled is False
assert s.social_auth_keycloak_client_id is None
assert s.social_auth_keycloak_server_url is None
assert s.social_auth_keycloak_realm is None
def test_generic_oauth2_config_defaults(self):
"""Test Generic OAuth2 config defaults."""
from app.config import Settings
s = Settings(_env_file=None, auth_enabled=True)
assert s.social_auth_generic_oauth2_enabled is False
assert s.social_auth_generic_oauth2_scope == "openid profile email"
assert s.social_auth_generic_oauth2_name == "OAuth2"
def test_saml2_config_defaults(self):
"""Test SAML2 config defaults."""
from app.config import Settings
s = Settings(_env_file=None, auth_enabled=True)
assert s.social_auth_saml2_enabled is False
assert s.social_auth_saml2_name == "SAML2"
def test_telegram_config_defaults(self):
"""Test Telegram config defaults."""
from app.config import Settings
s = Settings(_env_file=None, auth_enabled=True)
assert s.telegram_enabled is False
assert s.telegram_bot_token is None
assert s.telegram_chat_id is None
@pytest.mark.unit
class TestSettingsMetadata:
"""Tests that new settings have metadata entries."""
def test_github_settings_have_metadata(self):
"""Test GitHub settings are in SETTING_METADATA."""
from app.utils.settings_service import SETTING_METADATA
assert "social_auth_github_enabled" in SETTING_METADATA
assert "social_auth_github_client_id" in SETTING_METADATA
assert "social_auth_github_client_secret" in SETTING_METADATA
def test_keycloak_settings_have_metadata(self):
"""Test Keycloak settings are in SETTING_METADATA."""
from app.utils.settings_service import SETTING_METADATA
assert "social_auth_keycloak_enabled" in SETTING_METADATA
assert "social_auth_keycloak_client_id" in SETTING_METADATA
assert "social_auth_keycloak_server_url" in SETTING_METADATA
assert "social_auth_keycloak_realm" in SETTING_METADATA
def test_generic_oauth2_settings_have_metadata(self):
"""Test Generic OAuth2 settings are in SETTING_METADATA."""
from app.utils.settings_service import SETTING_METADATA
assert "social_auth_generic_oauth2_enabled" in SETTING_METADATA
assert "social_auth_generic_oauth2_authorize_url" in SETTING_METADATA
assert "social_auth_generic_oauth2_token_url" in SETTING_METADATA
def test_saml2_settings_have_metadata(self):
"""Test SAML2 settings are in SETTING_METADATA."""
from app.utils.settings_service import SETTING_METADATA
assert "social_auth_saml2_enabled" in SETTING_METADATA
assert "social_auth_saml2_sso_url" in SETTING_METADATA
assert "social_auth_saml2_entity_id" in SETTING_METADATA
def test_telegram_settings_have_metadata(self):
"""Test Telegram settings are in SETTING_METADATA."""
from app.utils.settings_service import SETTING_METADATA
assert "telegram_enabled" in SETTING_METADATA
assert "telegram_bot_token" in SETTING_METADATA
assert "telegram_chat_id" in SETTING_METADATA
def test_sso_auto_login_has_metadata(self):
"""Test SSO auto login has metadata."""
from app.utils.settings_service import SETTING_METADATA
assert "sso_auto_login" in SETTING_METADATA
meta = SETTING_METADATA["sso_auto_login"]
assert meta["category"] == "Authentication"
assert meta["type"] == "boolean"
def test_github_category_is_social_login(self):
"""Test GitHub settings are in Social Login category."""
from app.utils.settings_service import SETTING_METADATA
assert SETTING_METADATA["social_auth_github_enabled"]["category"] == "Social Login"
def test_github_secret_is_sensitive(self):
"""Test GitHub client secret is marked sensitive."""
from app.utils.settings_service import SETTING_METADATA
assert SETTING_METADATA["social_auth_github_client_secret"]["sensitive"] is True
def test_github_has_help_link(self):
"""Test GitHub has a help link to developer settings."""
from app.utils.settings_service import SETTING_METADATA
assert "help_link" in SETTING_METADATA["social_auth_github_enabled"]
@pytest.mark.unit
class TestConnectionsPageRoute:
"""Tests for the /admin/connections route."""
@pytest.mark.asyncio
async def test_connections_page_non_admin_redirected(self):
"""Test that non-admin users are redirected from connections page."""
mock_request = MagicMock()
mock_request.session = {"user": {"is_admin": False}}
mock_db = MagicMock()
# The require_admin_access decorator should handle this, so we test the decorator
from app.views.settings import require_admin_access
@require_admin_access
async def dummy_view(request):
return "success"
result = await dummy_view(mock_request)
assert result.status_code == status.HTTP_302_FOUND
@pytest.mark.asyncio
async def test_connections_page_returns_services(self):
"""Test that connections page includes expected services in context."""
from app.views.settings import connections_page
mock_request = MagicMock()
mock_request.session = {"user": {"is_admin": True}}
mock_db = MagicMock()
with (
patch("app.views.settings.get_all_settings_from_db", return_value={}),
patch("app.views.settings.templates") as mock_templates,
patch("app.views.settings.SETTING_METADATA", {}),
patch("app.views.settings.get_setting_metadata", return_value={}),
):
mock_templates.TemplateResponse.return_value = "response"
result = await connections_page(mock_request, db=mock_db)
# Check TemplateResponse was called
mock_templates.TemplateResponse.assert_called_once()
call_args = mock_templates.TemplateResponse.call_args
template_name = call_args[0][0]
context = call_args[0][1]
assert template_name == "admin_connections.html"
assert "services" in context
assert "service_settings" in context
assert "sso_auto_login" in context
# Verify expected service keys
service_keys = [s["key"] for s in context["services"]]
assert "google" in service_keys
assert "github" in service_keys
assert "keycloak" in service_keys
assert "generic_oauth2" in service_keys
assert "saml2" in service_keys
assert "smtp" in service_keys
assert "telegram" in service_keys
@pytest.mark.asyncio
async def test_connections_page_linked_status_from_db(self):
"""Linked status is derived from DB/effective settings, not SOCIAL_PROVIDERS."""
from app.views.settings import connections_page
mock_request = MagicMock()
mock_request.session = {"user": {"is_admin": True}}
mock_db = MagicMock()
# Simulate GitHub configured only in DB (not in SOCIAL_PROVIDERS yet)
db_values = {
"social_auth_github_enabled": "true",
"social_auth_github_client_id": "gh-id",
"social_auth_github_client_secret": "gh-secret",
}
with (
patch("app.views.settings.get_all_settings_from_db", return_value=db_values),
patch("app.views.settings.templates") as mock_templates,
patch("app.views.settings.SETTING_METADATA", {}),
patch("app.views.settings.get_setting_metadata", return_value={}),
):
mock_templates.TemplateResponse.return_value = "response"
await connections_page(mock_request, db=mock_db)
context = mock_templates.TemplateResponse.call_args[0][1]
services_by_key = {s["key"]: s for s in context["services"]}
# GitHub should be linked because DB values say so
assert services_by_key["github"]["linked"] is True
@pytest.mark.asyncio
async def test_connections_page_unlinked_when_credentials_missing(self):
"""Provider is unlinked when enabled=true but credentials are absent."""
from app.views.settings import connections_page
mock_request = MagicMock()
mock_request.session = {"user": {"is_admin": True}}
mock_db = MagicMock()
# enabled but no credentials
db_values = {"social_auth_github_enabled": "true"}
with (
patch("app.views.settings.get_all_settings_from_db", return_value=db_values),
patch("app.views.settings.templates") as mock_templates,
patch("app.views.settings.SETTING_METADATA", {}),
patch("app.views.settings.get_setting_metadata", return_value={}),
):
mock_templates.TemplateResponse.return_value = "response"
await connections_page(mock_request, db=mock_db)
context = mock_templates.TemplateResponse.call_args[0][1]
services_by_key = {s["key"]: s for s in context["services"]}
assert services_by_key["github"]["linked"] is False
@pytest.mark.asyncio
async def test_connections_page_oidc_linked_from_db(self):
"""OIDC linked status derives from DB effective settings."""
from app.views.settings import connections_page
mock_request = MagicMock()
mock_request.session = {"user": {"is_admin": True}}
mock_db = MagicMock()
db_values = {
"authentik_client_id": "my-client-id",
"authentik_client_secret": "my-secret",
"oauth_provider_name": "My SSO",
}
with (
patch("app.views.settings.get_all_settings_from_db", return_value=db_values),
patch("app.views.settings.templates") as mock_templates,
patch("app.views.settings.SETTING_METADATA", {}),
patch("app.views.settings.get_setting_metadata", return_value={}),
):
mock_templates.TemplateResponse.return_value = "response"
await connections_page(mock_request, db=mock_db)
context = mock_templates.TemplateResponse.call_args[0][1]
services_by_key = {s["key"]: s for s in context["services"]}
assert services_by_key["oidc"]["linked"] is True
assert services_by_key["oidc"]["name"] == "My SSO"
# oauth_configured template var should also reflect the DB state
assert context["oauth_configured"] is True
@pytest.mark.unit
class TestRefreshSocialProviders:
"""Tests for the refresh_social_providers() mechanism."""
def test_refresh_social_providers_exists(self):
"""refresh_social_providers is importable from app.auth."""
from app.auth import refresh_social_providers
assert callable(refresh_social_providers)
def test_refresh_social_providers_clears_and_repopulates(self):
"""After refresh, SOCIAL_PROVIDERS reflects current settings."""
import app.auth as auth_module
with (
patch.object(auth_module, "AUTH_ENABLED", True),
patch.object(auth_module, "settings") as mock_settings,
):
mock_settings.authentik_client_id = None
mock_settings.authentik_client_secret = None
mock_settings.social_auth_google_enabled = True
mock_settings.social_auth_google_client_id = "gid"
mock_settings.social_auth_google_client_secret = "gsecret"
mock_settings.social_auth_google_use_global_credentials = False
# All other providers disabled
for attr in (
"social_auth_microsoft_enabled",
"social_auth_apple_enabled",
"social_auth_dropbox_enabled",
"social_auth_github_enabled",
"social_auth_keycloak_enabled",
"social_auth_generic_oauth2_enabled",
):
setattr(mock_settings, attr, False)
with patch.object(auth_module, "_register_oauth_client"):
auth_module._setup_social_providers()
assert "google" in auth_module.SOCIAL_PROVIDERS
assert auth_module.OAUTH_CONFIGURED is False
def test_refresh_clears_previous_providers(self):
"""Providers removed from settings are cleared after refresh."""
import app.auth as auth_module
# Pre-populate with a stale entry
auth_module.SOCIAL_PROVIDERS["stale_provider"] = {"name": "Stale", "icon": "", "color": ""}
with (
patch.object(auth_module, "AUTH_ENABLED", True),
patch.object(auth_module, "settings") as mock_settings,
):
mock_settings.authentik_client_id = None
mock_settings.authentik_client_secret = None
for attr in (
"social_auth_google_enabled",
"social_auth_microsoft_enabled",
"social_auth_apple_enabled",
"social_auth_dropbox_enabled",
"social_auth_github_enabled",
"social_auth_keycloak_enabled",
"social_auth_generic_oauth2_enabled",
):
setattr(mock_settings, attr, False)
with patch.object(auth_module, "_register_oauth_client"):
auth_module._setup_social_providers()
assert "stale_provider" not in auth_module.SOCIAL_PROVIDERS
def test_register_oauth_client_clears_cache(self):
"""_register_oauth_client removes the cached client before re-registering."""
import app.auth as auth_module
# Inject a fake cached client
auth_module.oauth._clients["test_provider"] = object()
with patch.object(auth_module.oauth, "register"):
auth_module._register_oauth_client("test_provider", client_id="x", client_secret="y")
assert "test_provider" not in auth_module.oauth._clients
@pytest.mark.unit
class TestTranslationKeys:
"""Tests for new translation keys."""
def test_connections_translation_keys_exist(self):
"""Test that connections translation keys are in en.json."""
import json
from pathlib import Path
en_path = Path(__file__).parents[1] / "frontend" / "translations" / "en.json"
translations = json.loads(en_path.read_text())
expected_keys = [
"connections.title",
"connections.description",
"connections.configure",
"connections.linked",
"connections.unlinked",
"connections.sso_auto_login",
"connections.sso_auto_login_title",
"connections.sso_auto_login_description",
"connections.mobile_upload_title",
"connections.qr_code_enabled",
"connections.unlinked_services",
"nav.connections",
]
for key in expected_keys:
assert key in translations, f"Missing translation key: {key}"