# pylint: disable=C0413
# flake8: noqa: E402
from contextlib import asynccontextmanager
from typing import Generator
import httpx
import nest_asyncio
import pytest
from fastapi.testclient import TestClient
from httpx import AsyncClient
nest_asyncio.apply = lambda: None
from chatsky_ui.main import app
from chatsky_ui.schemas.pagination import Pagination
from chatsky_ui.schemas.preset import Preset
from chatsky_ui.services.process import RunProcess
from chatsky_ui.services.process_manager import BuildManager, RunManager
from chatsky_ui.services.websocket_manager import WebSocketManager
DUMMY_BUILD_ID = -1
[docs]async def start_process(async_client: AsyncClient, endpoint, preset_end_status) -> httpx.Response:
return await async_client.post(
endpoint,
json={"wait_time": 0.1, "end_status": preset_end_status},
)
[docs]@asynccontextmanager
async def override_dependency(mocker_obj, get_manager_func):
process_manager = get_manager_func()
process_manager.check_status = mocker_obj.AsyncMock()
app.dependency_overrides[get_manager_func] = lambda: process_manager
try:
yield process_manager
finally:
for _, process in process_manager.processes.items():
if process.process.returncode is None:
await process.stop()
app.dependency_overrides = {}
[docs]@pytest.fixture
def client() -> Generator:
with TestClient(app=app) as client:
yield client
[docs]@pytest.fixture(scope="session")
def preset() -> Preset:
return Preset(
wait_time=0,
end_status="loop",
)
[docs]@pytest.fixture()
def run_process():
async def _run_process(cmd_to_run):
process = RunProcess(id_=0, build_id=DUMMY_BUILD_ID)
await process.start(cmd_to_run)
return process
return _run_process
[docs]@pytest.fixture()
def run_manager():
manager = RunManager()
manager.set_logger()
return manager
[docs]@pytest.fixture()
def build_manager():
return BuildManager()
[docs]@pytest.fixture
def websocket_manager():
manager = WebSocketManager()
manager.set_logger()
return manager