codex-app-server/cloud/tests/test_skill_seed_worker.py

"""Worker wiring for ChatGPT-Codex-alignment skills: tenant seeding.

Mirrors test_websearch_worker.py: no Docker, no network, no real
model. _tenant_dirs runs against temp dirs; the skill pack used
here is the real cloud/skills/ pack, so a broken pack fails here
before it ever reaches a tenant.
"""

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

try:
    from cloud.server import init_db
    from cloud.worker import Worker
    from cloud.skill_seed import ensure_codex_home_skills, list_skills
except ImportError:
    sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
    from cloud.server import init_db
    from cloud.worker import Worker
    from cloud.skill_seed import ensure_codex_home_skills, list_skills


class SkillSeedHelperTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()

    def tearDown(self):
        self.tmp.cleanup()

    def _src(self):
        src = os.path.join(self.tmp.name, "pack")
        for name in ("live-info", "document-tools"):
            d = os.path.join(src, name)
            os.makedirs(d, exist_ok=True)
            open(os.path.join(d, "SKILL.md"), "w").write(
                "---\nname: %s\ndescription: test\n---\nbody\n" % name)
        return src

    def test_copy_idempotent_and_missing_source(self):
        home = os.path.join(self.tmp.name, "home")
        out = ensure_codex_home_skills(home, self._src())
        self.assertEqual(sorted(out["skills_copied"]),
                         ["document-tools", "live-info"])
        self.assertEqual(out["skills_skipped_differ"], [])
        out2 = ensure_codex_home_skills(home, self._src())
        self.assertEqual(out2["skills_copied"], [])
        out3 = ensure_codex_home_skills(
            os.path.join(self.tmp.name, "home2"),
            os.path.join(self.tmp.name, "no-such-pack"))
        self.assertEqual(out3["skills_copied"], [])

    def test_tenant_modified_skill_is_never_overwritten(self):
        home = os.path.join(self.tmp.name, "home")
        src = self._src()
        ensure_codex_home_skills(home, src)
        dest = os.path.join(home, "skills",
                            "live-info", "SKILL.md")
        open(dest, "w").write("tenant edited version\n")
        out = ensure_codex_home_skills(home, src)
        self.assertEqual(out["skills_skipped_differ"], ["live-info"])
        self.assertEqual(open(dest).read(), "tenant edited version\n")

    def test_real_pack_lists_all_skills(self):
        names = [n for n, _p in list_skills()]
        self.assertEqual(names,
                         ["browser-automation", "document-tools",
                          "image-workflow", "live-info",
                          "software-engineering"])


class WorkerSkillSeedTestCase(unittest.TestCase):
    def setUp(self):
        self.tmp = tempfile.TemporaryDirectory()
        self.db_path = os.path.join(self.tmp.name, "control.db")
        init_db(self.db_path)
        self.template_home = os.path.join(self.tmp.name, "tpl-home")
        os.makedirs(self.template_home)
        open(os.path.join(self.template_home, "config.toml"),
             "w").write('[model_provider.xcmapi]\n'
                         'base_url="https://xcmapi.org/v1"\n')
        self.template_repo = os.path.join(self.tmp.name, "tpl-repo")
        os.makedirs(os.path.join(self.template_repo, ".git"))
        open(os.path.join(self.template_repo, "sample.py"),
             "w").write("def add(a, b):\n    return a + b\n")

    def tearDown(self):
        self.tmp.cleanup()

    def _worker(self, **kwargs):
        kwargs.setdefault("env", {"XCM_API_KEY": "test-placeholder"})
        kwargs.setdefault("tenants_root",
                          os.path.join(self.tmp.name, "tenants"))
        kwargs.setdefault("tenant_template_home", self.template_home)
        kwargs.setdefault("tenant_template_repo", self.template_repo)
        kwargs.setdefault("tenant_isolation", True)
        return Worker(self.db_path, "/fake/host.py", **kwargs)

    def _skill_path(self, dirs, name):
        return os.path.join(dirs["codex_home"], "skills",
                            name, "SKILL.md")

    def test_flag_off_is_old_behaviour(self):
        worker = self._worker()
        self.assertFalse(worker.seed_skills)
        dirs = worker._tenant_dirs("projA")
        self.assertFalse(os.path.exists(
            os.path.join(dirs["codex_home"], "skills")))

    def test_flag_on_seeds_real_pack(self):
        worker = self._worker(seed_skills=True)
        dirs = worker._tenant_dirs("projA")
        for name in ("live-info", "document-tools", "image-workflow"):
            path = self._skill_path(dirs, name)
            self.assertTrue(os.path.exists(path), path)
            self.assertIn("name: %s" % name, open(path).read())

    def test_flag_on_seeds_software_engineering_and_user_edit_survives(self):
        worker = self._worker(seed_skills=True)
        dirs = worker._tenant_dirs("projA")
        path = self._skill_path(dirs, "software-engineering")
        self.assertTrue(os.path.exists(path), path)
        text = open(path).read()
        self.assertIn("name: software-engineering", text)
        # v0.2 P0-1: the appendix travels inside SKILL.md itself,
        # because skill_seed only copies SKILL.md, not templates/.
        self.assertIn("附录角色：00-coordinator", text)
        self.assertIn("附录角色：08-shared-core", text)
        open(path, "w").write("tenant edited version\n")
        worker._tenant_dirs("projA")
        self.assertEqual(open(path).read(), "tenant edited version\n")

    def test_old_tenant_merges_and_user_edit_survives(self):
        worker = self._worker()
        dirs = worker._tenant_dirs("projA")
        worker2 = self._worker(seed_skills=True)
        worker2._tenant_dirs("projA")
        path = self._skill_path(dirs, "live-info")
        self.assertTrue(os.path.exists(path))
        open(path, "w").write("tenant edited version\n")
        worker2._tenant_dirs("projA")
        self.assertEqual(open(path).read(), "tenant edited version\n")


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