playbook/test/test_build_tsl_playbook.py

import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path

from scripts.build_tsl_playbook import build_agents_text


ROOT = Path(__file__).resolve().parents[1]
SCRIPT = ROOT / "scripts" / "build_tsl_playbook.py"
SYNC_WORKFLOW = ROOT / ".gitea" / "workflows" / "sync-tsl-playbook.yml"
README = ROOT / "README.md"


class BuildTslPlaybookTests(unittest.TestCase):
    def test_readme_names_tsl_syntax_skill_as_unique_fact_owner(self):
        text = README.read_text(encoding="utf-8")

        self.assertIn("TSL 语法事实唯一由 `tsl-syntax-reference` 管理", text)
        self.assertIn("TSL 领域路由与事实边界", text)

    def test_builds_minimal_tsl_playbook_tree(self):
        with tempfile.TemporaryDirectory() as tmp_dir:
            output = Path(tmp_dir) / "tsl-playbook"

            result = subprocess.run(
                [sys.executable, str(SCRIPT), "--output", str(output)],
                capture_output=True,
                text=True,
            )

            self.assertEqual(result.returncode, 0, msg=result.stderr)
            self.assertEqual(
                sorted(path.name for path in output.iterdir()),
                ["AGENTS.md", "docs", "skills"],
            )
            self.assertFalse((output / "playbook.toml").exists())
            self.assertFalse((output / ".agents").exists())
            self.assertFalse((output / "docs" / "index.md").exists())

            self.assertTrue((output / "docs" / "tsl" / "index.md").is_file())
            syntax_skill = output / "skills" / "tsl-syntax-reference"
            self.assertTrue((syntax_skill / "SKILL.md").is_file())
            self.assertTrue((syntax_skill / "scripts" / "lookup.py").is_file())
            self.assertFalse((syntax_skill / "references" / "index.md").exists())
            self.assertTrue(
                (output / "skills" / "tsl-syntax-reference" / "SKILL.md").is_file()
            )
            self.assertTrue(
                (output / "skills" / "tsl-api-reference" / "SKILL.md").is_file()
            )
            self.assertTrue(
                (
                    output
                    / "skills"
                    / "tsl-api-reference"
                    / "scripts"
                    / "lookup.py"
                ).is_file()
            )

            agents_text = (output / "AGENTS.md").read_text(encoding="utf-8")
            self.assertIn("# TSL Agent Instructions", agents_text)
            self.assertIn("tsl-syntax-reference", agents_text)
            self.assertIn("tsl-api-reference", agents_text)
            self.assertNotIn(".agents/index.md", agents_text)
            self.assertNotIn(".agents/tsl/index.md", agents_text)

            source_docs = count_files(ROOT / "docs" / "tsl")
            output_docs = count_files(output / "docs" / "tsl")
            self.assertEqual(output_docs, source_docs)

            for skill_name in ("tsl-syntax-reference", "tsl-api-reference"):
                source_skill = count_files(ROOT / "skills" / skill_name)
                output_skill = count_files(output / "skills" / skill_name)
                self.assertEqual(output_skill, source_skill)

    def test_missing_tsl_skill_reports_specific_source_path(self):
        for skill_name in ("tsl-syntax-reference", "tsl-api-reference"):
            with self.subTest(skill=skill_name), tempfile.TemporaryDirectory() as tmp_dir:
                repo = Path(tmp_dir) / "repo"
                repo.mkdir()
                copy_required_sources(repo)
                missing_path = repo / "skills" / skill_name
                shutil.rmtree(missing_path)

                result = subprocess.run(
                    [
                        sys.executable,
                        str(repo / "scripts" / "build_tsl_playbook.py"),
                        "--output",
                        str(repo / "output"),
                    ],
                    capture_output=True,
                    text=True,
                )

                self.assertNotEqual(result.returncode, 0)
                self.assertIn(str(missing_path), result.stderr)

    def test_build_agents_text_only_rewrites_title_and_trailing_newline(self):
        marker = (
            "- 需要任何 TSL 语法、文件模型、函数、模块或金融数据事实时，"
            "第一跳统一进 `docs/tsl/index.md`，由它路由到具体页；"
            "不在本文件内猜文件路径，也不全目录搜索。"
        )
        with tempfile.TemporaryDirectory() as tmp_dir:
            ruleset = Path(tmp_dir) / "index.md"
            ruleset.write_text(
                f"# TSL 智能体规则\n\n{marker}\n\n",
                encoding="utf-8",
                newline="\n",
            )

            agents_text = build_agents_text(ruleset)

            self.assertEqual(
                agents_text,
                f"# TSL Agent Instructions\n\n{marker}\n",
            )

    def test_sync_workflow_does_not_remove_entire_target_branch(self):
        text = SYNC_WORKFLOW.read_text(encoding="utf-8")

        self.assertNotRegex(text, r"git rm -rf --quiet\s+\.")
        self.assertIn("managed_paths=(", text)
        for path in (
            "AGENTS.md",
            "docs/tsl",
            "skills/tsl-syntax-reference",
            "skills/tsl-api-reference",
        ):
            self.assertIn(f'"{path}"', text)
        self.assertNotIn("generated_paths=(AGENTS.md docs skills)", text)
        self.assertIn('rm -rf -- "${managed_paths[@]}"', text)
        self.assertIn('git add -A -- "${managed_paths[@]}"', text)
        self.assertNotIn('cp -R "$bundle"/. "$REPO_DIR"/', text)
        self.assertNotIn(".gitea/ci/", text)
        self.assertNotIn("https://oauth2", text)
        self.assertNotIn("oauth2:${TOKEN}", text)
        self.assertNotRegex(text, r"REPO_URL=.*(TOKEN|WORKFLOW)")
        self.assertNotIn("git remote set-url", text)
        self.assertIn("GIT_ASKPASS", text)
        self.assertIn('REPO_URL="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}.git"', text)

    def test_sync_preserves_files_outside_generated_paths(self):
        if shutil.which("bash") is None:
            self.skipTest("bash is required to run sync workflow script")

        with tempfile.TemporaryDirectory() as tmp_dir:
            repo = create_source_repo(Path(tmp_dir))

            git(repo, "checkout", "--orphan", "tsl-playbook")
            git(repo, "rm", "-rf", ".")
            unmanaged_files = {
                "README.md": "manual branch note\n",
                "docs/python/index.md": "manual python docs\n",
                "skills/manual-skill/SKILL.md": "manual skill\n",
            }
            for relative, content in unmanaged_files.items():
                path = repo / relative
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text(content, encoding="utf-8", newline="\n")

            stale_managed_files = (
                "docs/tsl/stale.md",
                "skills/tsl-syntax-reference/stale.md",
                "skills/tsl-api-reference/stale.md",
            )
            for relative in stale_managed_files:
                path = repo / relative
                path.parent.mkdir(parents=True, exist_ok=True)
                path.write_text("stale\n", encoding="utf-8", newline="\n")

            git(repo, "add", ".")
            git(repo, "commit", "-m", "manual target branch content")
            git(repo, "push", "-u", "origin", "tsl-playbook")

            git(repo, "checkout", "main")
            run_sync(repo)

            for relative, expected in unmanaged_files.items():
                result = run(
                    ["git", "show", f"HEAD:{relative}"],
                    cwd=repo,
                    check=False,
                )
                self.assertEqual(result.returncode, 0, msg=result.stderr)
                self.assertEqual(result.stdout, expected)

            for relative in stale_managed_files:
                result = run(
                    ["git", "cat-file", "-e", f"HEAD:{relative}"],
                    cwd=repo,
                    check=False,
                )
                self.assertNotEqual(result.returncode, 0, msg=relative)

            for path in (
                "AGENTS.md",
                "docs/tsl/index.md",
                "skills/tsl-syntax-reference/SKILL.md",
                "skills/tsl-api-reference/SKILL.md",
            ):
                git(repo, "cat-file", "-e", f"HEAD:{path}")

            first_publish = git(repo, "rev-parse", "HEAD").stdout.strip()
            git(repo, "checkout", "main")
            run_sync(repo)
            second_publish = git(repo, "rev-parse", "HEAD").stdout.strip()
            self.assertEqual(second_publish, first_publish)

    def test_sync_creates_new_branch_without_source_files(self):
        if shutil.which("bash") is None:
            self.skipTest("bash is required to run sync workflow script")

        with tempfile.TemporaryDirectory() as tmp_dir:
            repo = create_source_repo(Path(tmp_dir))

            run_sync(repo)

            for path in (
                "AGENTS.md",
                "docs/tsl/index.md",
                "skills/tsl-syntax-reference/SKILL.md",
                "skills/tsl-api-reference/SKILL.md",
            ):
                git(repo, "cat-file", "-e", f"HEAD:{path}")

            for path in (
                ".gitea/workflows/sync-tsl-playbook.yml",
                "scripts/build_tsl_playbook.py",
                "rulesets/tsl/index.md",
            ):
                result = run(
                    ["git", "cat-file", "-e", f"HEAD:{path}"],
                    cwd=repo,
                    check=False,
                )
                self.assertNotEqual(result.returncode, 0, msg=f"{path} leaked")

    def test_sync_ignores_inherited_repo_dir_from_environment(self):
        if shutil.which("bash") is None:
            self.skipTest("bash is required to run sync workflow script")

        with tempfile.TemporaryDirectory() as tmp_dir:
            tmp = Path(tmp_dir)
            repo = create_source_repo(tmp)

            # Simulate the CI environment exporting REPO_DIR pointing at the
            # real checkout. The sync must operate on the repo passed to
            # run_sync, never the inherited value.
            decoy = tmp / "decoy-checkout"
            decoy.mkdir()
            sentinel = decoy / "sentinel.txt"
            sentinel.write_text("untouched\n", encoding="utf-8", newline="\n")

            previous = os.environ.get("REPO_DIR")
            os.environ["REPO_DIR"] = str(decoy)
            try:
                run_sync(repo)
            finally:
                if previous is None:
                    os.environ.pop("REPO_DIR", None)
                else:
                    os.environ["REPO_DIR"] = previous

            # The decoy must be left completely alone.
            self.assertEqual(
                sorted(item.name for item in decoy.iterdir()),
                ["sentinel.txt"],
            )
            self.assertEqual(sentinel.read_text(encoding="utf-8"), "untouched\n")

            # And the intended repo must actually have been published.
            git(repo, "cat-file", "-e", "HEAD:AGENTS.md")


