kite/backend/tests/test_cloud_analysis_contract.py

"""
Phase 14 Plan 01 — RED contract tests for cloud analysis scope, estimates, and
idempotency (no bytes downloaded during estimate or already-current checks).

Requirements: ANALYZE-01, ANALYZE-02, ANALYZE-03, ANALYZE-06, ANALYZE-07

Threat coverage:
  T-14-01 — IDOR: foreign user cannot estimate/enqueue another user's connection
  T-14-03 — Hidden provider mutation: fake adapter mutation counters stay zero
  T-14-04 — Hidden byte download: estimate and already-current checks never call
             provider byte fetch

All tests import non-existent Phase 14 modules and must FAIL due to
ImportError or missing endpoints. They define the implementation contract.
"""
from __future__ import annotations

import uuid
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

pytestmark = pytest.mark.asyncio

from tests.conftest import _TEST_USER_AGENT


# ─── Minimal fake adapter ────────────────────────────────────────────────────

class FakeCloudAdapter:
    """Counts calls to mutation and byte-download methods.

    Zero counts on all methods after analysis operations confirms
    T-14-03 (no mutation) and T-14-04 (no byte download).
    """

    def __init__(self):
        self.upload_calls = 0
        self.delete_calls = 0
        self.rename_calls = 0
        self.move_calls = 0
        self.create_folder_calls = 0
        self.get_object_calls = 0       # byte download
        self.put_object_calls = 0       # byte upload

    # Mutations — must stay zero during analysis
    async def upload(self, *a, **kw):
        self.upload_calls += 1
        raise AssertionError("upload must not be called during analysis")

    async def delete(self, *a, **kw):
        self.delete_calls += 1
        raise AssertionError("delete must not be called during analysis")

    async def rename(self, *a, **kw):
        self.rename_calls += 1
        raise AssertionError("rename must not be called during analysis")

    async def move(self, *a, **kw):
        self.move_calls += 1
        raise AssertionError("move must not be called during analysis")

    async def create_folder(self, *a, **kw):
        self.create_folder_calls += 1
        raise AssertionError("create_folder must not be called during analysis")

    # Byte downloads — must stay zero during estimate and already-current checks
    async def get_object(self, *a, **kw):
        self.get_object_calls += 1
        raise AssertionError("get_object must not be called during estimate or already-current check")

    async def put_object(self, *a, **kw):
        self.put_object_calls += 1
        raise AssertionError("put_object must not be called during analysis path")

    @property
    def mutation_call_count(self):
        return (
            self.upload_calls + self.delete_calls + self.rename_calls +
            self.move_calls + self.create_folder_calls
        )

    @property
    def byte_call_count(self):
        return self.get_object_calls + self.put_object_calls


# ─── Shared test helpers ─────────────────────────────────────────────────────

async def _create_user_and_token(session, role: str = "user"):
    from db.models import User, Quota
    from services.auth import hash_password, create_access_token

    user_id = uuid.uuid4()
    user = User(
        id=user_id,
        handle=f"ana_user_{user_id.hex[:8]}",
        email=f"ana_{user_id.hex[:8]}@example.com",
        password_hash=hash_password("Testpassword123!"),
        role=role,
        is_active=True,
        password_must_change=False,
    )
    quota = Quota(user_id=user_id, limit_bytes=104857600, used_bytes=0)
    session.add(user)
    session.add(quota)
    await session.commit()
    await session.refresh(user)

    token = create_access_token(str(user_id), role, user_agent=_TEST_USER_AGENT)
    return {
        "user": user,
        "token": token,
        "headers": {
            "Authorization": f"Bearer {token}",
            "User-Agent": _TEST_USER_AGENT,
        },
    }


async def _create_cloud_connection(session, user_id, provider="google_drive", status="ACTIVE"):
    from db.models import CloudConnection
    from storage.cloud_utils import encrypt_credentials
    from config import settings

    master_key = settings.cloud_creds_key.encode()
    creds_enc = encrypt_credentials(
        master_key,
        str(user_id),
        {"access_token": "tok", "refresh_token": "ref"},
    )
    conn = CloudConnection(
        id=uuid.uuid4(),
        user_id=user_id,
        provider=provider,
        display_name="Test Connection",
        credentials_enc=creds_enc,
        status=status,
    )
    session.add(conn)
    await session.commit()
    return conn


async def _create_cloud_item(session, user_id, connection_id, *, name="doc.pdf",
                              kind="file", etag="etag-v1", analysis_status="pending"):
    from db.models import CloudItem

    item = CloudItem(
        id=uuid.uuid4(),
        user_id=user_id,
        connection_id=connection_id,
        provider_item_id=f"pitem-{uuid.uuid4().hex[:8]}",
        name=name,
        kind=kind,
        content_type="application/pdf",
        provider_size=102400,
        etag=etag,
        analysis_status=analysis_status,
    )
    session.add(item)
    await session.commit()
    return item


