"""
Tests for cloud item metadata schema, model isolation, and reconciliation service.

Task 2 tests (model/schema/isolation/quota):
  - ORM model fields match migration schema expectations.
  - Metadata, extracted text, topic links, and semantic data persist without
    requiring an object key or byte cache row.
  - Identical provider item IDs coexist across different connections.
  - Foreign-owner queries do not return rows for other users.
  - Provider size never alters Quota.used_bytes.
  - Root folder state can be represented without a CloudItem parent row.

Task 3 tests (reconciliation service):
  - Repeated reconciliation creates no duplicate rows.
  - Rename/move updates metadata without changing the CloudItem UUID.
  - Incomplete or failed listing cannot mark unseen children deleted.
  - All queries include both owner and connection scope.
  - Idempotent repeated reconciliation.
"""
from __future__ import annotations

import uuid
from datetime import datetime, timezone
from typing import Optional

import pytest
import pytest_asyncio
from sqlalchemy import Text, event
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.pool import StaticPool
from sqlalchemy.dialects.postgresql import UUID

from db.models import Base, CloudItem, CloudItemTopic, CloudFolderState, CloudConnection, Topic, User, Quota
from storage.cloud_base import CloudListing, CloudResource, CloudCapability, STATE_SUPPORTED
from services.cloud_items import (
    resolve_owned_connection,
    list_cloud_children,
    upsert_cloud_item,
    reconcile_cloud_listing,
    get_or_create_folder_state,
    update_folder_state,
    CloudItemNotFound,
    ConnectionNotFound,
)


# ── SQLite test engine setup ──────────────────────────────────────────────────

@pytest_asyncio.fixture
async def db_session():
    """In-memory SQLite session with PostgreSQL-type shims.

    Patches INET/JSONB to Text for SQLite compatibility.
    UUID(as_uuid=True) is left as-is: SQLAlchemy renders it as CHAR(32) and
    the bind processor handles uuid.UUID ↔ 32-char hex automatically.
    """
    from sqlalchemy.dialects.sqlite.base import SQLiteTypeCompiler
    from sqlalchemy.dialects.postgresql import INET, JSONB

    _orig_visit_INET = getattr(SQLiteTypeCompiler, "visit_INET", None)
    _orig_visit_JSONB = getattr(SQLiteTypeCompiler, "visit_JSONB", None)

    def _visit_inet(self, type_, **kw):
        return "TEXT"

    def _visit_jsonb(self, type_, **kw):
        return "TEXT"

    SQLiteTypeCompiler.visit_INET = _visit_inet  # type: ignore[attr-defined]
    SQLiteTypeCompiler.visit_JSONB = _visit_jsonb  # type: ignore[attr-defined]

    engine = create_async_engine(
        "sqlite+aiosqlite:///:memory:",
        connect_args={"check_same_thread": False},
        poolclass=StaticPool,
    )

    try:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.create_all)

        session_factory = async_sessionmaker(engine, expire_on_commit=False)
        async with session_factory() as session:
            yield session
    finally:
        async with engine.begin() as conn:
            await conn.run_sync(Base.metadata.drop_all)
        await engine.dispose()
        if _orig_visit_INET is not None:
            SQLiteTypeCompiler.visit_INET = _orig_visit_INET  # type: ignore
        else:
            try:
                del SQLiteTypeCompiler.visit_INET  # type: ignore
            except AttributeError:
                pass
        if _orig_visit_JSONB is not None:
            SQLiteTypeCompiler.visit_JSONB = _orig_visit_JSONB  # type: ignore
        else:
            try:
                del SQLiteTypeCompiler.visit_JSONB  # type: ignore
            except AttributeError:
                pass


# ── Helpers ───────────────────────────────────────────────────────────────────

def _user_id() -> uuid.UUID:
    return uuid.uuid4()


def _conn_id() -> uuid.UUID:
    return uuid.uuid4()


