File size: 6,593 Bytes
df41178 | 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 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | #!/usr/bin/env python3
"""会話整理案のローカル生成・検証・実行。"""
from __future__ import annotations
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
HERE = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(HERE))
import chats # noqa: E402
class _Response:
def __init__(self, data):
self.data = json.dumps(data, ensure_ascii=False).encode("utf-8")
def __enter__(self):
return self
def __exit__(self, *_args):
return False
def read(self):
return self.data
class ChatSeiriTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
root = Path(self.tmp.name)
self.old = chats.DIR, chats.DB, chats.TITLE_EXAMPLES, chats.SEIRI_OBOE, chats._READY
chats.DIR = str(root / "legacy")
chats.DB = str(root / "chats.sqlite3")
chats.TITLE_EXAMPLES = str(root / "title_examples.json")
chats.SEIRI_OBOE = str(root / "seiri_oboe.json")
chats._READY = False
def tearDown(self):
chats.DIR, chats.DB, chats.TITLE_EXAMPLES, chats.SEIRI_OBOE, chats._READY = self.old
self.tmp.cleanup()
def _chat(self, title, first, updated=None):
conversation = chats.create(title)
chats.add_turn(conversation["id"], "user", first)
if updated is not None:
with chats._db() as conn:
conn.execute("UPDATE conversations SET updated=? WHERE id=?", (updated, conversation["id"]))
return chats.load(conversation["id"])
def test_qwen_json_is_read_after_think_tag_and_code_fence(self):
payload = {"choices": [{"message": {"content": '<think>考え中</think>```json\n{"組":[],"ゴミ箱":[],"しまう":[]}\n```'}}]}
with patch("urllib.request.urlopen", side_effect=[_Response([]), _Response(payload)]):
answer = chats._ask_seiri_qwen([])
self.assertEqual(answer, {"組": [], "ゴミ箱": [], "しまう": []})
def test_broken_json_returns_none(self):
payload = {"choices": [{"message": {"content": '```json\n{"組": ['}}]}
with patch("urllib.request.urlopen", side_effect=[_Response([]), _Response(payload)]):
self.assertIsNone(chats._ask_seiri_qwen([]))
def test_nonexistent_ids_are_ignored_and_recent_chat_cannot_be_trashed(self):
old = self._chat("古い会話", "内容", chats._now() - 9 * 86400)
recent = self._chat("最近の会話", "別の内容")
answer = {"組": [], "ゴミ箱": [
{"id": "not-real", "理由": "存在しない"}, {"id": recent["id"], "理由": "最近"},
{"id": old["id"], "理由": "不要"}], "しまう": []}
proposals = chats._seiri_qwen_proposals([old, recent], answer)
self.assertEqual([x["id"] for x in proposals], [old["id"]])
self.assertEqual(proposals[0]["出どころ"], "Qwen3.6")
def test_recent_chat_is_not_archived_by_qwen(self):
recent = self._chat("今の作業", "続きをお願い", updated=chats._now() - 86400)
old = self._chat("前の作業", "もう終わった", updated=chats._now() - 5 * 86400)
answer = {"組": [], "ゴミ箱": [], "しまう": [{"id": recent["id"], "理由": "短い"}, {"id": old["id"], "理由": "終わった"}]}
with patch.object(chats, "_ask_seiri_qwen", return_value=answer):
proposals = [p for p in chats.seiri_an() if p.get("出どころ") == "Qwen3.6"]
self.assertEqual([(p["id"], p["区分"]) for p in proposals], [(old["id"], "しまう")])
def test_archived_chat_is_not_archived_again(self):
done = self._chat("終わった作業", "もう済んだ", updated=chats._now() - 9 * 86400)
chats.archive(done["id"], True)
answer = {"組": [], "ゴミ箱": [], "しまう": [{"id": done["id"], "理由": "既にしまった"}]}
with patch.object(chats, "_ask_seiri_qwen", return_value=answer):
self.assertFalse([p for p in chats.seiri_an() if p.get("出どころ") == "Qwen3.6"])
def test_timeout_falls_back_to_rules_only(self):
chat = self._chat("試験: 古い会話", "古い試験", chats._now() - 8 * 86400)
with patch.object(chats, "_ask_seiri_qwen", return_value=None):
proposals = chats.seiri_an()
self.assertEqual([(p["id"], p["区分"], p["出どころ"]) for p in proposals],
[(chat["id"], "ゴミ箱", "規則")])
def test_qwen_group_name_reuses_existing_name_and_limits_fields(self):
first = self._chat("旅の計画", "旅の計画を相談")
second = self._chat("旅の荷物", "旅の荷物を相談")
existing = self._chat("既存", "既存会話")
chats.set_group(existing["id"], "旅行")
answer = {"組": [{"名": "旅行メモ", "ids": [first["id"], second["id"], "missing"], "理由": "理由" * 30}],
"ゴミ箱": [], "しまう": []}
proposals = chats._seiri_qwen_proposals([first, second], answer)
self.assertEqual({p["組"] for p in proposals}, {"旅行"})
self.assertTrue(all(len(p["理由"]) <= 40 for p in proposals))
self.assertTrue(all(len(p["組"]) <= 12 for p in proposals))
def test_rejected_proposal_is_not_shown_again(self):
chat = self._chat("分類候補", "相談")
Path(chats.SEIRI_OBOE).write_text(json.dumps({"外した": [{"id": chat["id"], "区分": "しまう"}], "組名": []}))
answer = {"組": [], "ゴミ箱": [], "しまう": [{"id": chat["id"], "理由": "しまう"}]}
self.assertEqual(chats._seiri_qwen_proposals([chat], answer), [])
def test_seiri_suru_executes_remembered_plan_without_calling_qwen_again(self):
first = self._chat("旅行の計画", "一つ目")
second = self._chat("旅行の荷物", "二つ目")
answer = {"組": [{"名": "旅行", "ids": [first["id"], second["id"]], "理由": "近い話"}],
"ゴミ箱": [], "しまう": []}
with patch.object(chats, "_ask_seiri_qwen", return_value=answer):
proposals = chats.seiri_an()
with patch.object(chats, "_ask_seiri_qwen", side_effect=AssertionError("再問い合わせした")):
done = chats.seiri_suru([{**p, "checked": True} for p in proposals])
self.assertEqual(done, 2)
self.assertEqual(chats.load(first["id"])["組"], "旅行")
self.assertEqual(chats.load(second["id"])["組"], "旅行")
if __name__ == "__main__":
unittest.main()
|