# ─── ANALYZE-01: Single file analysis estimate ───────────────────────────────

async def test_estimate_single_file_returns_supported_count_and_bytes(async_client, db_session):
    """ANALYZE-01: Single-file estimate returns count=1 and provider_bytes without downloading.

    T-14-04: no bytes fetched during estimate.
    """
    # Requires Phase 14 endpoint: POST /api/cloud/analysis/connections/{id}/estimate
    from api.cloud.analysis import router as analysis_router  # Phase 14 — not yet implemented

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/estimate",
            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
            headers=ctx["headers"],
        )

    assert resp.status_code == 200
    body = resp.json()
    assert body["supported_count"] == 1
    assert body["unsupported_count"] == 0
    assert "total_provider_bytes" in body
    assert body["total_provider_bytes"] == item.provider_size

    # T-14-04: no byte downloads during estimate
    assert adapter.byte_call_count == 0, "Estimate must not download provider bytes"
    # T-14-03: no mutations during estimate
    assert adapter.mutation_call_count == 0, "Estimate must not call any provider mutation"


# ─── ANALYZE-02: Selection estimate ─────────────────────────────────────────

async def test_estimate_selection_aggregates_multiple_files(async_client, db_session):
    """ANALYZE-02: Multi-file selection estimate aggregates counts and bytes.

    T-14-04: bytes not fetched during selection estimate.
    """
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    item1 = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="a.pdf")
    item2 = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="b.docx")

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/estimate",
            json={
                "scope": "selection",
                "provider_item_ids": [item1.provider_item_id, item2.provider_item_id],
            },
            headers=ctx["headers"],
        )

    assert resp.status_code == 200
    body = resp.json()
    assert body["supported_count"] == 2
    # total bytes = sum of provider sizes
    assert body["total_provider_bytes"] == (item1.provider_size or 0) + (item2.provider_size or 0)

    assert adapter.byte_call_count == 0
    assert adapter.mutation_call_count == 0


async def test_estimate_includes_unsupported_item_count(async_client, db_session):
    """ANALYZE-02: Estimate reports unsupported items without blocking supportable ones.

    Unsupported items are reported but do not cause estimate to fail (D-23).
    """
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    pdf_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="report.pdf")
    # An item with an unsupported content type (e.g. .exe) should count as unsupported
    exe_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="setup.exe")

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/estimate",
            json={
                "scope": "selection",
                "provider_item_ids": [pdf_item.provider_item_id, exe_item.provider_item_id],
            },
            headers=ctx["headers"],
        )

    assert resp.status_code == 200
    body = resp.json()
    # At least one supported (pdf), at least one unsupported (exe)
    assert body["supported_count"] >= 1
    assert body["unsupported_count"] >= 1
    assert adapter.byte_call_count == 0


# ─── ANALYZE-03: Whole-connection estimate ───────────────────────────────────

async def test_whole_connection_estimate_always_required(async_client, db_session):
    """ANALYZE-03: Whole-connection scope always produces an estimate response before enqueue.

    The estimate endpoint must be called before enqueue for connection-scope (D-02).
    T-14-04: bytes not fetched.
    """
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    await _create_cloud_item(db_session, ctx["user"].id, conn.id, name="file1.pdf")

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/estimate",
            json={"scope": "connection", "recursive": True},
            headers=ctx["headers"],
        )

    assert resp.status_code == 200
    body = resp.json()
    # Whole-connection always includes a recursive flag in the response
    assert "recursive" in body or body.get("scope_kind") == "connection"
    assert "supported_count" in body
    assert adapter.byte_call_count == 0


