test_app.py

from fastapi.testclient import TestClient
import httpx
import json
import tomllib

from src import app
from src import configure_codex


def test_send_accepts_worktool_success_code_200(monkeypatch):
    request = httpx.Request("POST", app.WORKTOOL_SEND_URL)
    response = httpx.Response(
        200,
        request=request,
        json={"code": 200, "message": "指令已加入机器人离线待发队列中！"},
    )
    monkeypatch.setattr(app.httpx, "post", lambda *args, **kwargs: response)

    app.send_worktool_message("robot-send", "发送测试用户", "send-api-test")


def test_callback_ack_then_codex_result(monkeypatch):
    sent = []
    monkeypatch.setattr(app, "is_registered_robot", lambda robot_id: True)
    monkeypatch.setattr(app, "record_message", lambda *args, **kwargs: None)
    monkeypatch.setattr(
        app,
        "send_worktool_message",
        lambda robot_id, target, text: sent.append((robot_id, target, text)),
    )
    monkeypatch.setattr(
        app, "run_codex", lambda robot_id, key, prompt: f"Codex: {prompt}"
    )

    response = TestClient(app.app).post(
        "/worktool/callback/robot-a",
        json={
            "spoken": "hi",
            "rawSpoken": "@机器人 hi",
            "receivedName": "测试用户",
            "groupName": "测试群",
            "groupRemark": "",
            "roomType": 1,
            "atMe": True,
            "textType": 1,
            "messageId": "simulation-001",
        },
    )

    assert response.json() == {"code": 0, "message": "success"}
    assert sent == [
        ("robot-a", "测试群", app.ACK_MESSAGE),
        ("robot-a", "测试群", "Codex: hi"),
    ]


def test_each_callback_is_processed(monkeypatch):
    sent = []
    monkeypatch.setattr(app, "is_registered_robot", lambda robot_id: True)
    monkeypatch.setattr(app, "record_message", lambda *args, **kwargs: None)
    monkeypatch.setattr(
        app,
        "send_worktool_message",
        lambda robot_id, target, text: sent.append((robot_id, target, text)),
    )
    monkeypatch.setattr(app, "run_codex", lambda robot_id, key, prompt: "done")
    client = TestClient(app.app)
    payload = {
        "spoken": "duplicate-callback-test",
        "receivedName": "测试用户",
        "roomType": 2,
        "textType": 1,
        "messageId": "simulation-duplicate",
    }

    assert client.post("/worktool/callback/robot-a", json=payload).status_code == 200
    assert client.post("/worktool/callback/robot-a", json=payload).status_code == 200
    assert sent == [
        ("robot-a", "测试用户", app.ACK_MESSAGE),
        ("robot-a", "测试用户", "done"),
        ("robot-a", "测试用户", app.ACK_MESSAGE),
        ("robot-a", "测试用户", "done"),
    ]


def test_codex_config_generated_from_environment(monkeypatch, tmp_path):
    monkeypatch.setenv("CODEX_CONFIG_HOME", str(tmp_path))
    monkeypatch.setenv("OPENAI_API_KEY", 'key-with-"quote')
    monkeypatch.setenv("OPENAI_BASE_URL", "https://example.test/v1")
    monkeypatch.setenv("CODEX_MODEL_PROVIDER", "test_provider")
    monkeypatch.setenv("CODEX_MODEL", "test-model")

    configure_codex.configure_from_env()

    assert json.loads((tmp_path / "auth.json").read_text()) == {
        "OPENAI_API_KEY": 'key-with-"quote'
    }
    config = tomllib.loads((tmp_path / "config.toml").read_text())
    assert config["model"] == "test-model"
    assert config["model_providers"]["test_provider"]["base_url"] == (
        "https://example.test/v1"
    )


