tests/test_models.py

from __future__ import annotations

from sqlalchemy import select

from app.models import ChatMessage, ChatSession, KnowledgeChunk, KnowledgeSource


async def test_chat_session_defaults_and_messages(db_session):
    session = ChatSession(shop="x.myshopify.com")
    db_session.add(session)
    await db_session.flush()

    assert session.id  # uuid generated
    assert session.verified is False
    assert session.verify_attempts == 0
    assert session.locked_until is None

    sid = session.id
    db_session.add(ChatMessage(session_id=sid, role="user", content="hola"))
    await db_session.commit()
    db_session.expunge_all()

    # lazy="selectin" loads messages during the query, safe under async.
    loaded = (
        await db_session.execute(select(ChatSession).where(ChatSession.id == sid))
    ).scalar_one()
    assert len(loaded.messages) == 1
    assert loaded.messages[0].role == "user"


async def test_knowledge_source_chunks_cascade(db_session):
    src = KnowledgeSource(kind="url", name="docs", location="https://e.com")
    src.chunks.append(KnowledgeChunk(ordinal=0, text="hi", embedding=[0.1, 0.2, 0.3]))
    db_session.add(src)
    await db_session.flush()
    assert src.status == "pending"

    await db_session.delete(src)
    await db_session.flush()
    remaining = (await db_session.execute(select(KnowledgeChunk))).scalars().all()
    assert remaining == []