async def test_folder_estimate_asks_about_recursion(async_client, db_session):
    """ANALYZE-03/D-04: Folder-scope estimate accepts recursive flag and reports it back."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    folder_item = await _create_cloud_item(db_session, ctx["user"].id, conn.id,
                                            name="Reports", kind="folder")

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/estimate",
            json={
                "scope": "folder",
                "provider_item_ids": [folder_item.provider_item_id],
                "recursive": True,
            },
            headers=ctx["headers"],
        )

    assert resp.status_code == 200
    body = resp.json()
    assert "recursive" in body
    assert adapter.byte_call_count == 0


# ─── ANALYZE-06: Duplicate / already-current skip ────────────────────────────

async def test_already_current_item_skipped_without_byte_fetch(async_client, db_session):
    """ANALYZE-06: Item with unchanged etag is skipped without downloading provider bytes.

    T-14-04: get_object must not be called for already-current items.
    """
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    # Item already indexed with same etag
    item = await _create_cloud_item(
        db_session, ctx["user"].id, conn.id,
        etag="etag-stable-v1", analysis_status="indexed"
    )

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/jobs",
            json={
                "scope": "file",
                "provider_item_ids": [item.provider_item_id],
            },
            headers=ctx["headers"],
        )

    # Enqueue succeeds (202 Accepted or 200 with job status)
    assert resp.status_code in (200, 202)
    body = resp.json()

    # Item reported as already_current — not re-downloaded
    # The job should report 0 queued items (or the item appears as already_current)
    job_id = body.get("job_id")
    assert job_id is not None

    # Fetch job status to verify already_current
    status_resp = await async_client.get(
        f"/api/cloud/analysis/jobs/{job_id}",
        headers=ctx["headers"],
    )
    assert status_resp.status_code == 200
    status_body = status_resp.json()

    # Item should not be in "queued" state — it's already current
    assert status_body.get("already_current_count", 0) >= 1
    assert status_body.get("queued_count", 0) == 0

    # Critical: no bytes were fetched
    assert adapter.byte_call_count == 0, "already-current check must not download provider bytes"
    # Critical: no mutations occurred
    assert adapter.mutation_call_count == 0


async def test_changed_etag_triggers_reanalysis(async_client, db_session):
    """ANALYZE-06: Item with changed etag is queued for reanalysis (not skipped)."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    # Item previously indexed but etag has changed
    item = await _create_cloud_item(
        db_session, ctx["user"].id, conn.id,
        etag="etag-OLD", analysis_status="indexed"
    )

    adapter = FakeCloudAdapter()
    # Provider now reports a different etag
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        with patch("services.cloud_analysis.resolve_provider_metadata",
                   new_callable=AsyncMock,
                   return_value={"etag": "etag-NEW", "size": 102400}):
            resp = await async_client.post(
                f"/api/cloud/analysis/connections/{conn.id}/jobs",
                json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
                headers=ctx["headers"],
            )

    assert resp.status_code in (200, 202)
    body = resp.json()
    job_id = body.get("job_id")
    assert job_id is not None

    status_resp = await async_client.get(
        f"/api/cloud/analysis/jobs/{job_id}",
        headers=ctx["headers"],
    )
    assert status_resp.status_code == 200
    status_body = status_resp.json()
    # Item is queued for re-analysis (not marked already_current)
    assert status_body.get("queued_count", 0) >= 1


# ─── ANALYZE-07: Provider file not mutated ───────────────────────────────────

async def test_analysis_never_mutates_provider_file(async_client, db_session):
    """ANALYZE-07: Analysis job completion never calls any provider mutation method.

    T-14-03: mutation counters must remain zero for the full analysis lifecycle.
    """
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/jobs",
            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
            headers=ctx["headers"],
        )

    assert resp.status_code in (200, 202)
    # No mutations at any point in analysis
    assert adapter.mutation_call_count == 0, (
        f"Analysis called provider mutations: upload={adapter.upload_calls}, "
        f"delete={adapter.delete_calls}, rename={adapter.rename_calls}, "
        f"move={adapter.move_calls}, create_folder={adapter.create_folder_calls}"
    )


# ─── T-14-01: IDOR — foreign user cannot estimate/enqueue ───────────────────

async def test_foreign_user_cannot_estimate_another_users_connection(async_client, db_session):
    """T-14-01: Estimate request against another user's connection returns 403/404."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    owner_ctx = await _create_user_and_token(db_session)
    foreign_ctx = await _create_user_and_token(db_session)
    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)

    resp = await async_client.post(
        f"/api/cloud/analysis/connections/{owner_conn.id}/estimate",
        json={"scope": "connection", "recursive": True},
        headers=foreign_ctx["headers"],  # Foreign user's token
    )

    assert resp.status_code in (403, 404), (
        f"Foreign user estimate must be rejected — got {resp.status_code}"
    )


async def test_foreign_user_cannot_enqueue_another_users_connection(async_client, db_session):
    """T-14-01: Enqueue request against another user's connection is rejected."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    owner_ctx = await _create_user_and_token(db_session)
    foreign_ctx = await _create_user_and_token(db_session)
    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
    owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)

    resp = await async_client.post(
        f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
        json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
        headers=foreign_ctx["headers"],
    )

    assert resp.status_code in (403, 404), (
        f"Foreign user enqueue must be rejected — got {resp.status_code}"
    )


