feat(integrations): add quota enforcement, connection test, and quota endpoint

Co-authored-by: christianlouis <361235+christianlouis@users.noreply.github.com>

app/api/integrations.py

@@ -7,6 +7,15 @@ storage destination (e.g. S3, Dropbox, Google Drive) configured by a user.
 Sensitive credentials are encrypted at rest using Fernet symmetric encryption
 (keyed from ``SESSION_SECRET``) via :mod:`app.utils.encryption`.  Credential
 values are **never** returned in API responses.
+
+Subscription quota enforcement
+------------------------------
+On creation, the endpoint checks the user's subscription tier limits:
+
+* **Destinations** — ``max_storage_destinations`` from the plan.
+* **Sources (IMAP)** — ``max_mailboxes`` from the plan.
+
+Exceeding the quota returns HTTP 403 with an actionable error message.
 """
 
 import json
@@ -20,6 +29,7 @@ from sqlalchemy.orm import Session
 from app.database import get_db
 from app.models import IntegrationDirection, IntegrationType, UserIntegration
 from app.utils.encryption import decrypt_value, encrypt_value
+from app.utils.subscription import get_tier, get_user_tier_id
 from app.utils.user_scope import get_current_owner_id
 
 logger = logging.getLogger(__name__)
@@ -42,6 +52,119 @@ def _get_owner_id(request: Request) -> str:
 
 CurrentOwner = Annotated[str, Depends(_get_owner_id)]
 
+# ---------------------------------------------------------------------------
+# Quota helpers
+# ---------------------------------------------------------------------------
+
+_FREE_TIER_ID = "free"
+
+# Source types that consume the mailbox quota
+_MAILBOX_SOURCE_TYPES = {IntegrationType.IMAP}
+
+
+def _get_max_destinations(tier: dict[str, Any]) -> int | None:
+    """Return the maximum number of storage destinations allowed by *tier*.
+
+    Returns:
+        ``None``  — unlimited (paid tiers with ``max_storage_destinations == 0``)
+        positive  — the configured limit
+    """
+    tier_id: str = tier.get("id", _FREE_TIER_ID)
+    max_dest: int = tier.get("max_storage_destinations", 0)
+
+    # Free tier: the value itself is the limit (e.g. 1)
+    if tier_id == _FREE_TIER_ID:
+        return max_dest if max_dest > 0 else 1  # safe default
+
+    # Paid tiers: 0 means unlimited
+    if max_dest == 0:
+        return None
+
+    return max_dest
+
+
+def _get_max_sources(tier: dict[str, Any]) -> int | None:
+    """Return the maximum number of IMAP source integrations allowed by *tier*.
+
+    Returns:
+        ``None``  — unlimited (paid tiers with ``max_mailboxes == 0``)
+        ``0``     — no mailboxes allowed (free tier)
+        positive  — the configured limit
+    """
+    tier_id: str = tier.get("id", _FREE_TIER_ID)
+    max_mb: int = tier.get("max_mailboxes", 0)
+
+    # Free tier: 0 means "no access" (not "unlimited")
+    if tier_id == _FREE_TIER_ID:
+        return 0
+
+    # Paid tiers: 0 means unlimited
+    if max_mb == 0:
+        return None
+
+    return max_mb
+
+
+def _check_quota(db: Session, owner_id: str, direction: str, integration_type: str) -> None:
+    """Raise 403 if the user has reached their integration quota.
+
+    Quota rules:
+    * DESTINATION integrations are limited by ``max_storage_destinations``.
+    * SOURCE integrations of type IMAP are limited by ``max_mailboxes``.
+    * Other SOURCE types (WATCH_FOLDER, WEBHOOK) are not quota-limited yet.
+    """
+    tier_id = get_user_tier_id(db, owner_id)
+    tier = get_tier(tier_id, db)
+
+    if direction == IntegrationDirection.DESTINATION:
+        max_dest = _get_max_destinations(tier)
+        if max_dest is not None:
+            current_count = (
+                db.query(UserIntegration)
+                .filter(
+                    UserIntegration.owner_id == owner_id,
+                    UserIntegration.direction == IntegrationDirection.DESTINATION,
+                )
+                .count()
+            )
+            if current_count >= max_dest:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail=(
+                        f"You have reached your plan limit of {max_dest} storage destination(s). "
+                        "Please remove an existing destination or upgrade your plan."
+                    ),
+                )
+
+    elif direction == IntegrationDirection.SOURCE and integration_type in _MAILBOX_SOURCE_TYPES:
+        max_src = _get_max_sources(tier)
+
+        if max_src == 0:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Your current plan does not include email ingestion. Upgrade to a paid plan to add IMAP sources.",
+            )
+
+        if max_src is not None:
+            current_count = (
+                db.query(UserIntegration)
+                .filter(
+                    UserIntegration.owner_id == owner_id,
+                    UserIntegration.direction == IntegrationDirection.SOURCE,
+                    UserIntegration.integration_type.in_(list(_MAILBOX_SOURCE_TYPES)),
+                )
+                .count()
+            )
+            if current_count >= max_src:
+                raise HTTPException(
+                    status_code=status.HTTP_403_FORBIDDEN,
+                    detail=(
+                        f"You have reached your plan limit of {max_src} IMAP source(s). "
+                        "Please remove an existing source or upgrade your plan."
+                    ),
+                )
+
+
 # ---------------------------------------------------------------------------
 # Pydantic schemas
 # ---------------------------------------------------------------------------
@@ -72,6 +195,14 @@ class IntegrationUpdate(BaseModel):
     is_active: bool | None = None
 
 
+class IntegrationTestRequest(BaseModel):
+    """Schema for testing an integration connection without saving it."""
+
+    integration_type: str = Field(..., description="Integration type (e.g. 'IMAP', 'S3', 'DROPBOX')")
+    config: dict[str, Any] | None = Field(default=None, description="Non-sensitive configuration")
+    credentials: dict[str, Any] | None = Field(default=None, description="Credentials for the connection test")
+
+
 # ---------------------------------------------------------------------------
 # Validation helpers
 # ---------------------------------------------------------------------------
@@ -199,10 +330,15 @@ def create_integration(
 
     ``credentials`` are encrypted at rest using Fernet symmetric encryption
     before being persisted and are **never** returned in API responses.
+
+    Quota is enforced against the user's subscription plan before the
+    integration is persisted.
     """
     _validate_direction(body.direction)
     _validate_integration_type(body.integration_type)
 
