gh-christianlouis-docuelevate/tests/test_api_integrations.py

"""Tests for the per-user integrations API (app/api/integrations.py)."""

import pytest
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app.database import Base, get_db
from app.models import UserIntegration

# ---------------------------------------------------------------------------
# Test data constants
# ---------------------------------------------------------------------------

_OWNER = "test_user@example.com"
_OTHER_OWNER = "other_user@example.com"

_IMAP_SOURCE = {
    "direction": "SOURCE",
    "integration_type": "IMAP",
    "name": "Work Gmail",
    "config": {"host": "imap.gmail.com", "port": 993, "username": "work@example.com", "use_ssl": True},
    "credentials": {"password": "s3cr3t"},
    "is_active": True,
}

_S3_DESTINATION = {
    "direction": "DESTINATION",
    "integration_type": "S3",
    "name": "Archive Bucket",
    "config": {"bucket": "my-bucket", "region": "us-east-1"},
    "credentials": {
        "access_key_id": "AKIAIOSFODNN7EXAMPLE",
        "secret_access_key": "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY",
    },
    "is_active": True,
}


# ---------------------------------------------------------------------------
# Shared fixture helpers
# ---------------------------------------------------------------------------


@pytest.fixture()
def int_engine():
    """In-memory SQLite engine for integration tests."""
    engine = create_engine(
        "sqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )
    Base.metadata.create_all(bind=engine)
    yield engine
    Base.metadata.drop_all(bind=engine)


@pytest.fixture()
def int_session(int_engine):
    """DB session scoped to one test."""
    Session = sessionmaker(bind=int_engine)
    session = Session()
    yield session
    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

    def override_db():
        Session = sessionmaker(bind=int_engine)
        session = Session()
        try:
            yield session
        finally:
            session.close()

    def override_owner():
        return owner_id

    app.dependency_overrides[get_db] = override_db
    app.dependency_overrides[_get_owner_id] = override_owner
    with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
        yield client
    app.dependency_overrides.clear()


@pytest.fixture()
def int_client(int_engine):
    """TestClient authenticated as _OWNER."""
    yield from _make_client(int_engine, _OWNER)


# ---------------------------------------------------------------------------
# CRUD tests
# ---------------------------------------------------------------------------


@pytest.mark.integration
class TestCreateIntegration:
    """Tests for POST /api/integrations/."""

    def test_create_source_integration(self, int_client):
        """Create a SOURCE integration and verify the response."""
        resp = int_client.post("/api/integrations/", json=_IMAP_SOURCE)
        assert resp.status_code == 201
        data = resp.json()
        assert data["direction"] == "SOURCE"
        assert data["integration_type"] == "IMAP"
        assert data["name"] == "Work Gmail"
        assert data["config"]["host"] == "imap.gmail.com"
        assert data["has_credentials"] is True
        # Credentials must never appear in the response
        assert "credentials" not in data
        assert "password" not in data

    def test_create_destination_integration(self, int_client):
        """Create a DESTINATION integration and verify the response."""
        resp = int_client.post("/api/integrations/", json=_S3_DESTINATION)
        assert resp.status_code == 201
        data = resp.json()
        assert data["direction"] == "DESTINATION"
        assert data["integration_type"] == "S3"
        assert data["has_credentials"] is True

    def test_create_without_credentials(self, int_client):
        """Integration without credentials sets has_credentials to False."""
        payload = dict(_IMAP_SOURCE)
        payload["credentials"] = None
        resp = int_client.post("/api/integrations/", json=payload)
        assert resp.status_code == 201
        assert resp.json()["has_credentials"] is False

    def test_create_invalid_direction(self, int_client):
        """An unknown direction returns 400."""
        payload = dict(_IMAP_SOURCE, direction="INVALID")
        resp = int_client.post("/api/integrations/", json=payload)
        assert resp.status_code == 400

    def test_create_invalid_type(self, int_client):
        """An unknown integration_type returns 400."""
        payload = dict(_IMAP_SOURCE, integration_type="UNKNOWN_TYPE")
        resp = int_client.post("/api/integrations/", json=payload)
        assert resp.status_code == 400

    def test_credentials_are_encrypted_in_db(self, int_client, int_session):
        """Stored credentials must be encrypted (enc: prefix)."""
        resp = int_client.post("/api/integrations/", json=_IMAP_SOURCE)
        assert resp.status_code == 201
        rec = int_session.query(UserIntegration).first()
        assert rec is not None
        assert rec.credentials is not None
        assert rec.credentials.startswith("enc:")


@pytest.mark.integration
class TestListIntegrations:
    """Tests for GET /api/integrations/."""

    def test_list_empty(self, int_client):
        """No integrations returns empty list."""
        resp = int_client.get("/api/integrations/")
        assert resp.status_code == 200
        assert resp.json() == []

    def test_list_returns_own_records(self, int_client, int_session):
        """Users only see their own integrations."""
        # Create an integration for _OWNER via the API client
        int_client.post("/api/integrations/", json=_IMAP_SOURCE)

        # Create an integration for a different owner directly in the DB
        other_integration = UserIntegration(
            owner_id=_OTHER_OWNER,
            direction="DESTINATION",
            integration_type="S3",
            name="Archive Bucket",
            config='{"bucket": "other-bucket"}',
            is_active=True,
        )
        int_session.add(other_integration)
        int_session.commit()

        resp = int_client.get("/api/integrations/")
        assert resp.status_code == 200
        assert len(resp.json()) == 1
        assert resp.json()[0]["integration_type"] == "IMAP"

    def test_filter_by_direction(self, int_client):
        """direction query param filters results."""
        int_client.post("/api/integrations/", json=_IMAP_SOURCE)
        int_client.post("/api/integrations/", json=_S3_DESTINATION)

        resp = int_client.get("/api/integrations/?direction=SOURCE")
        assert resp.status_code == 200
        results = resp.json()
        assert len(results) == 1
        assert results[0]["direction"] == "SOURCE"

    def test_filter_by_integration_type(self, int_client):
        """integration_type query param filters results."""
        int_client.post("/api/integrations/", json=_IMAP_SOURCE)
        int_client.post("/api/integrations/", json=_S3_DESTINATION)

        resp = int_client.get("/api/integrations/?integration_type=S3")
        assert resp.status_code == 200
        results = resp.json()
        assert len(results) == 1
        assert results[0]["integration_type"] == "S3"

    def test_filter_invalid_direction_returns_400(self, int_client):
        """Unknown direction filter returns 400."""
        resp = int_client.get("/api/integrations/?direction=BAD")
        assert resp.status_code == 400

    def test_filter_invalid_type_returns_400(self, int_client):
        """Unknown integration_type filter returns 400."""
        resp = int_client.get("/api/integrations/?integration_type=BAD")
        assert resp.status_code == 400


@pytest.mark.integration
class TestGetIntegration:
    """Tests for GET /api/integrations/{id}."""

    def test_get_existing(self, int_client):
        """Retrieve a single integration by ID."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        resp = int_client.get(f"/api/integrations/{created['id']}")
        assert resp.status_code == 200
        assert resp.json()["id"] == created["id"]

    def test_get_not_found(self, int_client):
        """Non-existent ID returns 404."""
        resp = int_client.get("/api/integrations/9999")
        assert resp.status_code == 404

    def test_get_other_users_integration(self, int_client, int_session):
        """Cannot retrieve another user's integration."""
        other_integration = UserIntegration(
            owner_id=_OTHER_OWNER,
            direction="SOURCE",
            integration_type="IMAP",
            name="Other Mailbox",
            is_active=True,
        )
        int_session.add(other_integration)
        int_session.commit()
        resp = int_client.get(f"/api/integrations/{other_integration.id}")
        assert resp.status_code == 404


@pytest.mark.integration
class TestUpdateIntegration:
    """Tests for PUT /api/integrations/{id}."""

    def test_update_name(self, int_client):
        """Update the name of an integration."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        resp = int_client.put(f"/api/integrations/{created['id']}", json={"name": "Personal Gmail"})
        assert resp.status_code == 200
        assert resp.json()["name"] == "Personal Gmail"

    def test_update_config(self, int_client):
        """Update the config dict."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        new_config = {"host": "imap.new.com", "port": 143, "username": "new@example.com", "use_ssl": False}
        resp = int_client.put(f"/api/integrations/{created['id']}", json={"config": new_config})
        assert resp.status_code == 200
        assert resp.json()["config"]["host"] == "imap.new.com"

    def test_update_credentials_re_encrypts(self, int_client, int_session):
        """Updating credentials stores the new value encrypted."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        first_enc = int_session.query(UserIntegration).get(created["id"]).credentials

        resp = int_client.put(f"/api/integrations/{created['id']}", json={"credentials": {"password": "new_password"}})
        assert resp.status_code == 200
        int_session.expire_all()
        second_enc = int_session.query(UserIntegration).get(created["id"]).credentials
        # Both must be encrypted
        assert second_enc.startswith("enc:")
        # They should be different ciphertexts (Fernet uses random IV)
        assert first_enc != second_enc

    def test_update_is_active(self, int_client):
        """Deactivate an integration."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        resp = int_client.put(f"/api/integrations/{created['id']}", json={"is_active": False})
        assert resp.status_code == 200
        assert resp.json()["is_active"] is False

    def test_update_not_found(self, int_client):
        """Updating a non-existent integration returns 404."""
        resp = int_client.put("/api/integrations/9999", json={"name": "Ghost"})
        assert resp.status_code == 404

    def test_update_clears_last_error(self, int_client, int_session):
        """Updating an integration resets last_error."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        rec = int_session.query(UserIntegration).get(created["id"])
        rec.last_error = "Previous failure"
        int_session.commit()

        int_client.put(f"/api/integrations/{created['id']}", json={"name": "Updated"})
        int_session.expire_all()
        rec = int_session.query(UserIntegration).get(created["id"])
        assert rec.last_error is None


@pytest.mark.integration
class TestDeleteIntegration:
    """Tests for DELETE /api/integrations/{id}."""

    def test_delete_existing(self, int_client, int_session):
        """Delete an integration and confirm it is removed from the DB."""
        created = int_client.post("/api/integrations/", json=_IMAP_SOURCE).json()
        resp = int_client.delete(f"/api/integrations/{created['id']}")
        assert resp.status_code == 204
        assert int_session.query(UserIntegration).get(created["id"]) is None

    def test_delete_not_found(self, int_client):
        """Deleting a non-existent integration returns 404."""
        resp = int_client.delete("/api/integrations/9999")
        assert resp.status_code == 404

    def test_delete_other_users_integration(self, int_client, int_session):
        """Cannot delete another user's integration."""
        other_integration = UserIntegration(
            owner_id=_OTHER_OWNER,
            direction="SOURCE",
            integration_type="IMAP",
            name="Other Mailbox",
            is_active=True,
        )
        int_session.add(other_integration)
        int_session.commit()
        resp = int_client.delete(f"/api/integrations/{other_integration.id}")
        assert resp.status_code == 404


@pytest.mark.integration
class TestGetIntegrationCredentials:
    """Tests for GET /api/integrations/{id}/credentials."""

    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

    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"] == {}

    def test_not_found(self, int_client):
        """Non-existent integration returns 404."""
        resp = int_client.get("/api/integrations/9999/credentials")
        assert resp.status_code == 404

    def test_other_users_credentials_returns_404(self, int_client, int_session):
        """Cannot retrieve another user's credentials."""
        other_integration = UserIntegration(
            owner_id=_OTHER_OWNER,
            direction="SOURCE",
            integration_type="IMAP",
            name="Other Mailbox",
            credentials='{"password": "secret"}',
            is_active=True,
        )
        int_session.add(other_integration)
        int_session.commit()
        resp = int_client.get(f"/api/integrations/{other_integration.id}/credentials")
        assert resp.status_code == 404


@pytest.mark.unit
class TestIntegrationModel:
    """Unit tests for the UserIntegration model and helper constants."""

    def test_integration_direction_constants(self):
        """IntegrationDirection has expected values."""
        from app.models import IntegrationDirection

        assert IntegrationDirection.SOURCE == "SOURCE"
        assert IntegrationDirection.DESTINATION == "DESTINATION"
        assert "SOURCE" in IntegrationDirection.ALL
        assert "DESTINATION" in IntegrationDirection.ALL

    def test_integration_type_constants(self):
        """IntegrationType has expected values."""
        from app.models import IntegrationType

        assert IntegrationType.IMAP == "IMAP"
        assert IntegrationType.S3 == "S3"
        assert IntegrationType.DROPBOX == "DROPBOX"
        assert IntegrationType.GOOGLE_DRIVE == "GOOGLE_DRIVE"
        assert IntegrationType.ONEDRIVE == "ONEDRIVE"
        assert IntegrationType.WEBDAV == "WEBDAV"
        assert IntegrationType.NEXTCLOUD == "NEXTCLOUD"
        assert IntegrationType.WATCH_FOLDER == "WATCH_FOLDER"
        assert IntegrationType.WEBHOOK == "WEBHOOK"
        assert IntegrationType.FTP == "FTP"
        assert IntegrationType.SFTP == "SFTP"
        assert IntegrationType.EMAIL == "EMAIL"
        assert IntegrationType.PAPERLESS == "PAPERLESS"
        assert IntegrationType.RCLONE == "RCLONE"

    def test_user_integration_tablename(self):
        """UserIntegration uses the correct table name."""
        assert UserIntegration.__tablename__ == "user_integrations"

    def test_user_integration_fields(self):
        """UserIntegration has all required columns."""
        cols = {c.key for c in UserIntegration.__table__.columns}
        expected = {
            "id",
            "owner_id",
            "direction",
            "integration_type",
            "name",
            "config",
            "credentials",
            "is_active",
            "last_used_at",
            "last_error",
            "created_at",
            "updated_at",
        }
        assert expected <= cols


@pytest.mark.unit
class TestCredentialHelpers:
    """Unit tests for the credential encode/decode helpers in integrations.py."""

    def test_encode_decode_round_trip(self):
        """Encoding then decoding returns the original dict."""
        from app.api.integrations import _decode_credentials, _encode_credentials

        original = {"password": "super_secret", "token": "abc123"}
        encoded = _encode_credentials(original)
        assert encoded is not None
        decoded = _decode_credentials(encoded)
        assert decoded == original

    def test_encode_none_returns_none(self):
        """Encoding None returns None."""
        from app.api.integrations import _encode_credentials

        assert _encode_credentials(None) is None

    def test_encode_empty_dict_returns_none(self):
        """Encoding an empty dict returns None."""
        from app.api.integrations import _encode_credentials

        assert _encode_credentials({}) is None

    def test_decode_none_returns_none(self):
        """Decoding None returns None."""
        from app.api.integrations import _decode_credentials

        assert _decode_credentials(None) is None

    def test_decode_empty_string_returns_none(self):
        """Decoding an empty string returns None."""
        from app.api.integrations import _decode_credentials

        assert _decode_credentials("") is None

    def test_decode_invalid_json_returns_none(self):
        """Decoding a non-JSON plaintext string returns None."""
        from app.api.integrations import _decode_credentials

        # A non-JSON plaintext string (no enc: prefix) that decrypt_value returns as-is
        assert _decode_credentials("not-valid-json") is None


@pytest.mark.unit
class TestImapPasswordEncryption:
    """Unit tests verifying that IMAP passwords are encrypted at rest."""

    def test_create_encrypts_password(self, int_engine):
        """Creating an IMAP account stores the password encrypted."""
        from app.api.imap_accounts import _get_owner_id
        from app.main import app
        from app.models import SubscriptionPlan, UserImapAccount, UserProfile

        Session = sessionmaker(bind=int_engine)

        def override_db():
            session = Session()
            try:
                yield session
            finally:
                session.close()

        def override_owner():
            return _OWNER

        # Seed a paid plan (max_mailboxes=0 means unlimited for paid) and profile
        setup_session = Session()
        plan = SubscriptionPlan(
            plan_id="paid",
            name="Paid",
            price_monthly=9.99,
            price_yearly=99.99,
            max_mailboxes=0,  # 0 = unlimited for paid plans
            is_active=True,
        )
        setup_session.add(plan)
        profile = UserProfile(user_id=_OWNER, subscription_tier="paid")
        setup_session.add(profile)
        setup_session.commit()
        setup_session.close()

        app.dependency_overrides[get_db] = override_db
        app.dependency_overrides[_get_owner_id] = override_owner
        try:
            with TestClient(app, base_url="http://localhost", raise_server_exceptions=False) as client:
                payload = {
                    "name": "Test",
                    "host": "imap.example.com",
                    "port": 993,
                    "username": "user@example.com",
                    "password": "plaintext_password",
                    "use_ssl": True,
                    "delete_after_process": False,
                    "is_active": True,
                }
                resp = client.post("/api/imap-accounts/", json=payload)
                assert resp.status_code == 201

                # Verify the stored password is encrypted
                verify_session = Session()
                acct = verify_session.query(UserImapAccount).first()
                assert acct is not None
                assert acct.password != "plaintext_password"
                assert acct.password.startswith("enc:")
                verify_session.close()
        finally:
            app.dependency_overrides.clear()