from __future__ import annotations import contextlib import io import json import os import sys import tempfile import unittest from pathlib import Path from unittest import mock REPO_ROOT = Path(__file__).resolve().parents[1] ACK_SCRIPTS = REPO_ROOT / "skills" / "ack" / "scripts" sys.path.insert(0, str(ACK_SCRIPTS)) import feishu_bug_intake # noqa: E402 BOARD = """version: 1 project: name: demo bugIntake: provider: feishu-base profile: tenant-b baseToken: bascnDemo tableId: tblDemo viewId: vewReady fields: title: fldTitle actual: fldActual expected: fldExpected stepsToReproduce: fldSteps acceptance: fldAcceptance priority: fldPriority attachments: fldAttachments updatedAt: fldUpdated tasks: [] """ CLARIFIED_BOARD = """version: 1 project: name: demo bugIntake: provider: feishu-base workflow: clarified-writeback-v1 profile: tenant-b baseToken: bascnDemo tableId: tblDemo viewId: vewReady fields: title: 标题 details: 详细描述 problemStatement: 问题说明 expectedOutcome: 期望效果 acceptance: 验收标准 intakeStatus: 处理状态 ackTaskId: ACK任务ID attachments: 附件 updatedAt: 更新时间 tasks: [] """ class FeishuBugIntakeUnitTests(unittest.TestCase): def test_clarified_workflow_normalizes_only_source_and_clarification_fields(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: board_path = Path(temp_dir) / "tasks.yaml" board_path.write_text(CLARIFIED_BOARD, encoding="utf-8") config = feishu_bug_intake.config_from_board( feishu_bug_intake.load_board(board_path) ) row = [ "Bug", "用户描述", "问题说明", "期望效果", "1. 可观察结果", "待审核", "", [], "2026-08-03T12:00:00Z", ] with mock.patch.object(feishu_bug_intake, "profile_check"), mock.patch.object( feishu_bug_intake, "fetch_pages", return_value=[("recA", row)] ): record = feishu_bug_intake.fetch(config, None)["records"][0] self.assertEqual(record["details"], "用户描述") self.assertEqual(record["expectedOutcome"], "期望效果") self.assertNotIn("fixLogic", record) self.assertNotIn("priority", record) def test_clarified_draft_contract_rejects_fix_logic(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: path = Path(temp_dir) / "draft.json" path.write_text(json.dumps({ "problemStatement": "问题说清楚", "expectedOutcome": "期望说清楚", "acceptance": ["结果可从界面观察"], }), encoding="utf-8") draft = feishu_bug_intake.load_draft(path, "clarified-writeback-v1") self.assertEqual(set(draft), {"problemStatement", "expectedOutcome", "acceptance"}) path.write_text(json.dumps({**draft, "fixLogic": "不应出现"}), encoding="utf-8") with self.assertRaisesRegex(feishu_bug_intake.IntakeError, "exactly"): feishu_bug_intake.load_draft(path, "clarified-writeback-v1") def test_schema_plan_preserves_legacy_fields_and_adds_new_contract(self) -> None: config = { "workflow": "clarified-writeback-v1", "profile": "tenant-b", "baseToken": "bascnDemo", "tableId": "tblDemo", "viewId": "vewReady", } fields = [ {"id": "a", "name": "标题", "type": "text"}, {"id": "b", "name": "详细描述", "type": "text"}, {"id": "c", "name": "附件", "type": "attachment"}, {"id": "d", "name": "验收标准", "type": "text"}, {"id": "e", "name": "更新时间", "type": "updated_at"}, {"id": "f", "name": "期望结果", "type": "text"}, ] with mock.patch.object(feishu_bug_intake, "field_list", return_value=fields): plan = feishu_bug_intake.schema_plan(config) self.assertEqual(plan["missingFields"], ["问题说明", "期望效果", "处理状态", "ACK任务ID"]) self.assertIn("期望结果", plan["legacyFieldsPreserved"]) self.assertEqual(plan["typeConflicts"], []) self.assertEqual(plan["target"]["tableId"], "tblDemo") changed_target = {**config, "tableId": "tblOther"} self.assertNotEqual( feishu_bug_intake.schema_fingerprint(config, fields), feishu_bug_intake.schema_fingerprint(changed_target, fields), ) with self.assertRaisesRegex(feishu_bug_intake.IntakeError, "requires clarified"): feishu_bug_intake.schema_plan({**config, "workflow": "read-only-v1"}) def make_fake_cli(self, root: Path) -> tuple[Path, Path]: log_path = root / "calls.jsonl" fake = root / "lark-cli" fake.write_text( "#!" + sys.executable + "\n" "import json, pathlib, sys\n" f"log = pathlib.Path({str(log_path)!r})\n" "args = sys.argv[1:]\n" "with log.open('a') as f: f.write(json.dumps(args) + '\\n')\n" "if args[:2] == ['profile', 'list']:\n" " if '--format' in args: raise SystemExit(8)\n" " print(json.dumps([{'name':'tenant-a','appId':'cli_a','brand':'feishu','active':True},{'name':'tenant-b','appId':'cli_b','brand':'feishu','active':False}]))\n" "elif args[:2] == ['base', '+record-list']:\n" " fields=['fldTitle','fldActual','fldExpected','fldSteps','fldAcceptance','fldPriority','fldAttachments','fldUpdated']; row=[' Bug\\n title ',' actual ','expected',['one','two'],'1. save succeeds\\n2. the crash no longer occurs','P1',[{'file_token':'fileA','name':'shot.png','type':'image/png','size':3}],'2026-08-01']\n" " if 'fldFixLogic' in args: fields.append('fldFixLogic'); row.append('change parser without widening input')\n" " print(json.dumps({'code': 0, 'data': {'fields': fields, 'record_id_list': ['recA'], 'data': [row]}}))\n" "elif args[:2] == ['base', '+record-download-attachment']:\n" " out = pathlib.Path(args[args.index('--output') + 1]); out.mkdir(parents=True, exist_ok=True); (out / 'shot.png').write_bytes(b'png'); print(json.dumps({'ok': True, 'data': {}}))\n" "elif args[:2] == ['base', '+record-upsert']:\n" " print(json.dumps({'code': 0, 'data': {}}))\n" "else: raise SystemExit(2)\n", encoding="utf-8", ) fake.chmod(0o755) return fake, log_path def invoke(self, argv: list[str], executable: Path) -> tuple[int, str, str]: stdout, stderr = io.StringIO(), io.StringIO() with mock.patch.object(feishu_bug_intake, "resolve_lark_cli", return_value=executable), contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): code = feishu_bug_intake.main(argv) return code, stdout.getvalue(), stderr.getvalue() def test_reader_exposes_check_and_fetch_commands(self) -> None: output = io.StringIO() with self.assertRaises(SystemExit) as exited, contextlib.redirect_stdout(output): feishu_bug_intake.main(["--help"]) self.assertEqual(exited.exception.code, 0) self.assertIn("mark-imported", output.getvalue()) self.assertNotIn("confirm", output.getvalue()) def test_fetch_uses_mocked_trusted_executable_and_official_wire_shapes(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) board_path = temp / "tasks.yaml" board_path.write_text(BOARD, encoding="utf-8") fake, log_path = self.make_fake_cli(temp) hostile = temp / "hostile" hostile.mkdir() (hostile / "lark-cli").write_text("#!/bin/sh\nexit 99\n", encoding="utf-8") (hostile / "lark-cli").chmod(0o755) with mock.patch.dict(os.environ, {"PATH": str(hostile)}, clear=False): code, output, error = self.invoke(["fetch", str(board_path), "--output-dir", str(temp / "downloads")], fake) self.assertEqual(code, 0, error) payload = json.loads(output) self.assertEqual(payload["records"][0]["title"], "Bug title") self.assertEqual(payload["records"][0]["steps"], "one\ntwo") self.assertRegex(payload["records"][0]["sourceRef"], r"^feishu-base:sha256:[0-9a-f]{64}$") self.assertNotIn("bascnDemo", output) self.assertNotIn("fileA", output) self.assertNotIn("token", payload["records"][0]["attachments"][0]) self.assertTrue(Path(payload["records"][0]["attachments"][0]["localPath"]).is_file()) calls = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines()] self.assertEqual(calls[0], ["profile", "list"]) for call in calls: if call[:1] == ["base"]: self.assertEqual(call[call.index("--profile") + 1], "tenant-b") def test_resolver_ignores_hostile_path_and_rejects_unsafe_target(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) trusted, hostile = temp / "trusted", temp / "hostile" trusted.mkdir() hostile.mkdir() safe = trusted / "lark-cli" safe.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") safe.chmod(0o755) (hostile / "lark-cli").write_text("#!/bin/sh\nexit 99\n", encoding="utf-8") (hostile / "lark-cli").chmod(0o755) with mock.patch.object(feishu_bug_intake, "trusted_lark_cli_dirs", return_value=[trusted]), mock.patch.dict(os.environ, {"PATH": str(hostile)}, clear=False): self.assertEqual(feishu_bug_intake.resolve_lark_cli(), safe) safe.chmod(0o775) with mock.patch.object(feishu_bug_intake, "trusted_lark_cli_dirs", return_value=[trusted]): with self.assertRaisesRegex(feishu_bug_intake.IntakeError, "trusted"): feishu_bug_intake.resolve_lark_cli() def test_resolver_accepts_only_the_official_npm_wrapper_shape(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) trusted = temp / "trusted" scripts = temp / "node_modules" / "@larksuite" / "cli" / "scripts" trusted.mkdir() scripts.mkdir(parents=True) wrapper = scripts / "run.js" wrapper.write_text( "#!" + sys.executable + "\nimport json\nprint(json.dumps({'ok': True}))\n", encoding="utf-8", ) wrapper.chmod(0o755) manifest = scripts.parent / "package.json" manifest.write_text(json.dumps({ "name": "@larksuite/cli", "bin": {"lark-cli": "scripts/run.js"}, }), encoding="utf-8") native = scripts.parent / "bin" / "lark-cli" native.parent.mkdir() native.write_text( "#!" + sys.executable + "\nimport json\nprint(json.dumps({'ok': True}))\n", encoding="utf-8", ) native.chmod(0o755) (trusted / "lark-cli").symlink_to(wrapper) with mock.patch.object(feishu_bug_intake, "trusted_lark_cli_dirs", return_value=[trusted]): self.assertEqual(feishu_bug_intake.resolve_lark_cli(), native) self.assertEqual(feishu_bug_intake.run_cli(["probe"]), {"ok": True}) manifest.write_text(json.dumps({ "name": "lookalike", "bin": {"lark-cli": "scripts/run.js"}, }), encoding="utf-8") with mock.patch.object(feishu_bug_intake, "trusted_lark_cli_dirs", return_value=[trusted]): with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.resolve_lark_cli() def test_cli_environment_drops_credential_and_runtime_overrides(self) -> None: hostile = { "LARKSUITE_CLI_APP_ID": "wrong-app", "LARKSUITE_CLI_APP_SECRET": "wrong-secret", "LARKSUITE_CLI_CONFIG_DIR": "/tmp/wrong-config", "LARKSUITE_CLI_BRAND": "lark", "FEISHU_APP_SECRET": "wrong-feishu-secret", "NODE_OPTIONS": "--require=/tmp/inject.js", "PYTHONPATH": "/tmp/inject", } with mock.patch.dict(os.environ, hostile, clear=False): environment = feishu_bug_intake.cli_environment() with tempfile.TemporaryDirectory() as temp_dir: fake = Path(temp_dir) / "lark-cli" fake.write_text( "#!" + sys.executable + "\n" "import json, os\n" "keys = ['LARKSUITE_CLI_APP_ID','LARKSUITE_CLI_APP_SECRET','LARKSUITE_CLI_CONFIG_DIR','LARKSUITE_CLI_BRAND','FEISHU_APP_SECRET','NODE_OPTIONS','PYTHONPATH']\n" "print(json.dumps({'ok': True, 'data': {key: os.environ.get(key) for key in keys}}))\n", encoding="utf-8", ) fake.chmod(0o755) with mock.patch.object(feishu_bug_intake, "resolve_lark_cli", return_value=fake): child = feishu_bug_intake.run_cli(["probe"]) self.assertEqual(set(environment) - {"HOME", "PATH"}, set(environment) & {"LANG", "LC_ALL", "LC_CTYPE"}) for name in hostile: self.assertNotIn(name, environment) self.assertIsNone(child["data"][name]) def test_run_cli_rejects_error_and_ambiguous_envelopes(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) fake = temp / "lark-cli" fake.write_text( "#!" + sys.executable + "\nimport json, sys\nprint(sys.argv[1])\n", encoding="utf-8", ) fake.chmod(0o755) with mock.patch.object(feishu_bug_intake, "resolve_lark_cli", return_value=fake): for response in ('{"ok":false}', '{"code":7}', '{"data":{}}'): with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.run_cli([response]) def test_matrix_accepts_official_ok_envelope_and_data_rows(self) -> None: fields = ["fldTitle"] response = { "ok": True, "data": {"fields": fields, "record_id_list": ["recA"], "data": [["Bug"]]}, } self.assertEqual(feishu_bug_intake.matrix_from_response(response, fields), (["recA"], [["Bug"]])) def test_matrix_reorders_same_field_projection(self) -> None: response = { "ok": True, "data": { "fields": ["fldExpected", "fldTitle"], "record_id_list": ["recA"], "data": [["expected", "Bug"]], }, } self.assertEqual( feishu_bug_intake.matrix_from_response( response, ["fldTitle", "fldExpected"], ), (["recA"], [["Bug", "expected"]]), ) def test_matrix_rejects_different_field_projection_with_diagnostics(self) -> None: response = { "ok": True, "data": { "fields": ["fldActual"], "record_id_list": ["recA"], "data": [["actual"]], }, } with self.assertRaisesRegex( feishu_bug_intake.IntakeError, "expected=.*fldTitle.*actual=.*fldActual", ): feishu_bug_intake.matrix_from_response(response, ["fldTitle"]) def test_missing_priority_mapping_is_normalized_as_enrichment(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: board_path = Path(temp_dir) / "tasks.yaml" board_path.write_text( BOARD.replace(" priority: fldPriority\n", ""), encoding="utf-8", ) config = feishu_bug_intake.config_from_board( feishu_bug_intake.load_board(board_path) ) row = [ "Bug", "actual", "expected", "steps", "acceptance", [], "2026-08-01T12:00:00Z", ] with mock.patch.object(feishu_bug_intake, "profile_check"), mock.patch.object( feishu_bug_intake, "fetch_pages", return_value=[("recA", row)] ): payload = feishu_bug_intake.fetch(config, None) self.assertEqual(payload["records"][0]["priority"], "") self.assertIn("priority", payload["records"][0]["enrichmentRequired"]) def test_reviewed_workflow_requires_priority_mapping(self) -> None: board = feishu_bug_intake.load_yaml_subset( BOARD.replace( " profile: tenant-b\n", " workflow: reviewed-writeback-v1\n profile: tenant-b\n", ).replace( " acceptance: fldAcceptance\n", " acceptance: fldAcceptance\n fixLogic: fldFixLogic\n", ).replace(" priority: fldPriority\n", "") ) with self.assertRaisesRegex(feishu_bug_intake.IntakeError, "fixLogic and priority"): feishu_bug_intake.config_from_board(board) def test_fully_blank_clarified_row_is_skipped_without_key_error(self) -> None: config = feishu_bug_intake.config_from_board( feishu_bug_intake.load_yaml_subset(CLARIFIED_BOARD) ) row = ["", "", "", "", "", "", "", [], ""] with mock.patch.object(feishu_bug_intake, "profile_check"), mock.patch.object( feishu_bug_intake, "fetch_pages", return_value=[("recA", row)] ): payload = feishu_bug_intake.fetch(config, None) self.assertEqual(payload["records"], []) self.assertEqual(payload["warnings"][0]["code"], "blank_record_skipped") def test_mark_imported_binds_confirmed_record_to_existing_task(self) -> None: source_ref = "feishu-base:sha256:" + "a" * 64 revision = "sha256:" + "b" * 64 task = { "id": "BUG-001", "title": "Bug", "description": "问题说明", "actual": "用户描述", "expected": "期望效果", "acceptanceCriteria": ["结果可观察"], "source": { "kind": "feishu-base", "workflow": "clarified-writeback-v1", "ref": source_ref, "recordId": "recA", "updatedAt": "2026-08-04T10:00:00Z", "approvedRevision": revision, }, } task["source"]["approvedPayloadHash"] = feishu_bug_intake.approval_payload_hash(task) board = {"tasks": [task]} config = feishu_bug_intake.config_from_board( feishu_bug_intake.load_yaml_subset(CLARIFIED_BOARD) ) confirmed = { "recordId": "recA", "sourceRef": source_ref, "draftRevision": revision, "intakeStatus": "已确认", "ackTaskId": "", "title": "Bug", "details": "用户描述", "problemStatement": "问题说明", "expectedOutcome": "期望效果", "acceptance": "1. 结果可观察", "updatedAt": "2026-08-04T10:00:00Z", } imported = { **confirmed, "intakeStatus": "已导入", "ackTaskId": "BUG-001", } with mock.patch.object( feishu_bug_intake, "review_record", return_value=confirmed, ), mock.patch.object( feishu_bug_intake, "validate_task_board", return_value=[], ), mock.patch.object(feishu_bug_intake, "profile_check"), mock.patch.object( feishu_bug_intake, "run_cli", return_value={"ok": True}, ) as run_cli, mock.patch.object( feishu_bug_intake, "fetch", return_value={"records": [imported]}, ): result = feishu_bug_intake.mark_imported( board, config, "recA", source_ref, revision, "BUG-001", ) self.assertTrue(result["ok"]) patch = json.loads(run_cli.call_args.args[0][run_cli.call_args.args[0].index("--json") + 1]) self.assertEqual(patch, {"处理状态": "已导入", "ACK任务ID": "BUG-001"}) tampered = json.loads(json.dumps(board, ensure_ascii=False)) tampered["tasks"][0]["description"] = "未批准的改写" tampered["tasks"][0]["source"]["approvedPayloadHash"] = ( feishu_bug_intake.approval_payload_hash(tampered["tasks"][0]) ) with mock.patch.object( feishu_bug_intake, "review_record", return_value=confirmed, ), mock.patch.object( feishu_bug_intake, "validate_task_board", return_value=[], ), mock.patch.object(feishu_bug_intake, "run_cli") as blocked_write: with self.assertRaisesRegex( feishu_bug_intake.IntakeError, "does not match the approved", ): feishu_bug_intake.mark_imported( tampered, config, "recA", source_ref, revision, "BUG-001", ) blocked_write.assert_not_called() with mock.patch.object( feishu_bug_intake, "validate_task_board", return_value=["invalid"], ), mock.patch.object(feishu_bug_intake, "review_record") as blocked_read: with self.assertRaisesRegex(feishu_bug_intake.IntakeError, "board is invalid"): feishu_bug_intake.mark_imported( board, config, "recA", source_ref, revision, "BUG-001", ) blocked_read.assert_not_called() def test_optional_fix_logic_field_is_normalized_for_preapproval_review(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: board_path = Path(temp_dir) / "tasks.yaml" board_path.write_text( BOARD.replace( " profile: tenant-b\n", " workflow: reviewed-writeback-v1\n profile: tenant-b\n", ).replace( " acceptance: fldAcceptance\n", " acceptance: fldAcceptance\n fixLogic: fldFixLogic\n", ), encoding="utf-8", ) config = feishu_bug_intake.config_from_board( feishu_bug_intake.load_board(board_path) ) row = [ "Bug", "actual", "expected", "steps", "acceptance", "P1", [], "2026-08-01T12:00:00Z", "change parser without widening input", ] with mock.patch.object(feishu_bug_intake, "profile_check"), mock.patch.object( feishu_bug_intake, "fetch_pages", return_value=[("recA", row)] ): payload = feishu_bug_intake.fetch(config, None) record = payload["records"][0] self.assertEqual(record["fixLogic"], "change parser without widening input") self.assertNotIn("fixLogic", record["enrichmentRequired"]) def test_write_draft_uses_trusted_cli_and_only_configured_review_fields(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) board_path = temp / "tasks.yaml" board_path.write_text( BOARD.replace( " profile: tenant-b\n", " workflow: reviewed-writeback-v1\n profile: tenant-b\n", ).replace( " acceptance: fldAcceptance\n", " acceptance: fldAcceptance\n fixLogic: fldFixLogic\n", ), encoding="utf-8", ) draft_path = temp / "draft.json" draft_path.write_text(json.dumps({ "fixLogic": "change parser without widening input", "acceptance": ["save succeeds", "the crash no longer occurs"], }), encoding="utf-8") fake, log_path = self.make_fake_cli(temp) fetched_code, fetched_output, fetched_error = self.invoke([ "fetch", str(board_path), ], fake) self.assertEqual(fetched_code, 0, fetched_error) fetched_record = json.loads(fetched_output)["records"][0] code, output, error = self.invoke([ "write-draft", str(board_path), "--record-id", "recA", "--expected-source-ref", fetched_record["sourceRef"], "--expected-draft-revision", fetched_record["draftRevision"], "--input", str(draft_path), ], fake) self.assertEqual(code, 0, error) self.assertEqual(json.loads(output)["written"], ["fixLogic", "acceptance"]) self.assertRegex(json.loads(output)["draftRevision"], r"^sha256:[0-9a-f]{64}$") calls = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines()] write_call = next(call for call in calls if call[:2] == ["base", "+record-upsert"]) self.assertEqual(write_call[write_call.index("--profile") + 1], "tenant-b") patch = json.loads(write_call[write_call.index("--json") + 1]) self.assertEqual(set(patch), {"fldFixLogic", "fldAcceptance"}) self.assertIn("1. save succeeds", patch["fldAcceptance"]) imported_code, imported_output, imported_error = self.invoke([ "import-approved", str(board_path), "--record-id", "recA", "--expected-source-ref", fetched_record["sourceRef"], "--expected-draft-revision", fetched_record["draftRevision"], ], fake) self.assertEqual(imported_code, 0, imported_error) task_draft = json.loads(imported_output)["taskDraft"] self.assertEqual(task_draft["source"]["workflow"], "reviewed-writeback-v1") self.assertEqual( task_draft["acceptanceCriteria"], ["save succeeds", "the crash no longer occurs"], ) self.assertRegex( task_draft["source"]["approvedPayloadHash"], r"^sha256:[0-9a-f]{64}$", ) def test_write_draft_rejects_stale_revision_before_mutation(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) board_path = temp / "tasks.yaml" board_path.write_text( BOARD.replace( " profile: tenant-b\n", " workflow: reviewed-writeback-v1\n profile: tenant-b\n", ).replace( " acceptance: fldAcceptance\n", " acceptance: fldAcceptance\n fixLogic: fldFixLogic\n", ), encoding="utf-8", ) draft_path = temp / "draft.json" draft_path.write_text(json.dumps({ "fixLogic": "fix", "acceptance": ["pass"], }), encoding="utf-8") fake, log_path = self.make_fake_cli(temp) fetched = json.loads(self.invoke(["fetch", str(board_path)], fake)[1])["records"][0] code, _, _ = self.invoke([ "write-draft", str(board_path), "--record-id", "recA", "--expected-source-ref", fetched["sourceRef"], "--expected-draft-revision", "sha256:" + "0" * 64, "--input", str(draft_path), ], fake) self.assertNotEqual(code, 0) calls = [json.loads(line) for line in log_path.read_text(encoding="utf-8").splitlines()] self.assertFalse(any(call[:2] == ["base", "+record-upsert"] for call in calls)) def test_write_draft_rejects_symlink_and_unknown_fields(self) -> None: with tempfile.TemporaryDirectory() as temp_dir: temp = Path(temp_dir) target = temp / "draft.json" target.write_text(json.dumps({ "fixLogic": "fix", "acceptance": ["pass"], "unexpected": "do not write", }), encoding="utf-8") with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.load_draft(target) target.write_text(json.dumps({ "fixLogic": "fix", "acceptance": ["pass"], }), encoding="utf-8") linked = temp / "linked.json" linked.symlink_to(target) with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.load_draft(linked) def test_draft_revision_binds_attachment_identity(self) -> None: record = { "sourceRef": "feishu-base:sha256:" + "a" * 64, "updatedAt": "2026-08-03T12:00:00+08:00", "title": "Bug", "actual": "crash", "expected": "save", "steps": "open then save", "fixLogic": "fix parser", "acceptance": "save succeeds", "priority": "P1", "attachments": [{"name": "same.png", "type": "image/png", "size": 4}], } first = feishu_bug_intake.draft_revision(record, ["fileTokenA"]) second = feishu_bug_intake.draft_revision(record, ["fileTokenB"]) self.assertNotEqual(first, second) def test_attachments_and_required_text_are_resource_bounded_and_strict(self) -> None: with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.attachment_items([{ "file_token": "fileA", "name": "huge.png", "type": "image/png", "size": feishu_bug_intake.MAX_ATTACHMENT_BYTES + 1, }]) too_many = [ {"file_token": f"file{index}", "name": f"{index}.png", "size": 1} for index in range(feishu_bug_intake.MAX_ATTACHMENTS_PER_RECORD + 1) ] with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.attachment_items(too_many) with self.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.text({"unexpected": "value"}) if __name__ == "__main__": unittest.main()