def _item_id() -> uuid.UUID:
    return uuid.uuid4()


async def _make_user(session: AsyncSession, user_id: Optional[uuid.UUID] = None) -> uuid.UUID:
    uid = user_id or _user_id()
    u = User(
        id=uid,
        handle=f"user_{uid.hex[:8]}",
        email=f"{uid.hex[:8]}@example.com",
        password_hash="hash",
        role="user",
    )
    session.add(u)
    await session.flush()
    return uid


async def _make_connection(
    session: AsyncSession, user_id: uuid.UUID, conn_id: Optional[uuid.UUID] = None
) -> uuid.UUID:
    cid = conn_id or _conn_id()
    conn = CloudConnection(
        id=cid,
        user_id=user_id,
        provider="onedrive",
        display_name="My OneDrive",
        credentials_enc="enc",
        status="ACTIVE",
    )
    session.add(conn)
    await session.flush()
    return cid


def _cloud_resource(
    connection_id: uuid.UUID,
    user_id: uuid.UUID,
    provider_item_id: str = "item-001",
    name: str = "test.pdf",
    kind: str = "file",
    parent_ref: Optional[str] = None,
    size: Optional[int] = None,
    etag: Optional[str] = None,
) -> CloudResource:
    return CloudResource(
        id=uuid.uuid4(),
        provider_item_id=provider_item_id,
        connection_id=connection_id,
        user_id=user_id,
        name=name,
        kind=kind,
        parent_ref=parent_ref,
        size=size,
        etag=etag,
    )


# ── Task 2: model/schema tests ────────────────────────────────────────────────

@pytest.mark.asyncio
async def test_model_cloud_item_basic_fields(db_session: AsyncSession):
    """CloudItem persists metadata without requiring an object key."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)
    item_id = _item_id()

    item = CloudItem(
        id=item_id,
        user_id=uid,
        connection_id=cid,
        provider_item_id="driveitem-123",
        name="report.pdf",
        kind="file",
        content_type="application/pdf",
        provider_size=204800,
        etag="etag-v1",
    )
    db_session.add(item)
    await db_session.flush()
    await db_session.refresh(item)

    assert item.id == item_id
    assert item.provider_item_id == "driveitem-123"
    assert item.name == "report.pdf"
    assert item.kind == "file"
    assert item.provider_size == 204800
    assert item.analysis_status == "pending"
    assert item.semantic_index_status == "none"
    assert item.deleted_at is None


@pytest.mark.asyncio
async def test_model_cloud_item_extracted_text_without_object_key(db_session: AsyncSession):
    """Extracted text and semantic data persist with no MinIO object key field."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    item = CloudItem(
        id=_item_id(),
        user_id=uid,
        connection_id=cid,
        provider_item_id="item-text",
        name="notes.txt",
        kind="file",
        extracted_text="Sample extracted text",
        semantic_index_status="indexed",
        # No object_key field exists — provider bytes are not retained here
    )
    db_session.add(item)
    await db_session.flush()
    await db_session.refresh(item)

    assert item.extracted_text == "Sample extracted text"
    assert item.semantic_index_status == "indexed"
    # Confirm no object_key attribute at all
    assert not hasattr(item, "object_key")


