File size: 3,849 Bytes
8ede856 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 | from __future__ import annotations
import asyncio
from pathlib import Path
import pytest
from astrbot.core.skills.neo_skill_sync import NeoSkillSyncManager
class _FakeSkills:
async def list_releases(self, **kwargs):
_ = kwargs
return {
"items": [
{
"id": "sr-1",
"skill_key": "etl/loader@v1",
"candidate_id": "sc-1",
"stage": "stable",
}
],
"total": 1,
}
async def get_candidate(self, candidate_id: str):
assert candidate_id == "sc-1"
return {
"id": "sc-1",
"payload_ref": "blob:blob-1",
}
async def get_payload(self, payload_ref: str):
assert payload_ref == "blob:blob-1"
return {
"payload_ref": payload_ref,
"kind": "astrbot_skill_v1",
"payload": {
"skill_markdown": "---\ndescription: test\n---\n# title\ncontent",
},
}
class _FakeClient:
def __init__(self):
self.skills = _FakeSkills()
def test_sync_release_writes_skill_and_map(monkeypatch, tmp_path: Path):
calls = {"active": [], "sandbox_sync": 0}
def _fake_set_skill_active(self, name, active):
calls["active"].append((name, active))
async def _fake_sync_sandboxes():
calls["sandbox_sync"] += 1
monkeypatch.setattr(
"astrbot.core.skills.neo_skill_sync.SkillManager.set_skill_active",
_fake_set_skill_active,
)
monkeypatch.setattr(
"astrbot.core.skills.neo_skill_sync.sync_skills_to_active_sandboxes",
_fake_sync_sandboxes,
)
skills_root = tmp_path / "skills"
map_path = skills_root / "neo_skill_map.json"
mgr = NeoSkillSyncManager(skills_root=str(skills_root), map_path=str(map_path))
result = asyncio.run(
mgr.sync_release(_FakeClient(), release_id="sr-1", require_stable=True)
)
assert result.skill_key == "etl/loader@v1"
assert result.release_id == "sr-1"
assert result.local_skill_name.startswith("neo_")
assert calls["active"] == [(result.local_skill_name, True)]
assert calls["sandbox_sync"] == 1
skill_md = skills_root / result.local_skill_name / "SKILL.md"
assert skill_md.exists()
assert "description: test" in skill_md.read_text(encoding="utf-8")
assert map_path.exists()
map_text = map_path.read_text(encoding="utf-8")
assert "etl/loader@v1" in map_text
assert result.local_skill_name in map_text
def test_sync_release_rejects_non_stable(monkeypatch, tmp_path: Path):
class _CanarySkills(_FakeSkills):
async def list_releases(self, **kwargs):
_ = kwargs
return {
"items": [
{
"id": "sr-1",
"skill_key": "etl",
"candidate_id": "sc-1",
"stage": "canary",
}
],
"total": 1,
}
class _CanaryClient:
def __init__(self):
self.skills = _CanarySkills()
async def _fake_sync_sandboxes():
return
monkeypatch.setattr(
"astrbot.core.skills.neo_skill_sync.sync_skills_to_active_sandboxes",
_fake_sync_sandboxes,
)
monkeypatch.setattr(
"astrbot.core.skills.neo_skill_sync.SkillManager.set_skill_active",
lambda self, name, active: None,
)
mgr = NeoSkillSyncManager(
skills_root=str(tmp_path / "skills"),
map_path=str(tmp_path / "skills" / "neo_skill_map.json"),
)
with pytest.raises(ValueError, match="Only stable releases"):
asyncio.run(
mgr.sync_release(_CanaryClient(), release_id="sr-1", require_stable=True)
)
|