ai-video-fullstack/backend/tests/test_runtime_variables.py

from __future__ import annotations

import unittest

from models import AssistantConfig
from services.runtime_variables import (
    DynamicVariableError,
    DynamicVariableStore,
    prepare_dynamic_config,
    value_at_path,
)


class DynamicVariableTests(unittest.TestCase):
    def test_defaults_client_values_and_system_values_are_merged(self):
        cfg = AssistantConfig(
            type="prompt",
            runtimeMode="pipeline",
            prompt="服务 {{user_name}} / {{tier}} / {{system__conversation_id}}",
            greeting="您好 {{user_name}}",
            dynamic_variable_definitions={
                "user_name": {"type": "string", "required": True},
                "tier": {"type": "string", "default": "普通"},
            },
        )
        prepared = prepare_dynamic_config(
            cfg, {"user_name": "王先生"}, assistant_id="asst_1"
        )
        store = DynamicVariableStore.from_config(prepared)
        self.assertEqual(store.render(prepared.greeting), "您好 王先生")
        self.assertIn("普通", store.render(prepared.prompt))
        self.assertTrue(prepared.conversation_id.startswith("conv_"))
        self.assertEqual(
            prepared.dynamic_variables["system__conversation_id"],
            prepared.conversation_id,
        )

    def test_realtime_prompt_supports_dynamic_variables(self):
        cfg = AssistantConfig(
            type="prompt",
            runtimeMode="realtime",
            prompt="请称呼用户为 {{user_name}}",
            greeting="您好，{{user_name}}",
            dynamic_variable_definitions={
                "user_name": {"type": "string", "required": True}
            },
        )
        prepared = prepare_dynamic_config(
            cfg, {"user_name": "王先生"}, assistant_id="asst_realtime"
        )
        store = DynamicVariableStore.from_config(prepared)
        self.assertEqual(store.render(prepared.greeting), "您好，王先生")
        self.assertEqual(store.render(prepared.prompt), "请称呼用户为 王先生")

    def test_client_cannot_set_reserved_or_wrong_typed_values(self):
        cfg = AssistantConfig(
            type="prompt",
            runtimeMode="pipeline",
            dynamic_variable_definitions={
                "count": {"type": "number", "required": True}
            },
        )
        with self.assertRaisesRegex(DynamicVariableError, "保留变量"):
            prepare_dynamic_config(
                cfg, {"system__agent_turns": 99}, assistant_id="asst_1"
            )
        with self.assertRaisesRegex(DynamicVariableError, "类型"):
            prepare_dynamic_config(cfg, {"count": "1"}, assistant_id="asst_1")

    def test_secret_is_header_only_and_expansion_is_single_pass(self):
        store = DynamicVariableStore(
            {"name": "{{secret__token}}"}, {"secret__token": "private"}
        )
        self.assertEqual(store.render("{{name}}"), "{{secret__token}}")
        with self.assertRaisesRegex(DynamicVariableError, "只能用于 HTTP Header"):
            store.render("{{secret__token}}")
        self.assertEqual(
            store.render("Bearer {{secret__token}}", allow_secrets=True),
            "Bearer private",
        )

    def test_optional_workflow_variable_renders_empty_but_remains_unset(self):
        cfg = AssistantConfig(
            type="workflow",
            graph={
                "specVersion": 3,
                "settings": {"globalPrompt": "称呼 {{nickname}}"},
                "nodes": [
                    {
                        "id": "start",
                        "type": "start",
                        "data": {"greeting": "您好 {{nickname}}"},
                    }
                ],
                "edges": [],
            },
            greeting="",
            dynamic_variable_definitions={
                "nickname": {
                    "type": "string",
                    "required": False,
                    "default": None,
                }
            },
        )
        prepared = prepare_dynamic_config(cfg, {}, assistant_id="asst_workflow")
        store = DynamicVariableStore.from_config(prepared)

        self.assertEqual(store.render("您好 {{nickname}}"), "您好 ")
        self.assertNotIn("nickname", store.values)
        with self.assertRaisesRegex(DynamicVariableError, "缺少动态变量"):
            store.render("{{not_defined}}")

    def test_exact_placeholder_keeps_action_argument_type(self):
        store = DynamicVariableStore(
            {"count": 3, "confirmed": True},
            variable_types={"count": "number", "confirmed": "boolean"},
        )

        rendered = store.render_data(
            {
                "count": "{{count}}",
                "confirmed": "{{confirmed}}",
                "label": "数量 {{count}}",
            }
        )
        self.assertEqual(rendered["count"], 3)
        self.assertIs(rendered["confirmed"], True)
        self.assertEqual(rendered["label"], "数量 3")

        with self.assertRaisesRegex(DynamicVariableError, "类型应为 number"):
            store.assign("count", "four")

    def test_tool_assignment_and_system_history(self):
        store = DynamicVariableStore(
            {
                "order_status": "unknown",
                "system__agent_turns": 0,
                "system__conversation_history": "",
            }
        )
        store.assign("order_status", value_at_path({"order": {"status": "paid"}}, "order.status"))
        store.record("user", "查订单")
        store.record("agent", "已付款", completed_agent_turn=True)
        self.assertEqual(store.values["order_status"], "paid")
        self.assertEqual(store.values["system__agent_turns"], 1)
        self.assertIn("查订单", store.values["system__conversation_history"])


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