@pytest.mark.asyncio
async def test_model_cloud_item_topics_no_document_row(db_session: AsyncSession):
    """CloudItemTopic links item to topic without requiring a Document row."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    topic = Topic(id=_item_id(), user_id=uid, name="Finance", description="", color="#ff0000")
    db_session.add(topic)
    await db_session.flush()

    item = CloudItem(
        id=_item_id(),
        user_id=uid,
        connection_id=cid,
        provider_item_id="item-topic",
        name="budget.xlsx",
        kind="file",
    )
    db_session.add(item)
    await db_session.flush()

    link = CloudItemTopic(cloud_item_id=item.id, topic_id=topic.id)
    db_session.add(link)
    await db_session.flush()

    # Verify link persisted
    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItemTopic).where(CloudItemTopic.cloud_item_id == item.id)
    )
    rows = result.scalars().all()
    assert len(rows) == 1
    assert rows[0].topic_id == topic.id


@pytest.mark.asyncio
async def test_model_schema_isolation_same_provider_item_different_connections(db_session: AsyncSession):
    """Identical provider_item_id can coexist across different connections."""
    uid = await _make_user(db_session)
    cid1 = await _make_connection(db_session, uid)
    cid2 = await _make_connection(db_session, uid)
    shared_pid = "shared-item-001"

    item1 = CloudItem(
        id=_item_id(), user_id=uid, connection_id=cid1,
        provider_item_id=shared_pid, name="file.pdf", kind="file",
    )
    item2 = CloudItem(
        id=_item_id(), user_id=uid, connection_id=cid2,
        provider_item_id=shared_pid, name="file.pdf", kind="file",
    )
    db_session.add_all([item1, item2])
    await db_session.flush()

    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItem).where(CloudItem.provider_item_id == shared_pid)
    )
    rows = result.scalars().all()
    assert len(rows) == 2


@pytest.mark.asyncio
async def test_model_schema_foreign_owner_query_returns_nothing(db_session: AsyncSession):
    """Items for user A are not returned when querying for user B."""
    uid_a = await _make_user(db_session)
    uid_b = await _make_user(db_session)
    cid_a = await _make_connection(db_session, uid_a)

    item = CloudItem(
        id=_item_id(), user_id=uid_a, connection_id=cid_a,
        provider_item_id="item-a", name="private.pdf", kind="file",
    )
    db_session.add(item)
    await db_session.flush()

    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItem).where(
            CloudItem.user_id == uid_b,
            CloudItem.provider_item_id == "item-a",
        )
    )
    assert result.scalars().first() is None


@pytest.mark.asyncio
async def test_model_quota_unchanged_after_item_upsert(db_session: AsyncSession):
    """Upserting a cloud item with provider_size must not alter Quota.used_bytes."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    quota = Quota(user_id=uid, limit_bytes=104857600, used_bytes=0)
    db_session.add(quota)
    await db_session.flush()

    item = CloudItem(
        id=_item_id(), user_id=uid, connection_id=cid,
        provider_item_id="large-file", name="video.mp4", kind="file",
        provider_size=5_000_000_000,  # 5 GB provider-reported size
    )
    db_session.add(item)
    await db_session.flush()
    await db_session.refresh(quota)

    # Quota must remain unchanged
    assert quota.used_bytes == 0


@pytest.mark.asyncio
async def test_model_folder_state_root_no_parent_item(db_session: AsyncSession):
    """Root folder state can be created with parent_ref='' without a CloudItem row."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    fs = CloudFolderState(
        id=_item_id(),
        user_id=uid,
        connection_id=cid,
        parent_ref="",  # root
        refresh_state="fresh",
    )
    db_session.add(fs)
    await db_session.flush()
    await db_session.refresh(fs)

    assert fs.parent_ref == ""
    assert fs.refresh_state == "fresh"
    assert fs.last_refreshed_at is None


# ── Task 3: reconciliation service tests ─────────────────────────────────────

@pytest.mark.asyncio
async def test_service_resolve_owned_connection_found(db_session: AsyncSession):
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    conn = await resolve_owned_connection(db_session, connection_id=cid, user_id=uid)
    assert conn.id == cid


@pytest.mark.asyncio
async def test_service_resolve_owned_connection_wrong_owner_raises(db_session: AsyncSession):
    uid_a = await _make_user(db_session)
    uid_b = await _make_user(db_session)
    cid = await _make_connection(db_session, uid_a)

    with pytest.raises(ConnectionNotFound):
        await resolve_owned_connection(db_session, connection_id=cid, user_id=uid_b)


@pytest.mark.asyncio
async def test_service_upsert_creates_item(db_session: AsyncSession):
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)
    resource = _cloud_resource(cid, uid, provider_item_id="new-001", name="doc.pdf")

    item = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
    assert item.provider_item_id == "new-001"
    assert item.name == "doc.pdf"


@pytest.mark.asyncio
async def test_service_upsert_idempotent_same_uuid(db_session: AsyncSession):
    """Repeated upsert with unchanged data creates no duplicate row."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)
    resource = _cloud_resource(cid, uid, provider_item_id="idem-001")

    item1 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
    item2 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)

    assert item1.id == item2.id  # same DocuVault UUID preserved


