From c1d1d046b6115475b1b46f56aef44fd737dd1ba8 Mon Sep 17 00:00:00 2001 From: yoshikazu Date: Tue, 13 Oct 2026 03:13:01 +0900 Subject: [PATCH] =?UTF-8?q?n8n=20=E9=80=A3=E6=90=BA:=20=E5=8F=97=E3=81=91?= =?UTF-8?q?=E5=8F=A3=E3=82=92=E5=91=BC=E3=81=B6=E3=82=AF=E3=83=A9=E3=82=A4?= =?UTF-8?q?=E3=82=A2=E3=83=B3=E3=83=88=E3=81=A8=E8=A8=AD=E5=AE=9A=E3=81=AE?= =?UTF-8?q?=E4=BF=9D=E6=8C=81=E3=82=92=E8=BF=BD=E5=8A=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- chktime.py | 114 ++++++++++++++++++++++++++++++++++++++++++++++ tests/test_n8n.py | 106 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 220 insertions(+) diff --git a/chktime.py b/chktime.py index 22ff7c6..fd81a33 100644 --- a/chktime.py +++ b/chktime.py @@ -1330,6 +1330,120 @@ def mark_blocks(blocks, rows, deleted_sources, now=None): return blocks +class N8n: + """n8n の work-tracker--import を呼ぶ最小のクライアント (標準ライブラリのみ)。 + + LAN からは n8n.yfox.jp の証明書名が一致しないので、verify=False なら検証を外す。 + 合言葉は X-WT-Token ヘッダーで送り、ログにも例外の文言にも出さない。 + """ + + def __init__(self, url=None, token=None, verify=False): + self.url = (url or "").rstrip("/") + self.token = token or "" + self.ctx = None + if not verify: + self.ctx = ssl.create_default_context() + self.ctx.check_hostname = False + self.ctx.verify_mode = ssl.CERT_NONE + + def call(self, path, method="GET", payload=None): + if not self.url: + raise N8nError("n8n の URL が未設定です") + if not self.token: + raise N8nError("合言葉が未設定です") + body = json.dumps(payload, ensure_ascii=False).encode("utf-8") if payload is not None else None + req = urllib.request.Request(self.url + path, data=body, method=method) + req.add_header("X-WT-Token", self.token) + req.add_header("Accept", "application/json") + if body is not None: + req.add_header("Content-Type", "application/json") + try: + with urllib.request.urlopen(req, timeout=60, context=self.ctx) as res: + raw = res.read() + except urllib.error.HTTPError as exc: + if exc.code == 403: + raise N8nError("合言葉が違います (HTTP 403)") from None + detail = one_line(exc.read().decode("utf-8", "replace"), 200) + raise N8nError(f"HTTP {exc.code}: {detail or exc.reason}") from None + except urllib.error.URLError as exc: + raise N8nError(f"接続できません: {exc.reason}") from None + except OSError as exc: + raise N8nError(f"接続できません: {exc}") from None + try: + data = json.loads(raw or b"{}") + except ValueError: + raise N8nError("JSON として読めない応答が返りました") from None + if isinstance(data, dict) and data.get("ok") is False and data.get("error"): + raise N8nError(str(data["error"])) + return data if isinstance(data, dict) else {} + + def rows(self, start, end): + return self.call(f"/webhook/wt-import-rows?from={start.isoformat()}&to={end.isoformat()}") + + def import_rows(self, rows): + return self.call("/webhook/wt-import", "POST", {"rows": rows}) + + +class N8nHolder: + """n8n 連携の設定ファイルと、そこから作ったクライアントを保持する。合言葉は UI へ返さない。""" + + LIMITS = {"gapMinutes": (1, 240), "minMinutes": (0, 240)} + + def __init__(self, path): + self.path = Path(path) + self.lock = threading.Lock() + self.cfg = load_redmine_config(self.path) # 中身は汎用の JSON 読み込み + self.client = self._build() + + def _build(self): + return N8n(self.cfg.get("url"), self.cfg.get("token"), self.cfg.get("verifySsl", False)) + + def settings(self): + with self.lock: + return int(self.cfg.get("gapMinutes", N8N_GAP_MIN)), int(self.cfg.get("minMinutes", N8N_MIN_MIN)) + + def public(self): + with self.lock: + cfg = self.cfg + return { + "url": cfg.get("url", ""), + "hasToken": bool(cfg.get("token")), + "verifySsl": bool(cfg.get("verifySsl", False)), + "gapMinutes": int(cfg.get("gapMinutes", N8N_GAP_MIN)), + "minMinutes": int(cfg.get("minMinutes", N8N_MIN_MIN)), + "configPath": str(self.path), + } + + def update(self, patch): + """URL・合言葉・区間の設定を保存する。token が空文字なら既存の合言葉を残す。""" + with self.lock: + cfg = dict(self.cfg) + if "url" in patch: + cfg["url"] = str(patch["url"] or "").strip() + if "verifySsl" in patch: + cfg["verifySsl"] = bool(patch["verifySsl"]) + for key, (low, high) in self.LIMITS.items(): + if key in patch: + try: + value = int(patch[key]) + except (TypeError, ValueError): + raise N8nError(f"{key} は整数で指定してください") from None + if not low <= value <= high: + raise N8nError(f"{key} は {low}〜{high} で指定してください") + cfg[key] = value + if patch.get("clearToken"): + cfg.pop("token", None) + elif patch.get("token"): + cfg["token"] = str(patch["token"]).strip() + self.cfg = cfg + self.client = self._build() + try: + self.path.write_text(json.dumps(cfg, ensure_ascii=False, indent=2), encoding="utf-8") + except OSError as exc: + raise N8nError(f"設定を保存できません: {exc}") from None + return self.public() + + # ---------------------------------------------------------------- HTTP diff --git a/tests/test_n8n.py b/tests/test_n8n.py index 9352d32..b52c861 100644 --- a/tests/test_n8n.py +++ b/tests/test_n8n.py @@ -170,5 +170,111 @@ class CollectAndMarkTest(unittest.TestCase): 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()