"""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"]) # 進行中の行 (終了 = 今) とも重なる class FakeN8n: """wt-import の受け口を真似る偽のサーバー。受け取った要求を記録する。""" TOKEN = "test-token" def __init__(self): self.requests = [] fake = self class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass def reply(self, status, payload): body = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def do_GET(self): fake.requests.append(("GET", self.path, self.headers.get("X-WT-Token"))) if self.headers.get("X-WT-Token") != FakeN8n.TOKEN: return self.reply(403, {"message": "Authorization data is wrong!"}) if "from=2099" in self.path: return self.reply(200, {"ok": False, "error": "from が to より後です"}) self.reply(200, {"ok": True, "rows": [{"id": 9, "source": "chktime:s1:1", "start_time": None}], "deletedSources": []}) def do_POST(self): length = int(self.headers.get("Content-Length") or 0) body = json.loads(self.rfile.read(length)) fake.requests.append(("POST", self.path, body)) if self.headers.get("X-WT-Token") != FakeN8n.TOKEN: return self.reply(403, {}) inserted = [{"source": r["source"], "id": 100 + i} for i, r in enumerate(body["rows"])] self.reply(200, {"ok": True, "inserted": inserted, "skipped": [], "errors": []}) self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.server.server_address[1]}" threading.Thread(target=self.server.serve_forever, daemon=True).start() def close(self): self.server.shutdown() self.server.server_close() class N8nClientTest(unittest.TestCase): def setUp(self): self.fake = FakeN8n() def tearDown(self): self.fake.close() def test_rows_sends_token_and_range(self): client = chktime.N8n(self.fake.url, FakeN8n.TOKEN) data = client.rows(date(2026, 1, 1), date(2026, 1, 7)) self.assertEqual(data["rows"][0]["id"], 9) method, path, token = self.fake.requests[-1] self.assertEqual(path, "/webhook/wt-import-rows?from=2026-01-01&to=2026-01-07") self.assertEqual(token, FakeN8n.TOKEN) def test_wrong_token_is_403_message(self): client = chktime.N8n(self.fake.url, "wrong") with self.assertRaisesRegex(chktime.N8nError, "合言葉が違います"): client.rows(date(2026, 1, 1), date(2026, 1, 1)) def test_error_payload_raises(self): client = chktime.N8n(self.fake.url, FakeN8n.TOKEN) with self.assertRaisesRegex(chktime.N8nError, "from が to より後です"): client.rows(date(2099, 1, 2), date(2099, 1, 1)) def test_unreachable(self): client = chktime.N8n("http://127.0.0.1:9", FakeN8n.TOKEN) with self.assertRaisesRegex(chktime.N8nError, "接続できません"): client.rows(date(2026, 1, 1), date(2026, 1, 1)) def test_missing_settings(self): with self.assertRaisesRegex(chktime.N8nError, "URL が未設定"): chktime.N8n("", "x").rows(date(2026, 1, 1), date(2026, 1, 1)) with self.assertRaisesRegex(chktime.N8nError, "合言葉が未設定"): chktime.N8n(self.fake.url, "").rows(date(2026, 1, 1), date(2026, 1, 1)) class N8nHolderTest(unittest.TestCase): def test_public_hides_token_and_update_keeps_it(self): with tempfile.TemporaryDirectory() as tmp: path = Path(tmp) / "n8n.json" holder = chktime.N8nHolder(path) self.assertEqual(holder.settings(), (30, 5)) pub = holder.update({"url": "https://n8n.example/", "token": "secret", "gapMinutes": 20, "minMinutes": 3}) self.assertTrue(pub["hasToken"]) self.assertNotIn("secret", json.dumps(pub)) self.assertEqual(holder.settings(), (20, 3)) holder.update({"token": ""}) # 空なら既存を残す self.assertEqual(json.loads(path.read_text(encoding="utf-8"))["token"], "secret") self.assertEqual(holder.client.url, "https://n8n.example") def test_rejects_bad_minutes(self): with tempfile.TemporaryDirectory() as tmp: holder = chktime.N8nHolder(Path(tmp) / "n8n.json") for patch in ({"gapMinutes": 0}, {"gapMinutes": 241}, {"minMinutes": -1}, {"gapMinutes": "abc"}): with self.assertRaises(chktime.N8nError): holder.update(patch) if __name__ == "__main__": unittest.main()