chktime/tests/test_n8n.py

"""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_continued_session_copies_are_listed_once(self):
        # 会話を続けると、新しいセッションのファイルに前の履歴が同じ時刻のまま写される
        old = [(0, "user", "設計して"), (10, "assistant", "ok")]
        write_session(self.root, "C--debug-demo", "old", old, r"C:\debug\demo")
        write_session(self.root, "C--debug-demo", "new",
                      old + [(20, "user", "続き"), (25, "assistant", "ok"), (90, "user", "次"), (100, "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([(b["sessionId"], b["minutes"]) for b in blocks], [("new", 25), ("new", 10)])

    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)


class ApiFunctionsTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / "projects"
        self.store = chktime.Store(self.root)
        write_session(self.root, "C--debug-demo", "s1",
                      [(0, "user", "設計して"), (10, "assistant", "ok")], r"C:\debug\demo")
        self.fake = FakeN8n()
        self.holder = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
        self.holder.update({"url": self.fake.url, "token": FakeN8n.TOKEN})
        self.query = {"from": ["2026-01-05"], "to": ["2026-01-05"], "project": [""]}

    def tearDown(self):
        self.fake.close()
        self.tmp.cleanup()

    def source(self):
        return f"chktime:s1:{int(round(BASE * 1000))}"

    def test_range_defaults_and_limits(self):
        start, end = chktime.n8n_range({})
        self.assertEqual((end - start).days, 6)
        with self.assertRaises(chktime.N8nError):
            chktime.n8n_range({"from": ["2026-01-05"], "to": ["2026-01-04"]})
        with self.assertRaises(chktime.N8nError):
            chktime.n8n_range({"from": ["2026-01-01"], "to": ["2026-03-31"]})

    def test_blocks_marks_with_n8n_rows(self):
        data = chktime.n8n_blocks(self.store, self.holder, self.query, now=LATER)
        self.assertIsNone(data["n8nError"])
        self.assertEqual(len(data["blocks"]), 1)
        self.assertFalse(data["blocks"][0]["sent"])

    def test_blocks_when_n8n_down(self):
        self.holder.update({"url": "http://127.0.0.1:9"})
        data = chktime.n8n_blocks(self.store, self.holder, self.query, now=LATER)
        self.assertIn("接続できません", data["n8nError"])
        self.assertEqual(len(data["blocks"]), 1)
        self.assertFalse(data["blocks"][0]["sent"])

    def test_send_recomputes_times_on_server(self):
        item = {"project": "C--debug-demo", "source": self.source(), "task": "[demo] #12 設計", "memo": "m",
                "start_time": "1999-01-01T00:00:00Z"}  # ブラウザの時刻は使わない
        result = chktime.n8n_send(self.store, self.holder, {"items": [item]}, now=LATER)
        self.assertTrue(result["ok"])
        method, path, body = self.fake.requests[-1]
        row = body["rows"][0]
        self.assertEqual(path, "/webhook/wt-import")
        self.assertEqual(row["start_time"], chktime.iso(BASE))
        self.assertEqual(row["end_time"], chktime.iso(BASE + 10 * MIN))
        self.assertEqual(row["date"], "2026-01-05")
        self.assertEqual(row["task"], "[demo] #12 設計")

    def test_send_rejects_blank_task(self):
        item = {"project": "C--debug-demo", "source": self.source(), "task": "   "}
        result = chktime.n8n_send(self.store, self.holder, {"items": [item]}, now=LATER)
        self.assertFalse(result["ok"])
        self.assertEqual(result["errors"][0]["error"], "タスク名が空です")
        self.assertEqual([r for r in self.fake.requests if r[0] == "POST"], [])

    def test_send_rejects_unknown_and_in_progress(self):
        bad = {"project": "C--debug-demo", "source": "chktime:s1:123", "task": "x"}
        gone = {"project": "C--debug-demo", "source": "chktime:nope:1", "task": "x"}
        result = chktime.n8n_send(self.store, self.holder, {"items": [bad, gone]}, now=LATER)
        self.assertEqual([e["error"] for e in result["errors"]],
                         ["区間が見つかりません (ログか設定が変わりました)", "セッションが見つかりません"])
        live = {"project": "C--debug-demo", "source": self.source(), "task": "x"}
        result = chktime.n8n_send(self.store, self.holder, {"items": [live]}, now=BASE + 11 * MIN)
        self.assertEqual(result["errors"][0]["error"], "進行中の区間は送れません")

    def test_send_requires_items(self):
        with self.assertRaises(chktime.N8nError):
            chktime.n8n_send(self.store, self.holder, {"items": []}, now=LATER)


class RedmineSearchTest(unittest.TestCase):
    """チケット選択用の検索。偽の Redmine に対して要求の形と結果の変換を確かめる。"""

    def setUp(self):
        self.paths = []
        test = self

        class Handler(BaseHTTPRequestHandler):
            def log_message(self, *args):
                pass

            def do_GET(self):
                test.paths.append(self.path)
                if self.path.startswith("/issues/404.json"):
                    self.send_response(404)
                    self.end_headers()
                    return
                if self.path.startswith("/issues/"):
                    payload = {"issue": {"id": 7, "subject": "七", "project": {"name": "P"}}}
                else:
                    payload = {"issues": [{"id": 3, "subject": "設計", "project": {"name": "ImgChk"}}]}
                body = json.dumps(payload).encode()
                self.send_response(200)
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        threading.Thread(target=self.server.serve_forever, daemon=True).start()
        self.client = chktime.Redmine(f"http://127.0.0.1:{self.server.server_address[1]}", "key", interval=0)

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()

    def test_search_by_subject(self):
        self.assertEqual(self.client.search("設 計"), [{"id": 3, "subject": "設計", "project": "ImgChk"}])
        self.assertIn("status_id=open", self.paths[-1])
        self.assertIn("op[subject]=~", self.paths[-1])
        self.assertIn("v[subject][]=%E8%A8%AD%20%E8%A8%88", self.paths[-1])

    def test_search_by_number_and_missing_number(self):
        self.assertEqual(self.client.search("7"), [{"id": 7, "subject": "七", "project": "P"}])
        self.assertEqual(self.client.search("404"), [])


def write_raw(root, project_dir, sid, lines, mtime):
    """jsonl の行 (dict) をそのまま書く。mtime を指定する。"""
    import os
    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 obj in lines:
            # 本物のログと同じく区切りの空白なしで書く (tool_result 行の高速判定がこの形を前提にしている)
            fh.write(json.dumps(obj, ensure_ascii=False, separators=(",", ":")) + "\n")
    os.utime(path, (mtime, mtime))
    return path


def ts(minute):
    return chktime.iso(BASE + minute * MIN)


def user_line(minute, text, sid="p1", sidechain=False):
    return {"type": "user", "timestamp": ts(minute), "sessionId": sid, "cwd": r"C:\debug\demo",
            "isSidechain": sidechain, "message": {"role": "user", "content": text}}


def reply_line(minute, sid="p1", tool=None):
    content = [{"type": "tool_use", "id": "t1", "name": tool, "input": {}}] if tool else [{"type": "text", "text": "ok"}]
    return {"type": "assistant", "timestamp": ts(minute), "sessionId": sid, "message": {"model": "m", "content": content}}


def tool_result_line(minute, sid="p1"):
    return {"type": "user", "timestamp": ts(minute), "sessionId": sid,
            "message": {"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "done"}]}}


class FinalReviewFixTest(unittest.TestCase):
    """最終レビューで見つかった問題の再現テスト。"""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / "projects"
        self.store = chktime.Store(self.root)

    def tearDown(self):
        self.tmp.cleanup()

    def blocks(self, now):
        return chktime.collect_blocks(self.store, "", date(2026, 1, 5), date(2026, 1, 5), 30, 5, now=now)

    # Important 1: 最後がローカルのコマンド (/exit など) だと永遠に進行中になっていた
    def test_trailing_local_command_is_not_in_progress(self):
        write_raw(self.root, "C--debug-demo", "p1",
                  [user_line(0, "作業して"), reply_line(20), user_line(21, "<command-name>/exit</command-name>")],
                  BASE + 21 * MIN)
        self.assertEqual([b["inProgress"] for b in self.blocks(LATER)], [False])

    def test_prompt_without_reply_long_ago_is_not_in_progress(self):
        write_raw(self.root, "C--debug-demo", "p1", [user_line(0, "作業して"), reply_line(20), user_line(21, "次")],
                  BASE + 21 * MIN)
        self.assertEqual([b["inProgress"] for b in self.blocks(BASE + 21 * MIN + 25 * 3600)], [False])
        self.assertEqual([b["inProgress"] for b in self.blocks(BASE + 21 * MIN + 2 * 3600)], [True])

    # Important 2: ツールの結果を待っている間に 30 分静かだと、途中で切れた区間が送れていた
    def test_open_tool_call_keeps_block_in_progress(self):
        write_raw(self.root, "C--debug-demo", "p1", [user_line(0, "調べて"), reply_line(5, tool="Agent")],
                  BASE + 5 * MIN)
        self.assertEqual([b["inProgress"] for b in self.blocks(BASE + 40 * MIN)], [True])
        write_raw(self.root, "C--debug-demo", "p1",
                  [user_line(0, "調べて"), reply_line(5, tool="Agent"), tool_result_line(45), reply_line(46)],
                  BASE + 46 * MIN)
        self.store.clear()
        self.assertEqual([(b["inProgress"], b["minutes"]) for b in self.blocks(BASE + 100 * MIN)], [(False, 46)])

    def test_sent_block_that_grew_is_flagged(self):
        block = {"source": "chktime:a:1", "startEpoch": BASE, "endEpoch": BASE + 46 * MIN, "start": ts(0), "end": ts(46),
                 "taskName": "A"}
        rows = [{"id": 1, "source": "chktime:a:1", "start_time": ts(0), "end_time": ts(5), "task": "A"}]
        chktime.mark_blocks([block], rows, [], now=LATER)
        self.assertTrue(block["sent"])
        self.assertTrue(block["changedSinceSent"])

    # Important 3: 写しの同着で、送信済みの区間が未送信に見えていた
    def test_copy_with_same_end_keeps_sent_state(self):
        lines_old = [user_line(0, "設計して", sid="aaa"), reply_line(10, sid="aaa")]
        lines_new = [user_line(0, "設計して", sid="zzz"), reply_line(10, sid="zzz"), user_line(60, "続き", sid="zzz"),
                     reply_line(70, sid="zzz")]
        write_raw(self.root, "C--debug-demo", "zzz", lines_new, BASE + 70 * MIN)
        write_raw(self.root, "C--debug-demo", "aaa", lines_old, BASE + 10 * MIN)
        blocks = self.blocks(LATER)
        first = blocks[0]
        self.assertEqual(len(blocks), 2)
        sent_source = f"chktime:aaa:{int(round(BASE * 1000))}"
        self.assertIn(sent_source, [first["source"]] + first["aliases"])
        rows = [{"id": 1, "source": sent_source, "start_time": ts(0), "end_time": ts(10), "task": "A"}]
        chktime.mark_blocks(blocks, rows, [], now=LATER)
        self.assertTrue(first["sent"])
        self.assertEqual(first["overlaps"], [])

    # Important 5: 送れない文字の合言葉で一覧ごと 500 になっていた
    def test_non_ascii_token_is_n8n_error_without_token_text(self):
        client = chktime.N8n("http://127.0.0.1:9", "合言葉です")
        with self.assertRaises(chktime.N8nError) as ctx:
            client.rows(date(2026, 1, 1), date(2026, 1, 1))
        self.assertNotIn("合言葉です", str(ctx.exception))

    def test_holder_rejects_non_ascii_token(self):
        holder = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
        with self.assertRaises(chktime.N8nError):
            holder.update({"token": "合言葉"})

    # Important 4: URL だけ変えると、保存済みの合言葉が別のサーバーへ送られていた
    def test_changing_url_host_clears_token(self):
        holder = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
        holder.update({"url": "https://n8n.example", "token": "secret"})
        self.assertFalse(holder.update({"url": "https://attacker.example"})["hasToken"])
        holder.update({"url": "https://n8n.example", "token": "secret"})
        self.assertTrue(holder.update({"url": "https://n8n.example/"})["hasToken"])
        self.assertTrue(holder.update({"url": "https://other.example", "token": "secret2"})["hasToken"])


class MinorFixTest(unittest.TestCase):
    """後回しにした小さな問題の再現テスト。"""

    def test_zero_length_block_has_problem(self):
        blocks = chktime.make_blocks(session(turn(0, 0)), min_min=0, now=LATER)
        self.assertEqual(blocks[0]["problem"], "長さが 0 分です")

    def test_over_24h_block_has_problem(self):
        turns = [turn(m, m + 20) for m in range(0, 25 * 60, 40)]  # 40 分ごとにつながって 25 時間
        blocks = chktime.make_blocks(session(*turns), now=LATER + 2 * 86400)
        self.assertEqual(len(blocks), 1)
        self.assertEqual(blocks[0]["problem"], "24 時間を超えています")

    def test_normal_block_has_no_problem(self):
        self.assertIsNone(chktime.make_blocks(session(turn(0, 10)), now=LATER)[0]["problem"])


class MinorSendTest(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name) / "projects"
        self.store = chktime.Store(self.root)
        write_session(self.root, "C--debug-demo", "s1", [(0, "user", "設計して"), (0, "assistant", "ok")],
                      r"C:\debug\demo")
        self.fake = FakeN8n()
        self.holder = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
        self.holder.update({"url": self.fake.url, "token": FakeN8n.TOKEN, "minMinutes": 0})

    def tearDown(self):
        self.fake.close()
        self.tmp.cleanup()

    def test_send_rejects_block_with_problem(self):
        item = {"project": "C--debug-demo", "source": f"chktime:s1:{int(round(BASE * 1000))}", "task": "x"}
        result = chktime.n8n_send(self.store, self.holder, {"items": [item]}, now=LATER)
        self.assertEqual(result["errors"][0]["error"], "長さが 0 分です")
        self.assertEqual([r for r in self.fake.requests if r[0] == "POST"], [])

    def test_send_reports_non_dict_item(self):
        result = chktime.n8n_send(self.store, self.holder, {"items": ["abc", 3]}, now=LATER)
        self.assertFalse(result["ok"])
        self.assertEqual([e["error"] for e in result["errors"]], ["項目の形が不正です", "項目の形が不正です"])


class PostGuardTest(unittest.TestCase):
    """Important 4: 他のサイトから POST で設定を書き換えられないこと (Content-Type と Origin)。"""

    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        server = chktime.Server(("127.0.0.1", 0), chktime.Handler)
        server.store = chktime.Store(Path(self.tmp.name))
        server.redmine = chktime.RedmineHolder(Path(self.tmp.name) / "redmine.json")
        server.n8n = chktime.N8nHolder(Path(self.tmp.name) / "n8n.json")
        server.jobs = chktime.Jobs()
        self.server = server
        self.port = server.server_address[1]
        threading.Thread(target=server.serve_forever, daemon=True).start()

    def tearDown(self):
        self.server.shutdown()
        self.server.server_close()
        self.tmp.cleanup()

    def post(self, body, ctype="application/json", origin=None, host=None):
        import http.client
        conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
        headers = {"Content-Type": ctype}
        if origin:
            headers["Origin"] = origin
        if host:
            headers["Host"] = host
        conn.request("POST", "/api/n8n/config", body=json.dumps(body).encode(), headers=headers)
        res = conn.getresponse()
        status = res.status
        res.read()
        conn.close()
        return status

    def test_rejects_simple_cross_site_post(self):
        self.assertEqual(self.post({"url": "https://attacker.example"}, ctype="text/plain"), 415)
        self.assertEqual(self.server.n8n.public()["url"], "")

    def test_rejects_foreign_origin_and_host(self):
        self.assertEqual(self.post({"url": "https://a.example"}, origin="https://evil.example"), 403)
        self.assertEqual(self.post({"url": "https://a.example"}, host="evil.example"), 403)
        self.assertEqual(self.server.n8n.public()["url"], "")

    def test_accepts_same_origin_json(self):
        origin = f"http://127.0.0.1:{self.port}"
        self.assertEqual(self.post({"url": "https://n8n.example"}, origin=origin), 200)
        self.assertEqual(self.server.n8n.public()["url"], "https://n8n.example")


if __name__ == "__main__":
    unittest.main()