@pytest.mark.asyncio
async def test_service_upsert_rename_preserves_uuid(db_session: AsyncSession):
    """Rename updates name but keeps the same CloudItem UUID."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)
    resource = _cloud_resource(cid, uid, provider_item_id="rename-001", name="old.pdf")

    item1 = await upsert_cloud_item(db_session, user_id=uid, resource=resource)
    original_uuid = item1.id

    renamed = _cloud_resource(cid, uid, provider_item_id="rename-001", name="new.pdf")
    item2 = await upsert_cloud_item(db_session, user_id=uid, resource=renamed)

    assert item2.id == original_uuid
    assert item2.name == "new.pdf"


@pytest.mark.asyncio
async def test_service_list_children_owner_scoped(db_session: AsyncSession):
    """list_cloud_children only returns items for the correct (user, connection, parent)."""
    uid_a = await _make_user(db_session)
    uid_b = await _make_user(db_session)
    cid_a = await _make_connection(db_session, uid_a)

    res = _cloud_resource(cid_a, uid_a, provider_item_id="child-001", parent_ref="parent-ref")
    await upsert_cloud_item(db_session, user_id=uid_a, resource=res)

    # User A sees the item
    children_a = await list_cloud_children(
        db_session, user_id=uid_a, connection_id=cid_a, parent_ref="parent-ref"
    )
    assert len(children_a) == 1

    # User B does not
    children_b = await list_cloud_children(
        db_session, user_id=uid_b, connection_id=cid_a, parent_ref="parent-ref"
    )
    assert len(children_b) == 0


@pytest.mark.asyncio
async def test_service_reconcile_complete_marks_missing_deleted(db_session: AsyncSession):
    """Complete listing marks items not in the listing as deleted."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    # Pre-existing item
    existing = _cloud_resource(cid, uid, provider_item_id="keep-001", parent_ref="root")
    gone = _cloud_resource(cid, uid, provider_item_id="gone-001", parent_ref="root")
    await upsert_cloud_item(db_session, user_id=uid, resource=existing)
    await upsert_cloud_item(db_session, user_id=uid, resource=gone)

    # Reconcile with only "keep-001" in the complete listing
    listing = CloudListing(items=[existing], complete=True)
    await reconcile_cloud_listing(
        db_session, user_id=uid, connection_id=cid, parent_ref="root", listing=listing
    )

    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItem).where(CloudItem.provider_item_id == "gone-001")
    )
    gone_item = result.scalars().first()
    assert gone_item is not None
    assert gone_item.deleted_at is not None