+    _check_quota(db, owner_id, body.direction, body.integration_type)
+
     integration = UserIntegration(
         owner_id=owner_id,
         direction=body.direction,
@@ -342,3 +478,193 @@ def get_integration_credentials(
 
     credentials = _decode_credentials(integration.credentials)
     return {"credentials": credentials or {}}
+
+
+# ---------------------------------------------------------------------------
+# Connection test helpers
+# ---------------------------------------------------------------------------
+
+
+def _test_imap_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
+    """Test an IMAP connection using the provided config and credentials."""
+    import imaplib
+
+    cfg = config or {}
+    creds = credentials or {}
+    host = cfg.get("host", "")
+    port = int(cfg.get("port", 993))
+    username = cfg.get("username", "")
+    password = creds.get("password", "")
+    use_ssl = cfg.get("use_ssl", True)
+
+    if not host or not username or not password:
+        return {"success": False, "message": "Missing required fields: host, username, and password"}
+
+    try:
+        if use_ssl:
+            mail = imaplib.IMAP4_SSL(host, port)
+        else:
+            mail = imaplib.IMAP4(host, port)
+        mail.login(username, password)
+        mail.logout()
+        return {"success": True, "message": "IMAP connection successful"}
+    except OSError as exc:
+        return {"success": False, "message": f"Connection error: {exc}"}
+    except Exception as exc:  # noqa: BLE001
+        return {"success": False, "message": f"IMAP error: {exc}"}
+
+
+def _test_s3_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
+    """Test an S3 connection by calling HeadBucket."""
+    try:
+        import boto3
+        from botocore.exceptions import BotoCoreError, ClientError
+    except ImportError:
+        return {"success": False, "message": "boto3 is not installed"}
+
+    cfg = config or {}
+    creds = credentials or {}
+    bucket = cfg.get("bucket", "")
+    region = cfg.get("region", "us-east-1")
+
+    if not bucket:
+        return {"success": False, "message": "Missing required field: bucket"}
+
+    try:
+        client = boto3.client(
+            "s3",
+            region_name=region,
+            aws_access_key_id=creds.get("access_key_id", ""),
+            aws_secret_access_key=creds.get("secret_access_key", ""),
+            endpoint_url=cfg.get("endpoint_url"),
+        )
+        client.head_bucket(Bucket=bucket)
+        return {"success": True, "message": f"S3 bucket '{bucket}' is accessible"}
+    except (BotoCoreError, ClientError) as exc:
+        return {"success": False, "message": f"S3 error: {exc}"}
+    except Exception as exc:  # noqa: BLE001
+        return {"success": False, "message": f"Unexpected error: {exc}"}
+
+
+def _test_webdav_connection(config: dict[str, Any] | None, credentials: dict[str, Any] | None) -> dict[str, Any]:
+    """Test a WebDAV/Nextcloud connection by issuing an HTTP PROPFIND."""
+    import urllib.request
+
+    cfg = config or {}
+    creds = credentials or {}
+    url = cfg.get("url", "")
+    username = creds.get("username", "")
+    password = creds.get("password", "")
+
+    if not url:
+        return {"success": False, "message": "Missing required field: url"}
+
+    # Only allow http/https to prevent file:// or other custom scheme attacks
+    from urllib.parse import urlparse
+
+    parsed = urlparse(url)
+    if parsed.scheme not in ("http", "https"):
+        return {"success": False, "message": "URL must use http or https scheme"}
+
+    try:
+        import base64
+
+        req = urllib.request.Request(url, method="PROPFIND")  # noqa: S310
+        if username and password:
+            token = base64.b64encode(f"{username}:{password}".encode()).decode()
+            req.add_header("Authorization", f"Basic {token}")
+        req.add_header("Depth", "0")
+        with urllib.request.urlopen(req, timeout=10) as resp:  # noqa: S310
+            if resp.status < 400:
+                return {"success": True, "message": "WebDAV connection successful"}
+            return {"success": False, "message": f"WebDAV returned HTTP {resp.status}"}
+    except Exception as exc:  # noqa: BLE001
+        return {"success": False, "message": f"WebDAV error: {exc}"}
+
+
+_CONNECTION_TESTERS: dict[str, Any] = {
+    IntegrationType.IMAP: _test_imap_connection,
+    IntegrationType.S3: _test_s3_connection,
+    IntegrationType.WEBDAV: _test_webdav_connection,
+    IntegrationType.NEXTCLOUD: _test_webdav_connection,
+}
+
+
+# ---------------------------------------------------------------------------
+# Test & quota endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.post("/test", summary="Test an integration connection without saving")
+def test_integration_connection(
+    request: Request,
+    body: IntegrationTestRequest,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Test integration credentials without persisting anything.
+
+    Useful for the "Test connection" button in the UI before the user saves
+    a new integration.  Returns ``{"success": bool, "message": str}``.
+    """
+    _validate_integration_type(body.integration_type)
+
+    tester = _CONNECTION_TESTERS.get(body.integration_type)
+    if tester is None:
+        return {
+            "success": False,
+            "message": f"Connection testing is not yet supported for '{body.integration_type}'. "
+            "The integration can still be saved and will be validated on first use.",
+        }
+
+    return tester(body.config, body.credentials)
+
+
+@router.get("/quota/", summary="Get integration quota information for the current user")
+def get_integration_quota(
+    request: Request,
+    db: DbSession,
+    owner_id: CurrentOwner,
+) -> dict[str, Any]:
+    """Return the user's current integration usage vs. their plan quota.
+
+    Includes separate counts for destinations and IMAP sources.
+    """
+    tier_id = get_user_tier_id(db, owner_id)
+    tier = get_tier(tier_id, db)
+
+    max_dest = _get_max_destinations(tier)
+    max_src = _get_max_sources(tier)
+
+    dest_count = (
+        db.query(UserIntegration)
+        .filter(
+            UserIntegration.owner_id == owner_id,
+            UserIntegration.direction == IntegrationDirection.DESTINATION,
+        )
+        .count()
+    )
+
+    src_count = (
+        db.query(UserIntegration)
+        .filter(
+            UserIntegration.owner_id == owner_id,
+            UserIntegration.direction == IntegrationDirection.SOURCE,
+            UserIntegration.integration_type.in_(list(_MAILBOX_SOURCE_TYPES)),
+        )
+        .count()
+    )
+
+    return {
+        "tier_id": tier_id,
+        "tier_name": tier.get("name", tier_id),
+        "destinations": {
+            "current_count": dest_count,
+            "max_allowed": max_dest,
+            "can_add": max_dest is None or dest_count < max_dest,
+        },
+        "sources": {
+            "current_count": src_count,
+            "max_allowed": max_src,
+            "can_add": max_src is None or (max_src > 0 and src_count < max_src),
+        },
+    }

tests/test_api_integrations.py

@@ -7,7 +7,7 @@ from sqlalchemy.orm import sessionmaker
 from sqlalchemy.pool import StaticPool
 
 from app.database import Base, get_db
-from app.models import UserIntegration
+from app.models import SubscriptionPlan, UserIntegration, UserProfile
 
 # ---------------------------------------------------------------------------
 # Test data constants
@@ -43,6 +43,37 @@ _S3_DESTINATION = {
 # ---------------------------------------------------------------------------
 
 
+def _make_profile(session, owner: str = _OWNER, tier: str = "business") -> UserProfile:
+    """Create a UserProfile row for the given user and tier."""
+    profile = UserProfile(user_id=owner, subscription_tier=tier)
+    session.add(profile)
+    session.commit()
+    session.refresh(profile)
+    return profile
+
+
+def _make_plan(
+    session,
+    tier: str = "business",
+    max_storage_destinations: int = 10,
+    max_mailboxes: int = 0,
+) -> SubscriptionPlan:
+    """Create a SubscriptionPlan row."""
+    plan = SubscriptionPlan(
+        plan_id=tier,
+        name=tier.title(),
+        price_monthly=7.99,
+        price_yearly=76.99,
+        max_storage_destinations=max_storage_destinations,
+        max_mailboxes=max_mailboxes,
+        is_active=True,
+    )
+    session.add(plan)
+    session.commit()
+    session.refresh(plan)
+    return plan
+
+
 @pytest.fixture()
 def int_engine():
     """In-memory SQLite engine for integration tests."""
@@ -65,11 +96,41 @@ def int_session(int_engine):
     session.close()
 
 
+def _seed_default_plan(engine, owner_id: str = _OWNER) -> None:
+    """Seed a generous (business-tier) plan and profile for the given user.
+
+    Called automatically by ``_make_client`` so that existing CRUD tests keep
+    working after quota enforcement was added.
+    """
+    Session = sessionmaker(bind=engine)
+    session = Session()
+    try:
+        if not session.query(SubscriptionPlan).filter(SubscriptionPlan.plan_id == "business").first():
+            session.add(
+                SubscriptionPlan(
+                    plan_id="business",
+                    name="Power",
+                    price_monthly=7.99,
+                    price_yearly=76.99,
+                    max_storage_destinations=10,
+                    max_mailboxes=0,  # 0 = unlimited for paid tiers
+                    is_active=True,
+                )
+            )
+        if not session.query(UserProfile).filter(UserProfile.user_id == owner_id).first():
+            session.add(UserProfile(user_id=owner_id, subscription_tier="business"))
+        session.commit()
+    finally:
+        session.close()
+
+
 def _make_client(int_engine, owner_id: str = _OWNER):
     """Return a TestClient with *owner_id* injected as the authenticated user."""
     from app.api.integrations import _get_owner_id
     from app.main import app
 
+    _seed_default_plan(int_engine, owner_id)
+
     def override_db():
         Session = sessionmaker(bind=int_engine)
         session = Session()
@@ -548,3 +609,399 @@ class TestImapPasswordEncryption:
                 verify_session.close()
         finally:
             app.dependency_overrides.clear()
+
+
+# ---------------------------------------------------------------------------
+# Quota enforcement tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestQuotaEnforcementDestinations:
+    """Tests for destination quota enforcement on POST /api/integrations/."""
+
+    def test_create_destination_blocked_at_limit(self, int_engine, int_session):
+        """Users at the destination quota limit receive a 403."""
+        _make_profile(int_session, tier="starter")
+        _make_plan(int_session, tier="starter", max_storage_destinations=1, max_mailboxes=1)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                # First destination should succeed
+                resp1 = client.post("/api/integrations/", json=_S3_DESTINATION)
+                assert resp1.status_code == 201
+
+                # Second destination should be blocked
+                second = dict(_S3_DESTINATION, name="Second Bucket")
+                resp2 = client.post("/api/integrations/", json=second)
+                assert resp2.status_code == 403
+                assert "limit" in resp2.json()["detail"].lower()
+        finally:
+            app.dependency_overrides.clear()
+
+    def test_create_destination_allowed_under_limit(self, int_engine, int_session):
+        """Users under the destination quota can create integrations."""
+        _make_profile(int_session, tier="professional")
+        _make_plan(int_session, tier="professional", max_storage_destinations=5, max_mailboxes=3)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                resp = client.post("/api/integrations/", json=_S3_DESTINATION)
+                assert resp.status_code == 201
+        finally:
+            app.dependency_overrides.clear()
+
+    def test_free_tier_allows_one_destination(self, int_engine, int_session):
+        """Free tier allows exactly 1 destination."""
+        _make_profile(int_session, tier="free")
+        _make_plan(int_session, tier="free", max_storage_destinations=1, max_mailboxes=0)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                resp1 = client.post("/api/integrations/", json=_S3_DESTINATION)
+                assert resp1.status_code == 201
+
+                second = dict(_S3_DESTINATION, name="Second")
+                resp2 = client.post("/api/integrations/", json=second)
+                assert resp2.status_code == 403
+        finally:
+            app.dependency_overrides.clear()
+
+
+@pytest.mark.integration
+class TestQuotaEnforcementSources:
+    """Tests for IMAP source quota enforcement on POST /api/integrations/."""
+
+    def test_create_imap_source_blocked_on_free_tier(self, int_engine, int_session):
+        """Free-tier users cannot add IMAP source integrations."""
+        _make_profile(int_session, tier="free")
+        _make_plan(int_session, tier="free", max_storage_destinations=1, max_mailboxes=0)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                resp = client.post("/api/integrations/", json=_IMAP_SOURCE)
+                assert resp.status_code == 403
+                assert "plan" in resp.json()["detail"].lower()
+        finally:
+            app.dependency_overrides.clear()
+
+    def test_create_imap_source_blocked_at_limit(self, int_engine, int_session):
+        """Starter-tier users with 1 IMAP source cannot add a second."""
+        _make_profile(int_session, tier="starter")
+        _make_plan(int_session, tier="starter", max_storage_destinations=2, max_mailboxes=1)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                resp1 = client.post("/api/integrations/", json=_IMAP_SOURCE)
+                assert resp1.status_code == 201
+
+                second = dict(_IMAP_SOURCE, name="Second Mailbox")
+                resp2 = client.post("/api/integrations/", json=second)
+                assert resp2.status_code == 403
+        finally:
+            app.dependency_overrides.clear()
+
+    def test_create_imap_source_unlimited_on_power_tier(self, int_engine, int_session):
+        """Power-tier users can add multiple IMAP sources (unlimited)."""
+        _make_profile(int_session, tier="business")
+        _make_plan(int_session, tier="business", max_storage_destinations=10, max_mailboxes=0)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                resp1 = client.post("/api/integrations/", json=_IMAP_SOURCE)
+                resp2 = client.post("/api/integrations/", json=dict(_IMAP_SOURCE, name="Second"))
+                resp3 = client.post("/api/integrations/", json=dict(_IMAP_SOURCE, name="Third"))
+            assert resp1.status_code == 201
+            assert resp2.status_code == 201
+            assert resp3.status_code == 201
+        finally:
+            app.dependency_overrides.clear()
+
+    def test_watch_folder_source_not_quota_limited(self, int_engine, int_session):
+        """WATCH_FOLDER sources are not subject to mailbox quota limits."""
+        _make_profile(int_session, tier="free")
+        _make_plan(int_session, tier="free", max_storage_destinations=1, max_mailboxes=0)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                payload = {
+                    "direction": "SOURCE",
+                    "integration_type": "WATCH_FOLDER",
+                    "name": "My Folder",
+                    "config": {"path": "/tmp/watch"},
+                    "is_active": True,
+                }
+                resp = client.post("/api/integrations/", json=payload)
+                assert resp.status_code == 201
+        finally:
+            app.dependency_overrides.clear()
+
+
+# ---------------------------------------------------------------------------
+# Quota helpers unit tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.unit
+class TestQuotaHelpers:
+    """Unit tests for the quota helper functions."""
+
+    def test_get_max_destinations_free_tier(self):
+        from app.api.integrations import _get_max_destinations
+
+        assert _get_max_destinations({"id": "free", "max_storage_destinations": 1}) == 1
+
+    def test_get_max_destinations_paid_explicit(self):
+        from app.api.integrations import _get_max_destinations
+
+        assert _get_max_destinations({"id": "starter", "max_storage_destinations": 2}) == 2
+
+    def test_get_max_destinations_paid_unlimited(self):
+        from app.api.integrations import _get_max_destinations
+
+        assert _get_max_destinations({"id": "business", "max_storage_destinations": 0}) is None
+
+    def test_get_max_sources_free_tier(self):
+        from app.api.integrations import _get_max_sources
+
+        assert _get_max_sources({"id": "free", "max_mailboxes": 0}) == 0
+
+    def test_get_max_sources_paid_explicit(self):
+        from app.api.integrations import _get_max_sources
+
+        assert _get_max_sources({"id": "starter", "max_mailboxes": 1}) == 1
+
+    def test_get_max_sources_paid_unlimited(self):
+        from app.api.integrations import _get_max_sources
+
+        assert _get_max_sources({"id": "business", "max_mailboxes": 0}) is None
+
+
+# ---------------------------------------------------------------------------
+# Connection test endpoint tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestConnectionTestEndpoint:
+    """Tests for POST /api/integrations/test."""
+
+    def test_test_unsupported_type(self, int_client):
+        """Unsupported integration types return a helpful non-error message."""
+        payload = {
+            "integration_type": "DROPBOX",
+            "config": {},
+            "credentials": {"token": "abc"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "not yet supported" in data["message"]
+
+    def test_test_invalid_type_returns_400(self, int_client):
+        """Invalid integration_type returns 400."""
+        payload = {
+            "integration_type": "INVALID",
+            "config": {},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 400
+
+    def test_test_imap_missing_fields(self, int_client):
+        """IMAP test with missing fields returns failure."""
+        payload = {
+            "integration_type": "IMAP",
+            "config": {"host": ""},
+            "credentials": {},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "Missing" in data["message"]
+
+    def test_test_s3_missing_bucket(self, int_client):
+        """S3 test with missing bucket returns failure."""
+        payload = {
+            "integration_type": "S3",
+            "config": {},
+            "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 "bucket" in data["message"].lower()
+
+    def test_test_webdav_missing_url(self, int_client):
+        """WebDAV test with missing URL returns failure."""
+        payload = {
+            "integration_type": "WEBDAV",
+            "config": {},
+            "credentials": {"username": "u", "password": "p"},
+        }
+        resp = int_client.post("/api/integrations/test", json=payload)
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["success"] is False
+        assert "url" in data["message"].lower()
+
+
+# ---------------------------------------------------------------------------
+# Quota endpoint tests
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.integration
+class TestQuotaEndpoint:
+    """Tests for GET /api/integrations/quota/."""
+
+    def test_quota_returns_tier_info(self, int_client):
+        """Quota endpoint returns tier information and counts."""
+        resp = int_client.get("/api/integrations/quota/")
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "tier_id" in data
+        assert "tier_name" in data
+        assert "destinations" in data
+        assert "sources" in data
+        assert "current_count" in data["destinations"]
+        assert "max_allowed" in data["destinations"]
+        assert "can_add" in data["destinations"]
+
+    def test_quota_reflects_created_integrations(self, int_client):
+        """Quota counts update after creating integrations."""
+        int_client.post("/api/integrations/", json=_S3_DESTINATION)
+        resp = int_client.get("/api/integrations/quota/")
+        data = resp.json()
+        assert data["destinations"]["current_count"] == 1
+
+    def test_quota_free_tier(self, int_engine, int_session):
+        """Free tier shows correct quota limits."""
+        _make_profile(int_session, tier="free")
+        _make_plan(int_session, tier="free", max_storage_destinations=1, max_mailboxes=0)
+
+        from app.api.integrations import _get_owner_id
+        from app.main import app
+
+        def override_db():
+            Session = sessionmaker(bind=int_engine)
+            session = Session()
+            try:
+                yield session
+            finally:
+                session.close()
+
+        app.dependency_overrides[get_db] = override_db
+        app.dependency_overrides[_get_owner_id] = lambda: _OWNER
+
+        try:
+            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
+                resp = client.get("/api/integrations/quota/")
+                data = resp.json()
+                assert data["tier_id"] == "free"
+                assert data["destinations"]["max_allowed"] == 1
+                assert data["destinations"]["can_add"] is True
+                assert data["sources"]["max_allowed"] == 0
+                assert data["sources"]["can_add"] is False
+        finally:
+            app.dependency_overrides.clear()