55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
|
import pytest
|
||
|
|
import json
|
||
|
|
import asyncio
|
||
|
|
from aiohttp import web
|
||
|
|
from becomingone import KAIROSTemporalEngine
|
||
|
|
from becomingone.transducers.master import MasterConfig
|
||
|
|
|
||
|
|
# Import from the parent directory by mocking it temporarily
|
||
|
|
import sys
|
||
|
|
import os
|
||
|
|
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
|
import kairos_server
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def mock_engine():
|
||
|
|
from becomingone.core.engine import TemporalConfig
|
||
|
|
config = TemporalConfig(coherence_threshold=0.95)
|
||
|
|
engine = KAIROSTemporalEngine(config=config, name="test_master")
|
||
|
|
return engine
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def cli(aiohttp_client, mock_engine):
|
||
|
|
kairos_server.engine = mock_engine
|
||
|
|
app = await kairos_server.create_app()
|
||
|
|
return await aiohttp_client(app)
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_health_check(cli):
|
||
|
|
kairos_server._engine_components = {"ready": True}
|
||
|
|
resp = await cli.get('/health')
|
||
|
|
assert resp.status == 200
|
||
|
|
data = await resp.json()
|
||
|
|
assert data["status"] == "ready"
|
||
|
|
|
||
|
|
@pytest.mark.asyncio
|
||
|
|
async def test_openai_chat_completions_non_streaming(cli, monkeypatch):
|
||
|
|
# Mock the run_agentic_loop so we don't hit real external APIs during the test
|
||
|
|
async def mock_loop(prompt, c, d, i, m):
|
||
|
|
yield {"agent": {"messages": [type('obj', (object,), {'content': 'CMD: test command'})()]}}
|
||
|
|
|
||
|
|
import agent
|
||
|
|
monkeypatch.setattr(agent, "run_agentic_loop", mock_loop)
|
||
|
|
|
||
|
|
payload = {
|
||
|
|
"model": "becomingone-chorus",
|
||
|
|
"messages": [{"role": "user", "content": "Test prompt"}],
|
||
|
|
"stream": False
|
||
|
|
}
|
||
|
|
|
||
|
|
resp = await cli.post('/v1/chat/completions', json=payload)
|
||
|
|
assert resp.status == 200
|
||
|
|
data = await resp.json()
|
||
|
|
|
||
|
|
content = data["choices"][0]["message"]["content"]
|
||
|
|
assert "CMD: test command" in content
|