Method test_pipeline_with_codec

python/tests/test_codecs.py:254–287

Full pipeline: pass codec instance directly + annotated intercept, execute LLM call.

(self)

class TestCodecPipeline:
    async def test_pipeline_with_codec(self):
        """Full pipeline: pass codec instance directly + annotated intercept, execute LLM call."""
        codec = SimpleCodec()

        intercept_called = []

        def annotated_intercept(name, request, annotated):
            intercept_called.append(True)
            assert annotated is not None
            assert isinstance(annotated, AnnotatedLLMRequest)
            # Add a system message via the annotated request
            new_messages = [
                {"role": "system", "content": "injected by intercept"},
                *annotated.messages,
            ]
            annotated.messages = new_messages
            return (request, annotated)

        intercepts.register_llm_request("test-annot-intercept-pipeline", 1, False, annotated_intercept)

        try:

            def func(request):
                return {"messages": request.content.get("messages", []), "model": request.content.get("model")}

            request = make_request()
            result = await llm.execute("pipeline-llm", request, func, codec=codec)

            assert len(intercept_called) == 1
            # The encode step should have written modified messages back
            assert result["messages"][0]["role"] == "system"
            assert result["messages"][0]["content"] == "injected by intercept"
        finally:
            intercepts.deregister_llm_request("test-annot-intercept-pipeline")

    async def test_codec_parameter(self):
        """codec parameter passes the specified codec instance directly."""