PhoneWork/tests/test_commands.py

"""Tests for bot slash commands (replaces BDD feature tests).

Covers: //help, //new, //close, //switch, //status, //perm,
        //direct, //smart, //shell, //remind, //tasks, //stop, //progress, //nodes
"""
from __future__ import annotations

import time

import pytest

from agent.manager import manager, Session
from orchestrator.agent import agent
from orchestrator.tools import set_current_user, set_current_chat


def _setup_user(user_id="user_abc123", chat_id=None):
    set_current_user(user_id)
    if chat_id:
        set_current_chat(chat_id)


def _add_session(conv_id, cwd="/tmp/proj", user_id="user_abc123", activate=False):
    session = Session(conv_id=conv_id, cwd=cwd, owner_id=user_id)
    manager._sessions[conv_id] = session
    if activate:
        agent._active_conv[user_id] = conv_id
    return session


# ── //help ──────────────────────────────────────────────────────────────────

class TestHelp:
    @pytest.mark.asyncio
    async def test_help_lists_commands(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//help")
        for cmd in ("//new", "//list", "//close", "//switch", "//perm",
                     "//stop", "//progress", "//direct", "//smart", "//shell",
                     "//help"):
            assert cmd in reply, f"Missing {cmd} in help"
        # Should NOT list //retry (unimplemented)
        assert "//retry" not in reply
        # Should list aliases
        assert "alias" in reply.lower()

    @pytest.mark.asyncio
    async def test_h_alias(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//h")
        assert "//new" in reply

    @pytest.mark.asyncio
    async def test_unknown_command_returns_none(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//unknown_xyz")
        assert reply is None


# ── //new ───────────────────────────────────────────────────────────────────

class TestNew:
    @pytest.mark.asyncio
    async def test_no_args_shows_usage(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//new")
        assert "Usage" in reply

    @pytest.mark.asyncio
    async def test_creates_session(self, tmp_working_dir):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//new myproject")
        assert "myproject" in reply
        assert len(manager.list_sessions(user_id="user_abc123")) == 1

    @pytest.mark.asyncio
    async def test_path_traversal_blocked(self, tmp_working_dir):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//new ../../etc")
        assert "Error" in reply
        assert len(manager.list_sessions(user_id="user_abc123")) == 0

    @pytest.mark.asyncio
    async def test_with_perm_flag(self, tmp_working_dir):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//new myproject --perm plan")
        sessions = manager.list_sessions(user_id="user_abc123")
        assert len(sessions) == 1
        assert sessions[0]["permission_mode"] == "plan"

    @pytest.mark.asyncio
    async def test_sends_card_when_chat_set(self, tmp_working_dir, feishu_calls):
        from bot.commands import handle_command
        _setup_user(chat_id="chat1")
        reply = await handle_command("user_abc123", "//new myproject")
        assert reply == ""  # card was sent instead
        assert len(feishu_calls["cards"]) >= 1


# ── //close ─────────────────────────────────────────────────────────────────

class TestClose:
    @pytest.mark.asyncio
    async def test_no_sessions(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//close")
        assert "No sessions" in reply or "No active" in reply

    @pytest.mark.asyncio
    async def test_close_active(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//close")
        assert "Closed" in reply
        assert len(manager.list_sessions(user_id="user_abc123")) == 0

    @pytest.mark.asyncio
    async def test_close_by_number(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1")
        _add_session("s2")
        reply = await handle_command("user_abc123", "//close 1")
        assert "Closed" in reply
        assert len(manager.list_sessions(user_id="user_abc123")) == 1

    @pytest.mark.asyncio
    async def test_invalid_number(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1")
        reply = await handle_command("user_abc123", "//close 9")
        assert "Invalid" in reply

    @pytest.mark.asyncio
    async def test_cannot_close_other_user(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", user_id="other_user")
        reply = await handle_command("user_abc123", "//close s1")
        assert "another user" in reply or "not found" in reply.lower()


# ── //switch ────────────────────────────────────────────────────────────────

class TestSwitch:
    @pytest.mark.asyncio
    async def test_no_sessions(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//switch 1")
        assert "No sessions" in reply

    @pytest.mark.asyncio
    async def test_valid_switch(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1")
        _add_session("s2")
        reply = await handle_command("user_abc123", "//switch 2")
        assert "Switched" in reply
        assert agent._active_conv["user_abc123"] == "s2"

    @pytest.mark.asyncio
    async def test_out_of_range(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1")
        reply = await handle_command("user_abc123", "//switch 5")
        assert "Invalid" in reply

    @pytest.mark.asyncio
    async def test_non_numeric(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1")
        reply = await handle_command("user_abc123", "//switch abc")
        assert "Invalid" in reply


# ── //status ────────────────────────────────────────────────────────────────

class TestList:
    @pytest.mark.asyncio
    async def test_no_sessions(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//list")
        assert "No active sessions" in reply

    @pytest.mark.asyncio
    async def test_shows_sessions(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1")
        _add_session("s2")
        reply = await handle_command("user_abc123", "//list")
        assert "s1" in reply
        assert "s2" in reply

    @pytest.mark.asyncio
    async def test_shows_active_marker(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//list")
        assert "→" in reply

    @pytest.mark.asyncio
    async def test_status_alias_still_works(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//status")
        assert "No active sessions" in reply


# ── //perm ──────────────────────────────────────────────────────────────────

class TestPerm:
    @pytest.mark.asyncio
    async def test_no_args_shows_usage(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//perm")
        assert "Usage" in reply

    @pytest.mark.asyncio
    async def test_set_edit(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//perm edit")
        assert "edit" in reply
        assert manager._sessions["s1"].permission_mode == "acceptEdits"

    @pytest.mark.asyncio
    async def test_set_plan(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//perm plan")
        assert "plan" in reply
        assert manager._sessions["s1"].permission_mode == "plan"

    @pytest.mark.asyncio
    async def test_set_auto(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//perm auto")
        assert "auto" in reply
        assert manager._sessions["s1"].permission_mode == "dontAsk"

    @pytest.mark.asyncio
    async def test_unknown_mode(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//perm xyz")
        assert "Unknown" in reply

    @pytest.mark.asyncio
    async def test_no_active_session(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//perm edit")
        assert "No active session" in reply


# ── //direct + //smart ──────────────────────────────────────────────────────

class TestDirectSmart:
    @pytest.mark.asyncio
    async def test_direct_requires_session(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//direct")
        assert "No active session" in reply

    @pytest.mark.asyncio
    async def test_direct_enables_passthrough(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        reply = await handle_command("user_abc123", "//direct")
        assert "Direct mode ON" in reply
        assert agent._passthrough["user_abc123"] is True

    @pytest.mark.asyncio
    async def test_smart_disables_passthrough(self):
        from bot.commands import handle_command
        _setup_user()
        _add_session("s1", activate=True)
        agent._passthrough["user_abc123"] = True
        reply = await handle_command("user_abc123", "//smart")
        assert "Smart mode ON" in reply
        assert agent._passthrough["user_abc123"] is False


# ── //shell ─────────────────────────────────────────────────────────────────

class TestShell:
    @pytest.mark.asyncio
    async def test_no_args_shows_usage(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//shell")
        assert "Usage" in reply

    @pytest.mark.asyncio
    async def test_echo(self, tmp_working_dir):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//shell echo hello")
        assert "hello" in reply
        assert "exit code: 0" in reply

    @pytest.mark.asyncio
    async def test_blocked_dangerous(self, tmp_working_dir):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//shell rm -rf /")
        assert "Blocked" in reply


# ── //remind ────────────────────────────────────────────────────────────────

class TestRemind:
    @pytest.mark.asyncio
    async def test_no_args(self):
        from bot.commands import handle_command
        _setup_user(chat_id="chat1")
        reply = await handle_command("user_abc123", "//remind")
        assert "Usage" in reply

    @pytest.mark.asyncio
    async def test_missing_message(self):
        from bot.commands import handle_command
        _setup_user(chat_id="chat1")
        reply = await handle_command("user_abc123", "//remind 10m")
        assert "Usage" in reply

    @pytest.mark.asyncio
    async def test_valid_reminder(self):
        from bot.commands import handle_command
        from agent.scheduler import scheduler
        _setup_user(chat_id="chat1")
        reply = await handle_command("user_abc123", "//remind 30s check build")
        assert "Reminder" in reply
        assert len(scheduler._jobs) == 1


# ── //tasks ─────────────────────────────────────────────────────────────────

class TestTasks:
    @pytest.mark.asyncio
    async def test_no_tasks(self):
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//tasks")
        assert "No background tasks" in reply

    @pytest.mark.asyncio
    async def test_shows_running_task(self):
        from bot.commands import handle_command
        from agent.task_runner import task_runner, BackgroundTask, TaskStatus
        _setup_user()
        task_runner._tasks["t1"] = BackgroundTask(
            task_id="t1", description="fix bug", started_at=time.time(),
            status=TaskStatus.RUNNING,
        )
        reply = await handle_command("user_abc123", "//tasks")
        assert "t1" in reply
        assert "⏳" in reply


# ── //nodes ─────────────────────────────────────────────────────────────────

class TestNodes:
    @pytest.mark.asyncio
    async def test_nodes_outside_router_mode(self, monkeypatch):
        import config
        monkeypatch.setattr(config, "ROUTER_MODE", False)
        from bot.commands import handle_command
        _setup_user()
        reply = await handle_command("user_abc123", "//nodes")
        assert "Not in router mode" in reply