@pytest.mark.asyncio
async def test_service_reconcile_incomplete_does_not_delete(db_session: AsyncSession):
    """Incomplete (partial/failed) listing must not mark unseen children deleted."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    item = _cloud_resource(cid, uid, provider_item_id="retain-001", parent_ref="root")
    await upsert_cloud_item(db_session, user_id=uid, resource=item)

    # Reconcile with empty incomplete listing
    listing = CloudListing(items=[], complete=False)
    await reconcile_cloud_listing(
        db_session, user_id=uid, connection_id=cid, parent_ref="root", listing=listing
    )

    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItem).where(CloudItem.provider_item_id == "retain-001")
    )
    retained = result.scalars().first()
    assert retained is not None
    assert retained.deleted_at is None


@pytest.mark.asyncio
async def test_service_folder_state_create_and_update(db_session: AsyncSession):
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    fs = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
    assert fs.refresh_state == "fresh"

    await update_folder_state(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref="",
        refresh_state="warning",
        error_code="token_expired",
        error_message="Re-authenticate to refresh.",
    )

    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == cid,
            CloudFolderState.parent_ref == "",
        )
    )
    updated = result.scalars().first()
    assert updated.refresh_state == "warning"
    assert updated.error_code == "token_expired"


@pytest.mark.asyncio
async def test_service_folder_state_idempotent(db_session: AsyncSession):
    """get_or_create is idempotent — no duplicate rows."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    fs1 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
    fs2 = await get_or_create_folder_state(db_session, user_id=uid, connection_id=cid, parent_ref="")
    assert fs1.id == fs2.id


# ── Task 3: refresh_cloud_folder Celery task tests ────────────────────────────

