1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
| class ServiceWorkerOfflineDebugger:
"""Service Worker 离线调试器——结合 Network 和 BackgroundService 域"""
def __init__(self, ws):
self.ws = ws
self.session_id = None
self.bg_monitor = None
async def initialize(self):
"""初始化调试环境"""
targets = await cdp(self.ws, "Target.getTargets")
sw_target = None
page_target = None
for target in targets.get("targetInfos", []):
if target["type"] == "service_worker":
sw_target = target
elif target["type"] == "page" and not page_target:
page_target = target
if page_target:
result = await cdp(self.ws, "Target.attachToTarget", {
"targetId": page_target["targetId"],
"flatten": True
})
self.session_id = result["sessionId"]
if sw_target:
sw_result = await cdp(self.ws, "Target.attachToTarget", {
"targetId": sw_target["targetId"],
"flatten": True
})
self.sw_session_id = sw_result["sessionId"]
print(f"[调试] 已附加到 Service Worker: {sw_target.get('url', '')[:60]}")
await cdp(self.ws, "Page.enable", {}, self.session_id)
await cdp(self.ws, "Runtime.enable", {}, self.session_id)
await cdp(self.ws, "Network.enable", {}, self.session_id)
self.bg_monitor = BackgroundServiceMonitor(self.ws, self.session_id)
await self.bg_monitor.enable_all()
return self.session_id
async def simulate_offline_scenario(self):
"""模拟离线场景并观察后台行为"""
print("\n" + "=" * 60)
print("模拟离线场景")
print("=" * 60)
print("\n[步骤 1] 注册后台同步...")
register_sync = """
(async () => {
if ('serviceWorker' in navigator) {
const reg = await navigator.serviceWorker.ready;
await reg.sync.register('offline-data-sync');
return 'Sync registered: offline-data-sync';
}
return 'Service Worker not available';
})();
"""
result = await cdp(self.ws, "Runtime.evaluate", {
"expression": register_sync,
"awaitPromise": True,
"returnByValue": True
}, self.session_id)
print(f" {result.get('result', {}).get('value', '')}")
await asyncio.sleep(1)
print("\n[步骤 2] 模拟离线...")
await cdp(self.ws, "Network.emulateNetworkConditions", {
"offline": True,
"latency": 0,
"downloadThroughput": 0,
"uploadThroughput": 0
}, self.session_id)
print(" 网络状态: 离线")
await asyncio.sleep(1)
print("\n[步骤 3] 在离线状态下发起请求...")
make_request = """
fetch('/api/data.json')
.then(r => r.json())
.then(d => console.log('[Online] Data received:', d))
.catch(e => console.log('[Offline] Fetch failed (expected):', e.message));
"""
await cdp(self.ws, "Runtime.evaluate", {
"expression": make_request,
"returnByValue": True
}, self.session_id)
await asyncio.sleep(2)
print("\n[步骤 4] 恢复在线...")
await cdp(self.ws, "Network.emulateNetworkConditions", {
"offline": False,
"latency": 0,
"downloadThroughput": -1,
"uploadThroughput": -1
}, self.session_id)
print(" 网络状态: 在线")
print("\n等待后台同步事件...")
await asyncio.sleep(5)
print("\n[步骤 5] 收集后台事件...")
events = self.bg_monitor.get_events("backgroundSync")
print(f" 捕获到 {len(events)} 个 Background Sync 事件:")
for event in events:
ename = event.get("eventName", "?")
meta = {
e.get("key", ""): e.get("value", "")
for e in event.get("eventMetadata", [])
}
print(f" - {ename}: {meta}")
return events
async def cleanup(self):
"""清理调试环境"""
if self.bg_monitor:
await self.bg_monitor.disable_all()
await cdp(self.ws, "Network.emulateNetworkConditions", {
"offline": False,
"latency": 0,
"downloadThroughput": -1,
"uploadThroughput": -1
}, self.session_id)
print("[调试] 环境已清理")
|
