Merge pull request #821 from christianlouis/copilot/sub-pr-816

fix(integrations): harden saved-integration test endpoint & resolve merge conflicts

.jules/sentinel.md

@@ -1,4 +1,8 @@
+## 2026-03-20 - Safe Path Traversal Prevention in Low-Level Utilities
+**Vulnerability:** The generic file utility `hash_file` in `app/utils/file_operations.py` accepted any file path and was vulnerable to reading arbitrary files via path traversal (e.g., `../../../etc/passwd`) or absolute paths if an attacker could control the `filepath` argument.
+**Learning:** Naively checking for `".." in path` breaks legitimate relative paths used internally by the application. Blocking absolute paths entirely also breaks functionality. Input validation should occur at the API boundary, but for defense-in-depth, low-level utilities must enforce expected boundaries (e.g., the application's `workdir`).
+**Prevention:** Use `pathlib.Path.resolve()` on both the target path and the allowed base directory (`settings.workdir`). Ensure the resolved target path is strictly within the allowed boundary using `filepath_obj.relative_to(workdir_obj)`, catching the `ValueError` that is raised when the path is out of bounds. This safely blocks both relative traversal attacks and arbitrary absolute paths.
 ## 2025-05-18 - [SSRF Bypass via DNS Resolution Failure]
 **Vulnerability:** The `is_private_ip` function in `app/utils/network.py` failed open (returned `False`) when a hostname could not be resolved (`socket.gaierror`).
 **Learning:** This fail-open pattern was originally added to allow external domains in tests, but in production, it created a severe SSRF risk. An attacker could bypass SSRF protections by providing a URL that fails to resolve during the security check but resolves later (DNS rebinding), or by exploiting internal routing behaviors via unresolvable addresses.
-**Prevention:** Always fail securely in network authorization functions. If a domain cannot be resolved to verify its safety, the request must be blocked (`return True` / default-deny). Tests should mock DNS resolution correctly instead of compromising production security logic.
+**Prevention:** Always fail securely in network authorization functions. If a domain cannot be resolved to verify its safety, the request must be blocked (`return True` / default-deny). Tests should mock DNS resolution correctly instead of compromising production security logic.

CHANGELOG.md

@@ -12,6 +12,35 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+### Documentation
+
+- **changelog**: Update changelog [skip ci]
+  ([`0497fbb`](https://github.com/christianlouis/DocuElevate/commit/0497fbbbad71fd728e528498508bbfc7802dab70))
+
+- **changelog**: Update changelog [skip ci]
+  ([`45d3ac8`](https://github.com/christianlouis/DocuElevate/commit/45d3ac8cf07d39d49930dd6866f76e6015067b08))
+
+### Testing
+
+- Add assertions for task enqueuing parameters
+  ([`eeae47d`](https://github.com/christianlouis/DocuElevate/commit/eeae47ddec01339421e503ba484157e798750b8a))
+
+
+## Unreleased
+
+### Documentation
+
+- **changelog**: Update changelog [skip ci]
+  ([`45d3ac8`](https://github.com/christianlouis/DocuElevate/commit/45d3ac8cf07d39d49930dd6866f76e6015067b08))
+
+### Testing
+
+- Add assertions for task enqueuing parameters
+  ([`eeae47d`](https://github.com/christianlouis/DocuElevate/commit/eeae47ddec01339421e503ba484157e798750b8a))
+
+
+## Unreleased
+
 
 ## v0.172.2 (2026-03-23)
 

SECURITY_AUDIT.md

@@ -1,6 +1,6 @@
 # Security Audit Report
 
-**Date:** 2026-02-12
+**Date:** 2026-03-23
 **Status:** Bandit Security Scan Completed - All Critical/High/Medium Issues Resolved
 
 ## Executive Summary
@@ -9,6 +9,15 @@ This document tracks security vulnerabilities found in DocuElevate and their rem
 
 ## Recent Security Fixes
 
+### Insecure API Endpoint Exposing Integration Credentials ✅ FIXED (2026-03-23)
+**Severity:** HIGH
+
+**Issue:** The endpoint `GET /api/integrations/{integration_id}/credentials` exposed integration credentials (e.g. passwords, API keys) in plaintext over the API. Although requiring login, this allowed anyone with an active user session to extract the raw credentials. The frontend used this endpoint for testing integration connections.
+
+**Remediation:**
+- Removed the `/credentials` endpoint entirely.
+- Added a new `POST /api/integrations/{integration_id}/test` endpoint that securely runs connection tests server-side without returning the decrypted credentials to the client.
+
 ### CVE-2023-36464: PyPDF2/pypdf Infinite Loop Vulnerability ✅ FIXED (2026-02-12)
 
 **Severity:** Moderate (CVSS: 5.5)

app/api/integrations.py

@@ -470,31 +470,6 @@ def delete_integration(
     logger.info("User %s deleted integration %d", owner_id, integration_id)
 
 
-@router.get("/{integration_id}/credentials", summary="Retrieve decrypted credentials for an integration")
-def get_integration_credentials(
-    integration_id: int,
-    request: Request,
-    db: DbSession,
-    owner_id: CurrentOwner,
-) -> dict[str, Any]:
-    """Return the decrypted credentials dict for a saved integration.
-
-    This endpoint is intended for internal use by background tasks that need
-    to authenticate with a third-party service.  Treat the response as
-    sensitive — it contains plaintext secrets.
-    """
-    integration = (
-        db.query(UserIntegration)
-        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
-        .first()
-    )
-    if not integration:
-        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
-
-    credentials = _decode_credentials(integration.credentials)
-    return {"credentials": credentials or {}}
-
-
 # ---------------------------------------------------------------------------
 # Connection test helpers
 # ---------------------------------------------------------------------------
@@ -662,6 +637,62 @@ _CONNECTION_TESTERS: dict[str, Any] = {
 # ---------------------------------------------------------------------------
 
 
+@router.post("/{integration_id}/test", summary="Test a saved integration connection")
+def test_saved_integration_connection(
+    integration_id: int,
+    request: Request,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Test connection for an already-saved integration."""
+    integration = (
+        db.query(UserIntegration)
+        .filter(UserIntegration.id == integration_id, UserIntegration.owner_id == owner_id)
+        .first()
+    )
+    if not integration:
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Integration not found")
+
+    tester = _CONNECTION_TESTERS.get(integration.integration_type)
+    if tester is None:
+        return {
+            "success": False,
+            "message": f"Connection testing is not yet supported for '{integration.integration_type}'. "
+            "The integration can still be saved and will be validated on first use.",
+        }
+
+    try:
+        config = json.loads(integration.config) if integration.config else {}
+    except (json.JSONDecodeError, TypeError):
+        logger.warning(
+            "Invalid JSON in integration.config for integration_id=%s, type=%s",
+            integration.id,
+            integration.integration_type,
+        )
+        return {
+            "success": False,
+            "message": (
+                "Saved integration configuration is invalid and cannot be tested. "
+                "Please edit and re-save the integration, then try again."
+            ),
+        }
+
+    credentials = _decode_credentials(integration.credentials) or {}
+
+    try:
+        return tester(config, credentials)
+    except Exception:
+        logger.exception(
+            "Unexpected error testing integration_id=%s, type=%s",
+            integration.id,
+            integration.integration_type,
+        )
+        return {
+            "success": False,
+            "message": "An unexpected error occurred while testing the connection. Please check your configuration.",
+        }
+
+
 @router.post("/test", summary="Test an integration connection without saving")
 def test_integration_connection(
     request: Request,

app/tasks/upload_to_nextcloud.py

@@ -1,141 +1,157 @@
-#!/usr/bin/env python3
-
-import logging
-import os
-
-import requests
-from requests.auth import HTTPBasicAuth
-
-from app.celery_app import celery
-from app.config import settings
-from app.tasks.retry_config import UploadTaskWithRetry
-from app.utils import log_task_progress
-from app.utils.filename_utils import extract_remote_path, get_unique_filename
-from app.utils.network import join_url
-
-logger = logging.getLogger(__name__)
-
-
-@celery.task(base=UploadTaskWithRetry, bind=True)
-def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
-    """
-    Upload a file to Nextcloud WebDAV.
-
-    Args:
-        file_path: Path to the file to upload
-        file_id: Optional file ID to associate with logs
-    """
-    task_id = self.request.id
-    logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
-    log_task_progress(
-        task_id,
-        "upload_to_nextcloud",
-        "in_progress",
-        f"Uploading to Nextcloud: {os.path.basename(file_path)}",
-        file_id=file_id,
-    )
-
-    if not os.path.exists(file_path):
-        error_msg = f"File not found: {file_path}"
-        logger.error(f"[{task_id}] {error_msg}")
-        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
-        raise FileNotFoundError(error_msg)
-
-    # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
-    # This is what's shown in your env view
-    if not (
-        getattr(settings, "nextcloud_upload_url", None)
-        and getattr(settings, "nextcloud_username", None)
-        and getattr(settings, "nextcloud_password", None)
-    ):
-        logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
-        log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
-        return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
-
-    filename = os.path.basename(file_path)
-
-    try:
-        # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
-        webdav_url = settings.nextcloud_upload_url
-        if not webdav_url.endswith("/"):
-            webdav_url += "/"
-
-        # Calculate remote path based on local file structure
-        remote_base = (
-            folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
-        )
-        remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
-        full_url = join_url(webdav_url, remote_path)
-
-        # Function to check if file exists in Nextcloud
-        def check_exists_in_nextcloud(path):
-            check_url = join_url(webdav_url, os.path.dirname(path))
-            try:
-                response = requests.request(
-                    "PROPFIND",
-                    check_url,
-                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
-                    headers={"Depth": "1"},
-                    timeout=10,
-                )
-
-                return path in response.text
-            except Exception:
-                # If we can't check, assume it doesn't exist
-                return False
-
-        # Check for potential file collision and get a unique name if needed
-        remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
-        full_url = join_url(webdav_url, remote_path)
-
-        # Create necessary parent folders
-        parent_dirs = os.path.dirname(remote_path)
-        if parent_dirs:
-            current_path = ""
-            for folder in parent_dirs.split("/"):
-                if not folder:
-                    continue
-                current_path += f"{folder}/"
-                mkdir_url = join_url(webdav_url, current_path)
-
-                requests.request(
-                    "MKCOL",
-                    mkdir_url,
-                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
-                    timeout=10,
-                )
-
-        # Upload the file
-        logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
-        log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
-        with open(file_path, "rb") as file_data:
-            response = requests.put(
-                full_url,
-                data=file_data,
-                auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
-                headers={"Content-Type": "application/octet-stream"},
-                timeout=settings.http_request_timeout,  # Use configured timeout for large files
-            )
-
-        if response.status_code in (201, 204):  # Created or No Content
-            logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
-            log_task_progress(
-                task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
-            )
-            return {
-                "status": "Completed",
-                "file_path": file_path,
-                "nextcloud_path": remote_path,
-                "response_code": response.status_code,
-            }
-        else:
-            error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
-            logger.error(f"[{task_id}] {error_msg}")
-            log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
-            raise Exception(error_msg)
-
-    except Exception as e:
-        error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
-        logger.error(f"[{task_id}] {error_msg}")
-        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
-        raise Exception(error_msg)
+#!/usr/bin/env python3
+
+import logging
+import os
+
+import requests
+from requests.auth import HTTPBasicAuth
+
+from app.celery_app import celery
+from app.config import settings
+from app.tasks.retry_config import UploadTaskWithRetry
+from app.utils import log_task_progress
+from app.utils.filename_utils import extract_remote_path, get_unique_filename
+
+logger = logging.getLogger(__name__)
+
+
+@celery.task(base=UploadTaskWithRetry, bind=True)
+def upload_to_nextcloud(self, file_path: str, file_id: int = None, folder_override: str = None):
+    """
+    Upload a file to Nextcloud WebDAV.
+
+    Args:
+        file_path: Path to the file to upload
+        file_id: Optional file ID to associate with logs
+    """
+    task_id = self.request.id
+    logger.info(f"[{task_id}] Starting Nextcloud upload: {file_path}")
+    log_task_progress(
+        task_id,
+        "upload_to_nextcloud",
+        "in_progress",
+        f"Uploading to Nextcloud: {os.path.basename(file_path)}",
+        file_id=file_id,
+    )
+
+    if not os.path.exists(file_path):
+        error_msg = f"File not found: {file_path}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
+        raise FileNotFoundError(error_msg)
+
+    # For Nextcloud, we need to check for 'nextcloud_upload_url' instead of 'nextcloud_url'
+    # This is what's shown in your env view
+    if not (
+        getattr(settings, "nextcloud_upload_url", None)
+        and getattr(settings, "nextcloud_username", None)
+        and getattr(settings, "nextcloud_password", None)
+    ):
+        logger.info(f"[{task_id}] Nextcloud upload skipped: Missing configuration")
+        log_task_progress(task_id, "upload_to_nextcloud", "success", "Skipped: Not configured", file_id=file_id)
+        return {"status": "Skipped", "reason": "Nextcloud settings not configured"}
+
+    filename = os.path.basename(file_path)
+
+    try:
+        # Prepare WebDAV URL - use nextcloud_upload_url instead of nextcloud_url
+        webdav_url = settings.nextcloud_upload_url
+        if not webdav_url.endswith("/"):
+            webdav_url += "/"
+
+        # Calculate remote path based on local file structure
+        remote_base = (
+            folder_override if folder_override is not None else (getattr(settings, "nextcloud_folder", "") or "")
+        )
+        remote_path = extract_remote_path(file_path, settings.workdir, remote_base)
+        full_url = f"{webdav_url}/{remote_path}"
+
+        # Remove any double slashes (except in http://)
+        full_url = full_url.replace("://", "$PLACEHOLDER$")
+        while "//" in full_url:
+            full_url = full_url.replace("//", "/")
+        full_url = full_url.replace("$PLACEHOLDER$", "://")
+
+        # Function to check if file exists in Nextcloud
+        def check_exists_in_nextcloud(path):
+            check_url = f"{webdav_url}{os.path.dirname(path)}"
+            try:
+                response = requests.request(
+                    "PROPFIND",
+                    check_url,
+                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
+                    headers={"Depth": "1"},
+                    timeout=10,
+                )
+
+                return path in response.text
+            except Exception:
+                # If we can't check, assume it doesn't exist
+                return False
+
+        # Check for potential file collision and get a unique name if needed
+        remote_path = get_unique_filename(remote_path, check_exists_in_nextcloud)
+        full_url = f"{webdav_url}/{remote_path}"
+
+        # Fix double slashes again
+        full_url = full_url.replace("://", "$PLACEHOLDER$")
+        while "//" in full_url:
+            full_url = full_url.replace("//", "/")
+        full_url = full_url.replace("$PLACEHOLDER$", "://")
+
+        # Create necessary parent folders
+        parent_dirs = os.path.dirname(remote_path)
+        if parent_dirs:
+            current_path = ""
+            for folder in parent_dirs.split("/"):
+                if not folder:
+                    continue
+                current_path += f"{folder}/"
+                mkdir_url = f"{webdav_url}/{current_path}"
+                # Fix double slashes
+                mkdir_url = mkdir_url.replace("://", "$PLACEHOLDER$")
+                while "//" in mkdir_url:
+                    mkdir_url = mkdir_url.replace("//", "/")
+                mkdir_url = mkdir_url.replace("$PLACEHOLDER$", "://")
+
+                requests.request(
+                    "MKCOL",
+                    mkdir_url,
+                    auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
+                    timeout=10,
+                )
+
+        # Upload the file
+        logger.info(f"[{task_id}] Uploading {filename} to Nextcloud at {full_url}")
+        log_task_progress(task_id, "upload_file", "in_progress", f"Uploading to {remote_path}", file_id=file_id)
+        with open(file_path, "rb") as file_data:
+            response = requests.put(
+                full_url,
+                data=file_data,
+                auth=HTTPBasicAuth(settings.nextcloud_username, settings.nextcloud_password),
+                headers={"Content-Type": "application/octet-stream"},
+                timeout=settings.http_request_timeout,  # Use configured timeout for large files
+            )
+
+        if response.status_code in (201, 204):  # Created or No Content
+            logger.info(f"[{task_id}] Successfully uploaded {filename} to Nextcloud at {remote_path}")
+            log_task_progress(
+                task_id, "upload_to_nextcloud", "success", f"Uploaded to Nextcloud: {remote_path}", file_id=file_id
+            )
+            return {
+                "status": "Completed",
+                "file_path": file_path,
+                "nextcloud_path": remote_path,
+                "response_code": response.status_code,
+            }
+        else:
+            error_msg = f"Failed to upload {filename} to Nextcloud: {response.status_code} - {response.text}"
+            logger.error(f"[{task_id}] {error_msg}")
+            log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
+            raise Exception(error_msg)
+
+    except Exception as e:
+        error_msg = f"Failed to upload {filename} to Nextcloud: {str(e)}"
+        logger.error(f"[{task_id}] {error_msg}")
+        log_task_progress(task_id, "upload_to_nextcloud", "failure", error_msg, file_id=file_id)
+        raise Exception(error_msg)

app/utils/file_operations.py

@@ -7,8 +7,19 @@ def hash_file(filepath: str | Path, chunk_size: int = 65536) -> str:
     Returns the SHA-256 hash of the file at 'filepath'.
     Reads the file in chunks to handle large files efficiently.
     """
+    from app.config import settings
+
+    filepath_obj = Path(filepath).resolve()
+    workdir_obj = Path(settings.workdir).resolve()
+
+    # Security check: Ensure the resolved path is strictly within the allowed workdir
+    try:
+        filepath_obj.relative_to(workdir_obj)
+    except ValueError:
+        raise FileNotFoundError(f"Access denied: path traversal attempt or file outside workdir '{filepath}'")
+
     sha256 = hashlib.sha256()
-    with open(filepath, "rb") as f:
+    with open(filepath_obj, "rb") as f:
         while True:
             data = f.read(chunk_size)
             if not data:

app/utils/network.py

@@ -32,16 +32,3 @@ def is_private_ip(hostname: str) -> bool:
             # and SSRF bypasses via unresolvable addresses.
             logger.warning(f"Could not resolve hostname (blocking securely): {hostname}")
             return True
-
-
-def join_url(base: str, *parts: str) -> str:
-    """
-    Safely join a base URL and multiple path parts.
-    Handles double slashes while preserving the protocol '://'.
-    """
-    url = "/".join([base, *parts])
-    url = url.replace("://", "$PLACEHOLDER$")
-    while "//" in url:
-        url = url.replace("//", "/")
-    url = url.replace("$PLACEHOLDER$", "://")
-    return url

frontend/templates/integrations_dashboard.html

@@ -1379,27 +1379,15 @@ function integrationsDashboard() {
     async testSavedIntegration(intg) {
       this.testingId = intg.id;
       try {
-        // Retrieve saved credentials to test
-        const credsResp = await fetch(`/api/integrations/${intg.id}/credentials`);
-        if (!credsResp.ok) {
-          this.showAlert('error', 'Test Failed', 'Could not retrieve saved credentials for testing.');
-          return;
-        }
-        const creds = await credsResp.json();
-        const resp = await fetch('/api/integrations/test', {
+        const resp = await fetch(`/api/integrations/${intg.id}/test`, {
           method: 'POST',
           headers: authHeaders(true),
-          body: JSON.stringify({
-            integration_type: intg.integration_type,
-            config: intg.config,
-            credentials: creds.credentials,
-          }),
         });
         const data = await resp.json();
-        if (data.success) {
+        if (resp.ok && data.success) {
           this.showAlert('success', `${intg.name}: Connection OK`, data.message);
         } else {
-          this.showAlert('error', `${intg.name}: Connection Failed`, data.message);
+          this.showAlert('error', `${intg.name}: Connection Failed`, data.message || {{ _("integrations.connection_test_failed_fallback")|tojson }});
         }
       } catch (err) {
         this.showAlert('error', 'Test Failed', `Network error: ${err.message || 'Unknown error'}`);

frontend/translations/en.json

@@ -1107,6 +1107,7 @@
   "integrations.connected": "Connected",
   "integrations.connection_failed": "Connection failed",
   "integrations.connection_success": "Connection successful",
+  "integrations.connection_test_failed_fallback": "Connection test failed.",
   "integrations.delete_confirm_are_you_sure": "Are you sure you want to delete",
   "integrations.delete_confirm_title": "Delete Integration?",
   "integrations.delete_confirm_undone": "This action cannot be undone.",

tests/test_api_integrations.py

@@ -1,6 +1,7 @@
 """Tests for the per-user integrations API (app/api/integrations.py)."""
 
 import unittest.mock
+from unittest.mock import MagicMock, patch
 
 import pytest
 from fastapi.testclient import TestClient
@@ -405,32 +406,33 @@ class TestDeleteIntegration:
 
 
 @pytest.mark.integration
-class TestGetIntegrationCredentials:
-    """Tests for GET /api/integrations/{id}/credentials."""
+class TestTestSavedIntegrationConnection:
+    """Tests for POST /api/integrations/{id}/test."""
+
+    @patch("app.api.integrations._CONNECTION_TESTERS")
+    def test_test_saved_integration_success(self, mock_testers, int_client):
+        """Test a saved integration successfully."""
+        mock_tester = MagicMock(return_value={"success": True, "message": "OK"})
+        mock_testers.get.return_value = mock_tester
 
-    def test_returns_decrypted_credentials(self, int_client):
-        """Credentials endpoint returns the decrypted dict."""
         created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
-        resp = int_client.get(f"/api/integrations/{created['id']}/credentials")
-        assert resp.status_code == 200
-        creds = resp.json()["credentials"]
-        assert creds["password"] == "s3cr3t"  # noqa: S105
+        resp = int_client.post(f"/api/integrations/{created['id']}/test")
 
-    def test_returns_empty_dict_when_no_credentials(self, int_client):
-        """No credentials stored returns empty dict."""
-        payload = dict(_IMAP_SOURCE, credentials=None)
-        created = int_client.post("/api/integrations/", json=payload).json()
-        resp = int_client.get(f"/api/integrations/{created['id']}/credentials")
         assert resp.status_code == 200
-        assert resp.json()["credentials"] == {}
+        assert resp.json()["success"] is True
+        mock_testers.get.assert_called_with("IMAP")
+        mock_tester.assert_called_once()
+        args = mock_tester.call_args[0]
+        assert args[0]["host"] == "imap.gmail.com"
+        assert args[1]["password"] == "s3cr3t"
 
     def test_not_found(self, int_client):
         """Non-existent integration returns 404."""
-        resp = int_client.get("/api/integrations/9999/credentials")
+        resp = int_client.post("/api/integrations/9999/test")
         assert resp.status_code == 404
 
-    def test_other_users_credentials_returns_404(self, int_client, int_session):
-        """Cannot retrieve another user's credentials."""
+    def test_other_users_integration_returns_404(self, int_client, int_session):
+        """Cannot test another user's integration."""
         other_integration = UserIntegration(
             owner_id=_OTHER_OWNER,
             direction="SOURCE",
@@ -441,9 +443,47 @@ class TestGetIntegrationCredentials:
         )
         int_session.add(other_integration)
         int_session.commit()
-        resp = int_client.get(f"/api/integrations/{other_integration.id}/credentials")
+        resp = int_client.post(f"/api/integrations/{other_integration.id}/test")
         assert resp.status_code == 404
 
+    @patch("app.api.integrations._CONNECTION_TESTERS")
+    def test_invalid_json_config_returns_failure(self, mock_testers, int_client, int_session):
+        """Invalid JSON in config returns a controlled failure response."""
+        mock_tester = MagicMock()
+        mock_testers.get.return_value = mock_tester
+
+        bad_integration = UserIntegration(
+            owner_id=_OWNER,
+            direction="SOURCE",
+            integration_type="IMAP",
+            name="Bad Config",
+            config="not-valid-json{{{",
+            credentials="{}",
+            is_active=True,
+        )
+        int_session.add(bad_integration)
+        int_session.commit()
+
+        resp = int_client.post(f"/api/integrations/{bad_integration.id}/test")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "invalid" in data["message"].lower()
+        mock_tester.assert_not_called()
+
+    @patch("app.api.integrations._CONNECTION_TESTERS")
+    def test_tester_raises_exception_returns_failure(self, mock_testers, int_client):
+        """Tester that raises an exception returns a controlled failure response."""
+        mock_testers.get.return_value = MagicMock(side_effect=ValueError("bad port"))
+
+        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
+        resp = int_client.post(f"/api/integrations/{created['id']}/test")
+
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "unexpected error" in data["message"].lower()
+
 
 @pytest.mark.unit
 class TestIntegrationModel:

tests/test_upload_to_nextcloud_join_url.py

@@ -1,54 +0,0 @@
-import os
-from unittest.mock import MagicMock, patch
-
-import pytest
-
-from app.tasks.upload_to_nextcloud import upload_to_nextcloud
-
-
-@pytest.fixture
-def mock_settings():
-    with patch("app.tasks.upload_to_nextcloud.settings") as mock:
-        mock.nextcloud_upload_url = "http://nextcloud.local/"
-        mock.nextcloud_username = "testuser"
-        mock.nextcloud_password = "testpassword"
-        mock.nextcloud_folder = "uploads"
-        mock.workdir = "/tmp/workdir"
-        mock.http_request_timeout = 30
-        yield mock
-
-
-@pytest.fixture
-def mock_requests():
-    with patch("app.tasks.upload_to_nextcloud.requests") as mock:
-        # Mock PROPFIND to always return false (file doesn't exist)
-        mock.request.return_value = MagicMock(text="<response></response>")
-
-        # Mock PUT to return success
-        put_response = MagicMock()
-        put_response.status_code = 201
-        mock.put.return_value = put_response
-        yield mock
-
-
-def test_upload_to_nextcloud_url_construction(mock_settings, mock_requests):
-    file_path = "/tmp/workdir/test_file.txt"
-
-    # Create dummy file
-    os.makedirs("/tmp/workdir", exist_ok=True)
-    with open(file_path, "w") as f:
-        f.write("test content")
-
-    # Call the task directly
-    with patch("celery.app.task.Task.request", new_callable=MagicMock) as mock_req:
-        mock_req.id = "test-task-123"
-        result = upload_to_nextcloud(file_path)
-
-    assert result["status"] == "Completed"
-    assert result["nextcloud_path"] == "uploads/test_file.txt"
-
-    # Verify requests.put was called with the correct URL
-    mock_requests.put.assert_called_once()
-    args, kwargs = mock_requests.put.call_args
-    url = args[0]
-    assert url == "http://nextcloud.local/uploads/test_file.txt"