zenfun
74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
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
|