# -*- coding: utf-8 -*- """手机远程控制台的桌面端：命令通道、黑名单缓存同步、以服务端为准的移出。""" import shutil import sqlite3 import tempfile import threading import time import unittest from pathlib import Path from unittest import mock import backend_client import conversation_blacklist as cb import remote_control class BlacklistCacheTest(unittest.TestCase): def setUp(self): self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, True) def legacy_database(self): """升级前的本地库：没有 sync_state 列。""" with sqlite3.connect(self.root / "conversation_blacklist.sqlite3") as db: db.execute("CREATE TABLE blocked (account TEXT NOT NULL, conversation TEXT NOT NULL," " name TEXT NOT NULL DEFAULT '', PRIMARY KEY(account,conversation))") db.execute("INSERT INTO blocked(account,conversation,name) VALUES('a1','S:old','老客户')") def test_legacy_entries_are_uploaded_once(self): self.legacy_database() self.assertEqual(cb.pending_entries(self.root)[0]["conversation"], "S:old") self.assertIsNone(cb.synced_version(self.root)) cb.apply_server_snapshot([{"account": "a1", "conversation": "S:old", "name": "老客户"}], 3, root=self.root) self.assertEqual(cb.pending_entries(self.root), []) self.assertEqual(cb.synced_version(self.root), 3) self.assertIn(("a1", "S:old"), cb.entries(self.root)) def test_snapshot_removes_synced_rows_but_keeps_unsent_local_adds(self): cb.apply_server_snapshot([ {"account": "a1", "conversation": "S:1", "reason": "投诉", "created_at": "2026-09-28T10:00:00+08:00"}, {"account": "a1", "conversation": "S:2"}, ], 2, root=self.root) cb.set_blocked("a1", "S:local", True, root=self.root, name="刚拉黑") summary = cb.apply_server_snapshot([{"account": "a1", "conversation": "S:2"}], 4, root=self.root) self.assertEqual(summary["removed"], 1) self.assertEqual(cb.entries(self.root), {("a1", "S:2"), ("a1", "S:local")}) states = cb.local_states(self.root) self.assertEqual(states[("a1", "S:local")], "pending") self.assertEqual(states[("a1", "S:2")], "synced") rows = {row["conversation"]: row for row in cb.list_entries(self.root)} self.assertEqual(rows["S:local"]["name"], "刚拉黑") self.assertEqual(rows["S:local"]["sync_state"], "pending") class RemoveEntriesTest(unittest.TestCase): def setUp(self): self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, True) cb.apply_server_snapshot([{"account": "a1", "conversation": "S:synced"}], 1, root=self.root) cb.set_blocked("a1", "S:pending", True, root=self.root) def test_unsent_entries_are_removed_locally_without_the_server(self): with mock.patch.object(backend_client, "remove_remote_blacklist") as remote: outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": "S:pending"}], root=self.root) remote.assert_not_called() self.assertEqual(outcome["removed"], 1) self.assertNotIn(("a1", "S:pending"), cb.entries(self.root)) def test_synced_entries_need_the_server_and_its_permission(self): denied = backend_client.RemotePermissionDenied("移出黑名单需要 ZYT 权限 wecom_rpa.blacklist/remove", status=403) with mock.patch.object(backend_client, "remove_remote_blacklist", side_effect=denied): outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": "S:synced"}], root=self.root) self.assertFalse(outcome["ok"]) self.assertIn("wecom_rpa.blacklist/remove", outcome["message"]) self.assertIn(("a1", "S:synced"), cb.entries(self.root), "没有权限时本机必须继续拦截") with mock.patch.object(backend_client, "remove_remote_blacklist", return_value={"ok": True}): outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": "S:synced"}], root=self.root) self.assertTrue(outcome["ok"]) self.assertNotIn(("a1", "S:synced"), cb.entries(self.root)) def test_offline_removal_keeps_blocking(self): offline = backend_client.BackendError("无法连接后台：timed out") with mock.patch.object(backend_client, "remove_remote_blacklist", side_effect=offline): outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": "S:synced"}], root=self.root) self.assertFalse(outcome["ok"]) self.assertIn(("a1", "S:synced"), cb.entries(self.root)) def test_already_removed_on_the_server_is_removed_here(self): gone = backend_client.BackendError("该会话不在黑名单中", status=404, code="not_blocked") with mock.patch.object(backend_client, "remove_remote_blacklist", side_effect=gone): outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": "S:synced"}], root=self.root) self.assertEqual(outcome["removed"], 1) def test_a_missing_route_is_not_mistaken_for_an_approved_removal(self): # 旧版或回滚后的管理端没有这个接口，同样是 404，但不能据此删掉本机拦截。 missing = backend_client.BackendError("Not Found", status=404) with mock.patch.object(backend_client, "remove_remote_blacklist", side_effect=missing): outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": "S:synced"}], root=self.root) self.assertFalse(outcome["ok"]) self.assertIn(("a1", "S:synced"), cb.entries(self.root)) def test_first_connection_failure_stops_the_batch(self): cb.apply_server_snapshot([{"account": "a1", "conversation": f"S:{index}"} for index in range(5)], 2, root=self.root) offline = backend_client.BackendError("无法连接后台") with mock.patch.object(backend_client, "remove_remote_blacklist", side_effect=offline) as remote: outcome = remote_control.remove_blacklist_entries( [{"account": "a1", "conversation": f"S:{index}"} for index in range(5)], root=self.root) self.assertEqual(remote.call_count, 1, "界面线程上不能对每一条都等一次超时") self.assertEqual(len(outcome["failed"]), 5) def test_local_only_delete_skips_rows_that_just_got_synced(self): cb.apply_server_snapshot([{"account": "a1", "conversation": "S:pending"}], 3, root=self.root) removed = cb.remove_entries([{"account": "a1", "conversation": "S:pending"}], root=self.root, only_pending=True) self.assertEqual(removed, 0) self.assertIn(("a1", "S:pending"), cb.entries(self.root)) class TenantBindingTest(unittest.TestCase): def setUp(self): self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, True) def test_switching_accounts_keeps_blocking_and_resyncs(self): self.assertTrue(cb.adopt_tenant("tenant-a", root=self.root)) cb.apply_server_snapshot([{"account": "a1", "conversation": "S:1"}], 5, root=self.root, tenant="tenant-a") self.assertEqual(cb.synced_version(self.root, "tenant-a"), 5) self.assertIsNone(cb.synced_version(self.root, "tenant-b")) self.assertTrue(cb.adopt_tenant("tenant-b", root=self.root)) self.assertEqual(cb.local_states(self.root)[("a1", "S:1")], "pending") self.assertIsNone(cb.synced_version(self.root, "tenant-b")) self.assertIn(("a1", "S:1"), cb.entries(self.root)) # 迟到的旧账号快照不能覆盖新绑定 stale = cb.apply_server_snapshot([], 9, root=self.root, tenant="tenant-a") self.assertTrue(stale.get("stale")) self.assertIn(("a1", "S:1"), cb.entries(self.root)) self.assertFalse(cb.adopt_tenant("tenant-b", root=self.root)) def test_rejected_rows_keep_blocking_but_stop_retrying(self): cb.set_blocked("a1", "S:bad", True, root=self.root) self.assertEqual(cb.pending_count(self.root), 1) cb.mark_rejected([{"account": "a1", "conversation": "S:bad"}], root=self.root) self.assertEqual(cb.pending_count(self.root), 0) self.assertEqual(cb.pending_entries(self.root), []) self.assertIn(("a1", "S:bad"), cb.entries(self.root)) class FakeServer: """按顺序回放长轮询响应，记录回执和同步请求。""" def __init__(self, responses): self.responses = list(responses) self.polls = [] self.acks = [] self.syncs = [] self.sync_error = None self.sync_delay = 0.0 self.version = 7 self.done = threading.Event() def poll(self, status, *, wait, blacklist_version=None, timeout=None): self.polls.append({"status": status, "blacklist_version": blacklist_version, "at": time.monotonic()}) if not self.responses: self.done.set() time.sleep(0.05) return {"commands": [], "blacklist_version": self.version} item = self.responses.pop(0) if isinstance(item, BaseException): raise item return item def ack(self, command_id, *, ok, result=None, error="", status=None, timeout=10.0): self.acks.append({"id": command_id, "ok": ok, "result": result, "error": error, "status": status}) return {"ok": True} def sync(self, adds, *, timeout=15.0): self.syncs.append(list(adds)) if self.sync_delay: time.sleep(self.sync_delay) if self.sync_error is not None: raise self.sync_error return {"version": self.version, "items": [{"account": "a1", "conversation": "S:server"}], "rejected": []} class ChannelTest(unittest.TestCase): def setUp(self): self.root = Path(tempfile.mkdtemp()) self.addCleanup(shutil.rmtree, self.root, True) patch = mock.patch.object(remote_control, "MIN_POLL_INTERVAL_SECONDS", 0.01) patch.start() self.addCleanup(patch.stop) def start_channel(self, server, executor=None, executed=None): executed = executed if executed is not None else [] def execute(command): executed.append(command["id"]) return {"ok": True, "result": {"listening": command["type"] == "start_listening", "message": "好了"}} patches = [ mock.patch.object(backend_client, "desktop_access_token", return_value="desk-token"), mock.patch.object(backend_client, "poll_remote_commands", side_effect=server.poll), mock.patch.object(backend_client, "ack_remote_command", side_effect=server.ack), mock.patch.object(backend_client, "sync_remote_blacklist", side_effect=server.sync), ] for patch in patches: patch.start() self.addCleanup(patch.stop) channel = remote_control.RemoteControlChannel( status_provider=lambda: {"listening": False, "state": "stopped"}, command_executor=executor or execute, root=self.root, poll_wait=0, tenant_provider=lambda: "tenant-a", ) channel.start() self.addCleanup(channel.stop) return channel, executed def run_channel(self, server, executor=None): channel, executed = self.start_channel(server, executor) self.assertTrue(server.done.wait(10), "通道没有把回放的响应消费完") channel.stop() return executed def test_commands_run_once_even_when_redelivered(self): command = {"id": "c1", "type": "stop_listening", "expires_at": time.time() + 600, "created_by": "主管"} server = FakeServer([ {"commands": [command], "blacklist_version": 7}, {"commands": [command], "blacklist_version": 7}, ]) executed = self.run_channel(server) self.assertEqual(executed, ["c1"]) self.assertEqual([ack["id"] for ack in server.acks], ["c1", "c1"]) self.assertTrue(all(ack["ok"] for ack in server.acks)) self.assertIn("commands_v1", server.polls[0]["status"]["capabilities"]) def test_expired_commands_are_not_executed(self): server = FakeServer([{"commands": [{"id": "old", "type": "start_listening", "expires_at": time.time() - 5}], "blacklist_version": 7}]) executed = self.run_channel(server) self.assertEqual(executed, []) self.assertFalse(server.acks[0]["ok"]) self.assertIn("有效期", server.acks[0]["error"]) def test_startup_uploads_pending_entries_and_adopts_the_server_version(self): cb.set_blocked("a1", "S:local", True, root=self.root) server = FakeServer([{"commands": [], "blacklist_version": 7}]) self.run_channel(server) self.assertEqual(server.syncs[0][0]["conversation"], "S:local") self.assertEqual(cb.synced_version(self.root), 7) self.assertIn(("a1", "S:server"), cb.entries(self.root)) # 同步之后的长轮询带上本机版本，服务端据此判断要不要立刻叫醒 self.assertEqual(server.polls[-1]["blacklist_version"], 7) def test_failed_sync_cools_down_instead_of_spinning(self): server = FakeServer([{"commands": [], "blacklist_version": 9}] * 3) server.sync_error = backend_client.BackendError("无法连接后台") self.run_channel(server) self.assertEqual(len(server.syncs), 1, "同步失败后 60 秒内不应反复重试") self.assertTrue(all(poll["blacklist_version"] is None for poll in server.polls[1:])) def test_executor_errors_become_failed_acks(self): def broken(_command): raise RuntimeError("界面线程没响应") server = FakeServer([{"commands": [{"id": "c2", "type": "start_listening", "expires_at": time.time() + 600}], "blacklist_version": 7}]) self.run_channel(server, executor=broken) self.assertFalse(server.acks[0]["ok"]) self.assertIn("界面线程没响应", server.acks[0]["error"]) def test_the_thread_survives_unexpected_network_errors(self): import http.client command = {"id": "c3", "type": "stop_listening", "expires_at": time.time() + 600} server = FakeServer([ ConnectionResetError("reset by peer"), http.client.RemoteDisconnected("closed"), {"commands": [command], "blacklist_version": 7}, ]) with mock.patch.object(remote_control, "ERROR_BACKOFF_SECONDS", 0.01): executed = self.run_channel(server) self.assertEqual(executed, ["c3"]) def test_server_clock_decides_expiry(self): # 电脑时钟快了 20 分钟：按服务端时间这条指令还有 5 分钟有效期，必须执行。 server_now = time.time() - 1200 server = FakeServer([{"commands": [{"id": "c4", "type": "start_listening", "expires_at": server_now + 300}], "blacklist_version": 7, "server_time": server_now}]) self.assertEqual(self.run_channel(server), ["c4"]) def test_pending_rows_are_retried_even_when_versions_match(self): server = FakeServer([{"commands": [], "blacklist_version": 7}] * 2) channel, _executed = self.start_channel(server) self.assertTrue(server.done.wait(10)) before = len(server.syncs) cb.set_blocked("a1", "S:later", True, root=self.root) server.done.clear() server.responses = [{"commands": [], "blacklist_version": 7}] * 2 self.assertTrue(server.done.wait(10)) channel.stop() self.assertGreater(len(server.syncs), before) self.assertEqual(cb.local_states(self.root).get(("a1", "S:later")), "pending") def test_no_spin_while_another_sync_is_in_flight(self): """本机刚拉黑触发的同步还没回来时，长轮询因版本变化立即返回，不能空转。""" server = FakeServer([]) server.version = 7 channel, _executed = self.start_channel(server) self.assertTrue(server.done.wait(10)) server.sync_delay = 1.0 server.version = 8 server.responses = [{"commands": [], "blacklist_version": 8}] * 200 syncs_before = len(server.syncs) channel.sync_blacklist_soon() time.sleep(0.15) polls_before = len(server.polls) time.sleep(0.7) # 帮手线程的同步还在进行 polls_during = len(server.polls) - polls_before time.sleep(0.6) channel.stop() self.assertLessEqual(polls_during, 2, "同步进行中不应反复长轮询") self.assertEqual(len(server.syncs) - syncs_before, 1, "等到的那一轮已对齐版本，不该再同步一次") def test_timed_out_commands_are_cancelled_so_they_never_run_late(self): import concurrent.futures future = concurrent.futures.Future() outcome = remote_control.future_result(future, timeout=0.05) self.assertFalse(outcome["ok"]) self.assertTrue(future.cancelled()) if __name__ == "__main__": unittest.main()