302 lines
16 KiB
Python
302 lines
16 KiB
Python
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',[],fixture['updatedAt']]]}})); raise SystemExit\n"
|
|
" if mode=='enrichment': print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':['recBug1'],'data':[['title','actual','expected',None,None,None,[],fixture['updatedAt']]]}})); raise SystemExit\n"
|
|
" if mode=='blank': print(json.dumps({'ok':True,'data':{'fields':['fTitle','fActual','fExpected','fSteps','fAcceptance','fPriority','fAttachments','fUpdated'],'record_id_list':['recBlank'],'data':[[None,None,None,None,None,None,[],fixture['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_missing_coordinator_fields_are_planned_for_enrichment(self) -> None:
|
|
fetched = self.invoke("fetch", mode="enrichment")
|
|
self.assertEqual(fetched.returncode, 0, fetched.stderr)
|
|
payload = json.loads(fetched.stdout)
|
|
self.assertEqual(payload["warnings"], [])
|
|
self.assertEqual(payload["records"][0]["enrichmentRequired"], [
|
|
"steps", "acceptance", "priority",
|
|
])
|
|
|
|
planned = self.invoke("plan", mode="enrichment")
|
|
self.assertEqual(planned.returncode, 0, planned.stderr)
|
|
action = json.loads(planned.stdout)["actions"][0]
|
|
self.assertEqual(action["action"], "create")
|
|
self.assertEqual(action["enrichmentRequired"], [
|
|
"steps", "acceptance", "priority",
|
|
])
|
|
|
|
def test_fully_blank_rows_are_skipped_with_a_batch_warning(self) -> None:
|
|
result = self.invoke("fetch", mode="blank")
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
payload = json.loads(result.stdout)
|
|
self.assertEqual(payload["records"], [])
|
|
self.assertEqual(payload["warnings"], [{
|
|
"recordId": "recBlank",
|
|
"code": "blank_record_skipped",
|
|
}])
|
|
|
|
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()
|