Cognitive_Agent/tests/unit/test_openai_responses.py

from __future__ import annotations

import json
import os
import unittest
import urllib.error
from unittest.mock import patch

from cognitive_agent.infrastructure.llm.openai_responses import (
    SYSTEM_PROMPT,
    OpenAIIntentInterpreter,
)


INTENT = {
    "detected_domains": ["seat"],
    "requests": [
        {
            "domain": "seat",
            "capability": "seat.set_ventilation",
            "target_scope": "passenger",
            "operation": "activate",
            "parameters": [
                {"name": "state", "value": "on"},
                {"name": "level", "value": 1},
            ],
            "reason": "Passenger requested seat ventilation.",
        }
    ],
    "ambiguities": [],
    "decision_summary": ["打开副驾座椅通风"],
    "confidence": 0.99,
}


class FakeResponse:
    def __init__(self, body: str) -> None:
        self.body = body.encode("utf-8")

    def __enter__(self) -> FakeResponse:
        return self

    def __exit__(self, *_args: object) -> None:
        return None

    def read(self) -> bytes:
        return self.body


class OpenAIResponsesTests(unittest.TestCase):
    def test_blank_model_and_base_url_use_documented_defaults(self) -> None:
        with patch.dict(
            os.environ,
            {
                "OPENAI_API_KEY": " test-key ",
                "OPENAI_MODEL": "  ",
                "OPENAI_BASE_URL": "",
                "OPENAI_REASONING_EFFORT": " ",
            },
            clear=True,
        ):
            interpreter = OpenAIIntentInterpreter()

        self.assertEqual(interpreter.api_key, "test-key")
        self.assertEqual(interpreter.model, "gpt-5.6-sol")
        self.assertEqual(interpreter.responses_endpoint, "https://api.openai.com/v1/responses")
        self.assertEqual(interpreter.reasoning_effort, "medium")

    def test_fuzzy_front_cabin_goal_has_deterministic_ergonomic_policy(self) -> None:
        normalized = " ".join(SYSTEM_PROMPT.split())
        self.assertIn("front cabin is cramped", normalized)
        self.assertIn("longitudinal 80%", normalized)
        self.assertIn("backrest recline 35%", normalized)
        self.assertIn("do not manufacture a driver-seat target", normalized)
        self.assertIn("boss", normalized)
        self.assertIn("longitudinal 20%", normalized)
        self.assertIn("do not invent catalog_only capabilities", normalized)

    def test_non_streaming_structured_output(self) -> None:
        response_body = json.dumps({"output_text": json.dumps(INTENT)})
        environment = {
            "OPENAI_API_KEY": "test-key",
            "OPENAI_MODEL": "gpt-5.6-sol",
            "OPENAI_BASE_URL": "https://api.openai.com/v1",
            "OPENAI_STREAM": "false",
        }

        with patch.dict(os.environ, environment, clear=False):
            interpreter = OpenAIIntentInterpreter()
            with patch(
                "cognitive_agent.infrastructure.llm.openai_responses.urllib.request.urlopen",
                return_value=FakeResponse(response_body),
            ) as urlopen:
                result = interpreter.interpret(
                    "副驾还有点闷",
                    {
                        "last_target": "passenger",
                        "devices": {
                            "passenger": {
                                "ventilation": {"state": "on", "level": 1}
                            }
                        },
                    },
                )

        request_payload = json.loads(urlopen.call_args.args[0].data.decode("utf-8"))
        self.assertEqual(result, INTENT)
        self.assertNotIn("stream", request_payload)
        self.assertEqual(request_payload["model"], "gpt-5.6-sol")
        self.assertEqual(
            urlopen.call_args.args[0].full_url,
            "https://api.openai.com/v1/responses",
        )
        self.assertIn("capability registry", request_payload["input"][1]["content"][0]["text"])
        self.assertIn('"level": 1', request_payload["input"][2]["content"][0]["text"])

    def test_streaming_structured_output(self) -> None:
        intent_text = json.dumps(INTENT, ensure_ascii=False)
        stream_body = "\n".join(
            [
                "event: response.output_text.delta",
                f"data: {json.dumps({'type': 'response.output_text.delta', 'delta': intent_text[:30]}, ensure_ascii=False)}",
                "",
                "event: response.output_text.done",
                f"data: {json.dumps({'type': 'response.output_text.done', 'text': intent_text}, ensure_ascii=False)}",
                "",
                "data: [DONE]",
            ]
        )
        environment = {
            "OPENAI_API_KEY": "test-key",
            "OPENAI_STREAM": "true",
        }

        with patch.dict(os.environ, environment, clear=False):
            interpreter = OpenAIIntentInterpreter()
            with patch(
                "cognitive_agent.infrastructure.llm.openai_responses.urllib.request.urlopen",
                return_value=FakeResponse(stream_body),
            ) as urlopen:
                result = interpreter.interpret("打开副驾座椅通风")

        request_payload = json.loads(urlopen.call_args.args[0].data.decode("utf-8"))
        self.assertEqual(result, INTENT)
        self.assertTrue(request_payload["stream"])

    def test_retries_transient_network_failure(self) -> None:
        response_body = json.dumps({"output_text": json.dumps(INTENT)})
        environment = {
            "OPENAI_API_KEY": "test-key",
            "OPENAI_STREAM": "false",
            "OPENAI_MAX_RETRIES": "1",
        }

        with patch.dict(os.environ, environment, clear=False):
            interpreter = OpenAIIntentInterpreter()
            with patch(
                "cognitive_agent.infrastructure.llm.openai_responses.urllib.request.urlopen",
                side_effect=[urllib.error.URLError("temporary TLS EOF"), FakeResponse(response_body)],
            ) as urlopen, patch(
                "cognitive_agent.infrastructure.llm.openai_responses.time.sleep"
            ) as sleep:
                result = interpreter.interpret("打开副驾座椅通风")

        self.assertEqual(result, INTENT)
        self.assertEqual(urlopen.call_count, 2)
        sleep.assert_called_once_with(1)

    def test_retries_transient_error_inside_stream(self) -> None:
        error_stream = "data: " + json.dumps(
            {
                "type": "error",
                "error": {
                    "type": "service_unavailable_error",
                    "code": "server_is_overloaded",
                    "message": "try again",
                },
            }
        )
        intent_text = json.dumps(INTENT, ensure_ascii=False)
        success_stream = "data: " + json.dumps(
            {"type": "response.output_text.done", "text": intent_text},
            ensure_ascii=False,
        )
        environment = {
            "OPENAI_API_KEY": "test-key",
            "OPENAI_STREAM": "true",
            "OPENAI_MAX_RETRIES": "1",
        }

        with patch.dict(os.environ, environment, clear=False):
            interpreter = OpenAIIntentInterpreter()
            with patch(
                "cognitive_agent.infrastructure.llm.openai_responses.urllib.request.urlopen",
                side_effect=[FakeResponse(error_stream), FakeResponse(success_stream)],
            ) as urlopen, patch(
                "cognitive_agent.infrastructure.llm.openai_responses.time.sleep"
            ) as sleep:
                result = interpreter.interpret("打开副驾座椅通风")

        self.assertEqual(result, INTENT)
        self.assertEqual(urlopen.call_count, 2)
        sleep.assert_called_once_with(1)


if __name__ == "__main__":
    unittest.main()