#!/usr/bin/env python3 """判定ファイルと判定の画面のテスト (依頼 183). KIOXIA にも API にも触らない (一時フォルダだけ). python3 pi05/verdict_viewer/test_verdicts.py """ import json import os import sys import tempfile import threading import unittest import urllib.error import urllib.request from pathlib import Path from unittest import mock HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) import verdicts as V # noqa: E402 import seed_verdicts as S # noqa: E402 import server # noqa: E402 def row(stem, outcome, labeler, **kw): r = {"run_stem": stem, "pick": "unknown", "place": "unknown", "outcome": outcome, "labeler": labeler} r.update(kw) return r def make_runs(root): """偽の記録: a = json に success, b = json の outcome null, c = frames と log.jsonl だけ, d = 何もない.""" root = Path(root) for stem in ("pi05_20261001_000001", "pi05_20261001_000002", "pi05_20261001_000003", "pi05_20261001_000004"): (root / stem).mkdir(parents=True) a = root / "pi05_20261001_000001" (a / f"{a.name}.json").write_text(json.dumps({"outcome": "success", "ended_at": "2026-10-01T00:00:10", "judgments": {"pick": {"verdict": "yes"}, "place": {"verdict": "yes"}}})) b = root / "pi05_20261001_000002" (b / f"{b.name}.json").write_text(json.dumps({"outcome": None})) c = root / "pi05_20261001_000003" (c / "frames").mkdir() for i in range(3): (c / "frames" / f"{i:04d}.jpg").write_bytes(b"\xff\xd8\xff\xd9") (c / "frames" / "._0000.jpg").write_bytes(b"x") (c / "log.jsonl").write_text("\n".join(json.dumps({"chunk": str(i), "t": str(100 + i), "executed": "True", "state": json.dumps([i, 1, 2, 3, 4, -30, 0 if i < 2 else 1])}) for i in range(3)) + "\n") return root class TestAppend(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.csv = Path(self.tmp.name) / "v.csv" def tearDown(self): self.tmp.cleanup() def test_append_only_keeps_existing_bytes(self): V.append_rows([row("pi05_1", "success", "run_input"), row("pi05_2", "failure", "ai")], self.csv) before = self.csv.read_bytes() V.append_rows([row("pi05_1", "failure", "human_only")], self.csv) after = self.csv.read_bytes() self.assertTrue(after.startswith(before), "既存の行が変わった") self.assertEqual(after.count(b"labeled_at,"), 1, "見出しが 2 回書かれた") self.assertEqual(len(V.read_rows(self.csv)), 3) def test_header_columns(self): V.append_rows([row("pi05_1", "success", "ai")], self.csv) self.assertEqual(self.csv.read_text().splitlines()[0].split(","), V.COLUMNS) def test_latest_row_wins(self): V.append_rows([row("pi05_1", "success", "ai"), row("pi05_2", "failure", "ai")], self.csv) V.append_rows([row("pi05_1", "failure", "human_fixed_ai")], self.csv) V.append_rows([row("pi05_1", "unknown", "human_only")], self.csv) lat = V.latest_by_stem(V.read_rows(self.csv)) self.assertEqual(lat["pi05_1"]["outcome"], "unknown") self.assertEqual(lat["pi05_2"]["outcome"], "failure") self.assertEqual(V.latest_by_stem(V.read_rows(self.csv), labelers=("ai",))["pi05_1"]["outcome"], "success") def test_bad_values_write_nothing(self): V.append_rows([row("pi05_1", "success", "ai")], self.csv) before = self.csv.read_bytes() for bad in (row("pi05_1", "great", "ai"), row("pi05_1", "success", "robot"), row("", "success", "ai"), row("pi05_1", "success", "ai", pick="maybe")): with self.assertRaises(ValueError): V.append_rows([row("pi05_9", "success", "ai"), bad], self.csv) self.assertEqual(self.csv.read_bytes(), before, "まちがった行のときに途中まで書いた") def test_place_not_applicable_only_for_place(self): # 依頼 193: place は not_applicable (そこまで行かなかった) を書ける. pick には書けない V.append_rows([row("pi05_1", "failure", "human_only", pick="no", place="not_applicable")], self.csv) self.assertEqual(V.read_rows(self.csv)[-1]["place"], "not_applicable") before = self.csv.read_bytes() with self.assertRaises(ValueError): V.append_rows([row("pi05_2", "failure", "human_only", pick="not_applicable", place="not_applicable")], self.csv) self.assertEqual(self.csv.read_bytes(), before) def test_comma_and_newline_in_reason(self): V.append_rows([row("pi05_1", "failure", "ai", ai_reason='箱の外, "落とした"\n2 行目')], self.csv) self.assertEqual(V.read_rows(self.csv)[0]["ai_reason"], '箱の外, "落とした"\n2 行目') def test_broken_file_stops(self): V.append_rows([row("pi05_1", "success", "ai")], self.csv) with open(self.csv, "ab") as f: f.write(b"2026,pi05_2,yes") # 書きかけ (改行なし) before = self.csv.read_bytes() with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_3", "success", "ai")], self.csv) self.assertEqual(self.csv.read_bytes(), before) other = Path(self.tmp.name) / "other.csv" other.write_text("a,b\n1,2\n") with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_3", "success", "ai")], other) self.assertEqual(other.read_text(), "a,b\n1,2\n") def test_unclosed_quote_stops(self): V.append_rows([row("pi05_1", "success", "ai")], self.csv) with open(self.csv, "ab") as f: f.write(b'2026,pi05_2,yes,yes,success,ai,,,,"unfinished\n') before = self.csv.read_bytes() with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_3", "success", "ai")], self.csv) self.assertEqual(self.csv.read_bytes(), before) def test_long_note_is_cut(self): V.append_rows([row("pi05_1", "success", "human_only", note="あ" * 200000, ai_reason="い" * 200000)], self.csv) r = V.read_rows(self.csv)[0] self.assertEqual(len(r["note"]), V.MAX_NOTE + 1) self.assertEqual(len(r["ai_reason"]), V.MAX_REASON + 1) V.append_rows([row("pi05_2", "success", "ai")], self.csv) def test_forbid_dirs(self): d = Path(self.tmp.name) / "myruns" d.mkdir() with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_1", "success", "ai")], d / "v.csv", forbid_dirs=(d,)) self.assertFalse((d / "v.csv").exists()) def test_refuses_symlink_and_record_dirs(self): real = Path(self.tmp.name) / "real.csv" V.append_rows([row("pi05_1", "success", "ai")], real) link = Path(self.tmp.name) / "link.csv" link.symlink_to(real) with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_2", "success", "ai")], link) runs = Path(self.tmp.name) / "runs" (runs / "pi05_1").mkdir(parents=True) with mock.patch.object(V, "RUNS_DIR", runs): with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_2", "success", "ai")], runs / "pi05_1" / "x.csv") with self.assertRaises(V.BrokenCsv): V.append_rows([row("pi05_2", "success", "ai")], Path(self.tmp.name) / "x.json") self.assertEqual(len(V.read_rows(real)), 1) def test_skip_if_labeled_by(self): V.append_rows([row("pi05_1", "failure", "human_only")], self.csv) out = V.append_rows([row("pi05_1", "success", "ai"), row("pi05_2", "success", "ai")], self.csv, skip_if_labeled_by=V.HUMAN_LABELERS) self.assertEqual([r["run_stem"] for r in out], ["pi05_2"]) self.assertEqual(V.latest_by_stem(V.read_rows(self.csv))["pi05_1"]["labeler"], "human_only") class TestMount(unittest.TestCase): def test_not_mounted_stops_and_writes_nothing(self): target = V.KIOXIA_MOUNT / "r2r2r-data" / "runs" / "pi05_policy_verdicts_v1_TEST_never.csv" with mock.patch.object(V.os.path, "ismount", lambda p: False), \ mock.patch("builtins.open", side_effect=AssertionError("open してはいけない")): with self.assertRaises(V.NotMounted): V.append_rows([row("pi05_1", "success", "ai")], target) def test_mounted_check_only_for_kioxia(self): with mock.patch.object(V.os.path, "ismount", lambda p: False): V.ensure_ssd_mounted("/tmp/whatever.csv") # KIOXIA の外は確かめない (テスト用) def test_seed_aborts_without_mount(self): with mock.patch.object(V.os.path, "ismount", lambda p: False), \ mock.patch.object(sys, "argv", ["seed", "--run-input"]): self.assertEqual(S.main(), 2) def test_server_returns_503_without_mount(self): with tempfile.TemporaryDirectory() as d: runs = make_runs(Path(d) / "runs") app = server.App(runs, V.KIOXIA_MOUNT / "r2r2r-data" / "runs" / "never_TEST.csv") with mock.patch.object(V.os.path, "ismount", lambda p: False): with self.assertRaises(V.NotMounted): app.add_verdict({"run_stem": "pi05_20261001_000003", "outcome": "success"}) class TestLabeler(unittest.TestCase): def test_labeler(self): self.assertEqual(V.human_labeler_for("success", None), "human_only") self.assertEqual(V.human_labeler_for("success", {"outcome": "success"}), "human") self.assertEqual(V.human_labeler_for("failure", {"outcome": "success"}), "human_fixed_ai") class TestSeed(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.runs = make_runs(Path(self.tmp.name) / "runs") self.csv = Path(self.tmp.name) / "v.csv" def tearDown(self): self.tmp.cleanup() def test_run_input_copy_once(self): c = S.do_run_input(self.csv, self.runs, dry_run=False) self.assertEqual(c["success"], 1) rows = V.read_rows(self.csv) self.assertEqual([(r["run_stem"], r["outcome"], r["labeler"], r["pick"]) for r in rows], [("pi05_20261001_000001", "success", "run_input", "yes")]) S.do_run_input(self.csv, self.runs, dry_run=False) self.assertEqual(len(V.read_rows(self.csv)), 1, "2 回流したら行が増えた") def test_ai_targets_and_not_judgeable(self): calls = [] def fake_api(key, imgs, evidence): calls.append((len(imgs), evidence)) return {"pick": "yes", "place": "no", "outcome": "failure", "reason": "箱の外"}, "" S.do_run_input(self.csv, self.runs, dry_run=False) with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", fake_api): c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) self.assertEqual(calls, [(3, "frames")]) # c だけ API を呼ぶ (frames 3 枚・._ は除く) self.assertEqual(c["failure"], 1) self.assertEqual(c["not_judgeable"], 2) # b (null・素材なし) と d (何もない) ai = V.latest_by_stem(V.read_rows(self.csv), labelers=("ai",)) self.assertNotIn("pi05_20261001_000001", ai) self.assertEqual(ai["pi05_20261001_000003"]["ai_reason"], "[送ったチャンク 3] 箱の外") n = len(V.read_rows(self.csv)) with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", fake_api): S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) self.assertEqual(len(V.read_rows(self.csv)), n, "AI の行が 2 回書かれた") def test_ai_skips_when_api_unavailable(self): def boom(*a): raise S.ApiUnavailable("http_401") with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", boom): c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) self.assertEqual(c["skipped"], 1) self.assertNotIn("pi05_20261001_000003", V.latest_by_stem(V.read_rows(self.csv))) def test_seed_never_adds_after_human(self): V.append_rows([row("pi05_20261001_000003", "success", "human_only"), row("pi05_20261001_000001", "failure", "human_only")], self.csv) with mock.patch.object(S, "api_key", lambda: "k"), \ mock.patch.object(S, "call_api", lambda *a: ({"outcome": "failure"}, "")): S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only=None) S.do_run_input(self.csv, self.runs, dry_run=False) lat = V.latest_by_stem(V.read_rows(self.csv)) self.assertEqual(lat["pi05_20261001_000003"]["labeler"], "human_only") self.assertEqual(lat["pi05_20261001_000001"]["labeler"], "human_only") def test_seed_rechecks_inside_lock(self): # API を待つ間に人が確定した → AI の行は足さない def api_then_human(*a): V.append_rows([row("pi05_20261001_000003", "failure", "human_only")], self.csv) return {"outcome": "success"}, "" with mock.patch.object(S, "api_key", lambda: "k"), mock.patch.object(S, "call_api", api_then_human): c = S.do_ai(self.csv, self.runs, dry_run=False, limit=0, only={"pi05_20261001_000003"}) self.assertEqual(c["skipped_human"], 1) self.assertEqual(V.latest_by_stem(V.read_rows(self.csv))["pi05_20261001_000003"]["labeler"], "human_only") def test_symlinked_side_files_are_ignored(self): outside = Path(self.tmp.name) / "outside.jsonl" outside.write_text(json.dumps({"chunk": "0", "executed": "True"}) + "\n") rd = self.runs / "pi05_20261001_000004" (rd / "log.jsonl").symlink_to(outside) (rd / f"{rd.name}.json").symlink_to(outside) self.assertEqual(V.read_log(rd), []) self.assertIsNone(V.read_run_json(rd)) def test_server_refuses_csv_inside_runs_dir(self): app = server.App(self.runs, self.runs / "pi05_20261001_000003" / "v.csv") with self.assertRaises(V.BrokenCsv): app.add_verdict({"run_stem": "pi05_20261001_000003", "outcome": "success"}) self.assertFalse((self.runs / "pi05_20261001_000003" / "v.csv").exists()) def test_symlinked_run_and_frames_are_ignored(self): outside = Path(self.tmp.name) / "outside" (outside / "frames").mkdir(parents=True) (outside / "frames" / "0000.jpg").write_bytes(b"secret") (self.runs / "pi05_20261001_000099").symlink_to(outside) (self.runs / "pi05_20261001_000003" / "frames" / "0009.jpg").symlink_to(outside / "frames" / "0000.jpg") self.assertNotIn("pi05_20261001_000099", V.list_run_stems(self.runs)) self.assertNotIn("0009.jpg", V.frame_files(self.runs / "pi05_20261001_000003")) def test_api_key_export_line(self): with tempfile.TemporaryDirectory() as d: p = Path(d) / ".anthropic_key" p.write_text("export ANTHROPIC_API_KEY='sk-test'\n") with mock.patch.dict(os.environ, {"HOME": d}, clear=False), \ mock.patch.dict(os.environ, {}, clear=False) as env: env.pop("ANTHROPIC_API_KEY", None) self.assertEqual(S.api_key(), "sk-test") class TestServer(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.runs = make_runs(Path(self.tmp.name) / "runs") self.csv = Path(self.tmp.name) / "v.csv" V.append_rows([row("pi05_20261001_000003", "success", "ai", ai_reason="入っている", evidence="frames")], self.csv) self.app = server.App(self.runs, self.csv) self.srv = server.ThreadingHTTPServer(("127.0.0.1", 0), server.make_handler(self.app)) threading.Thread(target=self.srv.serve_forever, daemon=True).start() self.base = f"http://127.0.0.1:{self.srv.server_address[1]}" def tearDown(self): self.srv.shutdown() self.srv.server_close() self.tmp.cleanup() def get(self, path): with urllib.request.urlopen(self.base + path) as r: return json.loads(r.read()) def post(self, body): req = urllib.request.Request(self.base + "/api/verdict", data=json.dumps(body).encode(), headers={"Content-Type": "application/json"}, method="POST") try: with urllib.request.urlopen(req) as r: return r.status, json.loads(r.read()) except urllib.error.HTTPError as e: return e.code, json.loads(e.read()) def test_list_and_detail(self): runs = {r["run_stem"]: r for r in self.get("/api/runs")["runs"]} self.assertEqual(len(runs), 4) self.assertEqual(runs["pi05_20261001_000003"]["ai_outcome"], "success") self.assertFalse(runs["pi05_20261001_000003"]["confirmed"]) d = self.get("/api/run/pi05_20261001_000003") self.assertEqual(d["frames"], ["0000.jpg", "0001.jpg", "0002.jpg"]) self.assertEqual(len(d["states"]["joints"]), 3) self.assertEqual(d["states"]["grip"][-1][1], 1.0) self.assertEqual(d["executed_chunks"], 3) def test_post_appends_and_labeler(self): before = self.csv.read_bytes() st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "place": "no", "note": "箱の縁"}) self.assertEqual(st, 200) self.assertTrue(self.csv.read_bytes().startswith(before)) last = V.read_rows(self.csv)[-1] self.assertEqual((last["labeler"], last["outcome"], last["pick"], last["place"], last["ai_outcome"], last["ai_reason"], last["evidence"], last["note"]), ("human_fixed_ai", "failure", "unknown", "no", "success", "入っている", "frames", "箱の縁")) st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "success"}) self.assertEqual((j["row"]["labeler"], j["row"]["pick"], j["row"]["place"]), ("human", "yes", "yes")) st, j = self.post({"run_stem": "pi05_20261001_000004", "outcome": "not_judgeable"}) self.assertEqual(j["row"]["labeler"], "human_only") runs = {r["run_stem"]: r for r in self.get("/api/runs")["runs"]} self.assertTrue(runs["pi05_20261001_000003"]["confirmed"]) self.assertEqual(runs["pi05_20261001_000003"]["outcome"], "success") def test_post_place_not_applicable(self): # 依頼 193: 画面から place=not_applicable を送れる. pick=not_applicable は 400 で書かない st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "pick": "no", "place": "not_applicable"}) self.assertEqual(st, 200) self.assertEqual((j["row"]["pick"], j["row"]["place"]), ("no", "not_applicable")) before = self.csv.read_bytes() st, j = self.post({"run_stem": "pi05_20261001_000003", "outcome": "failure", "pick": "not_applicable"}) self.assertEqual(st, 400) self.assertEqual(self.csv.read_bytes(), before) def test_same_op_id_writes_once(self): n = len(V.read_rows(self.csv)) b = {"op_id": "abc", "run_stem": "pi05_20261001_000003", "outcome": "failure"} self.assertEqual(self.post(b)[0], 200) self.assertEqual(self.post(b)[0], 200) self.assertEqual(len(V.read_rows(self.csv)), n + 1) def test_media_link_to_other_run_is_refused(self): other = self.runs / "pi05_20261001_000004" / "frames" other.mkdir() (other / "0000.jpg").write_bytes(b"other") (self.runs / "pi05_20261001_000003" / "frames" / "0008.jpg").symlink_to(other / "0000.jpg") with self.assertRaises(urllib.error.HTTPError) as cm: urllib.request.urlopen(self.base + "/media/pi05_20261001_000003/frames/0008.jpg") self.assertEqual(cm.exception.code, 404) def test_post_rejects_bad(self): n = len(V.read_rows(self.csv)) self.assertEqual(self.post({"run_stem": "pi05_20261001_000003", "outcome": "great"})[0], 400) self.assertEqual(self.post({"run_stem": "../etc", "outcome": "success"})[0], 400) self.assertEqual(self.post({"run_stem": "pi05_20990101_000000", "outcome": "success"})[0], 400) self.assertEqual(len(V.read_rows(self.csv)), n) def test_media_and_traversal(self): with urllib.request.urlopen(self.base + "/media/pi05_20261001_000003/frames/0001.jpg") as r: self.assertEqual(r.read(), b"\xff\xd8\xff\xd9") for p in ("/media/pi05_20261001_000003/frames/..%2Flog.jsonl", "/media/..%2F..%2Fetc/top.mp4"): with self.assertRaises(urllib.error.HTTPError) as cm: urllib.request.urlopen(self.base + p) self.assertEqual(cm.exception.code, 404) class TestConfirm192(unittest.TestCase): """依頼 192: y/n/u/x は選ぶだけ・確定 (ボタンか Enter) で 1 行だけ書く. 画面を Chrome で動かす (playwright がなければ飛ばす).""" @classmethod def setUpClass(cls): try: from playwright.sync_api import sync_playwright cls.pw = sync_playwright().start() cls.browser = cls.pw.chromium.launch(channel="chrome") except Exception as e: # noqa: BLE001 raise unittest.SkipTest(f"playwright / Chrome がない: {e}") @classmethod def tearDownClass(cls): cls.browser.close() cls.pw.stop() def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.runs = make_runs(Path(self.tmp.name) / "runs") self.csv = Path(self.tmp.name) / "v.csv" V.append_rows([row("pi05_20261001_000003", "success", "ai", ai_reason="入っている", evidence="frames")], self.csv) self.srv = server.ThreadingHTTPServer(("127.0.0.1", 0), server.make_handler(server.App(self.runs, self.csv))) threading.Thread(target=self.srv.serve_forever, daemon=True).start() self.pg = self.browser.new_page() self.pg.goto(f"http://127.0.0.1:{self.srv.server_address[1]}/") self.wait_loaded() def tearDown(self): self.pg.close() self.srv.shutdown() self.srv.server_close() self.tmp.cleanup() def wait_loaded(self): self.pg.wait_for_function("() => { const t = document.querySelector('#title').textContent;" " return t.startsWith('pi05_') && !t.includes('読み込み中'); }") def n_rows(self): return len(V.read_rows(self.csv)) def on_keys(self): return self.pg.eval_on_selector_all("button.key.on", "e => e.map(x => x.dataset.oc)") def disabled(self): return self.pg.eval_on_selector("#confirm", "e => e.disabled") def test_choose_only_does_not_write(self): n = self.n_rows() self.assertTrue(self.disabled()) self.pg.keyboard.press("n") self.assertEqual(self.on_keys(), ["failure"]) self.assertFalse(self.disabled()) self.pg.click('button.key[data-oc="success"]') # 選び直し self.assertEqual(self.on_keys(), ["success"]) self.pg.wait_for_timeout(600) self.assertEqual(self.n_rows(), n) def test_confirm_before_choose_does_nothing(self): n = self.n_rows() title = self.pg.text_content("#title") self.pg.keyboard.press("Enter") self.pg.click("#confirm", force=True) self.pg.wait_for_timeout(600) self.assertEqual(self.n_rows(), n) self.assertEqual(self.pg.text_content("#title"), title) def test_enter_writes_one_row(self): n = self.n_rows() stem = self.pg.text_content("#title") self.pg.keyboard.press("n") self.pg.keyboard.press("Enter") self.pg.keyboard.press("Enter") # 0.4 秒の間の 2 回目は書かない self.pg.wait_for_function("() => document.querySelector('#toast').textContent.includes('確定しました')") self.assertIn(stem, self.pg.text_content("#toast")) self.pg.wait_for_timeout(800) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["run_stem"], rows[-1]["outcome"]), (stem, "failure")) def test_button_writes_one_row(self): n = self.n_rows() stem = self.pg.text_content("#title") self.pg.keyboard.press("y") self.pg.click("#confirm") self.pg.click("#confirm", force=True) self.pg.wait_for_timeout(1000) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["run_stem"], rows[-1]["outcome"]), (stem, "success")) def test_enter_in_note_confirms_but_keys_do_not_choose(self): n = self.n_rows() self.pg.keyboard.press("u") self.pg.click("#note") self.pg.keyboard.type("yn") self.assertEqual(self.on_keys(), ["unknown"]) self.pg.keyboard.press("Enter") self.pg.wait_for_timeout(1000) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["outcome"], rows[-1]["note"]), ("unknown", "yn")) def test_moving_discards_choice(self): n = self.n_rows() stem = self.pg.text_content("#title") self.pg.keyboard.press("n") self.pg.keyboard.press("ArrowRight") self.wait_loaded() self.assertNotEqual(self.pg.text_content("#title"), stem) self.assertEqual(self.on_keys(), []) self.assertTrue(self.disabled()) self.pg.keyboard.press("ArrowLeft") self.wait_loaded() self.assertEqual(self.pg.text_content("#title"), stem) self.assertEqual(self.on_keys(), []) self.pg.keyboard.press("Enter") self.pg.wait_for_timeout(600) self.assertEqual(self.n_rows(), n) # ---- 依頼 193: つかむ / 置くを日本語に・つかめなかったで置くは自動でそこまで行かなかった ---- def pp_on(self): return self.pg.eval_on_selector_all("[data-pp].on", "e => e.map(x => x.dataset.pp + '=' + x.dataset.v)") def test_pp_labels_are_japanese(self): got = self.pg.eval_on_selector_all("[data-pp]", "e => e.map(x => x.dataset.pp + '=' + x.dataset.v + ':' + x.textContent)") self.assertEqual(got, ["pick=yes:つかめた", "pick=no:つかめなかった", "pick=unknown:分からない", "place=yes:置けた", "place=no:置けなかった", "place=not_applicable:そこまで行かなかった", "place=unknown:分からない"]) self.assertEqual(self.pg.eval_on_selector_all("#panel .row .lbl", "e => e.map(x => x.textContent)"), ["pick", "place"]) def test_pick_no_selects_place_not_applicable(self): n = self.n_rows() self.pg.click('[data-pp="pick"][data-v="no"]') self.assertEqual(self.pp_on(), ["pick=no", "place=not_applicable"]) self.pg.click('[data-pp="place"][data-v="unknown"]') # 選び直せる self.assertEqual(self.pp_on(), ["pick=no", "place=unknown"]) self.pg.click('[data-pp="pick"][data-v="yes"]') # つかめたに戻すと置くは選んでいない self.assertEqual(self.pp_on(), ["pick=yes"]) self.pg.click('[data-pp="pick"][data-v="no"]') self.pg.wait_for_timeout(300) self.assertEqual(self.n_rows(), n) # 選ぶだけでは書かない stem = self.pg.text_content("#title") self.pg.keyboard.press("n") self.pg.keyboard.press("Enter") self.pg.wait_for_timeout(1000) rows = V.read_rows(self.csv) self.assertEqual(len(rows), n + 1) self.assertEqual((rows[-1]["run_stem"], rows[-1]["pick"], rows[-1]["place"]), (stem, "no", "not_applicable")) class TestRule185(unittest.TestCase): """エイイチの決め (2026-11-14 依頼 185): run_input も確定・送ったチャンク 0 → not_applicable・素材不足 → not_judgeable.""" def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.runs = Path(self.tmp.name) / "runs" self.csv = Path(self.tmp.name) / "v.csv" # stem: (送ったチャンク, 先に入れる行) self.case = { "pi05_20261002_000001": (0, [row("pi05_20261002_000001", "not_judgeable", "ai", evidence="frames")]), # 2 と 3 → 2 "pi05_20261002_000002": (2, [row("pi05_20261002_000002", "not_judgeable", "ai", evidence="frames")]), # 3 "pi05_20261002_000003": (2, [row("pi05_20261002_000003", "not_judgeable", "ai", evidence="")]), # 3 (素材なし) "pi05_20261002_000004": (0, [row("pi05_20261002_000004", "failure", "ai", evidence="top.mp4")]), # 2 "pi05_20261002_000005": (0, [row("pi05_20261002_000005", "not_judgeable", "ai", evidence="frames"), row("pi05_20261002_000005", "failure", "human_fixed_ai")]), # 人の行 → 足さない "pi05_20261002_000006": (2, [row("pi05_20261002_000006", "not_judgeable", "ai", evidence=""), row("pi05_20261002_000006", "success", "human_only")]), # 人の行 → 足さない "pi05_20261002_000007": (0, [row("pi05_20261002_000007", "success", "run_input")]), # run_input → 足さない "pi05_20261002_000008": (2, [row("pi05_20261002_000008", "not_judgeable", "ai", evidence="top.mp4")]), # 当たらない "pi05_20261002_000009": (3, [row("pi05_20261002_000009", "failure", "ai", evidence="frames")]), # 当たらない } rows = [] for stem, (n, rs) in self.case.items(): d = self.runs / stem d.mkdir(parents=True) lines = [{"type": "model"}] + [{"chunk": str(i), "executed": "True"} for i in range(n)] + [{"chunk": "9", "executed": "False"}] (d / "log.jsonl").write_text("\n".join(json.dumps(x) for x in lines) + "\n") rows += rs V.append_rows(rows, self.csv) def tearDown(self): self.tmp.cleanup() def rule_rows(self): return {r["run_stem"]: r for r in V.read_rows(self.csv) if r["labeler"] == "rule"} def test_rule_rows(self): c = S.do_rule(self.csv, self.runs, dry_run=False) got = {k[-1]: v["outcome"] for k, v in self.rule_rows().items()} self.assertEqual(got, {"1": "not_applicable", "2": "not_judgeable", "3": "not_judgeable", "4": "not_applicable"}) self.assertEqual((c["rule2"], c["rule3"], c["both"]), (2, 2, 1)) for r in self.rule_rows().values(): self.assertIn("エイイチの決め 2026-11-14 依頼185", r["note"]) self.assertIn("決め2", self.rule_rows()["pi05_20261002_000001"]["note"]) self.assertIn("決め3", self.rule_rows()["pi05_20261002_000002"]["note"]) def test_rule_never_after_human_or_run_input(self): S.do_rule(self.csv, self.runs, dry_run=False) for stem in ("pi05_20261002_000005", "pi05_20261002_000006", "pi05_20261002_000007"): self.assertNotIn(stem, self.rule_rows()) def test_rule_twice_adds_nothing(self): S.do_rule(self.csv, self.runs, dry_run=False) before = self.csv.read_bytes() S.do_rule(self.csv, self.runs, dry_run=False) self.assertEqual(self.csv.read_bytes(), before) def test_rule_rechecks_inside_lock(self): # 一覧を読んだあとに人の行が足されたら, rule の行は足さない real = V.read_rows def late_human(p): rows = real(p) V.append_rows([row("pi05_20261002_000002", "failure", "human")], self.csv) return rows with mock.patch.object(V, "read_rows", side_effect=late_human): S.do_rule(self.csv, self.runs, dry_run=False) self.assertNotIn("pi05_20261002_000002", self.rule_rows()) def test_confirmed_and_success_rate(self): S.do_rule(self.csv, self.runs, dry_run=False) app = server.App(self.runs, self.csv) conf = {r["run_stem"][-1]: r["confirmed"] for r in app.runs_summary()} self.assertEqual(conf, {"1": True, "2": True, "3": True, "4": True, "5": True, "6": True, "7": True, "8": False, "9": False}) st = app.stats() # 確定: success 2 (6・7) / failure 1 (5). not_applicable 2・not_judgeable 2 は分母に入れない. AI の行も入れない self.assertEqual((st["success"], st["failure"], st["denominator"], st["confirmed"]), (2, 1, 3, 7)) self.assertAlmostEqual(st["rate"], 2 / 3) self.assertEqual(st["by_outcome"]["not_applicable"], 2) def test_success_rate_denominator(self): rows = [row("a", "success", "human"), row("b", "failure", "run_input"), row("c", "unknown", "human"), row("d", "not_judgeable", "rule"), row("e", "not_applicable", "rule"), row("f", "success", "ai"), row("b", "success", "human_fixed_ai")] # b は後ろの人の行が正 st = V.success_rate(rows) self.assertEqual((st["success"], st["failure"], st["denominator"]), (2, 0, 2)) self.assertEqual(st["rate"], 1.0) self.assertIsNone(V.success_rate([row("x", "not_applicable", "rule")])["rate"]) def test_ai_after_confirmed_does_not_unconfirm(self): rows = [row("a", "not_applicable", "rule"), row("a", "failure", "ai")] self.assertEqual(V.current_by_stem(rows)["a"]["labeler"], "rule") def test_ai_cannot_write_not_applicable(self): r = S.ai_row("pi05_x", {"outcome": "not_applicable", "pick": "no", "place": "no"}, "frames", 0) self.assertEqual(r["outcome"], "unknown") if __name__ == "__main__": unittest.main(verbosity=1)