From 470f08d89322f2904b78a8b0f820973611486c26 Mon Sep 17 00:00:00 2001
From: "google-labs-jules[bot]"
 <161369871+google-labs-jules[bot]@users.noreply.github.com>
Date: Thu, 26 Mar 2026 04:18:24 +0000
Subject: [PATCH] test: add tests for SSRF validation in integrations

Adds missing unit tests for `_test_imap_connection` and `_test_s3_connection` to cover the new `is_private_ip()` SSRF blocking logic and satisfy Codecov checks.

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>
---
 tests/test_api_integrations.py | 30 ++++++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/tests/test_api_integrations.py b/tests/test_api_integrations.py
index 958a1a86..561ac177 100644
--- a/tests/test_api_integrations.py
+++ b/tests/test_api_integrations.py
@@ -998,6 +998,23 @@ class TestConnectionTestEndpoint:
         assert data["success"] is False
         assert "Missing" in data["message"]
 
+    def test_test_imap_blocks_private_ip(self, int_client):
+        """IMAP test with private IP returns failure (SSRF protection)."""
+        payload = {
+            "integration_type": "IMAP",
+            "config": {
+                "host": "127.0.0.1",
+                "port": 993,
+                "username": "user",
+            },
+            "credentials": {"password": "pass"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "Invalid hostname or IP address" in data["message"]
+
     def test_test_s3_missing_bucket(self, int_client):
         """S3 test with missing bucket returns failure."""
         payload = {
@@ -1011,6 +1028,19 @@ class TestConnectionTestEndpoint:
         assert data["success"] is False
         assert "bucket" in data["message"].lower()
 
+    def test_test_s3_blocks_private_ip(self, int_client):
+        """S3 test with private IP endpoint returns failure (SSRF protection)."""
+        payload = {
+            "integration_type": "S3",
+            "config": {"bucket": "my-bucket", "endpoint_url": "http://127.0.0.1:9000"},
+            "credentials": {"access_key_id": "AKIA", "secret_access_key": "secret"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "Invalid endpoint URL or private IP" in data["message"]
+
     def test_test_webdav_missing_url(self, int_client):
         """WebDAV test with missing URL returns failure."""
         payload = {