184 lines
7.6 KiB
Python
184 lines
7.6 KiB
Python
"""
|
|
Shared Given/Then step definitions used across all feature files.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pytest_bdd import given, then, parsers
|
|
|
|
|
|
# ── Given: user identity ─────────────────────────────────────────────────────
|
|
|
|
@given(parsers.parse('user "{user_id}" is sending commands'))
|
|
def set_user(user_id, pytestconfig):
|
|
from orchestrator.tools import set_current_user
|
|
set_current_user(user_id)
|
|
pytestconfig._test_user_id = user_id
|
|
|
|
|
|
@given(parsers.parse('the current chat_id is "{chat_id}"'))
|
|
def set_chat(chat_id):
|
|
from orchestrator.tools import set_current_chat
|
|
set_current_chat(chat_id)
|
|
|
|
|
|
# ── Given: session setup ─────────────────────────────────────────────────────
|
|
|
|
@given(parsers.parse('user has session "{conv_id}" in "{cwd}"'))
|
|
def add_session(conv_id, cwd, pytestconfig, tmp_path):
|
|
from agent.manager import manager, Session
|
|
user_id = getattr(pytestconfig, "_test_user_id", "user_abc123")
|
|
session = Session(conv_id=conv_id, cwd=str(tmp_path / conv_id), owner_id=user_id, cc_timeout=50.0)
|
|
(tmp_path / conv_id).mkdir(exist_ok=True)
|
|
manager._sessions[conv_id] = session
|
|
|
|
|
|
@given(parsers.parse('session "{conv_id}" in "{cwd}" belongs to user "{owner}"'))
|
|
def add_foreign_session(conv_id, cwd, owner, tmp_path):
|
|
from agent.manager import manager, Session
|
|
session = Session(conv_id=conv_id, cwd=str(tmp_path / conv_id), owner_id=owner, cc_timeout=50.0)
|
|
(tmp_path / conv_id).mkdir(exist_ok=True)
|
|
manager._sessions[conv_id] = session
|
|
|
|
|
|
@given(parsers.parse('active session is "{conv_id}"'))
|
|
def set_active_session(conv_id, pytestconfig):
|
|
from orchestrator.agent import agent
|
|
user_id = getattr(pytestconfig, "_test_user_id", "user_abc123")
|
|
agent._active_conv[user_id] = conv_id
|
|
|
|
|
|
@given(parsers.parse('active session is "{conv_id}" which does not exist'))
|
|
def set_ghost_active_session(conv_id, pytestconfig):
|
|
from orchestrator.agent import agent
|
|
user_id = getattr(pytestconfig, "_test_user_id", "user_abc123")
|
|
agent._active_conv[user_id] = conv_id
|
|
# intentionally NOT added to manager._sessions
|
|
|
|
|
|
@given(parsers.parse('no active session for user "{user_id}"'))
|
|
def ensure_no_active_session(user_id):
|
|
from orchestrator.agent import agent
|
|
agent._active_conv[user_id] = None
|
|
|
|
|
|
# ── Given: mode toggles ──────────────────────────────────────────────────────
|
|
|
|
@given(parsers.parse('direct mode is enabled for user "{user_id}"'))
|
|
def enable_direct_mode(user_id):
|
|
from orchestrator.agent import agent
|
|
agent._passthrough[user_id] = True
|
|
|
|
|
|
# ── Given: mocks ─────────────────────────────────────────────────────────────
|
|
|
|
@given(parsers.parse('run_claude returns "{output}"'))
|
|
def set_run_claude_return(output, mock_run_claude):
|
|
mock_run_claude.return_value = output
|
|
|
|
|
|
# ── Given: config ────────────────────────────────────────────────────────────
|
|
|
|
@given("ROUTER_MODE is disabled")
|
|
def disable_router_mode(monkeypatch):
|
|
import config
|
|
monkeypatch.setattr(config, "ROUTER_MODE", False)
|
|
|
|
|
|
# ── Then: reply assertions ───────────────────────────────────────────────────
|
|
|
|
@then(parsers.parse('reply contains "{text}"'))
|
|
def reply_contains(text, pytestconfig):
|
|
reply = getattr(pytestconfig, "_reply", None)
|
|
assert text in (reply or ""), \
|
|
f"Expected {text!r} in reply, got: {reply!r}"
|
|
|
|
|
|
@then(parsers.parse('reply does not contain "{text}"'))
|
|
def reply_not_contains(text, pytestconfig):
|
|
reply = getattr(pytestconfig, "_reply", None)
|
|
assert text not in (reply or ""), \
|
|
f"Expected {text!r} NOT in reply, got: {reply!r}"
|
|
|
|
|
|
@then("reply is not empty")
|
|
def reply_not_empty(pytestconfig):
|
|
reply = getattr(pytestconfig, "_reply", None)
|
|
assert reply and reply.strip(), \
|
|
f"Expected non-empty reply, got: {reply!r}"
|
|
|
|
|
|
@then("text reply is empty")
|
|
def reply_is_empty(pytestconfig):
|
|
reply = getattr(pytestconfig, "_reply", None)
|
|
assert reply == "", \
|
|
f"Expected empty reply, got: {reply!r}"
|
|
|
|
|
|
@then("command is not handled")
|
|
def command_not_handled(pytestconfig):
|
|
reply = getattr(pytestconfig, "_reply", None)
|
|
assert reply is None
|
|
|
|
|
|
# ── Then: session state ──────────────────────────────────────────────────────
|
|
|
|
@then(parsers.parse('session manager has {count:d} session for user "{user_id}"'))
|
|
@then(parsers.parse('session manager has {count:d} sessions for user "{user_id}"'))
|
|
def check_session_count(count, user_id):
|
|
from agent.manager import manager
|
|
sessions = manager.list_sessions(user_id=user_id)
|
|
assert len(sessions) == count, \
|
|
f"Expected {count} sessions, got {len(sessions)}: {sessions}"
|
|
|
|
|
|
@then(parsers.parse('active session for user "{user_id}" is "{conv_id}"'))
|
|
def check_active_session(user_id, conv_id):
|
|
from orchestrator.agent import agent
|
|
assert agent._active_conv.get(user_id) == conv_id
|
|
|
|
|
|
@then(parsers.parse('active session for user "{user_id}" is None'))
|
|
def check_no_active_session(user_id):
|
|
from orchestrator.agent import agent
|
|
assert agent._active_conv.get(user_id) is None
|
|
|
|
|
|
# ── Then: mode state ─────────────────────────────────────────────────────────
|
|
|
|
@then(parsers.parse('passthrough mode is enabled for user "{user_id}"'))
|
|
def check_passthrough_on(user_id):
|
|
from orchestrator.agent import agent
|
|
assert agent._passthrough.get(user_id) is True
|
|
|
|
|
|
@then(parsers.parse('passthrough mode is disabled for user "{user_id}"'))
|
|
def check_passthrough_off(user_id):
|
|
from orchestrator.agent import agent
|
|
assert agent._passthrough.get(user_id) is False
|
|
|
|
|
|
# ── Then: Feishu output ──────────────────────────────────────────────────────
|
|
|
|
@then(parsers.parse('a sessions card is sent to chat "{chat_id}"'))
|
|
def check_card_sent(chat_id, feishu_calls):
|
|
cards = feishu_calls["cards"]
|
|
assert any(c["receive_id"] == chat_id for c in cards), \
|
|
f"No card sent to {chat_id!r}, captured: {cards}"
|
|
|
|
|
|
# ── Then: scheduler ──────────────────────────────────────────────────────────
|
|
|
|
@then(parsers.parse('scheduler has {count:d} pending job'))
|
|
@then(parsers.parse('scheduler has {count:d} pending jobs'))
|
|
def check_scheduler_jobs(count):
|
|
from agent.scheduler import scheduler
|
|
assert len(scheduler._jobs) == count, \
|
|
f"Expected {count} jobs, got {len(scheduler._jobs)}"
|
|
|
|
|
|
# ── Then: run_claude ─────────────────────────────────────────────────────────
|
|
|
|
@then("run_claude was called")
|
|
def check_run_claude_called(mock_run_claude):
|
|
assert mock_run_claude.call_count >= 1, "Expected run_claude to be called"
|