feat: add Pydantic models and EventEmitter for v2 core
- models.py: ProviderConfig, RoleConfig, DiscussionConfig, ScenarioConfig, Message, ToolCallRecord, Session — compatible with config.yaml format - events.py: Event constants, Event dataclass, EventEmitter with on/off/emit - 67 tests passing (37 models + 30 events) Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
311
tests/test_events.py
Normal file
311
tests/test_events.py
Normal file
@@ -0,0 +1,311 @@
|
||||
"""Tests for meeting_room.events — EventEmitter and event types."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
import pytest
|
||||
|
||||
from meeting_room.events import (
|
||||
AGENT_ERROR,
|
||||
AGENT_MESSAGE,
|
||||
AGENT_TURN_START,
|
||||
BOSS_TURN,
|
||||
DISCUSSION_END,
|
||||
DISCUSSION_START,
|
||||
Event,
|
||||
EventEmitter,
|
||||
ROUND_END,
|
||||
ROUND_START,
|
||||
TOOL_CALL,
|
||||
TOOL_RESULT,
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Event type constants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEventConstants:
|
||||
"""All event type constants must be lowercase strings."""
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"name, value",
|
||||
[
|
||||
("DISCUSSION_START", "discussion_start"),
|
||||
("ROUND_START", "round_start"),
|
||||
("AGENT_TURN_START", "agent_turn_start"),
|
||||
("AGENT_MESSAGE", "agent_message"),
|
||||
("TOOL_CALL", "tool_call"),
|
||||
("TOOL_RESULT", "tool_result"),
|
||||
("AGENT_ERROR", "agent_error"),
|
||||
("ROUND_END", "round_end"),
|
||||
("DISCUSSION_END", "discussion_end"),
|
||||
("BOSS_TURN", "boss_turn"),
|
||||
],
|
||||
)
|
||||
def test_constants(self, name: str, value: str) -> None:
|
||||
import meeting_room.events as mod
|
||||
|
||||
assert getattr(mod, name) == value
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Event dataclass
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEvent:
|
||||
def test_defaults(self) -> None:
|
||||
e = Event(type=AGENT_MESSAGE)
|
||||
assert e.type == AGENT_MESSAGE
|
||||
assert e.data == {}
|
||||
assert isinstance(e.timestamp, datetime)
|
||||
|
||||
def test_custom_data(self) -> None:
|
||||
e = Event(type=TOOL_CALL, data={"tool": "search", "query": "test"})
|
||||
assert e.data["tool"] == "search"
|
||||
assert e.data["query"] == "test"
|
||||
|
||||
def test_timestamp_preserved(self) -> None:
|
||||
ts = datetime(2025, 1, 1, 12, 0, 0)
|
||||
e = Event(type=AGENT_ERROR, data={"msg": "fail"}, timestamp=ts)
|
||||
assert e.timestamp == ts
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# EventEmitter — subscription
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEventEmitterOn:
|
||||
def test_on_returns_handler(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
pass
|
||||
|
||||
result = bus.on(AGENT_MESSAGE, handler)
|
||||
assert result is handler
|
||||
|
||||
def test_on_stores_handler(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
|
||||
assert len(bus.handlers(AGENT_MESSAGE)) == 1
|
||||
|
||||
def test_on_multiple_handlers_same_event(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
|
||||
assert len(bus.handlers(AGENT_MESSAGE)) == 3
|
||||
|
||||
def test_on_different_event_types(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
bus.on(TOOL_CALL, lambda e: None)
|
||||
|
||||
assert len(bus.handlers(AGENT_MESSAGE)) == 1
|
||||
assert len(bus.handlers(TOOL_CALL)) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# EventEmitter — unsubscription
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEventEmitterOff:
|
||||
def test_off_removes_handler(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
pass
|
||||
|
||||
bus.on(AGENT_MESSAGE, handler)
|
||||
bus.off(AGENT_MESSAGE, handler)
|
||||
|
||||
assert len(bus.handlers(AGENT_MESSAGE)) == 0
|
||||
|
||||
def test_off_unknown_event_type_is_noop(self) -> None:
|
||||
bus = EventEmitter()
|
||||
bus.off("nonexistent_event", lambda e: None) # should not raise
|
||||
|
||||
def test_off_missing_handler_is_noop(self) -> None:
|
||||
bus = EventEmitter()
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
bus.off(AGENT_MESSAGE, lambda e: None) # different lambda, no raise
|
||||
|
||||
def test_off_cleans_up_empty_list(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
pass
|
||||
|
||||
bus.on(AGENT_MESSAGE, handler)
|
||||
bus.off(AGENT_MESSAGE, handler)
|
||||
|
||||
# Internal dict should not keep empty lists
|
||||
assert AGENT_MESSAGE not in bus._handlers
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# EventEmitter — emission
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEventEmitterEmit:
|
||||
def test_emit_calls_handler(self) -> None:
|
||||
bus = EventEmitter()
|
||||
received: list[Event] = []
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: received.append(e))
|
||||
bus.emit(AGENT_MESSAGE, text="hello", agent="Alice")
|
||||
|
||||
assert len(received) == 1
|
||||
assert received[0].type == AGENT_MESSAGE
|
||||
assert received[0].data == {"text": "hello", "agent": "Alice"}
|
||||
|
||||
def test_emit_calls_all_handlers(self) -> None:
|
||||
bus = EventEmitter()
|
||||
results_a: list[str] = []
|
||||
results_b: list[str] = []
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: results_a.append(e.data["text"]))
|
||||
bus.on(AGENT_MESSAGE, lambda e: results_b.append(e.data["text"]))
|
||||
|
||||
bus.emit(AGENT_MESSAGE, text="hello")
|
||||
|
||||
assert results_a == ["hello"]
|
||||
assert results_b == ["hello"]
|
||||
|
||||
def test_emit_no_handlers_is_safe(self) -> None:
|
||||
bus = EventEmitter()
|
||||
event = bus.emit(DISCUSSION_START) # no subscribers — no crash
|
||||
assert event.type == DISCUSSION_START
|
||||
|
||||
def test_emit_returns_event(self) -> None:
|
||||
bus = EventEmitter()
|
||||
event = bus.emit(ROUND_START, round_number=3)
|
||||
assert event.type == ROUND_START
|
||||
assert event.data == {"round_number": 3}
|
||||
assert isinstance(event.timestamp, datetime)
|
||||
|
||||
def test_emit_handler_exception_does_not_crash(self) -> None:
|
||||
bus = EventEmitter()
|
||||
call_count = 0
|
||||
|
||||
def bad_handler(event: Event) -> None:
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
raise RuntimeError("boom")
|
||||
|
||||
def good_handler(event: Event) -> None:
|
||||
nonlocal call_count
|
||||
call_count += 1
|
||||
|
||||
bus.on(AGENT_ERROR, bad_handler)
|
||||
bus.on(AGENT_ERROR, good_handler)
|
||||
|
||||
# bad_handler raises, but good_handler still runs
|
||||
bus.emit(AGENT_ERROR, msg="fail")
|
||||
|
||||
assert call_count == 2 # both handlers were called
|
||||
|
||||
def test_emit_multiple_events_independently(self) -> None:
|
||||
bus = EventEmitter()
|
||||
msg_log: list[str] = []
|
||||
tool_log: list[str] = []
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: msg_log.append(e.data["text"]))
|
||||
bus.on(TOOL_CALL, lambda e: tool_log.append(e.data["tool"]))
|
||||
|
||||
bus.emit(AGENT_MESSAGE, text="hi")
|
||||
bus.emit(TOOL_CALL, tool="search")
|
||||
bus.emit(AGENT_MESSAGE, text="bye")
|
||||
|
||||
assert msg_log == ["hi", "bye"]
|
||||
assert tool_log == ["search"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# EventEmitter — clear
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestEventEmitterClear:
|
||||
def test_clear_removes_all_handlers(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
bus.on(AGENT_MESSAGE, lambda e: None)
|
||||
bus.on(TOOL_CALL, lambda e: None)
|
||||
bus.clear()
|
||||
|
||||
assert bus.handlers(AGENT_MESSAGE) == []
|
||||
assert bus.handlers(TOOL_CALL) == []
|
||||
assert bus._handlers == {}
|
||||
|
||||
def test_handlers_returns_copy(self) -> None:
|
||||
bus = EventEmitter()
|
||||
|
||||
def handler(event: Event) -> None:
|
||||
pass
|
||||
|
||||
bus.on(AGENT_MESSAGE, handler)
|
||||
h = bus.handlers(AGENT_MESSAGE)
|
||||
h.clear() # mutating the copy should not affect the bus
|
||||
|
||||
assert len(bus.handlers(AGENT_MESSAGE)) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Integration-style: full discussion lifecycle
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDiscussionLifecycle:
|
||||
"""Simulate a full discussion lifecycle to exercise all event types."""
|
||||
|
||||
def test_full_lifecycle(self) -> None:
|
||||
bus = EventEmitter()
|
||||
log: list[str] = []
|
||||
|
||||
bus.on(DISCUSSION_START, lambda e: log.append(f"DISCUSSION_START"))
|
||||
bus.on(ROUND_START, lambda e: log.append(f"ROUND_START:{e.data['round']}"))
|
||||
bus.on(AGENT_TURN_START, lambda e: log.append(f"TURN:{e.data['agent']}"))
|
||||
bus.on(AGENT_MESSAGE, lambda e: log.append(f"MSG:{e.data['text']}"))
|
||||
bus.on(TOOL_CALL, lambda e: log.append(f"TOOL:{e.data['tool']}"))
|
||||
bus.on(TOOL_RESULT, lambda e: log.append(f"RESULT:{e.data['tool']}"))
|
||||
bus.on(AGENT_ERROR, lambda e: log.append(f"ERR:{e.data['msg']}"))
|
||||
bus.on(ROUND_END, lambda e: log.append(f"ROUND_END:{e.data['round']}"))
|
||||
bus.on(BOSS_TURN, lambda e: log.append(f"BOSS:{e.data['decision']}"))
|
||||
bus.on(DISCUSSION_END, lambda e: log.append("DISCUSSION_END"))
|
||||
|
||||
bus.emit(DISCUSSION_START)
|
||||
bus.emit(ROUND_START, round=1)
|
||||
bus.emit(AGENT_TURN_START, agent="Alice")
|
||||
bus.emit(AGENT_MESSAGE, text="I think we should use Python.")
|
||||
bus.emit(TOOL_CALL, tool="search")
|
||||
bus.emit(TOOL_RESULT, tool="search")
|
||||
bus.emit(ROUND_END, round=1)
|
||||
bus.emit(BOSS_TURN, decision="proceed")
|
||||
bus.emit(AGENT_ERROR, msg="timeout")
|
||||
bus.emit(DISCUSSION_END)
|
||||
|
||||
assert log == [
|
||||
"DISCUSSION_START",
|
||||
"ROUND_START:1",
|
||||
"TURN:Alice",
|
||||
"MSG:I think we should use Python.",
|
||||
"TOOL:search",
|
||||
"RESULT:search",
|
||||
"ROUND_END:1",
|
||||
"BOSS:proceed",
|
||||
"ERR:timeout",
|
||||
"DISCUSSION_END",
|
||||
]
|
||||
396
tests/test_models.py
Normal file
396
tests/test_models.py
Normal file
@@ -0,0 +1,396 @@
|
||||
"""Tests for meeting_room.models — Pydantic v2 data models."""
|
||||
|
||||
import copy
|
||||
from datetime import datetime, timezone
|
||||
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from meeting_room.models import (
|
||||
DefaultsConfig,
|
||||
DiscussionConfig,
|
||||
Message,
|
||||
ProviderConfig,
|
||||
RoleConfig,
|
||||
ScenarioConfig,
|
||||
Session,
|
||||
ToolCallRecord,
|
||||
)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures — sample data matching config.yaml structure
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
SAMPLE_YAML = """\
|
||||
providers:
|
||||
routerai:
|
||||
base_url: "https://routerai.ru/api/v1"
|
||||
api_key: "sk-test-key"
|
||||
ollama_cloud:
|
||||
api_key: "02a8f6e9f9744088885982ac645f1ce1.HB1MFh4AlTpyic9oJGxQjNuA"
|
||||
base_url: "https://ollama.com/v1"
|
||||
|
||||
roles:
|
||||
moderator:
|
||||
name: "Moderator"
|
||||
provider: ollama_cloud
|
||||
model: "gpt-4o-mini"
|
||||
temperature: 0.7
|
||||
tools: "none"
|
||||
system_prompt: |
|
||||
You are the moderator of a discussion.
|
||||
skeptic:
|
||||
name: "Skeptic"
|
||||
provider: ollama_cloud
|
||||
model: "gpt-4o-mini"
|
||||
temperature: 0.9
|
||||
tools: "readonly"
|
||||
system_prompt: |
|
||||
You are a skeptic and critic.
|
||||
|
||||
defaults:
|
||||
max_rounds: 8
|
||||
language: "ru"
|
||||
framework: "custom"
|
||||
workdir: "."
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def yaml_config() -> dict:
|
||||
"""Parse SAMPLE_YAML the same way cli.py does."""
|
||||
return yaml.safe_load(SAMPLE_YAML)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# ProviderConfig
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestProviderConfig:
|
||||
def test_basic(self):
|
||||
p = ProviderConfig(base_url="https://api.example.com/v1", api_key="sk-123")
|
||||
assert p.base_url == "https://api.example.com/v1"
|
||||
assert p.api_key == "sk-123"
|
||||
|
||||
def test_extra_fields_forbidden(self):
|
||||
with pytest.raises(Exception):
|
||||
ProviderConfig(base_url="https://x", api_key="k", unknown_field="oops")
|
||||
|
||||
def test_missing_required(self):
|
||||
with pytest.raises(Exception):
|
||||
ProviderConfig(base_url="https://x") # missing api_key
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# RoleConfig
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestRoleConfig:
|
||||
def test_basic(self):
|
||||
r = RoleConfig(
|
||||
name="Moderator",
|
||||
provider="ollama_cloud",
|
||||
model="gpt-4o-mini",
|
||||
temperature=0.7,
|
||||
tools="none",
|
||||
system_prompt="You moderate.",
|
||||
)
|
||||
assert r.name == "Moderator"
|
||||
assert r.temperature == 0.7
|
||||
assert r.tools == "none"
|
||||
|
||||
def test_defaults(self):
|
||||
r = RoleConfig(name="Bot", provider="p", model="m")
|
||||
assert r.temperature == 0.7
|
||||
assert r.tools == "none"
|
||||
assert r.system_prompt == ""
|
||||
|
||||
def test_tools_string_values(self):
|
||||
for val in ("all", "none", "readonly", "web", "full", "files"):
|
||||
r = RoleConfig(name="X", provider="p", model="m", tools=val)
|
||||
assert r.tools == val
|
||||
|
||||
def test_tools_list(self):
|
||||
r = RoleConfig(name="X", provider="p", model="m", tools=["read_file", "web_search"])
|
||||
assert r.tools == ["read_file", "web_search"]
|
||||
|
||||
def test_tools_coercion_list_items_to_str(self):
|
||||
r = RoleConfig(name="X", provider="p", model="m", tools=["read_file"])
|
||||
assert isinstance(r.tools[0], str)
|
||||
|
||||
def test_tools_invalid_type(self):
|
||||
with pytest.raises(Exception):
|
||||
RoleConfig(name="X", provider="p", model="m", tools=42)
|
||||
|
||||
def test_extra_fields_forbidden(self):
|
||||
with pytest.raises(Exception):
|
||||
RoleConfig(name="X", provider="p", model="m", rogue_field=True)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# DefaultsConfig
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestDefaultsConfig:
|
||||
def test_defaults(self):
|
||||
d = DefaultsConfig()
|
||||
assert d.max_rounds == 8
|
||||
assert d.language == "ru"
|
||||
assert d.framework == "custom"
|
||||
assert d.workdir == "."
|
||||
|
||||
def test_custom(self):
|
||||
d = DefaultsConfig(max_rounds=12, language="en", framework="autogen", workdir="/tmp")
|
||||
assert d.max_rounds == 12
|
||||
|
||||
def test_extra_ignored(self):
|
||||
# DefaultsConfig uses extra="ignore" — unknown keys are silently dropped
|
||||
d = DefaultsConfig(max_rounds=5, unknown_key="oops")
|
||||
assert d.max_rounds == 5
|
||||
assert not hasattr(d, "unknown_key")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# DiscussionConfig — the critical one
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestDiscussionConfig:
|
||||
def test_from_yaml_config(self, yaml_config: dict):
|
||||
"""DiscussionConfig(**yaml_config) must work without changes."""
|
||||
cfg = DiscussionConfig(**yaml_config)
|
||||
assert "routerai" in cfg.providers
|
||||
assert "ollama_cloud" in cfg.providers
|
||||
assert cfg.providers["routerai"].base_url == "https://routerai.ru/api/v1"
|
||||
assert "moderator" in cfg.roles
|
||||
assert "skeptic" in cfg.roles
|
||||
assert cfg.roles["moderator"].name == "Moderator"
|
||||
assert cfg.roles["skeptic"].tools == "readonly"
|
||||
assert cfg.defaults.max_rounds == 8
|
||||
|
||||
def test_roundtrip_serialization(self, yaml_config: dict):
|
||||
"""Model can be built from yaml, exported, and rebuilt."""
|
||||
cfg = DiscussionConfig(**yaml_config)
|
||||
exported = cfg.model_dump()
|
||||
rebuilt = DiscussionConfig(**exported)
|
||||
assert rebuilt.providers.keys() == cfg.providers.keys()
|
||||
assert rebuilt.roles.keys() == cfg.roles.keys()
|
||||
|
||||
def test_empty_config(self):
|
||||
cfg = DiscussionConfig()
|
||||
assert cfg.providers == {}
|
||||
assert cfg.roles == {}
|
||||
assert cfg.defaults.max_rounds == 8
|
||||
|
||||
def test_extra_keys_ignored(self):
|
||||
# DiscussionConfig uses extra="ignore"
|
||||
cfg = DiscussionConfig(
|
||||
providers={},
|
||||
roles={},
|
||||
defaults={},
|
||||
unknown_top_level="should be dropped",
|
||||
)
|
||||
assert not hasattr(cfg, "unknown_top_level")
|
||||
|
||||
def test_nested_provider_is_typed(self, yaml_config: dict):
|
||||
cfg = DiscussionConfig(**yaml_config)
|
||||
prov = cfg.providers["routerai"]
|
||||
assert isinstance(prov, ProviderConfig)
|
||||
|
||||
def test_nested_role_is_typed(self, yaml_config: dict):
|
||||
cfg = DiscussionConfig(**yaml_config)
|
||||
role = cfg.roles["moderator"]
|
||||
assert isinstance(role, RoleConfig)
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# ScenarioConfig
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestScenarioConfig:
|
||||
def test_defaults(self):
|
||||
s = ScenarioConfig()
|
||||
assert s.name == "Untitled"
|
||||
assert s.participants == []
|
||||
assert s.max_rounds == 8
|
||||
assert s.problem == ""
|
||||
|
||||
def test_from_frontmatter(self):
|
||||
data = {
|
||||
"name": "SaaS Pricing",
|
||||
"participants": ["moderator", "skeptic"],
|
||||
"max_rounds": 6,
|
||||
"problem": "How should we price our SaaS?",
|
||||
}
|
||||
s = ScenarioConfig(**data)
|
||||
assert s.name == "SaaS Pricing"
|
||||
assert s.participants == ["moderator", "skeptic"]
|
||||
assert s.max_rounds == 6
|
||||
|
||||
def test_extra_ignored(self):
|
||||
s = ScenarioConfig(name="X", surprise="drop me")
|
||||
assert s.name == "X"
|
||||
assert not hasattr(s, "surprise")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Message
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestMessage:
|
||||
def test_basic(self):
|
||||
m = Message(role_id="moderator", name="Moderator", content="Hello everyone!")
|
||||
assert m.role_id == "moderator"
|
||||
assert m.name == "Moderator"
|
||||
assert m.content == "Hello everyone!"
|
||||
|
||||
def test_extra_forbidden(self):
|
||||
with pytest.raises(Exception):
|
||||
Message(role_id="mod", name="Mod", content="hi", extra=True)
|
||||
|
||||
def test_missing_required(self):
|
||||
with pytest.raises(Exception):
|
||||
Message(role_id="mod", name="Mod") # missing content
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# ToolCallRecord
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestToolCallRecord:
|
||||
def test_basic(self):
|
||||
t = ToolCallRecord(id="call_abc", name="read_file", arguments='{"path": "/tmp/x"}', result="file contents")
|
||||
assert t.id == "call_abc"
|
||||
assert t.name == "read_file"
|
||||
assert t.result == "file contents"
|
||||
|
||||
def test_default_result(self):
|
||||
t = ToolCallRecord(id="call_1", name="web_search", arguments='{"q": "test"}')
|
||||
assert t.result == ""
|
||||
|
||||
def test_extra_forbidden(self):
|
||||
with pytest.raises(Exception):
|
||||
ToolCallRecord(id="1", name="n", arguments="{}", extra="bad")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Session
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestSession:
|
||||
def test_defaults(self):
|
||||
s = Session()
|
||||
assert s.scenario.name == "Untitled"
|
||||
assert s.messages == []
|
||||
assert s.tool_calls == []
|
||||
assert isinstance(s.created_at, datetime)
|
||||
assert isinstance(s.updated_at, datetime)
|
||||
|
||||
def test_with_messages(self):
|
||||
msg = Message(role_id="moderator", name="Moderator", content="Let's begin.")
|
||||
s = Session(messages=[msg])
|
||||
assert len(s.messages) == 1
|
||||
assert s.messages[0].content == "Let's begin."
|
||||
|
||||
def test_timestamps_auto_set(self):
|
||||
before = datetime.now(timezone.utc)
|
||||
s = Session()
|
||||
after = datetime.now(timezone.utc)
|
||||
assert before <= s.created_at <= after
|
||||
assert before <= s.updated_at <= after
|
||||
|
||||
def test_explicit_timestamps(self):
|
||||
t = datetime(2025, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
|
||||
s = Session(created_at=t, updated_at=t)
|
||||
assert s.created_at.year == 2025
|
||||
|
||||
def test_serialization_roundtrip(self):
|
||||
msg = Message(role_id="mod", name="Mod", content="Hi")
|
||||
tc = ToolCallRecord(id="c1", name="search", arguments='{}', result="ok")
|
||||
s = Session(
|
||||
scenario=ScenarioConfig(name="Test"),
|
||||
messages=[msg],
|
||||
tool_calls=[tc],
|
||||
)
|
||||
data = s.model_dump()
|
||||
rebuilt = Session(**data)
|
||||
assert rebuilt.messages[0].content == "Hi"
|
||||
assert rebuilt.tool_calls[0].name == "search"
|
||||
assert rebuilt.scenario.name == "Test"
|
||||
|
||||
def test_extra_ignored(self):
|
||||
s = Session(unknown_field="drop me")
|
||||
assert not hasattr(s, "unknown_field")
|
||||
|
||||
|
||||
# ===========================================================================
|
||||
# Edge cases with real config.yaml structure
|
||||
# ===========================================================================
|
||||
|
||||
|
||||
class TestConfigYamlCompatibility:
|
||||
"""Verify that the models accept exactly what yaml.safe_load produces
|
||||
from the actual config.yaml in the package.
|
||||
"""
|
||||
|
||||
def test_full_config_yaml_structure(self):
|
||||
"""Build DiscussionConfig from the full config.yaml structure
|
||||
(minus env-var resolution) and verify every field."""
|
||||
raw = yaml.safe_load(SAMPLE_YAML)
|
||||
cfg = DiscussionConfig(**raw)
|
||||
|
||||
# Providers
|
||||
assert len(cfg.providers) == 2
|
||||
assert cfg.providers["routerai"].base_url == "https://routerai.ru/api/v1"
|
||||
assert cfg.providers["routerai"].api_key == "sk-test-key"
|
||||
assert cfg.providers["ollama_cloud"].base_url == "https://ollama.com/v1"
|
||||
|
||||
# Roles
|
||||
assert len(cfg.roles) == 2
|
||||
mod = cfg.roles["moderator"]
|
||||
assert mod.name == "Moderator"
|
||||
assert mod.provider == "ollama_cloud"
|
||||
assert mod.model == "gpt-4o-mini"
|
||||
assert mod.temperature == 0.7
|
||||
assert mod.tools == "none"
|
||||
assert "moderator" in mod.system_prompt.lower() or "moderator" in mod.system_prompt
|
||||
|
||||
sk = cfg.roles["skeptic"]
|
||||
assert sk.name == "Skeptic"
|
||||
assert sk.temperature == 0.9
|
||||
assert sk.tools == "readonly"
|
||||
|
||||
# Defaults
|
||||
assert cfg.defaults.max_rounds == 8
|
||||
assert cfg.defaults.language == "ru"
|
||||
|
||||
def test_tools_as_list_in_role(self):
|
||||
"""Roles can specify tools as a list of tool names."""
|
||||
cfg = DiscussionConfig(
|
||||
roles={
|
||||
"analyst": RoleConfig(
|
||||
name="Analyst",
|
||||
provider="ollama_cloud",
|
||||
model="gpt-4o-mini",
|
||||
tools=["read_file", "web_search", "list_directory"],
|
||||
),
|
||||
},
|
||||
)
|
||||
assert cfg.roles["analyst"].tools == ["read_file", "web_search", "list_directory"]
|
||||
|
||||
def test_minimal_config(self):
|
||||
"""Minimal valid config with just one provider and one role."""
|
||||
cfg = DiscussionConfig(
|
||||
providers={"local": ProviderConfig(base_url="http://localhost:11434/v1", api_key="ollama")},
|
||||
roles={"bot": RoleConfig(name="Bot", provider="local", model="llama3")},
|
||||
)
|
||||
assert cfg.providers["local"].api_key == "ollama"
|
||||
assert cfg.roles["bot"].model == "llama3"
|
||||
assert cfg.defaults.max_rounds == 8 # default
|
||||
Reference in New Issue
Block a user