File size: 1,396 Bytes
187966e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 | 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 == []
|