From dc04db65386d72f819f1484238605464bb1c1b69 Mon Sep 17 00:00:00 2001
From: jimi <1847930177@qq.com>
Date: Sat, 28 Feb 2026 19:13:23 +0800
Subject: [PATCH] test: add multi-worker ownership routing guard

---
 tests/test_multi_worker_routing.py | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)
 create mode 100644 tests/test_multi_worker_routing.py

diff --git a/tests/test_multi_worker_routing.py b/tests/test_multi_worker_routing.py
new file mode 100644
index 0000000..4157503
--- /dev/null
+++ b/tests/test_multi_worker_routing.py
@@ -0,0 +1,26 @@
+import os
+import unittest
+
+from core.websocket_client import QingjianAPIClient
+
+
+class MultiWorkerRoutingTest(unittest.TestCase):
+    def test_only_one_worker_owns_customer_when_no_explicit_shards(self):
+        os.environ["AI_CS_WORKER_COUNT"] = "4"
+        key = "shop_x:tb123456"
+        owners = 0
+        for wid in range(4):
+            os.environ["AI_CS_WORKER_ID"] = str(wid)
+            c = QingjianAPIClient(enable_agent=False)
+            c.shard_keys = set()  # 模拟当前无分片表
+            if c._is_owned_by_this_worker(key):
+                owners += 1
+        self.assertEqual(owners, 1)
+
+    def tearDown(self):
+        os.environ.pop("AI_CS_WORKER_COUNT", None)
+        os.environ.pop("AI_CS_WORKER_ID", None)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)