File size: 2,044 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 | from __future__ import annotations
import asyncio
from types import SimpleNamespace
from astrbot.core.agent.run_context import ContextWrapper
from astrbot.core.computer.tools.neo_skills import PromoteSkillCandidateTool
class _FakeSkills:
def __init__(self):
self.rollback_called_with = None
async def promote_candidate(self, candidate_id: str, stage: str = "canary"):
assert candidate_id == "cand-1"
assert stage == "stable"
return {
"id": "sr-1",
"skill_key": "k1",
"candidate_id": candidate_id,
"stage": stage,
}
async def rollback_release(self, release_id: str):
self.rollback_called_with = release_id
return {"id": "rb-1", "rollback_of": release_id}
class _FakeClient:
def __init__(self):
self.skills = _FakeSkills()
class _FakeBooter:
def __init__(self):
self.bay_client = _FakeClient()
self.sandbox = object()
def test_promote_stable_sync_failure_auto_rolls_back(monkeypatch):
async def _fake_get_booter(_ctx, _session_id):
return _FakeBooter()
async def _fake_sync_release(self, client, **kwargs):
_ = self, client, kwargs
raise ValueError("sync failed")
monkeypatch.setattr(
"astrbot.core.computer.tools.neo_skills.get_booter",
_fake_get_booter,
)
monkeypatch.setattr(
"astrbot.core.computer.tools.neo_skills.NeoSkillSyncManager.sync_release",
_fake_sync_release,
)
event = SimpleNamespace(role="admin", unified_msg_origin="session-1")
astr_ctx = SimpleNamespace(context=SimpleNamespace(), event=event)
run_ctx = ContextWrapper(context=astr_ctx)
tool = PromoteSkillCandidateTool()
result = asyncio.run(
tool.call(
run_ctx,
candidate_id="cand-1",
stage="stable",
sync_to_local=True,
)
)
assert isinstance(result, str)
assert "auto rollback succeeded" in result
assert "sync failed" in result
|