test(12.1-02): add failing tests for truthful freshness gate (RED)

- test_incomplete_listing_never_marks_folder_fresh
- test_incomplete_listing_retains_cached_rows_and_last_success
- test_complete_empty_listing_is_authoritative_and_fresh
- test_partial_items_upsert_without_deleting_unseen_children
- test_apply_listing_returns_warning_for_complete_false
- test_sync_browse_returns_warning_for_complete_false
- test_worker_returns_warning_for_complete_false
- security: browse_complete_false_never_sets_fresh, no_raw_provider_error, owner_scoped_on_incomplete

backend/tests/test_cloud_security.py

@@ -421,6 +421,126 @@ async def test_browse_malformed_connection_id_returns_422(async_client, db_sessi
     assert resp.status_code == 422, f"Expected 422 for malformed UUID, got {resp.status_code}"
 
 
+# ── Plan 02 (T-12.1-06..10): Incomplete listing security assertions ───────────
+
+async def test_browse_complete_false_never_sets_fresh(async_client, db_session):
+    """Browse with complete=False provider response must not expose refresh_state='fresh'.
+
+    T-12.1-06: false fresh state hides provider failure from the user.
+    The API must return the cached rows with a warning freshness state.
+    """
+    from storage.cloud_base import CloudListing
+
+    auth = await _create_user_and_token(db_session, role="user")
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    mock_adapter = AsyncMock()
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(
+        return_value=_make_mock_adapter().get_capabilities.return_value
+    )
+
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth["headers"],
+        )
+
+    assert resp.status_code == 200
+    freshness = resp.json().get("freshness", {})
+    assert freshness.get("refresh_state") != "fresh", (
+        "T-12.1-06: complete=False must never produce refresh_state='fresh'"
+    )
+
+
+async def test_browse_complete_false_no_raw_provider_error_in_response(async_client, db_session):
+    """Browse error messages must be controlled — no traceback, URL, or credential leakage.
+
+    T-12.1-09: provider errors must not leak secrets or raw exception text to API consumers.
+    """
+    from storage.cloud_base import CloudListing
+
+    auth = await _create_user_and_token(db_session, role="user")
+    conn = await _create_cloud_connection(db_session, auth["user"].id)
+
+    mock_adapter = AsyncMock()
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(
+        return_value=_make_mock_adapter().get_capabilities.return_value
+    )
+
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth["headers"],
+        )
+
+    assert resp.status_code == 200
+    body_lower = resp.text.lower()
+    # No raw exception / traceback in the response body
+    assert "traceback" not in body_lower, "T-12.1-09: traceback must not appear in API response"
+    assert "exception" not in body_lower, "T-12.1-09: exception text must not appear in API response"
+    # No credential fields leaked
+    assert "access_token" not in body_lower
+    assert "refresh_token" not in body_lower
+    assert "credentials_enc" not in body_lower
+
+
+async def test_browse_cached_items_remain_owner_scoped_on_incomplete(async_client, db_session):
+    """Cached items returned after incomplete refresh must still be owner-scoped.
+
+    T-12.1-07: incomplete refresh must not return other users' items.
+    """
+    from storage.cloud_base import CloudListing
+    from db.models import CloudItem
+
+    auth_owner = await _create_user_and_token(db_session, role="user")
+    auth_other = await _create_user_and_token(db_session, role="user")
+
+    conn = await _create_cloud_connection(db_session, auth_owner["user"].id)
+
+    # Seed a durable item for the owner
+    item = CloudItem(
+        id=_uuid.uuid4(),
+        user_id=auth_owner["user"].id,
+        connection_id=conn.id,
+        provider_item_id="owner-item-001",
+        name="owner.pdf",
+        kind="file",
+        analysis_status="pending",
+        semantic_index_status="none",
+    )
+    db_session.add(item)
+    await db_session.commit()
+
+    mock_adapter = AsyncMock()
+    mock_adapter.list_folder = AsyncMock(return_value=CloudListing(items=(), complete=False))
+    mock_adapter.get_capabilities = AsyncMock(
+        return_value=_make_mock_adapter().get_capabilities.return_value
+    )
+
+    # Other user tries to browse owner's connection — must be blocked
+    resp_other = await async_client.get(
+        f"/api/cloud/connections/{conn.id}/items",
+        headers=auth_other["headers"],
+    )
+    assert resp_other.status_code == 404, (
+        "foreign user must still get IDOR 404 even during incomplete refresh"
+    )
+
+    # Owner can still browse and gets their cached item
+    with patch("api.cloud.browse.build_cloud_resource_adapter", return_value=mock_adapter):
+        resp_owner = await async_client.get(
+            f"/api/cloud/connections/{conn.id}/items",
+            headers=auth_owner["headers"],
+        )
+
+    assert resp_owner.status_code == 200
+    items_returned = resp_owner.json().get("items", [])
+    item_ids = [i["provider_item_id"] for i in items_returned]
+    assert "owner-item-001" in item_ids, "owner's cached item must be returned after incomplete refresh"
+
+
 # ── D-18: No byte download during list_folder ─────────────────────────────────
 
 async def test_no_byte_download_during_browse(async_client, db_session):