wehub-resource-sync
fed8b2eed7
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Backend release / release (push) Has been cancelled
Bandit Security Scan / bandit_scan (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / manifest (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / manifest (push) Has been cancelled
Python linting / ruff (push) Has been cancelled
Run python tests with pytest / Run tests and count coverage (3.12) (push) Has been cancelled
React Widget Build / build (push) Has been cancelled
874 lines
31 KiB
Python
874 lines
31 KiB
Python
"""Tests for application/api/answer/services/conversation_service.py."""
|
|
|
|
from contextlib import contextmanager
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
|
|
@contextmanager
|
|
def _patch_db(conn):
|
|
@contextmanager
|
|
def _yield():
|
|
yield conn
|
|
|
|
with patch(
|
|
"application.api.answer.services.conversation_service.db_session",
|
|
_yield,
|
|
), patch(
|
|
"application.api.answer.services.conversation_service.db_readonly",
|
|
_yield,
|
|
):
|
|
yield
|
|
|
|
|
|
class TestConversationServiceGet:
|
|
def test_returns_none_when_no_conversation_id(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
assert ConversationService().get_conversation("", "u") is None
|
|
|
|
def test_returns_none_when_no_user_id(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
assert (
|
|
ConversationService().get_conversation(
|
|
"00000000-0000-0000-0000-000000000001", ""
|
|
)
|
|
is None
|
|
)
|
|
|
|
def test_returns_none_when_not_found(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with _patch_db(pg_conn):
|
|
got = ConversationService().get_conversation(
|
|
"00000000-0000-0000-0000-000000000000", "u",
|
|
)
|
|
assert got is None
|
|
|
|
def test_returns_conversation_with_messages(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-get"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="hi")
|
|
conv_id = str(conv["id"])
|
|
repo.append_message(
|
|
conv_id, {"prompt": "q1", "response": "r1"}
|
|
)
|
|
|
|
with _patch_db(pg_conn):
|
|
got = ConversationService().get_conversation(conv_id, user)
|
|
assert got is not None
|
|
assert got["_id"] == conv_id
|
|
assert len(got["queries"]) == 1
|
|
|
|
def test_handles_exception_returns_none(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
@contextmanager
|
|
def _broken():
|
|
raise RuntimeError("boom")
|
|
yield
|
|
|
|
with patch(
|
|
"application.api.answer.services.conversation_service.db_readonly",
|
|
_broken,
|
|
):
|
|
got = ConversationService().get_conversation("abc", "u")
|
|
assert got is None
|
|
|
|
|
|
class TestConversationServiceSave:
|
|
def test_raises_for_none_token(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with pytest.raises(ValueError):
|
|
ConversationService().save_conversation(
|
|
conversation_id=None,
|
|
question="q", response="r", thought="", sources=[],
|
|
tool_calls=[], llm=MagicMock(), model_id="gpt",
|
|
decoded_token=None,
|
|
)
|
|
|
|
def test_raises_when_no_user_in_token(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with pytest.raises(ValueError):
|
|
ConversationService().save_conversation(
|
|
conversation_id=None,
|
|
question="q", response="r", thought="", sources=[],
|
|
tool_calls=[], llm=MagicMock(), model_id="gpt",
|
|
decoded_token={},
|
|
)
|
|
|
|
def test_creates_new_conversation(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-save-new"
|
|
mock_llm = MagicMock()
|
|
mock_llm.gen.return_value = "Title"
|
|
|
|
with _patch_db(pg_conn):
|
|
conv_id = ConversationService().save_conversation(
|
|
conversation_id=None,
|
|
question="q1", response="r1", thought="",
|
|
sources=[{"text": "x" * 2000}], tool_calls=[],
|
|
llm=mock_llm, model_id="gpt-4",
|
|
decoded_token={"sub": user},
|
|
)
|
|
got = ConversationsRepository(pg_conn).get_any(conv_id, user)
|
|
assert got is not None
|
|
assert got["name"] == "Title"
|
|
|
|
def test_appends_message_to_existing(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-append"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="existing")
|
|
conv_id = str(conv["id"])
|
|
|
|
with _patch_db(pg_conn):
|
|
got = ConversationService().save_conversation(
|
|
conversation_id=conv_id,
|
|
question="q-new", response="r-new", thought="",
|
|
sources=[], tool_calls=[],
|
|
llm=MagicMock(), model_id="gpt-4",
|
|
decoded_token={"sub": user},
|
|
)
|
|
assert got == conv_id
|
|
messages = repo.get_messages(conv_id)
|
|
assert any(m["prompt"] == "q-new" for m in messages)
|
|
|
|
def test_updates_message_at_index(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-update-at"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="edit")
|
|
conv_id = str(conv["id"])
|
|
repo.append_message(conv_id, {"prompt": "old", "response": "old"})
|
|
|
|
with _patch_db(pg_conn):
|
|
got = ConversationService().save_conversation(
|
|
conversation_id=conv_id,
|
|
question="updated", response="reply", thought="",
|
|
sources=[], tool_calls=[],
|
|
llm=MagicMock(), model_id="gpt-4",
|
|
decoded_token={"sub": user},
|
|
index=0,
|
|
)
|
|
assert got == conv_id
|
|
messages = repo.get_messages(conv_id)
|
|
assert messages[0]["prompt"] == "updated"
|
|
|
|
def test_raises_when_conversation_missing(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with _patch_db(pg_conn), pytest.raises(ValueError):
|
|
ConversationService().save_conversation(
|
|
conversation_id="00000000-0000-0000-0000-000000000000",
|
|
question="q", response="r", thought="",
|
|
sources=[], tool_calls=[],
|
|
llm=MagicMock(), model_id="gpt",
|
|
decoded_token={"sub": "u"},
|
|
)
|
|
|
|
def test_save_with_empty_llm_title_falls_back(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-empty-title"
|
|
mock_llm = MagicMock()
|
|
mock_llm.gen.return_value = ""
|
|
|
|
with _patch_db(pg_conn):
|
|
conv_id = ConversationService().save_conversation(
|
|
conversation_id=None,
|
|
question="q-fallback", response="r", thought="",
|
|
sources=[], tool_calls=[],
|
|
llm=mock_llm, model_id="gpt-4",
|
|
decoded_token={"sub": user},
|
|
)
|
|
got = ConversationsRepository(pg_conn).get_any(conv_id, user)
|
|
assert got["name"] == "q-fallback"
|
|
|
|
|
|
class TestSaveUserQuestion:
|
|
def test_creates_conversation_and_reserves_message(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
TERMINATED_RESPONSE_PLACEHOLDER,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-wal-new"
|
|
with _patch_db(pg_conn):
|
|
result = ConversationService().save_user_question(
|
|
conversation_id=None,
|
|
question="what is python?",
|
|
decoded_token={"sub": user},
|
|
)
|
|
assert result["conversation_id"]
|
|
assert result["message_id"]
|
|
assert result["request_id"]
|
|
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.get_any(result["conversation_id"], user)
|
|
assert conv is not None
|
|
messages = repo.get_messages(result["conversation_id"])
|
|
assert len(messages) == 1
|
|
assert messages[0]["status"] == "pending"
|
|
assert messages[0]["prompt"] == "what is python?"
|
|
assert messages[0]["response"] == TERMINATED_RESPONSE_PLACEHOLDER
|
|
assert messages[0]["request_id"] == result["request_id"]
|
|
|
|
def test_appends_to_existing_conversation(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-wal-existing"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="hi")
|
|
conv_id = str(conv["id"])
|
|
|
|
with _patch_db(pg_conn):
|
|
result = ConversationService().save_user_question(
|
|
conversation_id=conv_id,
|
|
question="follow-up",
|
|
decoded_token={"sub": user},
|
|
)
|
|
assert result["conversation_id"] == conv_id
|
|
msgs = repo.get_messages(conv_id)
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["prompt"] == "follow-up"
|
|
|
|
def test_raises_when_token_missing(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with pytest.raises(ValueError):
|
|
ConversationService().save_user_question(
|
|
conversation_id=None,
|
|
question="q",
|
|
decoded_token=None,
|
|
)
|
|
|
|
def test_regenerate_at_index_replaces_old_message(self, pg_conn):
|
|
"""Regenerate at ``index`` truncates the old message *and
|
|
everything after* before reserving the placeholder, so the new
|
|
WAL row lands at ``position=index`` rather than appending at
|
|
the end. Pre-fix the WAL path appended unconditionally and the
|
|
old answer survived alongside the regenerated one.
|
|
"""
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-wal-regen"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="regen-test")
|
|
conv_id = str(conv["id"])
|
|
|
|
# Seed five completed messages at positions 0..4.
|
|
for i in range(5):
|
|
repo.append_message(
|
|
conv_id,
|
|
{
|
|
"prompt": f"q{i}",
|
|
"response": f"a{i}",
|
|
"thought": "",
|
|
"sources": [],
|
|
"tool_calls": [],
|
|
"metadata": {},
|
|
},
|
|
)
|
|
seeded = repo.get_messages(conv_id)
|
|
assert len(seeded) == 5
|
|
assert [m["position"] for m in seeded] == [0, 1, 2, 3, 4]
|
|
|
|
with _patch_db(pg_conn):
|
|
result = ConversationService().save_user_question(
|
|
conversation_id=conv_id,
|
|
question="q3-regen",
|
|
decoded_token={"sub": user},
|
|
index=3,
|
|
)
|
|
|
|
msgs = repo.get_messages(conv_id)
|
|
# Positions 0,1,2 from the seed plus the new placeholder at 3.
|
|
assert [m["position"] for m in msgs] == [0, 1, 2, 3]
|
|
# The placeholder carries the regenerated prompt.
|
|
regen = next(m for m in msgs if m["position"] == 3)
|
|
assert regen["prompt"] == "q3-regen"
|
|
assert regen["status"] == "pending"
|
|
assert str(regen["id"]) == result["message_id"]
|
|
# The old answer at index 3 is gone.
|
|
assert not any(m["response"] == "a3" for m in msgs)
|
|
# And anything after index 3 was truncated.
|
|
assert not any(m["prompt"] == "q4" for m in msgs)
|
|
|
|
def test_regenerate_at_index_zero_truncates_everything(self, pg_conn):
|
|
"""``index=0`` is a valid edge: it should drop every prior
|
|
message and reseat the placeholder at position 0.
|
|
"""
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-wal-regen-zero"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="regen-zero")
|
|
conv_id = str(conv["id"])
|
|
for i in range(3):
|
|
repo.append_message(
|
|
conv_id,
|
|
{
|
|
"prompt": f"old-{i}",
|
|
"response": f"old-a-{i}",
|
|
"thought": "",
|
|
"sources": [],
|
|
"tool_calls": [],
|
|
"metadata": {},
|
|
},
|
|
)
|
|
|
|
with _patch_db(pg_conn):
|
|
ConversationService().save_user_question(
|
|
conversation_id=conv_id,
|
|
question="fresh-from-start",
|
|
decoded_token={"sub": user},
|
|
index=0,
|
|
)
|
|
|
|
msgs = repo.get_messages(conv_id)
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["position"] == 0
|
|
assert msgs[0]["prompt"] == "fresh-from-start"
|
|
|
|
def test_regenerate_index_ignored_without_conversation_id(self, pg_conn):
|
|
"""``index`` only makes sense against an existing conversation;
|
|
the create-then-reserve path silently treats it as a no-op
|
|
rather than truncating a freshly-created conversation.
|
|
"""
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-wal-regen-no-conv"
|
|
with _patch_db(pg_conn):
|
|
result = ConversationService().save_user_question(
|
|
conversation_id=None,
|
|
question="brand new q",
|
|
decoded_token={"sub": user},
|
|
index=2,
|
|
)
|
|
|
|
repo = ConversationsRepository(pg_conn)
|
|
msgs = repo.get_messages(result["conversation_id"])
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["position"] == 0
|
|
assert msgs[0]["prompt"] == "brand new q"
|
|
|
|
def test_raises_when_conversation_unauthorized(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with _patch_db(pg_conn), pytest.raises(ValueError):
|
|
ConversationService().save_user_question(
|
|
conversation_id="00000000-0000-0000-0000-000000000000",
|
|
question="q",
|
|
decoded_token={"sub": "u"},
|
|
)
|
|
|
|
|
|
class TestFinalizeMessage:
|
|
def test_finalizes_complete(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-fin-ok"
|
|
with _patch_db(pg_conn):
|
|
svc = ConversationService()
|
|
res = svc.save_user_question(
|
|
conversation_id=None,
|
|
question="q",
|
|
decoded_token={"sub": user},
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
MessageUpdateOutcome,
|
|
)
|
|
assert svc.finalize_message(
|
|
res["message_id"],
|
|
"real answer",
|
|
thought="thinking",
|
|
sources=[{"text": "x" * 2000, "title": "doc"}],
|
|
tool_calls=[{"name": "search"}],
|
|
model_id="gpt-4",
|
|
metadata={"foo": "bar"},
|
|
status="complete",
|
|
) is MessageUpdateOutcome.UPDATED
|
|
|
|
msgs = ConversationsRepository(pg_conn).get_messages(
|
|
res["conversation_id"],
|
|
)
|
|
assert msgs[0]["response"] == "real answer"
|
|
assert msgs[0]["status"] == "complete"
|
|
assert msgs[0]["thought"] == "thinking"
|
|
assert msgs[0]["model_id"] == "gpt-4"
|
|
# source text trimmed to 1000 chars at finalize time
|
|
assert len(msgs[0]["sources"][0]["text"]) == 1000
|
|
assert msgs[0]["metadata"]["foo"] == "bar"
|
|
|
|
def test_finalizes_failed_records_error(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-fin-fail"
|
|
with _patch_db(pg_conn):
|
|
svc = ConversationService()
|
|
res = svc.save_user_question(
|
|
conversation_id=None,
|
|
question="q",
|
|
decoded_token={"sub": user},
|
|
)
|
|
err = RuntimeError("provider down")
|
|
from application.storage.db.repositories.conversations import (
|
|
MessageUpdateOutcome,
|
|
)
|
|
assert svc.finalize_message(
|
|
res["message_id"],
|
|
"fallback text",
|
|
status="failed",
|
|
error=err,
|
|
) is MessageUpdateOutcome.UPDATED
|
|
|
|
msgs = ConversationsRepository(pg_conn).get_messages(
|
|
res["conversation_id"],
|
|
)
|
|
assert msgs[0]["status"] == "failed"
|
|
assert msgs[0]["metadata"]["error"] == "RuntimeError: provider down"
|
|
|
|
def test_finalize_flips_executed_tool_calls(self, pg_conn):
|
|
"""finalize_message must mark tool_call_attempts.status='executed'
|
|
rows as 'confirmed' for the same message_id."""
|
|
from sqlalchemy import text as sql_text
|
|
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
user = "u-fin-tools"
|
|
with _patch_db(pg_conn):
|
|
svc = ConversationService()
|
|
res = svc.save_user_question(
|
|
conversation_id=None,
|
|
question="q",
|
|
decoded_token={"sub": user},
|
|
)
|
|
pg_conn.execute(
|
|
sql_text(
|
|
"INSERT INTO tool_call_attempts "
|
|
"(call_id, message_id, tool_name, action_name, arguments, status) "
|
|
"VALUES (:cid, CAST(:mid AS uuid), 't', 'a', '{}'::jsonb, 'executed')"
|
|
),
|
|
{"cid": "c1", "mid": res["message_id"]},
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
MessageUpdateOutcome,
|
|
)
|
|
assert svc.finalize_message(
|
|
res["message_id"], "ans", status="complete",
|
|
) is MessageUpdateOutcome.UPDATED
|
|
|
|
status = pg_conn.execute(
|
|
sql_text("SELECT status FROM tool_call_attempts WHERE call_id = :cid"),
|
|
{"cid": "c1"},
|
|
).scalar()
|
|
assert status == "confirmed"
|
|
|
|
def test_finalize_returns_not_found_for_unknown_message(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
MessageUpdateOutcome,
|
|
)
|
|
with _patch_db(pg_conn):
|
|
assert ConversationService().finalize_message(
|
|
"00000000-0000-0000-0000-000000000000",
|
|
"x",
|
|
status="complete",
|
|
) is MessageUpdateOutcome.NOT_FOUND
|
|
|
|
def test_finalize_rolls_back_tool_call_confirm_on_message_update_failure(
|
|
self, pg_conn
|
|
):
|
|
"""Atomicity: if ``update_message_by_id`` raises after the
|
|
tool_call_attempts confirm ran on the same connection, the
|
|
confirm rolls back with the rest of the transaction. The
|
|
``pg_conn`` fixture pins one connection inside an outer
|
|
rolled-back transaction; we patch ``db_session`` to wrap each
|
|
call in a SAVEPOINT so the production-code ``with`` block
|
|
actually rolls back when the message-update raises.
|
|
"""
|
|
from contextlib import contextmanager
|
|
|
|
from sqlalchemy import text as sql_text
|
|
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories import conversations as conv_module
|
|
|
|
user = "u-fin-rollback"
|
|
|
|
@contextmanager
|
|
def _savepoint_session():
|
|
nested = pg_conn.begin_nested()
|
|
try:
|
|
yield pg_conn
|
|
nested.commit()
|
|
except Exception:
|
|
nested.rollback()
|
|
raise
|
|
|
|
with patch(
|
|
"application.api.answer.services.conversation_service.db_session",
|
|
_savepoint_session,
|
|
), patch(
|
|
"application.api.answer.services.conversation_service.db_readonly",
|
|
_savepoint_session,
|
|
):
|
|
svc = ConversationService()
|
|
res = svc.save_user_question(
|
|
conversation_id=None,
|
|
question="q",
|
|
decoded_token={"sub": user},
|
|
)
|
|
pg_conn.execute(
|
|
sql_text(
|
|
"INSERT INTO tool_call_attempts "
|
|
"(call_id, message_id, tool_name, action_name, "
|
|
"arguments, status) VALUES (:cid, CAST(:mid AS uuid), "
|
|
"'t', 'a', '{}'::jsonb, 'executed')"
|
|
),
|
|
{"cid": "rb-1", "mid": res["message_id"]},
|
|
)
|
|
original = conv_module.ConversationsRepository.update_message_by_id
|
|
|
|
def boom(self, *args, **kwargs):
|
|
_ = (args, kwargs)
|
|
raise RuntimeError("simulated message-update failure")
|
|
|
|
conv_module.ConversationsRepository.update_message_by_id = boom
|
|
try:
|
|
with pytest.raises(RuntimeError):
|
|
svc.finalize_message(
|
|
res["message_id"], "answer", status="complete",
|
|
)
|
|
finally:
|
|
conv_module.ConversationsRepository.update_message_by_id = original
|
|
|
|
# The tool_call confirm rolled back: row stays at ``executed``.
|
|
status = pg_conn.execute(
|
|
sql_text(
|
|
"SELECT status FROM tool_call_attempts WHERE call_id = :cid"
|
|
),
|
|
{"cid": "rb-1"},
|
|
).scalar()
|
|
assert status == "executed"
|
|
msg_status = pg_conn.execute(
|
|
sql_text(
|
|
"SELECT status FROM conversation_messages "
|
|
"WHERE id = CAST(:mid AS uuid)"
|
|
),
|
|
{"mid": res["message_id"]},
|
|
).scalar()
|
|
assert msg_status == "pending"
|
|
|
|
def test_finalize_generates_title_when_provided(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-fin-title"
|
|
mock_llm = MagicMock()
|
|
mock_llm.gen.return_value = "Short Title"
|
|
with _patch_db(pg_conn):
|
|
svc = ConversationService()
|
|
res = svc.save_user_question(
|
|
conversation_id=None,
|
|
question="long question that becomes the fallback name",
|
|
decoded_token={"sub": user},
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
MessageUpdateOutcome,
|
|
)
|
|
assert svc.finalize_message(
|
|
res["message_id"],
|
|
"answer",
|
|
status="complete",
|
|
title_inputs={
|
|
"llm": mock_llm,
|
|
"question": "long question that becomes the fallback name",
|
|
"response": "answer",
|
|
"model_id": "gpt-4",
|
|
"fallback_name": (
|
|
"long question that becomes the fallback name"[:50]
|
|
),
|
|
},
|
|
) is MessageUpdateOutcome.UPDATED
|
|
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.get_any(res["conversation_id"], user)
|
|
assert conv["name"] == "Short Title"
|
|
|
|
|
|
class TestSaveUserQuestionFinalizeFailedFlow:
|
|
"""LLM fails immediately; question stays queryable with status='failed' + error metadata."""
|
|
|
|
def test_failed_llm_leaves_question_persisted(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-acceptance"
|
|
with _patch_db(pg_conn):
|
|
svc = ConversationService()
|
|
# Simulates the WAL pre-persist before LLM call.
|
|
res = svc.save_user_question(
|
|
conversation_id=None,
|
|
question="why did this fail?",
|
|
decoded_token={"sub": user},
|
|
)
|
|
# Simulates the LLM raising immediately, caught by complete_stream.
|
|
try:
|
|
raise RuntimeError("upstream 503")
|
|
except RuntimeError as e:
|
|
svc.finalize_message(
|
|
res["message_id"],
|
|
"",
|
|
status="failed",
|
|
error=e,
|
|
)
|
|
|
|
msgs = ConversationsRepository(pg_conn).get_messages(
|
|
res["conversation_id"],
|
|
)
|
|
assert len(msgs) == 1
|
|
assert msgs[0]["prompt"] == "why did this fail?"
|
|
assert msgs[0]["status"] == "failed"
|
|
assert "RuntimeError" in msgs[0]["metadata"]["error"]
|
|
assert "upstream 503" in msgs[0]["metadata"]["error"]
|
|
|
|
|
|
class TestCompressionMetadata:
|
|
def test_update_compression_metadata(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-compress"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="c")
|
|
conv_id = str(conv["id"])
|
|
|
|
with _patch_db(pg_conn):
|
|
ConversationService().update_compression_metadata(
|
|
conv_id,
|
|
{
|
|
"timestamp": datetime.now(timezone.utc),
|
|
"compressed_summary": "summary",
|
|
"model_used": "gpt",
|
|
},
|
|
)
|
|
|
|
def test_update_compression_raises_on_error(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
@contextmanager
|
|
def _broken():
|
|
raise RuntimeError("boom")
|
|
yield
|
|
|
|
with patch(
|
|
"application.api.answer.services.conversation_service.db_session",
|
|
_broken,
|
|
), pytest.raises(RuntimeError):
|
|
ConversationService().update_compression_metadata("abc", {})
|
|
|
|
def test_append_compression_message_skips_empty_summary(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
with _patch_db(pg_conn):
|
|
ConversationService().append_compression_message(
|
|
"any-id", {"compressed_summary": ""}
|
|
)
|
|
|
|
def test_append_compression_message_appends_summary(self, pg_conn):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
from application.storage.db.repositories.conversations import (
|
|
ConversationsRepository,
|
|
)
|
|
|
|
user = "u-append-cs"
|
|
repo = ConversationsRepository(pg_conn)
|
|
conv = repo.create(user, name="c")
|
|
conv_id = str(conv["id"])
|
|
|
|
with _patch_db(pg_conn):
|
|
ConversationService().append_compression_message(
|
|
conv_id,
|
|
{
|
|
"compressed_summary": "A summary",
|
|
"timestamp": datetime.now(timezone.utc),
|
|
"model_used": "gpt-4",
|
|
},
|
|
)
|
|
|
|
messages = repo.get_messages(conv_id)
|
|
assert any(m["response"] == "A summary" for m in messages)
|
|
|
|
def test_append_compression_message_swallows_error(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
@contextmanager
|
|
def _broken():
|
|
raise RuntimeError("boom")
|
|
yield
|
|
|
|
with patch(
|
|
"application.api.answer.services.conversation_service.db_session",
|
|
_broken,
|
|
):
|
|
ConversationService().append_compression_message(
|
|
"abc", {"compressed_summary": "x"}
|
|
)
|
|
|
|
def test_get_compression_metadata_returns_none_for_missing(
|
|
self, pg_conn,
|
|
):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
with _patch_db(pg_conn):
|
|
got = ConversationService().get_compression_metadata(
|
|
"00000000-0000-0000-0000-000000000000"
|
|
)
|
|
assert got is None
|
|
|
|
def test_get_compression_metadata_non_uuid_does_not_raise(
|
|
self, pg_conn, caplog,
|
|
):
|
|
"""Non-UUID ids (legacy Mongo ObjectIds with no legacy_mongo_id row)
|
|
must return None without hitting ``CAST(:id AS uuid)`` — which
|
|
would raise and pollute logs with a stack trace every call."""
|
|
import logging
|
|
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
with caplog.at_level(logging.ERROR):
|
|
with _patch_db(pg_conn):
|
|
got = ConversationService().get_compression_metadata(
|
|
"507f1f77bcf86cd799439011"
|
|
)
|
|
assert got is None
|
|
assert not any(
|
|
"Error getting compression metadata" in r.message for r in caplog.records
|
|
)
|
|
|
|
def test_get_compression_metadata_handles_exception(self):
|
|
from application.api.answer.services.conversation_service import (
|
|
ConversationService,
|
|
)
|
|
|
|
@contextmanager
|
|
def _broken():
|
|
raise RuntimeError("boom")
|
|
yield
|
|
|
|
with patch(
|
|
"application.api.answer.services.conversation_service.db_readonly",
|
|
_broken,
|
|
):
|
|
got = ConversationService().get_compression_metadata("abc")
|
|
assert got is None
|