from __future__ import annotations import json import os import subprocess import sys import tempfile import unittest import contextlib import io from unittest import mock from pathlib import Path ROOT = Path(__file__).resolve().parents[1] READER = ROOT / "skills/ack/scripts/feishu_bug_intake.py" sys.path.insert(0, str(READER.parent)) import feishu_bug_intake # noqa: E402 class FeishuIntakeBlackBox(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory(prefix="feishu-e2e-") self.d = Path(self.tmp.name) self.bin = self.d / "bin" self.bin.mkdir() self.fixture = self.d / "submitted-bug.json" self.fixture.write_text(json.dumps({ "recordId": "recBug1", "title": "Crash on save", "actual": "button crashes", "expected": "save succeeds", "steps": ["open app", "click Save"], "acceptance": "regression covered", "priority": "P1", "attachments": [{"file_token": "fileTok", "name": "screen.png", "type": "image/png", "size": 4}], "updatedAt": "2026-08-01T12:00:00Z", }), encoding="utf-8") self.mode_file = self.d / "mode.txt" self.mode_file.write_text("", encoding="utf-8") self.log = self.d / "argv.jsonl" self.fake = self.bin / "lark-cli" self.fake.write_text( "#!" + sys.executable + "\n" "import json, os, pathlib, sys\n" f"fixture=json.loads(pathlib.Path({str(self.fixture)!r}).read_text())\n" f"mode=pathlib.Path({str(self.mode_file)!r}).read_text().strip()\n" f"log=pathlib.Path({str(self.log)!r}); a=sys.argv[1:]\n" "with log.open('a') as f: f.write(json.dumps(a)+'\\n')\n" "p=a[a.index('--profile')+1] if '--profile' in a else None\n" "if a[:2]==['profile','list']:\n" " if '--format' in a: 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}])); raise SystemExit\n" "if p != 'tenant-b': print('wrong tenant',file=sys.stderr); raise SystemExit(9)\n" "if mode=='malformed': print('{bad'); raise SystemExit\n" "if a[:2]==['base','+record-list']:\n" " if mode=='okfalse': print(json.dumps({'ok':False})); raise SystemExit\n" " if mode=='code': print(json.dumps({'code':7})); raise SystemExit\n" " if mode=='ambiguous': print(json.dumps({'data':{}})); raise SystemExit\n" " if mode=='noprog': print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':[],'data':[],'has_more':True}})); raise SystemExit\n" " if mode=='max': print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':['recBug1'],'data':[['x']*8],'has_more':True}})); raise SystemExit\n" " if mode=='matrix': print(json.dumps({'code':0,'data':{'fields':['fTitle'],'record_id_list':['recBug1'],'records':[]}})); raise SystemExit\n" " if mode=='cell': print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':['recBug1'],'data':[[{'unexpected':'value'},'actual','expected','steps','accept','P1',[],f['updatedAt']]]}})); raise SystemExit\n" " if mode=='empty': print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':['recBug1'],'data':[['title',None,'expected','steps','accept','P1',[],f['updatedAt']]]}})); raise SystemExit\n" " f=fixture; print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':[f['recordId']],'data':[[f['title'],f['actual'],f['expected'],f['steps'],f['acceptance'],f['priority'],f['attachments'],f['updatedAt']]]}})); raise SystemExit\n" "if a[:2]==['base','+record-download-attachment']:\n" " out=pathlib.Path(a[a.index('--output')+1]); out.mkdir(parents=True,exist_ok=True)\n" " if mode=='escape': (out/'escape').symlink_to('/tmp'); raise SystemExit\n" " (out/'screen.png').write_bytes(b'fake'); print(json.dumps({'ok':True})); raise SystemExit\n" "raise SystemExit(2)\n", encoding="utf-8") self.fake.chmod(0o755) self.env = {**os.environ, "PATH": f"{self.bin}{os.pathsep}{os.environ.get('PATH','')}"} self.env["FAKE_FIXTURE"] = str(self.fixture) self.board = self.d / "tasks.yaml" self.board.write_text("""version: 1 project: name: isolated-fake bugIntake: provider: feishu-base profile: tenant-b baseToken: base-secret tableId: tbl-bugs viewId: view-ready fields: title: fTitle actual: fActual expected: fExpected stepsToReproduce: fSteps acceptance: fAcceptance priority: fPriority attachments: fAttachments updatedAt: fUpdated tasks: [] """, encoding="utf-8") def tearDown(self) -> None: self.tmp.cleanup() def invoke(self, command: str, *, mode: str | None = None, output: Path | None = None, resolver=None): env = dict(self.env) self.mode_file.write_text(mode or "", encoding="utf-8") args = [command, str(self.board)] if output: args += ["--output-dir", str(output)] stdout, stderr = io.StringIO(), io.StringIO() with mock.patch.object(feishu_bug_intake, "resolve_lark_cli", return_value=self.fake) if resolver is None else mock.patch.object(feishu_bug_intake, "resolve_lark_cli", side_effect=resolver), \ mock.patch.dict(os.environ, env, clear=True), \ contextlib.redirect_stdout(stdout), contextlib.redirect_stderr(stderr): try: code = feishu_bug_intake.main(args) except SystemExit as exc: code = int(exc.code or 0) return subprocess.CompletedProcess(args, code, stdout.getvalue(), stderr.getvalue()) def test_isolated_tenant_check_fetch_and_triage_contract(self) -> None: self.assertEqual(self.invoke("check").returncode, 0) downloads = self.d / "downloads" result = self.invoke("fetch", output=downloads) self.assertEqual(result.returncode, 0, result.stderr) payload = json.loads(result.stdout) record = payload["records"][0] self.assertEqual(record["recordId"], "recBug1") self.assertEqual(record["priority"], "P1") self.assertEqual(record["updatedAt"], "2026-08-01T12:00:00Z") self.assertIn("Crash on save", record["title"]) self.assertTrue(Path(record["attachments"][0]["localPath"]).is_relative_to(downloads)) self.assertRegex(record["sourceRef"], r"^feishu-base:sha256:[0-9a-f]{64}$") for secret in ("tenant-b", "base-secret", "tbl-bugs", "recBug1", "fileTok"): self.assertNotIn(secret, record["sourceRef"]) for secret in ("base-secret", "fileTok", "app-secret-sentinel"): self.assertNotIn(secret, result.stdout) second = self.invoke("fetch", output=self.d / "repeat") self.assertEqual(second.returncode, 0, second.stderr) second_record = json.loads(second.stdout)["records"][0] self.assertRegex(second_record["sourceRef"], r"^feishu-base:sha256:[0-9a-f]{64}$") self.assertEqual(record["sourceRef"], second_record["sourceRef"]) for secret in ("tenant-b", "base-secret", "tbl-bugs", "recBug1", "fileTok"): self.assertNotIn(secret, second_record["sourceRef"]) for secret in ("base-secret", "fileTok", "app-secret-sentinel"): self.assertNotIn(secret, second.stdout) calls = [json.loads(x) for x in self.log.read_text().splitlines()] for call in calls: if call[:1] == ["base"]: self.assertEqual(call[call.index("--profile") + 1], "tenant-b") self.assertNotIn("tenant-a", call) if call[:2] == ["base", "+record-download-attachment"]: self.assertFalse(Path(call[call.index("--output") + 1]).is_absolute()) self.assertEqual(sum(1 for c in calls if c[:2] == ["base", "+record-list"]), 2) list_call = next(c for c in reversed(calls) if c[:2] == ["base", "+record-list"]) self.assertEqual(list_call[list_call.index("--view-id") + 1], "view-ready") self.assertEqual(list_call[:2], ["base", "+record-list"]) self.assertEqual(list_call.count("--field-id"), 8) def test_fail_closed_wrong_profile_and_unsafe_response(self) -> None: bad = self.board.read_text().replace("profile: tenant-b", "profile: tenant-a") self.board.write_text(bad) result = self.invoke("fetch") self.assertNotEqual(result.returncode, 0) self.board.write_text(bad.replace("profile: tenant-a", "profile: tenant-b")) result = self.invoke("fetch", mode="malformed") self.assertNotEqual(result.returncode, 0) result = self.invoke("fetch", mode="matrix", output=self.d / "matrix-output") self.assertNotEqual(result.returncode, 0) result = self.invoke("fetch", mode="cell", output=self.d / "cell-output") self.assertNotEqual(result.returncode, 0) result = self.invoke("fetch", mode="empty", output=self.d / "empty-output") self.assertNotEqual(result.returncode, 0) result = self.invoke("fetch", mode="escape", output=self.d / "downloads") self.assertNotEqual(result.returncode, 0) def test_resolver_trust_and_no_executable_override_surface(self) -> None: trusted, hostile = self.d / "trusted", self.d / "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.assertRaises(feishu_bug_intake.IntakeError): feishu_bug_intake.resolve_lark_cli() help_result = self.invoke("--help") self.assertEqual(help_result.returncode, 0) self.assertNotIn("--lark-cli", help_result.stdout + help_result.stderr) self.assertNotIn("--executable", help_result.stdout + help_result.stderr) self.assertNotIn("executable_override", READER.read_text(encoding="utf-8")) def test_fail_closed_envelopes_pagination_and_missing_trusted_cli(self) -> None: for mode in ("okfalse", "code", "ambiguous", "noprog"): result = self.invoke("fetch", mode=mode) self.assertNotEqual(result.returncode, 0, mode) self.assertEqual(result.stdout, "", mode) with mock.patch.object(feishu_bug_intake, "MAX_PAGES", 2): result = self.invoke("fetch", mode="max") self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") with mock.patch.object(feishu_bug_intake, "resolve_lark_cli", side_effect=feishu_bug_intake.IntakeError("missing trusted")): result = self.invoke("fetch", resolver=feishu_bug_intake.IntakeError("missing trusted")) self.assertNotEqual(result.returncode, 0) self.assertEqual(result.stdout, "") def test_reference_template_example_contract_signals(self) -> None: paths = [ROOT / "skills/ack/references/feishu-bug-intake.md", ROOT / "skills/ack/templates/tasks.template.yaml", ROOT / "skills/ack/examples/tasks.example.yaml"] content = "\n".join(path.read_text(encoding="utf-8") for path in paths) self.assertIn("feishu-base:sha256:", content) self.assertIn('npm install --global --prefix "$HOME/.local" @larksuite/cli@latest', content) self.assertIn("profile add", content) self.assertIn("--brand feishu", content) self.assertTrue("scope" in content) self.assertIn("不要把 `lark-cli auth check`", content) self.assertNotIn("--lark-cli", content) self.assertNotIn("--executable", content) def test_plan_organizes_stable_sources_against_a_real_task_board(self) -> None: first = json.loads(self.invoke("fetch").stdout)["records"][0]["sourceRef"] second = json.loads(self.invoke("fetch").stdout)["records"][0]["sourceRef"] self.assertRegex(first, r"^feishu-base:sha256:[0-9a-f]{64}$") self.assertEqual(first, second) empty_plan = json.loads(self.invoke("plan").stdout) self.assertEqual(empty_plan["actions"], [{ "sourceRef": first, "recordId": "recBug1", "action": "create", }]) base = self.board.read_text(encoding="utf-8") def write_existing(status: str, updated_at: str, *, duplicate: bool = False) -> None: task = f"""tasks: - id: BUG-1 title: Existing imported bug status: {status} source: kind: feishu-base ref: "{first}" recordId: recBug1 updatedAt: "{updated_at}" """ if duplicate: task += f""" - id: BUG-2 title: Duplicate imported bug status: open source: kind: feishu-base ref: "{first}" recordId: recBug1 updatedAt: "{updated_at}" """ self.board.write_text(base.replace("tasks: []\n", task), encoding="utf-8") write_existing("open", "2026-07-31T12:00:00Z") validator = ROOT / "skills/ack/scripts/validate_tasks.py" validated = subprocess.run( [sys.executable, str(validator), str(self.board)], text=True, capture_output=True, check=False, ) self.assertEqual(validated.returncode, 0, validated.stderr) self.assertEqual(json.loads(self.invoke("plan").stdout)["actions"][0]["action"], "refresh") write_existing("verified", "2026-07-31T12:00:00Z") self.assertEqual(json.loads(self.invoke("plan").stdout)["actions"][0]["action"], "drift") write_existing("verified", "2026-08-01T12:00:00Z") self.assertEqual(json.loads(self.invoke("plan").stdout)["actions"][0]["action"], "unchanged") write_existing("open", "2026-07-31T12:00:00Z", duplicate=True) duplicate_result = self.invoke("plan") self.assertNotEqual(duplicate_result.returncode, 0) self.assertEqual(duplicate_result.stdout, "") if __name__ == "__main__": unittest.main()