def test_same_conversation_resumes_persisted_thread(monkeypatch, tmp_path):
    calls = []

    class FakeThread:
        id = "thread-123"

        def run(self, prompt, effort):
            text = prompt[-1].text if isinstance(prompt, list) else prompt
            calls.append(("run", text, effort))
            return type("Result", (), {"final_response": f"answer: {text}"})()

    class FakeClient:
        def __enter__(self):
            return self

        def __exit__(self, *args):
            return None

        def thread_start(self, **kwargs):
            calls.append(("start", kwargs))
            return FakeThread()

        def thread_resume(self, thread_id, **kwargs):
            calls.append(("resume", thread_id, kwargs))
            return FakeThread()

    monkeypatch.setattr(app, "CODEX_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
    monkeypatch.setattr(app, "CODEX_WORKSPACE_ROOT", str(tmp_path / "workspaces"))
    monkeypatch.setattr(app, "Codex", FakeClient)
    app.initialize_session_store()

    assert app.run_codex("robot-a", "项目群", "first") == "answer: first"
    assert app.run_codex("robot-a", "项目群", "second") == "answer: second"

    assert [call[0] for call in calls] == ["start", "run", "resume", "run"]
    assert calls[2][1] == "thread-123"
    assert calls[0][1]["cwd"] == str(tmp_path / "workspaces" / "robot-a")
    assert calls[2][2]["cwd"] == str(tmp_path / "workspaces" / "robot-a")
    assert calls[0][1]["sandbox"] is app.Sandbox.full_access
    assert calls[2][2]["sandbox"] is app.Sandbox.full_access


def test_skill_crud_and_disabled_skills_are_not_injected(monkeypatch, tmp_path):
    monkeypatch.setattr(app, "CODEX_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
    monkeypatch.setattr(app, "CODEX_WORKSPACE_ROOT", str(tmp_path / "workspaces"))
    app.initialize_session_store()
    app.register_robot("robot-a", "https://example.test/callback")
    client = TestClient(app.app)
    base = "/api/robots/robot-a/skills"
    scope = "?conversation_key=项目群"

    response = client.post(
        base + scope, json={"name": "deploy_check", "content": "# Deploy\n"}
    )
    assert response.status_code == 200
    assert response.json() == {"status": "created", "name": "deploy_check"}

    response = client.get(base + scope)
    assert response.json()["skills"] == [
        {"name": "deploy_check", "content": "# Deploy\n", "enabled": True}
    ]
    assert [item.name for item in app.robot_skill_inputs("robot-a", "项目群", "run")[:-1]] == [
        "deploy_check"
    ]

    response = client.patch(
        base + "/deploy_check" + scope, json={"enabled": False}
    )
    assert response.json()["enabled"] is False
    assert app.robot_skill_inputs("robot-a", "项目群", "run")[-1].text == "run"

    response = client.put(
        base + "/deploy_check" + scope, json={"content": "# Updated\n"}
    )
    assert response.json()["status"] == "saved"
    assert client.get(base + scope).json()["skills"][0]["content"] == "# Updated\n"

    response = client.delete(base + "/deploy_check" + scope)
    assert response.json() == {"status": "deleted", "name": "deploy_check"}
    assert client.get(base + scope).json() == {"skills": []}


def test_missing_thread_rollout_is_recreated(monkeypatch, tmp_path):
    calls = []

    class FakeThread:
        def __init__(self, thread_id):
            self.id = thread_id

        def run(self, prompt, effort):
            return type("Result", (), {"final_response": "recovered"})()

    class FakeClient:
        def __enter__(self):
            return self

        def __exit__(self, *args):
            return None

        def thread_resume(self, thread_id, **kwargs):
            calls.append(("resume", thread_id))
            raise RuntimeError("JSON-RPC error -32600: no rollout found for thread id old")

        def thread_start(self, **kwargs):
            calls.append(("start", kwargs))
            return FakeThread("new-thread")

    monkeypatch.setattr(app, "CODEX_SESSION_DB", str(tmp_path / "sessions.sqlite3"))
    monkeypatch.setattr(app, "CODEX_WORKSPACE_ROOT", str(tmp_path / "workspaces"))
    monkeypatch.setattr(app, "Codex", FakeClient)
    app.initialize_session_store()
    app.register_robot("robot-a", "https://example.test/callback")
    app.save_thread_id("robot-a", "项目群", "old-thread")

    assert app.run_codex("robot-a", "项目群", "retry") == "recovered"
    assert [call[0] for call in calls] == ["resume", "start"]
    assert app.get_thread_id("robot-a", "项目群") == "new-thread"


def test_robot_workspaces_are_isolated(monkeypatch, tmp_path):
    monkeypatch.setattr(app, "CODEX_WORKSPACE_ROOT", str(tmp_path))

    assert app.robot_working_directory("robot-a") == str(tmp_path / "robot-a")
    assert app.robot_working_directory("robot-b") == str(tmp_path / "robot-b")
    assert (tmp_path / "robot-a").is_dir()
    assert (tmp_path / "robot-b").is_dir()


def test_long_robot_ids_get_short_unique_workspace_names():
    first = app.default_workspace_name("abcdefgh-one-very-long-robot-id")
    second = app.default_workspace_name("abcdefgh-two-very-long-robot-id")

    assert first.startswith("abcdefgh-")
    assert second.startswith("abcdefgh-")
    assert len(first) == 15
    assert first != second