diff --git a/app/api/integrations.py b/app/api/integrations.py index d7aeb179..418e23b6 100644 --- a/app/api/integrations.py +++ b/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), + }, + } diff --git a/tests/test_api_integrations.py b/tests/test_api_integrations.py index 3b030cd0..63708cd0 100644 --- a/tests/test_api_integrations.py +++ b/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()