829 lines
31 KiB
Python
829 lines
31 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
import skiff.cli
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
class SkillInitTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.home = Path(self.temp_dir.name)
|
|
self.skills_home = self.home / ".skills"
|
|
skill = self.skills_home / "skills" / "ack"
|
|
(skill / "templates").mkdir(parents=True)
|
|
(skill / "scripts").mkdir()
|
|
(skill / "SKILL.md").write_text("---\nname: ack\n---\n", encoding="utf-8")
|
|
(skill / "VERSION").write_text("1.2.3\n", encoding="utf-8")
|
|
(skill / "templates" / "project.template.md").write_text(
|
|
"# <project_name>\nversion=<ack_version>\npath=<overlay_file_path>\n",
|
|
encoding="utf-8",
|
|
)
|
|
(skill / "templates" / "tasks.template.yaml").write_text(
|
|
'updatedAt: "<YYYY-MM-DDTHH:mm:ss+TZ>"\n'
|
|
'ackVersion: "<接入时的 ack skill 版本>"\n'
|
|
'project:\n'
|
|
' name: "<project_name>"\n'
|
|
' repoPath: "<repo_path>"\n'
|
|
' devWorktree: "<dev_worktree>"\n',
|
|
encoding="utf-8",
|
|
)
|
|
tasks_template = skill / "templates" / "tasks.template.yaml"
|
|
tasks_template.write_text(
|
|
tasks_template.read_text(encoding="utf-8")
|
|
+ ' deliveryFile: "docs/ack/delivery.yaml"\n'
|
|
+ 'deliveryRuns: []\n',
|
|
encoding="utf-8",
|
|
)
|
|
(skill / "templates" / "knowledge.template.yaml").write_text(
|
|
'updatedAt: "<YYYY-MM-DDTHH:mm:ss+TZ>"\n'
|
|
'project:\n'
|
|
' name: "<project_name>"\n'
|
|
' repoPath: "<repo_path>"\n',
|
|
encoding="utf-8",
|
|
)
|
|
(skill / "templates" / "delivery.template.yaml").write_text(
|
|
'version: 1\n'
|
|
'updatedAt: "<YYYY-MM-DDTHH:mm:ss+TZ>"\n'
|
|
'project:\n'
|
|
' name: "<project_name>"\n'
|
|
'enabled: false\n'
|
|
'defaultProfile: null\n'
|
|
'entrypoints: {}\n'
|
|
'artifacts: {}\n'
|
|
'destinations: {}\n'
|
|
'environments: {}\n'
|
|
'profiles: {}\n',
|
|
encoding="utf-8",
|
|
)
|
|
for validator_name in (
|
|
"validate_tasks.py",
|
|
"validate_knowledge.py",
|
|
"validate_delivery.py",
|
|
):
|
|
(skill / "scripts" / validator_name).write_text(
|
|
"raise SystemExit(0)\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temp_dir.cleanup()
|
|
|
|
def run_skiff(self, *args: str) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
env["HOME"] = str(self.home)
|
|
env["PYTHONPATH"] = str(REPO_ROOT)
|
|
return subprocess.run(
|
|
[sys.executable, "-m", "skiff", *args],
|
|
cwd=REPO_ROOT,
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
|
|
def test_init_creates_only_rendered_project_files(self) -> None:
|
|
project = self.home / "sample-app"
|
|
project.mkdir()
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
target = project / "docs" / "ack"
|
|
self.assertFalse((target / "kit").exists())
|
|
self.assertFalse((target / "framework").exists())
|
|
project_content = (target / "project.md").read_text(encoding="utf-8")
|
|
tasks_content = (target / "tasks.yaml").read_text(encoding="utf-8")
|
|
knowledge_content = (target / "knowledge.yaml").read_text(encoding="utf-8")
|
|
delivery_content = (target / "delivery.yaml").read_text(encoding="utf-8")
|
|
self.assertIn("# sample-app", project_content)
|
|
self.assertIn("version=1.2.3", project_content)
|
|
self.assertIn(f'repoPath: "{project}"', tasks_content)
|
|
self.assertIn(f'repoPath: "{project}"', knowledge_content)
|
|
self.assertIn('name: "sample-app"', delivery_content)
|
|
self.assertIn("enabled: false", delivery_content)
|
|
self.assertNotIn("<project_name>", tasks_content)
|
|
self.assertNotIn("<project_name>", knowledge_content)
|
|
self.assertNotIn("<project_name>", delivery_content)
|
|
|
|
def test_init_refuses_to_overwrite_existing_files(self) -> None:
|
|
project = self.home / "existing-app"
|
|
target = project / "docs" / "ack"
|
|
target.mkdir(parents=True)
|
|
existing = target / "project.md"
|
|
existing.write_text("keep me", encoding="utf-8")
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("拒绝覆盖已有路径", result.stderr)
|
|
self.assertEqual(existing.read_text(encoding="utf-8"), "keep me")
|
|
self.assertFalse((target / "tasks.yaml").exists())
|
|
self.assertFalse((target / "knowledge.yaml").exists())
|
|
self.assertFalse((target / "delivery.yaml").exists())
|
|
|
|
def test_init_refuses_to_overwrite_existing_knowledge_file(self) -> None:
|
|
project = self.home / "existing-knowledge-app"
|
|
target = project / "docs" / "ack"
|
|
target.mkdir(parents=True)
|
|
existing = target / "knowledge.yaml"
|
|
existing.write_text("keep me", encoding="utf-8")
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("拒绝覆盖已有路径", result.stderr)
|
|
self.assertEqual(existing.read_text(encoding="utf-8"), "keep me")
|
|
self.assertFalse((target / "project.md").exists())
|
|
self.assertFalse((target / "tasks.yaml").exists())
|
|
|
|
def test_init_refuses_to_overwrite_existing_delivery_file(self) -> None:
|
|
project = self.home / "existing-delivery-app"
|
|
target = project / "docs" / "ack"
|
|
target.mkdir(parents=True)
|
|
existing = target / "delivery.yaml"
|
|
existing.write_text("keep me", encoding="utf-8")
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("拒绝覆盖已有路径", result.stderr)
|
|
self.assertEqual(existing.read_text(encoding="utf-8"), "keep me")
|
|
self.assertFalse((target / "project.md").exists())
|
|
self.assertFalse((target / "tasks.yaml").exists())
|
|
self.assertFalse((target / "knowledge.yaml").exists())
|
|
|
|
def test_init_rejects_symlinked_destination_directories(self) -> None:
|
|
for symlink_level in ("docs", "ack"):
|
|
with self.subTest(symlink_level=symlink_level):
|
|
project = self.home / f"symlink-{symlink_level}-app"
|
|
outside = self.home / f"symlink-{symlink_level}-outside"
|
|
project.mkdir()
|
|
outside.mkdir()
|
|
if symlink_level == "docs":
|
|
(project / "docs").symlink_to(
|
|
outside,
|
|
target_is_directory=True,
|
|
)
|
|
else:
|
|
(project / "docs").mkdir()
|
|
(project / "docs" / "ack").symlink_to(
|
|
outside,
|
|
target_is_directory=True,
|
|
)
|
|
|
|
result = self.run_skiff(
|
|
"init",
|
|
"ack",
|
|
"--project",
|
|
str(project),
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("不能是软链接", result.stderr)
|
|
self.assertFalse((outside / "project.md").exists())
|
|
self.assertFalse((outside / "tasks.yaml").exists())
|
|
self.assertFalse((outside / "knowledge.yaml").exists())
|
|
self.assertFalse((outside / "delivery.yaml").exists())
|
|
|
|
def test_init_rejects_path_like_skill_name_before_resolving_targets(self) -> None:
|
|
project = self.home / "path-traversal-app"
|
|
outside = self.home / "path-traversal-outside"
|
|
project.mkdir()
|
|
outside.mkdir()
|
|
(project / "skills").symlink_to(outside, target_is_directory=True)
|
|
|
|
result = self.run_skiff(
|
|
"init",
|
|
"../skills/ack",
|
|
"--project",
|
|
str(project),
|
|
)
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("skill 名称无效", result.stderr)
|
|
self.assertFalse((outside / "ack").exists())
|
|
|
|
def test_atomic_publish_failure_never_exposes_partial_ack_directory(self) -> None:
|
|
project = self.home / "atomic-publish-app"
|
|
project.mkdir()
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_rename_directory_noreplace",
|
|
side_effect=RuntimeError("publish interrupted"),
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(RuntimeError, "publish interrupted"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
docs = project / "docs"
|
|
if docs.exists():
|
|
self.assertEqual(list(docs.iterdir()), [])
|
|
|
|
def test_atomic_publish_never_replaces_a_raced_destination(self) -> None:
|
|
project = self.home / "atomic-no-replace-app"
|
|
project.mkdir()
|
|
raced_inode: int | None = None
|
|
real_publish = skiff.cli._rename_directory_noreplace
|
|
|
|
def create_destination_then_publish(
|
|
source_parent_fd: int,
|
|
source_name: str,
|
|
destination_parent_fd: int,
|
|
destination_name: str,
|
|
) -> None:
|
|
nonlocal raced_inode
|
|
os.mkdir(destination_name, mode=0o711, dir_fd=destination_parent_fd)
|
|
raced_inode = os.stat(
|
|
destination_name,
|
|
dir_fd=destination_parent_fd,
|
|
follow_symlinks=False,
|
|
).st_ino
|
|
real_publish(
|
|
source_parent_fd,
|
|
source_name,
|
|
destination_parent_fd,
|
|
destination_name,
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_rename_directory_noreplace",
|
|
side_effect=create_destination_then_publish,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "拒绝覆盖已有路径"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
target = project / "docs" / "ack"
|
|
self.assertTrue(target.is_dir())
|
|
self.assertEqual(target.stat().st_ino, raced_inode)
|
|
self.assertEqual(list(target.iterdir()), [])
|
|
self.assertEqual(target.stat().st_mode & 0o777, 0o711)
|
|
|
|
def test_published_destination_replacement_never_reports_success(self) -> None:
|
|
project = self.home / "published-destination-app"
|
|
moved_target = project / "docs" / "ack-moved"
|
|
project.mkdir()
|
|
real_publish = skiff.cli._rename_directory_noreplace
|
|
|
|
def replace_destination_after_publish(
|
|
source_parent_fd: int,
|
|
source_name: str,
|
|
destination_parent_fd: int,
|
|
destination_name: str,
|
|
) -> None:
|
|
real_publish(
|
|
source_parent_fd,
|
|
source_name,
|
|
destination_parent_fd,
|
|
destination_name,
|
|
)
|
|
target = project / "docs" / destination_name
|
|
target.rename(moved_target)
|
|
target.mkdir()
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_rename_directory_noreplace",
|
|
side_effect=replace_destination_after_publish,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "ACK 目录已被替换"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
self.assertEqual(list((project / "docs" / "ack").iterdir()), [])
|
|
self.assertEqual(
|
|
sorted(path.name for path in moved_target.iterdir()),
|
|
["delivery.yaml", "knowledge.yaml", "project.md", "tasks.yaml"],
|
|
)
|
|
|
|
def test_transaction_container_replacement_cannot_forge_payload(self) -> None:
|
|
project = self.home / "transaction-source-app"
|
|
attacker = self.home / "transaction-attacker"
|
|
project.mkdir()
|
|
attacker.mkdir()
|
|
(attacker / "marker").write_text("forged", encoding="utf-8")
|
|
real_publish = skiff.cli._rename_directory_noreplace
|
|
|
|
def replace_outer_transaction_then_publish(
|
|
source_parent_fd: int,
|
|
source_name: str,
|
|
destination_parent_fd: int,
|
|
destination_name: str,
|
|
) -> None:
|
|
docs = project / "docs"
|
|
transactions = [
|
|
path
|
|
for path in docs.iterdir()
|
|
if path.name.startswith(".ack-init-")
|
|
]
|
|
self.assertEqual(len(transactions), 1)
|
|
transaction = transactions[0]
|
|
saved = docs / f"{transaction.name}.saved"
|
|
transaction.rename(saved)
|
|
transaction.symlink_to(attacker, target_is_directory=True)
|
|
real_publish(
|
|
source_parent_fd,
|
|
source_name,
|
|
destination_parent_fd,
|
|
destination_name,
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_rename_directory_noreplace",
|
|
side_effect=replace_outer_transaction_then_publish,
|
|
),
|
|
):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
target = project / "docs" / "ack"
|
|
self.assertTrue(target.is_dir())
|
|
self.assertFalse(target.is_symlink())
|
|
self.assertFalse((target / "marker").exists())
|
|
self.assertEqual(
|
|
sorted(path.name for path in target.iterdir()),
|
|
["delivery.yaml", "knowledge.yaml", "project.md", "tasks.yaml"],
|
|
)
|
|
|
|
def test_post_publish_fsync_failure_preserves_complete_state(self) -> None:
|
|
project = self.home / "post-publish-fsync-app"
|
|
project.mkdir()
|
|
real_fsync = os.fsync
|
|
calls = 0
|
|
|
|
def fail_directory_fsync_after_publish(file_descriptor: int) -> None:
|
|
nonlocal calls
|
|
calls += 1
|
|
if calls == 8:
|
|
raise OSError("simulated directory fsync failure")
|
|
real_fsync(file_descriptor)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli.os,
|
|
"fsync",
|
|
side_effect=fail_directory_fsync_after_publish,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "已完整发布"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
target = project / "docs" / "ack"
|
|
self.assertEqual(
|
|
sorted(path.name for path in target.iterdir()),
|
|
["delivery.yaml", "knowledge.yaml", "project.md", "tasks.yaml"],
|
|
)
|
|
|
|
def test_project_root_replacement_aborts_before_publish(self) -> None:
|
|
project = self.home / "root-replacement-app"
|
|
moved_project = self.home / "root-replacement-moved"
|
|
project.mkdir()
|
|
real_open_docs = skiff.cli._open_or_create_directory_at
|
|
replaced = False
|
|
|
|
def replace_root_then_open_docs(
|
|
parent_fd: int,
|
|
name: str,
|
|
) -> tuple[int, bool]:
|
|
nonlocal replaced
|
|
if not replaced:
|
|
project.rename(moved_project)
|
|
project.mkdir()
|
|
replaced = True
|
|
return real_open_docs(parent_fd, name)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_open_or_create_directory_at",
|
|
side_effect=replace_root_then_open_docs,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "项目目录已被替换"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
self.assertFalse((moved_project / "docs" / "ack").exists())
|
|
|
|
def test_project_root_replacement_at_publish_never_reports_success(self) -> None:
|
|
project = self.home / "publish-root-replacement-app"
|
|
moved_project = self.home / "publish-root-replacement-moved"
|
|
project.mkdir()
|
|
real_publish = skiff.cli._rename_directory_noreplace
|
|
|
|
def replace_root_then_publish(
|
|
source_parent_fd: int,
|
|
source_name: str,
|
|
destination_parent_fd: int,
|
|
destination_name: str,
|
|
) -> None:
|
|
project.rename(moved_project)
|
|
project.mkdir()
|
|
real_publish(
|
|
source_parent_fd,
|
|
source_name,
|
|
destination_parent_fd,
|
|
destination_name,
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_rename_directory_noreplace",
|
|
side_effect=replace_root_then_publish,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "ACK 目录已移动或不可访问"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
target = moved_project / "docs" / "ack"
|
|
self.assertEqual(
|
|
sorted(path.name for path in target.iterdir()),
|
|
["delivery.yaml", "knowledge.yaml", "project.md", "tasks.yaml"],
|
|
)
|
|
|
|
def test_docs_replacement_aborts_before_publish(self) -> None:
|
|
project = self.home / "docs-replacement-app"
|
|
moved_docs = project / "docs-moved"
|
|
project.mkdir()
|
|
real_assert_binding = skiff.cli._assert_open_directory_path
|
|
replaced = False
|
|
|
|
def replace_docs_at_publish_check(
|
|
directory_fd: int,
|
|
path: Path,
|
|
*,
|
|
phase: str,
|
|
label: str = "项目目录",
|
|
) -> None:
|
|
nonlocal replaced
|
|
if label == "docs 目录" and phase == "发布" and not replaced:
|
|
(project / "docs").rename(moved_docs)
|
|
(project / "docs").mkdir()
|
|
replaced = True
|
|
real_assert_binding(
|
|
directory_fd,
|
|
path,
|
|
phase=phase,
|
|
label=label,
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_assert_open_directory_path",
|
|
side_effect=replace_docs_at_publish_check,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "docs 目录已被替换"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
self.assertFalse((moved_docs / "ack").exists())
|
|
|
|
def test_docs_replacement_at_publish_never_reports_success(self) -> None:
|
|
project = self.home / "publish-docs-replacement-app"
|
|
moved_docs = project / "docs-moved"
|
|
project.mkdir()
|
|
real_publish = skiff.cli._rename_directory_noreplace
|
|
|
|
def replace_docs_then_publish(
|
|
source_parent_fd: int,
|
|
source_name: str,
|
|
destination_parent_fd: int,
|
|
destination_name: str,
|
|
) -> None:
|
|
(project / "docs").rename(moved_docs)
|
|
(project / "docs").mkdir()
|
|
real_publish(
|
|
source_parent_fd,
|
|
source_name,
|
|
destination_parent_fd,
|
|
destination_name,
|
|
)
|
|
|
|
with (
|
|
mock.patch.object(skiff.cli, "SKILLS_HOME", self.skills_home),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"SKILLS_DIR",
|
|
self.skills_home / "skills",
|
|
),
|
|
mock.patch.object(skiff.cli, "ensure_skills_home"),
|
|
mock.patch.object(
|
|
skiff.cli,
|
|
"_rename_directory_noreplace",
|
|
side_effect=replace_docs_then_publish,
|
|
),
|
|
):
|
|
with self.assertRaisesRegex(SystemExit, "ACK 目录已移动或不可访问"):
|
|
skiff.cli.cmd_init(
|
|
SimpleNamespace(name="ack", project=str(project))
|
|
)
|
|
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
target = moved_docs / "ack"
|
|
self.assertEqual(
|
|
sorted(path.name for path in target.iterdir()),
|
|
["delivery.yaml", "knowledge.yaml", "project.md", "tasks.yaml"],
|
|
)
|
|
|
|
def test_ack_init_requires_knowledge_template(self) -> None:
|
|
project = self.home / "missing-knowledge-template-app"
|
|
project.mkdir()
|
|
(
|
|
self.skills_home
|
|
/ "skills"
|
|
/ "ack"
|
|
/ "templates"
|
|
/ "knowledge.template.yaml"
|
|
).unlink()
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("knowledge.template.yaml", result.stderr)
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
|
|
def test_ack_init_requires_delivery_template(self) -> None:
|
|
project = self.home / "missing-delivery-template-app"
|
|
project.mkdir()
|
|
(
|
|
self.skills_home
|
|
/ "skills"
|
|
/ "ack"
|
|
/ "templates"
|
|
/ "delivery.template.yaml"
|
|
).unlink()
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("delivery.template.yaml", result.stderr)
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
|
|
def test_ack_init_requires_all_validators(self) -> None:
|
|
for validator_name in (
|
|
"validate_tasks.py",
|
|
"validate_knowledge.py",
|
|
"validate_delivery.py",
|
|
):
|
|
with self.subTest(validator_name=validator_name):
|
|
project = self.home / f"missing-{validator_name}-app"
|
|
project.mkdir()
|
|
validator = (
|
|
self.skills_home
|
|
/ "skills"
|
|
/ "ack"
|
|
/ "scripts"
|
|
/ validator_name
|
|
)
|
|
original = validator.read_text(encoding="utf-8")
|
|
validator.unlink()
|
|
try:
|
|
result = self.run_skiff(
|
|
"init",
|
|
"ack",
|
|
"--project",
|
|
str(project),
|
|
)
|
|
finally:
|
|
validator.write_text(original, encoding="utf-8")
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("缺少初始化校验器", result.stderr)
|
|
self.assertIn(validator_name, result.stderr)
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
|
|
def test_validator_failure_leaves_no_partial_initialization(self) -> None:
|
|
project = self.home / "invalid-knowledge-app"
|
|
project.mkdir()
|
|
validator = (
|
|
self.skills_home / "skills" / "ack" / "scripts" / "validate_knowledge.py"
|
|
)
|
|
validator.write_text("raise SystemExit(1)\n", encoding="utf-8")
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
self.assertIn("初始化知识库校验失败", result.stderr)
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
|
|
def test_delivery_validator_failure_leaves_no_partial_initialization(self) -> None:
|
|
project = self.home / "invalid-delivery-app"
|
|
project.mkdir()
|
|
validator = (
|
|
self.skills_home / "skills" / "ack" / "scripts" / "validate_delivery.py"
|
|
)
|
|
validator.write_text("raise SystemExit(1)\n", encoding="utf-8")
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertNotIn("Traceback", result.stderr)
|
|
self.assertIn("初始化交付契约校验失败", result.stderr)
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
|
|
def test_validator_cannot_replace_staged_bytes_before_install(self) -> None:
|
|
project = self.home / "mutated-staging-app"
|
|
project.mkdir()
|
|
validator = (
|
|
self.skills_home / "skills" / "ack" / "scripts" / "validate_knowledge.py"
|
|
)
|
|
validator.write_text(
|
|
"import sys\n"
|
|
"from pathlib import Path\n"
|
|
"Path(sys.argv[1]).write_text('forged: true\\n', encoding='utf-8')\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("临时文件在校验期间发生变化", result.stderr)
|
|
self.assertFalse((project / "docs" / "ack").exists())
|
|
|
|
def test_ack_init_validates_mirrored_staging_root(self) -> None:
|
|
project = self.home / "staged-knowledge-app"
|
|
project.mkdir()
|
|
validator = (
|
|
self.skills_home / "skills" / "ack" / "scripts" / "validate_knowledge.py"
|
|
)
|
|
validator.write_text(
|
|
"import sys\n"
|
|
"from pathlib import Path\n"
|
|
"required = ['--tasks', '--project-root']\n"
|
|
"if any(item not in sys.argv for item in required):\n"
|
|
" raise SystemExit(3)\n"
|
|
"root = Path(sys.argv[sys.argv.index('--project-root') + 1])\n"
|
|
"knowledge = Path(sys.argv[1])\n"
|
|
"tasks = Path(sys.argv[sys.argv.index('--tasks') + 1])\n"
|
|
"expected = root / 'docs' / 'ack'\n"
|
|
"raise SystemExit(0 if knowledge.parent == expected and "
|
|
"tasks.parent == expected else 4)\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue((project / "docs" / "ack" / "knowledge.yaml").is_file())
|
|
|
|
def test_ack_delivery_init_validates_mirrored_staging_root(self) -> None:
|
|
project = self.home / "staged-delivery-app"
|
|
project.mkdir()
|
|
validator = (
|
|
self.skills_home / "skills" / "ack" / "scripts" / "validate_delivery.py"
|
|
)
|
|
validator.write_text(
|
|
"import sys\n"
|
|
"from pathlib import Path\n"
|
|
"required = ['--tasks', '--project-root']\n"
|
|
"if any(item not in sys.argv for item in required):\n"
|
|
" raise SystemExit(3)\n"
|
|
"root = Path(sys.argv[sys.argv.index('--project-root') + 1])\n"
|
|
"delivery = Path(sys.argv[1])\n"
|
|
"tasks = Path(sys.argv[sys.argv.index('--tasks') + 1])\n"
|
|
"expected = root / 'docs' / 'ack'\n"
|
|
"raise SystemExit(0 if delivery.parent == expected and "
|
|
"tasks.parent == expected else 4)\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
self.assertTrue((project / "docs" / "ack" / "delivery.yaml").is_file())
|
|
|
|
def test_non_ack_init_still_requires_only_project_and_tasks_templates(self) -> None:
|
|
skill = self.skills_home / "skills" / "plain"
|
|
(skill / "templates").mkdir(parents=True)
|
|
(skill / "SKILL.md").write_text("---\nname: plain\n---\n", encoding="utf-8")
|
|
(skill / "templates" / "project.template.md").write_text(
|
|
"# <project_name>\n", encoding="utf-8"
|
|
)
|
|
(skill / "templates" / "tasks.template.yaml").write_text(
|
|
'project: "<project_name>"\n', encoding="utf-8"
|
|
)
|
|
project = self.home / "plain-app"
|
|
project.mkdir()
|
|
|
|
result = self.run_skiff("init", "plain", "--project", str(project))
|
|
|
|
self.assertEqual(result.returncode, 0, result.stderr)
|
|
target = project / "docs" / "plain"
|
|
self.assertTrue((target / "project.md").is_file())
|
|
self.assertTrue((target / "tasks.yaml").is_file())
|
|
self.assertFalse((target / "knowledge.yaml").exists())
|
|
self.assertFalse((target / "delivery.yaml").exists())
|
|
|
|
def test_init_rejects_missing_project_directory(self) -> None:
|
|
project = self.home / "missing-app"
|
|
|
|
result = self.run_skiff("init", "ack", "--project", str(project))
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("项目目录不存在", result.stderr)
|
|
self.assertFalse(project.exists())
|
|
|
|
def test_legacy_kit_command_is_not_exposed(self) -> None:
|
|
result = self.run_skiff("kit", "init", "ack")
|
|
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("invalid choice", result.stderr)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|