""" 私聊建单前违禁品：词表、改口、不建单。 不连企微、TMS、主账。群聊不走这条拦截。 """ from __future__ import annotations import os import sys import unittest ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) if ROOT not in sys.path: sys.path.insert(0, ROOT) os.environ.setdefault("YTD_ENV", "test") os.environ.setdefault("REDIS_BACKEND", "memory") os.environ.setdefault("CHECKPOINT_BACKEND", "memory") os.environ.setdefault("MESSAGE_STORE_BACKEND", "memory") os.environ.setdefault("LLM_DATA_USAGE_CONFIRMED", "false") os.environ.setdefault("LLM_ALLOW_NETWORK", "false") os.environ.setdefault("LEDGER_BACKEND", "memory") from agent.ledger.memory_ledger import MemoryLedger from agent.policy.sensitive_cargo import BLOCK_REPLY, contains_blocked, judge from agent.policy.sea_text_flow import SeaTextInquiryFlow, reset_sea_text_flow_for_test COMPLETE_SEA = { "起运港": "上海", "目的港": "洛杉矶", "品名": "普货", "货量": "20吨", "整柜或拼柜": "整柜", "箱型箱量": "1x40HQ", "贸易条款": "FOB", "运输分类": "Port to Port", } class WordListTests(unittest.TestCase): def test_real_weapons_and_drugs_block(self) -> None: for word in ("炸弹", "枪支", "手枪", "子弹", "海洛因", "火炮"): self.assertTrue(contains_blocked(f"上海到洛杉矶，品名{word}"), word) def test_toys_and_fireworks_pass(self) -> None: for word in ("玩具枪", "水枪", "仿真枪", "气枪", "烟花", "礼炮", "鞭炮"): self.assertFalse(contains_blocked(f"品名{word}"), word) def test_toy_prefix_does_not_hide_real_gun(self) -> None: self.assertTrue(contains_blocked("玩具枪支")) def test_ordinary_dangerous_goods_pass(self) -> None: self.assertFalse(contains_blocked("这票是危险品，含电，锂电池")) class JudgeTests(unittest.TestCase): def test_stray_sentence_sticks_until_cargo_rename(self) -> None: facts = {"品名": "服装", "起运港": "上海"} blocked, hold, _edits = judge(text="客户提到有炸弹", facts=dict(facts), hold="") self.assertTrue(blocked) self.assertIn("炸弹", hold) still, hold2, _edits = judge(text="重量100KGS", facts=dict(facts), hold=hold) self.assertTrue(still) self.assertEqual(hold2, hold) released, hold3, edits = judge(text="品名改为服装", facts=dict(facts), hold=hold) self.assertFalse(released) self.assertEqual(hold3, "") self.assertEqual(edits.get("品名"), "服装") def test_rename_sentence_that_still_says_the_word_stays_blocked(self) -> None: blocked, hold, _edits = judge(text="不是炸弹，品名服装", facts={"品名": "服装"}, hold="") self.assertTrue(blocked) self.assertIn("炸弹", hold) def test_origin_and_remark_remain_after_cargo_rename(self) -> None: facts = {"品名": "炸弹", "起运港": "上海", "备注": "内有枪支"} blocked, hold, _edits = judge(text="品名改为服装", facts=facts, hold="品名炸弹") self.assertTrue(blocked) self.assertEqual(hold, "") self.assertEqual(facts["品名"], "服装") self.assertIn("枪支", facts["备注"]) def test_remark_can_be_rewritten(self) -> None: facts = {"品名": "服装", "备注": "枪支"} blocked, hold, _edits = judge(text="备注改为无", facts=facts, hold="") self.assertFalse(blocked) self.assertEqual(facts["备注"], "无") self.assertEqual(hold, "") class SeaPrivateFlowTests(unittest.TestCase): def setUp(self) -> None: reset_sea_text_flow_for_test() self.ledger = MemoryLedger() self.flow = SeaTextInquiryFlow(self.ledger) self.replies: list[str] = [] def _reply(self, text: str, extra=None) -> None: _ = extra self.replies.append(text) def test_private_text_blocks_before_ticket(self) -> None: facts = dict(COMPLETE_SEA) facts["品名"] = "炸弹" phase = self.flow.on_text( sender_id="s-bomb", text="海运 上海到洛杉矶 品名炸弹", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "sensitive_block") self.assertEqual(self.replies, [BLOCK_REPLY]) self.assertEqual(self.ledger._tickets, {}) sess = self.flow.session_of("s-bomb") self.assertFalse((sess.work_order_no or "").strip()) def test_file_facts_block_even_if_caption_is_plain(self) -> None: facts = dict(COMPLETE_SEA) facts["品名"] = "枪支" phase = self.flow.on_text( sender_id="s-file", text="海运报价单", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "sensitive_block") self.assertEqual(self.ledger._tickets, {}) def test_rename_then_confirm_creates_ticket(self) -> None: facts = dict(COMPLETE_SEA) facts["品名"] = "炸弹" self.flow.on_text( sender_id="s-fix", text="海运 品名炸弹 上海到洛杉矶", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.replies.clear() fixed = dict(COMPLETE_SEA) phase = self.flow.on_text( sender_id="s-fix", text="品名改为服装", reply=self._reply, injected_facts=fixed, injected_mode="SEA", ) self.assertNotEqual(phase, "sensitive_block") self.assertNotIn(BLOCK_REPLY, self.replies) self.replies.clear() created = self.flow.on_text(sender_id="s-fix", text="确定", reply=self._reply) self.assertNotEqual(created, "sensitive_block") self.assertTrue(self.ledger._tickets) joined = "\n".join(self.replies) self.assertNotIn(BLOCK_REPLY, joined) def test_oral_restatement_clears_previous_bomb(self) -> None: """销售不说「品名改为」，直接重说运送衣服，或说改成衣服。""" hold = "10月30日，从广州去深圳，运送炸弹，请报价" facts = {"品名": "炸弹", "起运港": "广州"} for text in ( "10月30日，从广州去深圳，运送衣服，请报价", "炸弹错了，给我改成衣服", "炸弹换衣服", ): blocked, new_hold, edits = judge(text=text, facts=dict(facts), hold=hold) self.assertFalse(blocked, text) self.assertEqual(new_hold, "", text) self.assertEqual(edits.get("品名"), "衣服", text) def test_weight_alone_does_not_clear(self) -> None: facts = dict(COMPLETE_SEA) facts["品名"] = "弹药" self.flow.on_text( sender_id="s-hold", text="海运 品名弹药", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.replies.clear() phase = self.flow.on_text( sender_id="s-hold", text="重量 20吨", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "sensitive_block") self.assertEqual(self.replies, [BLOCK_REPLY]) self.assertEqual(self.ledger._tickets, {}) def test_toy_gun_can_continue(self) -> None: facts = dict(COMPLETE_SEA) facts["品名"] = "玩具枪" phase = self.flow.on_text( sender_id="s-toy", text="海运 品名玩具枪 上海到洛杉矶", reply=self._reply, injected_facts=facts, injected_mode="SEA", ) self.assertEqual(phase, "wait_confirm") self.assertNotIn(BLOCK_REPLY, self.replies)