def create_source_repo(tmp: Path) -> Path:
    repo = tmp / "repo"
    remote = tmp / "remote.git"

    run(["git", "init", "--bare", str(remote)])
    run(["git", "init", str(repo)])
    git(repo, "checkout", "-b", "main")
    git(repo, "config", "user.name", "test")
    git(repo, "config", "user.email", "test@example.invalid")

    copy_required_sources(repo)
    git(repo, "add", ".")
    git(repo, "commit", "-m", "initial sources")
    git(repo, "remote", "add", "origin", "../remote.git")
    git(repo, "push", "-u", "origin", "main")
    return repo


def run_sync(repo: Path) -> None:
    env = os.environ.copy()
    env.update(
        {
            "REPO_DIR": str(repo),
            "TARGET_BRANCH": "tsl-playbook",
            "COMMIT_AUTHOR_NAME": "test",
            "COMMIT_AUTHOR_EMAIL": "test@example.invalid",
        }
    )
    script_path = repo / ".sync-test.sh"
    script_path.write_text(
        extract_sync_workflow_script(), encoding="utf-8", newline="\n"
    )
    try:
        result = subprocess.run(
            ["bash", script_path.name],
            cwd=repo,
            env=env,
            capture_output=True,
            text=True,
            encoding="utf-8",
            errors="replace",
        )
    finally:
        script_path.unlink(missing_ok=True)
    if result.returncode != 0:
        raise AssertionError(result.stderr + result.stdout)


