106 lines
2.7 KiB
Python
106 lines
2.7 KiB
Python
import pytest
|
|
import asyncio
|
|
from pathlib import Path
|
|
import tempfile
|
|
import shutil
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
|
|
pytest_plugins = ("pytest_asyncio",)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def temp_dir():
|
|
temp_path = Path(tempfile.mkdtemp())
|
|
yield temp_path
|
|
shutil.rmtree(temp_path, ignore_errors=True)
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_settings(monkeypatch, temp_dir):
|
|
from app.core import config
|
|
|
|
class TestSettings(config.Settings):
|
|
data_dir: Path = temp_dir / "xcclaw_test"
|
|
opencode_host: str = "127.0.0.1"
|
|
opencode_port: int = 4096
|
|
opencode_password: str = ""
|
|
app_host: str = "0.0.0.0"
|
|
app_port: int = 3005
|
|
|
|
monkeypatch.setattr(config, "settings", TestSettings())
|
|
return config.settings
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_opencode_client():
|
|
mock = AsyncMock()
|
|
mock.sessions = {}
|
|
mock.call_count = 0
|
|
|
|
async def create_session(title=None, parent_id=None):
|
|
import uuid
|
|
|
|
session_id = str(uuid.uuid4())
|
|
mock.sessions[session_id] = {
|
|
"id": session_id,
|
|
"title": title,
|
|
"status": "created",
|
|
}
|
|
return {"id": session_id, "title": title}
|
|
|
|
async def get_session(session_id):
|
|
return mock.sessions.get(session_id, {})
|
|
|
|
async def delete_session(session_id):
|
|
if session_id in mock.sessions:
|
|
del mock.sessions[session_id]
|
|
return True
|
|
return False
|
|
|
|
async def send_message(session_id, text):
|
|
mock.call_count += 1
|
|
if session_id in mock.sessions:
|
|
mock.sessions[session_id]["status"] = "finished"
|
|
return {"session_id": session_id, "status": "finished"}
|
|
raise ValueError(f"Session {session_id} not found")
|
|
|
|
async def send_message_async(session_id, text):
|
|
mock.call_count += 1
|
|
if session_id in mock.sessions:
|
|
mock.sessions[session_id]["status"] = "running"
|
|
|
|
async def abort_session(session_id):
|
|
if session_id in mock.sessions:
|
|
mock.sessions[session_id]["status"] = "aborted"
|
|
return True
|
|
return False
|
|
|
|
async def get_session_status():
|
|
return {"sessions": mock.sessions}
|
|
|
|
async def health_check():
|
|
return {"status": "ok"}
|
|
|
|
async def close():
|
|
pass
|
|
|
|
mock.create_session = create_session
|
|
mock.get_session = get_session
|
|
mock.delete_session = delete_session
|
|
mock.send_message = send_message
|
|
mock.send_message_async = send_message_async
|
|
mock.abort_session = abort_session
|
|
mock.get_session_status = get_session_status
|
|
mock.health_check = health_check
|
|
mock.close = close
|
|
|
|
return mock
|