47 lines
1.8 KiB
Python
47 lines
1.8 KiB
Python
import pytest
|
|
import logging
|
|
|
|
from review_agent.file_summary.skills.base import BaseSkill, SkillResult, WorkflowContext
|
|
from review_agent.file_summary.skills.registry import SkillRegistry
|
|
|
|
|
|
class EchoSkill(BaseSkill):
|
|
name = "echo"
|
|
|
|
def run(self, context):
|
|
return SkillResult(success=True, data={"batch_id": context.batch.id})
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_skill_registry_executes_registered_skill(django_user_model):
|
|
from review_agent.models import Conversation, FileSummaryBatch
|
|
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
batch = FileSummaryBatch.objects.create(conversation=conversation, user=user, batch_no="FS-X")
|
|
registry = SkillRegistry()
|
|
registry.register(EchoSkill())
|
|
|
|
result = registry.execute("echo", WorkflowContext(batch=batch))
|
|
|
|
assert result.success is True
|
|
assert result.data == {"batch_id": batch.id}
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_skill_registry_logs_skill_lifecycle(caplog, django_user_model):
|
|
from review_agent.models import Conversation, FileSummaryBatch
|
|
|
|
user = django_user_model.objects.create_user(username="owner", password="pass")
|
|
conversation = Conversation.objects.create(user=user, title="会话")
|
|
batch = FileSummaryBatch.objects.create(conversation=conversation, user=user, batch_no="FS-LOG")
|
|
registry = SkillRegistry()
|
|
registry.register(EchoSkill())
|
|
|
|
with caplog.at_level(logging.INFO, logger="review_agent.file_summary"):
|
|
registry.execute("echo", WorkflowContext(batch=batch))
|
|
|
|
messages = [record.getMessage() for record in caplog.records]
|
|
assert any("Skill started" in message and "echo" in message for message in messages)
|
|
assert any("Skill finished" in message and "echo" in message for message in messages)
|