def copy_required_sources(repo: Path) -> None:
    (repo / ".gitea" / "workflows").mkdir(parents=True)
    shutil.copy2(
        SYNC_WORKFLOW, repo / ".gitea" / "workflows" / "sync-tsl-playbook.yml"
    )

    (repo / "scripts").mkdir()
    shutil.copy2(SCRIPT, repo / "scripts" / "build_tsl_playbook.py")

    (repo / "docs" / "tsl").mkdir(parents=True)
    (repo / "docs" / "tsl" / "index.md").write_text(
        "# TSL Index\n", encoding="utf-8", newline="\n"
    )

    syntax_skill = repo / "skills" / "tsl-syntax-reference"
    (syntax_skill / "scripts").mkdir(parents=True)
    (syntax_skill / "SKILL.md").write_text(
        "---\nname: tsl-syntax-reference\n---\n", encoding="utf-8", newline="\n"
    )
    (syntax_skill / "scripts" / "lookup.py").write_text(
        "print('lookup')\n", encoding="utf-8", newline="\n"
    )

    api_skill = repo / "skills" / "tsl-api-reference"
    (api_skill / "scripts").mkdir(parents=True)
    (api_skill / "SKILL.md").write_text(
        "---\nname: tsl-api-reference\n---\n", encoding="utf-8", newline="\n"
    )
    (api_skill / "scripts" / "lookup.py").write_text(
        "print('lookup')\n", encoding="utf-8", newline="\n"
    )

    syntax_skill = repo / "skills" / "tsl-syntax-reference"
    (syntax_skill / "references").mkdir(parents=True)
    (syntax_skill / "SKILL.md").write_text(
        "---\nname: tsl-syntax-reference\n---\n",
        encoding="utf-8",
        newline="\n",
    )
    (syntax_skill / "references" / "index.md").write_text(
        "# TSL Syntax Reference\n", encoding="utf-8", newline="\n"
    )

    (repo / "rulesets" / "tsl").mkdir(parents=True)
    (repo / "rulesets" / "tsl" / "index.md").write_text(
        "# TSL Agent Instructions\n", encoding="utf-8", newline="\n"
    )