@pytest.mark.asyncio
async def test_refresh_cloud_folder_idempotent_reconciliation(db_session: AsyncSession):
    """Calling reconcile_cloud_listing twice yields no duplicate rows (idempotency check)."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    resource = _cloud_resource(cid, uid, provider_item_id="stable-001", name="stable.pdf")
    listing = CloudListing(items=[resource], complete=True)

    # First reconcile
    await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing)
    # Second reconcile with same data
    await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing)

    from sqlalchemy import select, func
    result = await db_session.execute(
        select(func.count()).select_from(CloudItem).where(
            CloudItem.provider_item_id == "stable-001",
            CloudItem.deleted_at.is_(None),
        )
    )
    count = result.scalar()
    assert count == 1, "Idempotent reconciliation must not create duplicate rows"


@pytest.mark.asyncio
async def test_refresh_cloud_folder_failed_refresh_retains_cached_rows(db_session: AsyncSession):
    """On provider failure, previously cached rows must not be deleted (D-13)."""
    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    # Seed durable items
    resource = _cloud_resource(cid, uid, provider_item_id="durable-001", name="important.pdf")
    complete_listing = CloudListing(items=[resource], complete=True)
    await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=complete_listing)

    # Simulate provider failure: incomplete listing with no items
    empty_failed_listing = CloudListing(items=[], complete=False)
    await reconcile_cloud_listing(db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=empty_failed_listing)

    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItem).where(
            CloudItem.provider_item_id == "durable-001",
            CloudItem.deleted_at.is_(None),
        )
    )
    retained = result.scalars().first()
    assert retained is not None, "Incomplete listing must not delete previously cached rows"


@pytest.mark.asyncio
async def test_refresh_cloud_folder_task_structure():
    """refresh_cloud_folder is a registered Celery task with correct queue routing."""
    from tasks.cloud_tasks import refresh_cloud_folder
    from celery_app import celery_app

    assert hasattr(refresh_cloud_folder, "delay"), "Task must have .delay() method"
    assert hasattr(refresh_cloud_folder, "apply_async"), "Task must have .apply_async() method"
    assert refresh_cloud_folder.name == "tasks.cloud_tasks.refresh_cloud_folder"
    assert refresh_cloud_folder.max_retries == 3

    routes = celery_app.conf.task_routes
    assert "tasks.cloud_tasks.*" in routes
    assert routes["tasks.cloud_tasks.*"]["queue"] == "documents"


@pytest.mark.asyncio
async def test_refresh_cloud_folder_no_byte_calls(db_session: AsyncSession):
    """reconcile_cloud_listing never calls get_object or any download method — D-18/CACHE-01."""
    from unittest.mock import AsyncMock, patch

    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)
    resource = _cloud_resource(cid, uid, provider_item_id="nobytes-001")
    listing = CloudListing(items=[resource], complete=True)

    mock_get_object = AsyncMock()
    with patch("db.models.CloudItem", wraps=CloudItem) as _patched:
        # reconcile_cloud_listing must not call any external byte-download method
        await reconcile_cloud_listing(
            db_session, user_id=uid, connection_id=cid, parent_ref=None, listing=listing
        )
        mock_get_object.assert_not_called()


# ── Plan 02: Truthful freshness / listing-gate semantics ──────────────────────

@pytest.mark.asyncio
async def test_incomplete_listing_never_marks_folder_fresh(db_session: AsyncSession):
    """apply_listing_and_finalize must NOT set refresh_state=fresh when listing.complete=False.

    RED: cloud_items.py currently sets refresh_state="fresh" unconditionally after
    reconcile_cloud_listing.  This test must fail until the gate is in place.
    """
    from services.cloud_items import apply_listing_and_finalize

    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    # Seed an existing successful refresh
    await update_folder_state(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref="",
        refresh_state="fresh",
        last_refreshed_at=datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc),
    )
    await db_session.flush()

    incomplete_listing = CloudListing(items=(), complete=False)
    result = await apply_listing_and_finalize(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref=None,
        listing=incomplete_listing,
    )

    from sqlalchemy import select
    fs = (await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == cid,
            CloudFolderState.parent_ref == "",
        )
    )).scalars().first()

    assert fs is not None
    assert fs.refresh_state != "fresh", (
        "incomplete listing (complete=False) must never set refresh_state=fresh"
    )


@pytest.mark.asyncio
async def test_incomplete_listing_retains_cached_rows_and_last_success(db_session: AsyncSession):
    """Incomplete refresh must retain cached items AND preserve previous last_refreshed_at.

    T-12.1-07: cached metadata survives ambiguous/incomplete provider responses.
    T-12.1-08: last known-good timestamp is not overwritten on incomplete refresh.
    """
    from services.cloud_items import apply_listing_and_finalize

    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)
    prior_ts = datetime(2026, 6, 1, 12, 0, 0, tzinfo=timezone.utc)

    # Seed durable items from a prior successful refresh
    resource = _cloud_resource(cid, uid, provider_item_id="keeper-001", name="keep.pdf")
    await upsert_cloud_item(db_session, user_id=uid, resource=resource)

    await update_folder_state(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref="",
        refresh_state="fresh",
        last_refreshed_at=prior_ts,
    )
    await db_session.flush()

    # Incomplete listing with no items — must not delete keeper-001
    incomplete_listing = CloudListing(items=(), complete=False)
    await apply_listing_and_finalize(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref=None,
        listing=incomplete_listing,
    )

    # Cached items must survive
    from sqlalchemy import select
    result = await db_session.execute(
        select(CloudItem).where(
            CloudItem.provider_item_id == "keeper-001",
            CloudItem.deleted_at.is_(None),
        )
    )
    assert result.scalars().first() is not None, "incomplete listing must not delete cached items"

    # Prior successful timestamp must be preserved
    fs = (await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == cid,
            CloudFolderState.parent_ref == "",
        )
    )).scalars().first()
    assert fs is not None
    # SQLite stores naive datetimes; strip tzinfo for comparison
    stored_ts = fs.last_refreshed_at
    if stored_ts is not None and stored_ts.tzinfo is None:
        stored_ts = stored_ts.replace(tzinfo=timezone.utc)
    assert stored_ts == prior_ts, (
        f"incomplete listing must not overwrite last_refreshed_at: {fs.last_refreshed_at!r} != {prior_ts!r}"
    )


@pytest.mark.asyncio
async def test_complete_empty_listing_is_authoritative_and_fresh(db_session: AsyncSession):
    """A provider-confirmed empty listing (complete=True, items=[]) is authoritative and fresh.

    T-12.1-07: reconcile deletion runs only when complete=True.
    A complete empty listing soft-deletes all unseen children AND marks folder fresh.
    """
    from services.cloud_items import apply_listing_and_finalize

    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    # Pre-existing item that the provider no longer reports
    gone = _cloud_resource(cid, uid, provider_item_id="gone-002", name="deleted.pdf")
    await upsert_cloud_item(db_session, user_id=uid, resource=gone)

    complete_empty_listing = CloudListing(items=(), complete=True)
    result = await apply_listing_and_finalize(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref=None,
        listing=complete_empty_listing,
    )

    from sqlalchemy import select

    # Soft-deletion must have occurred
    gone_row = (await db_session.execute(
        select(CloudItem).where(CloudItem.provider_item_id == "gone-002")
    )).scalars().first()
    assert gone_row is not None
    assert gone_row.deleted_at is not None, "complete empty listing must soft-delete unseen items"

    # Folder must be marked fresh
    fs = (await db_session.execute(
        select(CloudFolderState).where(
            CloudFolderState.connection_id == cid,
            CloudFolderState.parent_ref == "",
        )
    )).scalars().first()
    assert fs is not None
    assert fs.refresh_state == "fresh", "complete empty listing must mark folder fresh"


@pytest.mark.asyncio
async def test_partial_items_upsert_without_deleting_unseen_children(db_session: AsyncSession):
    """Incomplete listing upserts seen items but leaves unseen children intact.

    T-12.1-07: soft-deletion of missing children requires complete=True.
    Partially-seen items (e.g. first page only) must not delete un-paged items.
    """
    from services.cloud_items import apply_listing_and_finalize

    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    # Two pre-existing items
    seen = _cloud_resource(cid, uid, provider_item_id="seen-001", name="seen.pdf")
    unseen = _cloud_resource(cid, uid, provider_item_id="unseen-002", name="hidden.pdf")
    await upsert_cloud_item(db_session, user_id=uid, resource=seen)
    await upsert_cloud_item(db_session, user_id=uid, resource=unseen)

    # Incomplete listing: only "seen-001" came back in partial page
    partial_listing = CloudListing(items=[seen], complete=False)
    await apply_listing_and_finalize(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref=None,
        listing=partial_listing,
    )

    from sqlalchemy import select
    unseen_row = (await db_session.execute(
        select(CloudItem).where(
            CloudItem.provider_item_id == "unseen-002",
            CloudItem.deleted_at.is_(None),
        )
    )).scalars().first()
    assert unseen_row is not None, "unseen items must survive when listing is incomplete"


@pytest.mark.asyncio
async def test_apply_listing_returns_warning_for_complete_false(db_session: AsyncSession):
    """apply_listing_and_finalize returns a result with warning when listing.complete=False.

    T-12.1-06: false fresh state must never be returned to the API caller.
    The result must carry a controlled warning code (incomplete_listing) and no
    raw provider exception text.
    """
    from services.cloud_items import apply_listing_and_finalize

    uid = await _make_user(db_session)
    cid = await _make_connection(db_session, uid)

    incomplete_listing = CloudListing(items=(), complete=False)
    result = await apply_listing_and_finalize(
        db_session,
        user_id=uid,
        connection_id=cid,
        parent_ref=None,
        listing=incomplete_listing,
    )

    # The result must signal warning, not fresh
    assert result.is_fresh is False, "incomplete listing must not report is_fresh=True"
    assert result.warning_code == "incomplete_listing", (
        f"expected warning_code='incomplete_listing', got {result.warning_code!r}"
    )
    # Warning message must be a stable controlled string (no traceback, no URLs, no credentials)
    assert result.warning_message is not None
    assert len(result.warning_message) > 0
    dangerous_patterns = ["traceback", "http://", "https://", "password", "token", "credentials"]
    for pattern in dangerous_patterns:
        assert pattern not in result.warning_message.lower(), (
            f"warning_message must not contain sensitive content: {pattern!r}"
        )