async def test_admin_cannot_enqueue_another_users_analysis_job(async_client, db_session):
    """T-14-01: Admin role cannot access cloud analysis routes (same as browse restriction)."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    user_ctx = await _create_user_and_token(db_session, role="user")
    admin_ctx = await _create_user_and_token(db_session, role="admin")
    user_conn = await _create_cloud_connection(db_session, user_ctx["user"].id)
    user_item = await _create_cloud_item(db_session, user_ctx["user"].id, user_conn.id)

    resp = await async_client.post(
        f"/api/cloud/analysis/connections/{user_conn.id}/jobs",
        json={"scope": "file", "provider_item_ids": [user_item.provider_item_id]},
        headers=admin_ctx["headers"],
    )

    assert resp.status_code in (403, 404), (
        f"Admin must not access cloud analysis — got {resp.status_code}"
    )


# ─── Job status and controls ─────────────────────────────────────────────────

async def test_job_status_response_excludes_credentials_and_object_keys(async_client, db_session):
    """T-14-02: Job status response never includes credentials or raw object keys."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        enqueue_resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/jobs",
            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
            headers=ctx["headers"],
        )

    assert enqueue_resp.status_code in (200, 202)
    job_id = enqueue_resp.json().get("job_id")
    assert job_id is not None

    status_resp = await async_client.get(
        f"/api/cloud/analysis/jobs/{job_id}",
        headers=ctx["headers"],
    )
    assert status_resp.status_code == 200
    body_text = status_resp.text

    # T-14-02: No credential fields or raw object keys in response
    forbidden_fields = [
        "credentials_enc", "access_token", "refresh_token",
        "client_secret", "object_key",
    ]
    for field in forbidden_fields:
        assert field not in body_text, (
            f"Sensitive field '{field}' must not appear in job status response"
        )


async def test_foreign_user_cannot_read_job_status(async_client, db_session):
    """T-14-01: Foreign user cannot read job status for another user's job."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    owner_ctx = await _create_user_and_token(db_session)
    foreign_ctx = await _create_user_and_token(db_session)
    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
    owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        enqueue_resp = await async_client.post(
            f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
            json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
            headers=owner_ctx["headers"],
        )

    job_id = enqueue_resp.json().get("job_id")

    # Foreign user tries to read the job
    resp = await async_client.get(
        f"/api/cloud/analysis/jobs/{job_id}",
        headers=foreign_ctx["headers"],
    )
    assert resp.status_code in (403, 404)


async def test_foreign_user_cannot_cancel_another_users_job(async_client, db_session):
    """T-14-01: Foreign user cannot cancel a job owned by another user."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    owner_ctx = await _create_user_and_token(db_session)
    foreign_ctx = await _create_user_and_token(db_session)
    owner_conn = await _create_cloud_connection(db_session, owner_ctx["user"].id)
    owner_item = await _create_cloud_item(db_session, owner_ctx["user"].id, owner_conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        enqueue_resp = await async_client.post(
            f"/api/cloud/analysis/connections/{owner_conn.id}/jobs",
            json={"scope": "file", "provider_item_ids": [owner_item.provider_item_id]},
            headers=owner_ctx["headers"],
        )

    job_id = enqueue_resp.json().get("job_id")

    resp = await async_client.post(
        f"/api/cloud/analysis/jobs/{job_id}/cancel",
        headers=foreign_ctx["headers"],
    )
    assert resp.status_code in (403, 404)


# ─── Cancel and retry state transitions ──────────────────────────────────────

async def test_cancel_job_transitions_to_cancelled_status(async_client, db_session):
    """ANALYZE-05: Cancel job transitions the job status to 'cancelled'."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        enqueue_resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/jobs",
            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
            headers=ctx["headers"],
        )

    job_id = enqueue_resp.json().get("job_id")

    cancel_resp = await async_client.post(
        f"/api/cloud/analysis/jobs/{job_id}/cancel",
        headers=ctx["headers"],
    )
    assert cancel_resp.status_code in (200, 202, 204)

    status_resp = await async_client.get(
        f"/api/cloud/analysis/jobs/{job_id}",
        headers=ctx["headers"],
    )
    assert status_resp.json().get("status") in ("cancelled", "cancelling")


async def test_retry_failed_item_requeues_it(async_client, db_session):
    """ANALYZE-05: Retry on a failed item transitions it back to queued status."""
    from api.cloud.analysis import router as analysis_router  # Phase 14

    ctx = await _create_user_and_token(db_session)
    conn = await _create_cloud_connection(db_session, ctx["user"].id)
    item = await _create_cloud_item(db_session, ctx["user"].id, conn.id)

    adapter = FakeCloudAdapter()
    with patch("services.cloud_analysis.get_adapter", return_value=adapter):
        enqueue_resp = await async_client.post(
            f"/api/cloud/analysis/connections/{conn.id}/jobs",
            json={"scope": "file", "provider_item_ids": [item.provider_item_id]},
            headers=ctx["headers"],
        )

    job_id = enqueue_resp.json().get("job_id")

    # Simulate item in failed state then retry
    resp = await async_client.post(
        f"/api/cloud/analysis/jobs/{job_id}/items/{item.id}/retry",
        headers=ctx["headers"],
    )
    # Must return a valid status that indicates retry was accepted
    assert resp.status_code in (200, 202, 204)