345 lines
17 KiB
Python
345 lines
17 KiB
Python
# -*- 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()
|