646 lines
30 KiB
Python
646 lines
30 KiB
Python
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()
|