wehub-resource-sync
e4dcfc49aa
Tests / Import Check (Python 3.13) (push) Has been cancelled
Tests / Import Check (Python 3.14) (push) Has been cancelled
Tests / Python Tests (Python 3.11) (push) Has been cancelled
Tests / Python Tests (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.14) (push) Has been cancelled
Tests / Test Summary (push) Has been cancelled
Tests / Lint and Format (push) Has been cancelled
Tests / Web Node Tests (push) Has been cancelled
Tests / Import Check (Python 3.11) (push) Has been cancelled
Tests / Import Check (Python 3.12) (push) Has been cancelled
Tests / Python Tests (Python 3.13) (push) Has been cancelled
374 lines
14 KiB
Python
374 lines
14 KiB
Python
"""Tests for TurnRuntimeManager helper functions and lightweight behaviour."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from deeptutor.core.stream import StreamEvent, StreamEventType
|
|
from deeptutor.services.session.turn_runtime import (
|
|
_artifact_attachments,
|
|
_clip_text,
|
|
_extract_followup_question_context,
|
|
_extract_memory_references,
|
|
_extract_persist_user_message,
|
|
_format_followup_question_context,
|
|
_narration_marker_call_id,
|
|
_should_capture_assistant_content,
|
|
)
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _should_capture_assistant_content
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestShouldCaptureAssistantContent:
|
|
def test_content_without_call_id_is_captured(self) -> None:
|
|
event = StreamEvent(type=StreamEventType.CONTENT, content="hello")
|
|
assert _should_capture_assistant_content(event) is True
|
|
|
|
def test_content_with_final_response_kind_is_captured(self) -> None:
|
|
event = StreamEvent(
|
|
type=StreamEventType.CONTENT,
|
|
content="answer",
|
|
metadata={"call_id": "c1", "call_kind": "llm_final_response"},
|
|
)
|
|
assert _should_capture_assistant_content(event) is True
|
|
|
|
def test_content_with_non_final_call_kind_not_captured(self) -> None:
|
|
event = StreamEvent(
|
|
type=StreamEventType.CONTENT,
|
|
content="internal",
|
|
metadata={"call_id": "c1", "call_kind": "llm_reasoning"},
|
|
)
|
|
assert _should_capture_assistant_content(event) is False
|
|
|
|
def test_agent_loop_round_content_is_captured(self) -> None:
|
|
# The single-loop chat agent streams the finish round's answer as
|
|
# ``content`` with ``agent_loop_round``; it must reach the persisted
|
|
# answer (regression: was dropped, so the bubble cleared on reload).
|
|
event = StreamEvent(
|
|
type=StreamEventType.CONTENT,
|
|
content="the answer",
|
|
metadata={"call_id": "c1", "call_kind": "agent_loop_round"},
|
|
)
|
|
assert _should_capture_assistant_content(event) is True
|
|
|
|
def test_non_content_event_not_captured(self) -> None:
|
|
event = StreamEvent(type=StreamEventType.THINKING, content="hmm")
|
|
assert _should_capture_assistant_content(event) is False
|
|
|
|
def test_tool_call_not_captured(self) -> None:
|
|
event = StreamEvent(type=StreamEventType.TOOL_CALL, content="web_search")
|
|
assert _should_capture_assistant_content(event) is False
|
|
|
|
|
|
class TestNarrationMarkerCallId:
|
|
def test_narration_marker_returns_call_id(self) -> None:
|
|
event = StreamEvent(
|
|
type=StreamEventType.PROGRESS,
|
|
metadata={
|
|
"call_id": "round-1",
|
|
"trace_kind": "call_status",
|
|
"call_state": "complete",
|
|
"call_role": "narration",
|
|
},
|
|
)
|
|
assert _narration_marker_call_id(event) == "round-1"
|
|
|
|
def test_finish_marker_is_not_narration(self) -> None:
|
|
event = StreamEvent(
|
|
type=StreamEventType.PROGRESS,
|
|
metadata={
|
|
"call_id": "round-2",
|
|
"trace_kind": "call_status",
|
|
"call_state": "complete",
|
|
"call_role": "finish",
|
|
},
|
|
)
|
|
assert _narration_marker_call_id(event) is None
|
|
|
|
def test_running_status_is_not_narration(self) -> None:
|
|
event = StreamEvent(
|
|
type=StreamEventType.PROGRESS,
|
|
metadata={
|
|
"call_id": "round-1",
|
|
"trace_kind": "call_status",
|
|
"call_state": "running",
|
|
"call_role": "narration",
|
|
},
|
|
)
|
|
assert _narration_marker_call_id(event) is None
|
|
|
|
|
|
class TestArtifactAttachments:
|
|
def _sources_event(self, sources: list[dict]) -> StreamEvent:
|
|
return StreamEvent(type=StreamEventType.SOURCES, metadata={"sources": sources})
|
|
|
|
def test_artifact_source_becomes_generated_attachment(self) -> None:
|
|
event = self._sources_event(
|
|
[
|
|
{
|
|
"type": "artifact",
|
|
"filename": "report.pdf",
|
|
"url": "/api/outputs/workspace/chat/chat/t1/exec/report.pdf",
|
|
"mime_type": "application/pdf",
|
|
"size_bytes": 2048,
|
|
}
|
|
]
|
|
)
|
|
atts = _artifact_attachments(event)
|
|
assert len(atts) == 1
|
|
a = atts[0]
|
|
assert a["type"] == "document"
|
|
assert a["filename"] == "report.pdf"
|
|
assert a["url"].endswith("report.pdf")
|
|
assert a["mime_type"] == "application/pdf"
|
|
assert a["generated"] is True
|
|
|
|
def test_image_artifact_typed_as_image(self) -> None:
|
|
event = self._sources_event(
|
|
[
|
|
{
|
|
"type": "artifact",
|
|
"filename": "chart.png",
|
|
"url": "/api/outputs/x/chart.png",
|
|
"mime_type": "image/png",
|
|
}
|
|
]
|
|
)
|
|
assert _artifact_attachments(event)[0]["type"] == "image"
|
|
|
|
def test_non_artifact_sources_ignored(self) -> None:
|
|
event = self._sources_event([{"type": "rag", "query": "q", "kb_name": "kb"}])
|
|
assert _artifact_attachments(event) == []
|
|
|
|
def test_tool_result_artifacts_extracted(self) -> None:
|
|
# tool_result events carry artifacts the moment exec finishes — the
|
|
# durable source for cancelled turns (the aggregate SOURCES event
|
|
# only fires when the loop completes).
|
|
event = StreamEvent(
|
|
type=StreamEventType.TOOL_RESULT,
|
|
content="Exit code: 0",
|
|
metadata={
|
|
"tool_metadata": {
|
|
"exit_code": 0,
|
|
"artifacts": [
|
|
{
|
|
"filename": "notes.pdf",
|
|
"url": "/api/outputs/workspace/chat/chat/t2/exec/notes.pdf",
|
|
"mime_type": "application/pdf",
|
|
"size_bytes": 1024,
|
|
}
|
|
],
|
|
}
|
|
},
|
|
)
|
|
atts = _artifact_attachments(event)
|
|
assert len(atts) == 1
|
|
assert atts[0]["filename"] == "notes.pdf"
|
|
assert atts[0]["generated"] is True
|
|
|
|
def test_tool_result_without_artifacts_ignored(self) -> None:
|
|
event = StreamEvent(
|
|
type=StreamEventType.TOOL_RESULT,
|
|
content="rag result",
|
|
metadata={"tool_metadata": {"kb_name": "kb"}},
|
|
)
|
|
assert _artifact_attachments(event) == []
|
|
|
|
def test_non_sources_event_ignored(self) -> None:
|
|
event = StreamEvent(type=StreamEventType.CONTENT, content="hello")
|
|
assert _artifact_attachments(event) == []
|
|
|
|
def test_artifact_without_url_skipped(self) -> None:
|
|
event = self._sources_event([{"type": "artifact", "filename": "x.pdf"}])
|
|
assert _artifact_attachments(event) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _clip_text
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestClipText:
|
|
def test_short_text_unchanged(self) -> None:
|
|
assert _clip_text("hello", limit=100) == "hello"
|
|
|
|
def test_long_text_truncated(self) -> None:
|
|
text = "x" * 5000
|
|
result = _clip_text(text, limit=100)
|
|
assert len(result) < 200
|
|
assert "[truncated]" in result
|
|
|
|
def test_empty_string(self) -> None:
|
|
assert _clip_text("") == ""
|
|
|
|
def test_none_becomes_empty(self) -> None:
|
|
assert _clip_text(None) == "" # type: ignore[arg-type]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_memory_references
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExtractMemoryReferences:
|
|
def test_extracts_valid_memory_files_in_order(self) -> None:
|
|
payload = {"memory_references": ["profile", "summary"]}
|
|
|
|
assert _extract_memory_references(payload) == ["profile", "summary"]
|
|
|
|
def test_filters_unknown_and_duplicate_memory_files(self) -> None:
|
|
payload = {"memory_references": ["summary", "unknown", "summary", "profile"]}
|
|
|
|
assert _extract_memory_references(payload) == ["summary", "profile"]
|
|
|
|
def test_non_list_memory_references_are_ignored(self) -> None:
|
|
assert _extract_memory_references({"memory_references": "summary"}) == []
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_followup_question_context
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExtractFollowupQuestionContext:
|
|
def test_none_config(self) -> None:
|
|
assert _extract_followup_question_context(None) is None
|
|
|
|
def test_missing_key(self) -> None:
|
|
assert _extract_followup_question_context({}) is None
|
|
|
|
def test_non_dict_value(self) -> None:
|
|
assert _extract_followup_question_context({"followup_question_context": "string"}) is None
|
|
|
|
def test_missing_question(self) -> None:
|
|
assert (
|
|
_extract_followup_question_context({"followup_question_context": {"question_id": "q1"}})
|
|
is None
|
|
)
|
|
|
|
def test_valid_context_extracted(self) -> None:
|
|
config = {
|
|
"followup_question_context": {
|
|
"question": "What is AI?",
|
|
"question_id": "q1",
|
|
"question_type": "mcq",
|
|
"options": {"A": "Choice A", "B": "Choice B"},
|
|
"correct_answer": "A",
|
|
"explanation": "AI is...",
|
|
"difficulty": "easy",
|
|
"user_answer": "B",
|
|
"is_correct": False,
|
|
}
|
|
}
|
|
result = _extract_followup_question_context(config)
|
|
assert result is not None
|
|
assert result["question"] == "What is AI?"
|
|
assert result["question_id"] == "q1"
|
|
assert result["options"]["A"] == "Choice A"
|
|
assert result["is_correct"] is False
|
|
assert "followup_question_context" not in config # popped
|
|
|
|
def test_options_normalized(self) -> None:
|
|
config = {
|
|
"followup_question_context": {
|
|
"question": "Q",
|
|
"options": {"a": "lower", "B": "upper", "c": ""},
|
|
}
|
|
}
|
|
result = _extract_followup_question_context(config)
|
|
assert "A" in result["options"]
|
|
assert "B" in result["options"]
|
|
assert "C" not in result["options"] # empty value excluded
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _extract_persist_user_message
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestExtractPersistUserMessage:
|
|
def test_default_is_true(self) -> None:
|
|
assert _extract_persist_user_message({}) is True
|
|
|
|
def test_none_config_is_true(self) -> None:
|
|
assert _extract_persist_user_message(None) is True
|
|
|
|
def test_false_bool(self) -> None:
|
|
config = {"_persist_user_message": False}
|
|
assert _extract_persist_user_message(config) is False
|
|
assert "_persist_user_message" not in config # popped
|
|
|
|
def test_false_string(self) -> None:
|
|
assert _extract_persist_user_message({"_persist_user_message": "false"}) is False
|
|
|
|
def test_zero_string(self) -> None:
|
|
assert _extract_persist_user_message({"_persist_user_message": "0"}) is False
|
|
|
|
def test_no_string(self) -> None:
|
|
assert _extract_persist_user_message({"_persist_user_message": "no"}) is False
|
|
|
|
def test_true_string(self) -> None:
|
|
assert _extract_persist_user_message({"_persist_user_message": "true"}) is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# _format_followup_question_context
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestFormatFollowupQuestionContext:
|
|
def _base_context(self) -> dict:
|
|
return {
|
|
"question_id": "q1",
|
|
"parent_quiz_session_id": "qs1",
|
|
"question_type": "mcq",
|
|
"difficulty": "medium",
|
|
"concentration": "math",
|
|
"question": "What is 2+2?",
|
|
"options": {"A": "3", "B": "4"},
|
|
"user_answer": "A",
|
|
"is_correct": False,
|
|
"correct_answer": "B",
|
|
"explanation": "2+2=4",
|
|
"knowledge_context": "",
|
|
}
|
|
|
|
def test_english_format(self) -> None:
|
|
text = _format_followup_question_context(self._base_context(), language="en")
|
|
assert "You are handling follow-up questions" in text
|
|
assert "What is 2+2?" in text
|
|
assert "A. 3" in text
|
|
assert "B. 4" in text
|
|
assert "incorrect" in text
|
|
|
|
def test_chinese_format(self) -> None:
|
|
text = _format_followup_question_context(self._base_context(), language="zh")
|
|
assert "你正在处理一道测验题的后续追问" in text
|
|
assert "What is 2+2?" in text
|
|
|
|
def test_correct_answer_shows_correct(self) -> None:
|
|
ctx = self._base_context()
|
|
ctx["is_correct"] = True
|
|
text = _format_followup_question_context(ctx, language="en")
|
|
assert "correct" in text.lower()
|
|
|
|
def test_unknown_correctness(self) -> None:
|
|
ctx = self._base_context()
|
|
ctx["is_correct"] = None
|
|
text = _format_followup_question_context(ctx, language="en")
|
|
assert "unknown" in text.lower()
|
|
|
|
def test_knowledge_context_included(self) -> None:
|
|
ctx = self._base_context()
|
|
ctx["knowledge_context"] = "Some KB knowledge"
|
|
text = _format_followup_question_context(ctx, language="en")
|
|
assert "Some KB knowledge" in text
|
|
|
|
def test_no_options(self) -> None:
|
|
ctx = self._base_context()
|
|
ctx["options"] = {}
|
|
text = _format_followup_question_context(ctx, language="en")
|
|
assert "Options:" not in text
|