tw/tests/test_regression_pipeline.py

import os
import unittest
from unittest.mock import AsyncMock

from core.pydantic_ai_agent import CustomerServiceAgent, CustomerMessage
from db.customer_db import db


class RegressionPipelineTest(unittest.IsolatedAsyncioTestCase):
    def setUp(self):
        self.customer_id = "__regression_test_customer__"
        db.clear_pending_quote_state(self.customer_id)
        os.environ["FEATURE_BATCH_QUOTE_ENABLED"] = "true"
        os.environ["FEATURE_BATCH_QUOTE_PERCENT"] = "100"
        os.environ["FEATURE_BATCH_QUOTE_SHOPS"] = ""

    async def test_collect_images_then_ack(self):
        agent = CustomerServiceAgent()
        msg = CustomerMessage(
            msg_id="m1",
            acc_id="test_shop",
            msg="https://img.alicdn.com/a.jpg#*#https://img.alicdn.com/b.jpg",
            from_id=self.customer_id,
            from_name="t",
            cy_id=self.customer_id,
            acc_type="AliWorkbench",
            msg_type=0,
            cy_name="t",
            goods_name="专业找图",
            goods_order="",
        )
        resp = await agent.process_message(msg)
        self.assertTrue(resp.should_reply)
        self.assertIn("张", resp.reply)
        st = agent._get_conversation_state(self.customer_id)
        self.assertEqual(len(st.pending_image_urls), 2)

    async def test_finish_signal_triggers_batch_quote(self):
        agent = CustomerServiceAgent()
        st = agent._get_conversation_state(self.customer_id)
        st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
        st.pending_requirements = ["去背景"]
        agent._sync_pending_quote_state(self.customer_id, st)
        agent._quote_pending_images = AsyncMock(return_value={"reply": "打包15元，确认我就安排", "need_transfer": False})

        msg = CustomerMessage(
            msg_id="m2",
            acc_id="test_shop",
            msg="发完了，报价吧",
            from_id=self.customer_id,
            from_name="t",
            cy_id=self.customer_id,
            acc_type="AliWorkbench",
            msg_type=0,
            cy_name="t",
            goods_name="专业找图",
            goods_order="",
        )
        resp = await agent.process_message(msg)
        self.assertTrue(resp.should_reply)
        self.assertIn("15", resp.reply)
        agent._quote_pending_images.assert_awaited()

    async def test_single_image_requirement_intent_triggers_quote(self):
        agent = CustomerServiceAgent()
        st = agent._get_conversation_state(self.customer_id)
        st.pending_image_urls = ["https://img.alicdn.com/a.jpg"]
        st.pending_requirements = []
        agent._sync_pending_quote_state(self.customer_id, st)
        agent._quote_pending_images = AsyncMock(return_value={"reply": "这张20元，定了我马上做", "need_transfer": False})

        msg = CustomerMessage(
            msg_id="m3",
            acc_id="test_shop",
            msg="这个门头上面的字做一下",
            from_id=self.customer_id,
            from_name="t",
            cy_id=self.customer_id,
            acc_type="AliWorkbench",
            msg_type=0,
            cy_name="t",
            goods_name="专业找图",
            goods_order="",
        )
        resp = await agent.process_message(msg)
        self.assertTrue(resp.should_reply)
        self.assertIn("20", resp.reply)
        agent._quote_pending_images.assert_awaited()

    async def test_multi_image_finish_intent_triggers_quote(self):
        agent = CustomerServiceAgent()
        st = agent._get_conversation_state(self.customer_id)
        st.pending_image_urls = ["https://img.alicdn.com/a.jpg", "https://img.alicdn.com/b.jpg"]
        st.pending_requirements = ["改字"]
        agent._sync_pending_quote_state(self.customer_id, st)
        agent._quote_pending_images = AsyncMock(return_value={"reply": "两张打包45，定了我就开做", "need_transfer": False})

        msg = CustomerMessage(
            msg_id="m4",
            acc_id="test_shop",
            msg="就这几张，先按这些报个价",
            from_id=self.customer_id,
            from_name="t",
            cy_id=self.customer_id,
            acc_type="AliWorkbench",
            msg_type=0,
            cy_name="t",
            goods_name="专业找图",
            goods_order="",
        )
        resp = await agent.process_message(msg)
        self.assertTrue(resp.should_reply)
        self.assertIn("45", resp.reply)
        agent._quote_pending_images.assert_awaited()

    async def test_cross_image_composite_intent_triggers_quote(self):
        agent = CustomerServiceAgent()
        st = agent._get_conversation_state(self.customer_id)
        st.pending_image_urls = ["https://img.alicdn.com/a.jpg", "https://img.alicdn.com/b.jpg"]
        st.pending_requirements = []
        agent._sync_pending_quote_state(self.customer_id, st)
        agent._quote_pending_images = AsyncMock(return_value={"reply": "这个跨图合成可以做，打包50元", "need_transfer": False})

        msg = CustomerMessage(
            msg_id="m5",
            acc_id="test_shop",
            msg="A图的图案转换到B图上去",
            from_id=self.customer_id,
            from_name="t",
            cy_id=self.customer_id,
            acc_type="AliWorkbench",
            msg_type=0,
            cy_name="t",
            goods_name="专业找图",
            goods_order="",
        )
        resp = await agent.process_message(msg)
        self.assertTrue(resp.should_reply)
        self.assertIn("50", resp.reply)
        agent._quote_pending_images.assert_awaited()

    async def test_pending_state_restore(self):
        db.update_pending_quote_state(self.customer_id, ["u1", "u2"], ["r1"])
        agent = CustomerServiceAgent()
        st = agent._get_conversation_state(self.customer_id)
        self.assertEqual(st.pending_image_urls, ["u1", "u2"])
        self.assertEqual(st.pending_requirements, ["r1"])

    def tearDown(self):
        db.clear_pending_quote_state(self.customer_id)


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