RandomZ / app /tests /test_lcel_invoke_stub.py
StormShadow308's picture
feat: async pipeline, job queue, generation hardening, and docs
732b14f
Raw
History Blame Contribute Delete
562 Bytes
"""Coverage anchor for LCEL invoke helpers."""
from __future__ import annotations
import pytest
from app.llm.lcel_invoke import invoke_lcel_chain
@pytest.mark.asyncio
async def test_invoke_lcel_chain_mock_chain() -> None:
class _Chain:
async def ainvoke(self, variables: dict) -> str:
return f"ok:{variables.get('x')}"
def invoke(self, variables: dict) -> str:
return f"sync:{variables.get('x')}"
out = await invoke_lcel_chain(_Chain(), {"x": "1"}, phase="test")
assert "ok:1" in out or "sync:1" in out