175 lines
8.4 KiB
Python
175 lines
8.4 KiB
Python
"""n8n 連携 (作業区間の計算・重なり・送信) のテスト。標準ライブラリの unittest だけを使う。
|
|
|
|
python -m unittest discover -s tests -v
|
|
"""
|
|
|
|
import json
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from datetime import date, datetime
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
import chktime # noqa: E402
|
|
|
|
MIN = 60.0
|
|
BASE = datetime(2026, 1, 5, 9, 0).timestamp() # ローカル時刻 2026-01-05 09:00
|
|
|
|
|
|
def turn(start_min, end_min, kind="prompt", reply=True, interrupted=False):
|
|
"""BASE からの分で 1 ターンを作る。reply=False なら応答なし。"""
|
|
ts = BASE + start_min * MIN
|
|
end = BASE + end_min * MIN
|
|
return {"ts": ts, "endTs": end if reply else ts, "firstReplyTs": ts + 1 if reply else None,
|
|
"kind": kind, "interrupted": interrupted}
|
|
|
|
|
|
def session(*turns, sid="s1", mtime=None):
|
|
last_end = max(t["endTs"] for t in turns) if turns else BASE
|
|
return {"id": sid, "turns": list(turns), "mtime": mtime if mtime is not None else last_end}
|
|
|
|
|
|
LATER = BASE + 24 * 3600 # 「今」: 最後のログから十分経っている
|
|
|
|
|
|
class MakeBlocksTest(unittest.TestCase):
|
|
def test_gap_29_minutes_keeps_block(self):
|
|
blocks = chktime.make_blocks(session(turn(0, 10), turn(39, 50)), now=LATER)
|
|
self.assertEqual(len(blocks), 1)
|
|
self.assertEqual(blocks[0]["minutes"], 50)
|
|
self.assertEqual(blocks[0]["turns"], 2)
|
|
|
|
def test_gap_30_minutes_splits_block(self):
|
|
blocks = chktime.make_blocks(session(turn(0, 10), turn(40, 50)), now=LATER)
|
|
self.assertEqual([b["minutes"] for b in blocks], [10, 10])
|
|
|
|
def test_parallel_prompt_keeps_block(self):
|
|
# 前の応答 (0〜20 分) の途中の 5 分に次の発言
|
|
blocks = chktime.make_blocks(session(turn(0, 20), turn(5, 30)), now=LATER)
|
|
self.assertEqual(len(blocks), 1)
|
|
self.assertEqual(blocks[0]["minutes"], 30)
|
|
|
|
def test_short_block_dropped_but_exactly_min_kept(self):
|
|
short = chktime.make_blocks(session(turn(0, 4.9)), now=LATER)
|
|
exact = chktime.make_blocks(session(turn(0, 5)), now=LATER)
|
|
self.assertEqual(short, [])
|
|
self.assertEqual(len(exact), 1)
|
|
|
|
def test_last_block_without_reply_is_in_progress(self):
|
|
blocks = chktime.make_blocks(session(turn(0, 10), turn(60, 60, reply=False)), now=LATER)
|
|
self.assertEqual([b["inProgress"] for b in blocks], [False, True])
|
|
|
|
def test_recent_log_marks_last_block_in_progress(self):
|
|
s = session(turn(0, 10))
|
|
blocks = chktime.make_blocks(s, now=s["mtime"] + 29 * MIN)
|
|
self.assertTrue(blocks[0]["inProgress"])
|
|
blocks = chktime.make_blocks(s, now=s["mtime"] + 30 * MIN)
|
|
self.assertFalse(blocks[0]["inProgress"])
|
|
|
|
def test_interrupted_last_turn_is_not_in_progress(self):
|
|
blocks = chktime.make_blocks(session(turn(0, 10, reply=False, interrupted=True), turn(1, 10)), now=LATER)
|
|
self.assertFalse(blocks[0]["inProgress"])
|
|
only = chktime.make_blocks(session(turn(0, 10), turn(11, 11, reply=False, interrupted=True)), now=LATER)
|
|
self.assertFalse(only[-1]["inProgress"])
|
|
|
|
def test_source_is_stable(self):
|
|
a = chktime.make_blocks(session(turn(0, 10)), now=LATER)
|
|
b = chktime.make_blocks(session(turn(0, 10)), now=LATER + 999)
|
|
self.assertEqual(a[0]["source"], b[0]["source"])
|
|
self.assertEqual(a[0]["source"], f"chktime:s1:{int(round(BASE * 1000))}")
|
|
|
|
def test_custom_gap_and_min(self):
|
|
blocks = chktime.make_blocks(session(turn(0, 3), turn(14, 17)), gap_min=10, min_min=3, now=LATER)
|
|
self.assertEqual([b["minutes"] for b in blocks], [3, 3])
|
|
|
|
def test_date_is_local_start_date(self):
|
|
blocks = chktime.make_blocks(session(turn(0, 10)), now=LATER)
|
|
self.assertEqual(blocks[0]["date"], "2026-01-05")
|
|
|
|
|
|
def jsonl_line(obj):
|
|
return json.dumps(obj, ensure_ascii=False) + "\n"
|
|
|
|
|
|
def write_session(root, project_dir, sid, events, cwd):
|
|
"""events: [(分, 'user'|'assistant'|'sidechain', 本文)] を jsonl にする。"""
|
|
folder = Path(root) / project_dir
|
|
folder.mkdir(parents=True, exist_ok=True)
|
|
path = folder / f"{sid}.jsonl"
|
|
with path.open("w", encoding="utf-8") as fh:
|
|
for minute, kind, text in events:
|
|
ts = chktime.iso(BASE + minute * MIN)
|
|
if kind == "assistant":
|
|
fh.write(jsonl_line({"type": "assistant", "timestamp": ts, "sessionId": sid, "cwd": cwd,
|
|
"message": {"model": "m", "content": [{"type": "text", "text": text}]}}))
|
|
else:
|
|
fh.write(jsonl_line({"type": "user", "timestamp": ts, "sessionId": sid, "cwd": cwd,
|
|
"isSidechain": kind == "sidechain", "message": {"role": "user", "content": text}}))
|
|
end = BASE + max(m for m, _, _ in events) * MIN
|
|
import os
|
|
os.utime(path, (end, end))
|
|
return path
|
|
|
|
|
|
class CollectAndMarkTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.tmp = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.tmp.name)
|
|
self.store = chktime.Store(self.root)
|
|
|
|
def tearDown(self):
|
|
self.tmp.cleanup()
|
|
|
|
def test_collect_blocks_names_and_range(self):
|
|
write_session(self.root, "C--debug-demo", "s1",
|
|
[(0, "user", "設計して"), (10, "assistant", "ok"), (60, "user", "続き"), (70, "assistant", "ok")],
|
|
r"C:\debug\demo")
|
|
blocks = chktime.collect_blocks(self.store, "", date(2026, 1, 5), date(2026, 1, 5), 30, 5, now=LATER)
|
|
self.assertEqual(len(blocks), 2)
|
|
self.assertEqual(blocks[0]["taskName"], "[demo] 設計して")
|
|
self.assertEqual(blocks[0]["project"], "C--debug-demo")
|
|
none = chktime.collect_blocks(self.store, "", date(2026, 1, 6), date(2026, 1, 6), 30, 5, now=LATER)
|
|
self.assertEqual(none, [])
|
|
|
|
def test_sidechain_does_not_start_turn(self):
|
|
write_session(self.root, "C--debug-demo", "s2",
|
|
[(0, "user", "調べて"), (10, "assistant", "ok"), (45, "sidechain", "sub"), (80, "user", "次"), (90, "assistant", "ok")],
|
|
r"C:\debug\demo")
|
|
blocks = chktime.collect_blocks(self.store, "", date(2026, 1, 5), date(2026, 1, 5), 30, 5, now=LATER)
|
|
# サブエージェントの行は直前のターンを 45 分まで伸ばすだけ。80 分の発言はその 35 分後なので別の区間。
|
|
self.assertEqual([b["minutes"] for b in blocks], [45, 10])
|
|
|
|
def test_midnight_block_belongs_to_start_date(self):
|
|
start = datetime(2026, 1, 5, 23, 50).timestamp()
|
|
minutes = (start - BASE) / MIN
|
|
write_session(self.root, "C--debug-demo", "s3",
|
|
[(minutes, "user", "夜"), (minutes + 20, "assistant", "ok")], r"C:\debug\demo")
|
|
day1 = chktime.collect_blocks(self.store, "", date(2026, 1, 5), date(2026, 1, 5), 30, 5, now=LATER + 86400)
|
|
day2 = chktime.collect_blocks(self.store, "", date(2026, 1, 6), date(2026, 1, 6), 30, 5, now=LATER + 86400)
|
|
self.assertEqual([b["date"] for b in day1], ["2026-01-05"])
|
|
self.assertEqual(day2, [])
|
|
|
|
def test_mark_blocks(self):
|
|
b1 = {"source": "chktime:a:1", "startEpoch": BASE, "endEpoch": BASE + 600, "start": "", "end": "", "taskName": "A"}
|
|
b2 = {"source": "chktime:b:2", "startEpoch": BASE + 600, "endEpoch": BASE + 1200, "start": "", "end": "", "taskName": "B"}
|
|
b3 = {"source": "chktime:c:3", "startEpoch": BASE + 1199, "endEpoch": BASE + 1800, "start": "", "end": "", "taskName": "C"}
|
|
rows = [
|
|
{"id": 1, "source": "chktime:a:1", "start_time": chktime.iso(BASE), "end_time": chktime.iso(BASE + 600), "task": "A"},
|
|
{"id": 2, "source": "", "start_time": chktime.iso(BASE + 1700), "end_time": None, "task": "手で記録"},
|
|
]
|
|
chktime.mark_blocks([b1, b2, b3], rows, ["chktime:b:2"], now=BASE + 3600)
|
|
self.assertTrue(b1["sent"])
|
|
self.assertEqual(b1["overlaps"], []) # 自分の行は除く。b2 とは接しているだけ
|
|
self.assertFalse(b2["sent"])
|
|
self.assertTrue(b2["deleted"])
|
|
self.assertEqual([o["source"] for o in b2["overlaps"]], ["chktime:c:3"]) # 1 秒重なる
|
|
self.assertEqual(sorted(o["kind"] for o in b3["overlaps"]), ["block", "row"]) # 進行中の行 (終了 = 今) とも重なる
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|