n8n 連携: 区間の一覧・送信・設定の API と Redmine のチケット検索を追加

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_n8n.py

@@ -276,5 +276,126 @@ class N8nHolderTest(unittest.TestCase):
                     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"), [])
+
+
 if __name__ == "__main__":
     unittest.main()