def count_files(path: Path) -> int:
    return sum(
        1
        for item in path.rglob("*")
        if item.is_file()
        and "__pycache__" not in item.parts
        and item.suffix != ".pyc"
    )


def extract_sync_workflow_script() -> str:
    lines = SYNC_WORKFLOW.read_text(encoding="utf-8").splitlines()
    in_sync_step = False
    in_run_block = False
    script_lines: list[str] = []

    for line in lines:
        if line.startswith("      - name: 📦 Build and publish tsl-playbook"):
            in_sync_step = True
            continue
        if in_sync_step and line.startswith("      - name: "):
            break
        if in_sync_step and line == "        run: |":
            in_run_block = True
            continue
        if not in_run_block:
            continue
        if line.startswith("          "):
            script_lines.append(line[10:])
            continue
        if line.strip() == "":
            script_lines.append("")
            continue
        break

    if not script_lines:
        raise AssertionError("sync workflow run block was not found")
    return "\n".join(script_lines) + "\n"


def git(repo: Path, *args: str) -> subprocess.CompletedProcess[str]:
    return run(["git", *args], cwd=repo)


def run(
    args: list[str],
    cwd: Path | None = None,
    check: bool = True,
) -> subprocess.CompletedProcess[str]:
    result = subprocess.run(
        args,
        cwd=cwd,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    if check and result.returncode != 0:
        raise AssertionError(
            f"command failed: {' '.join(args)}\n{result.stderr}{result.stdout}"
        )
    return result


if __name__ == "__main__":
    unittest.main()