1#!/usr/bin/env python3
2import asyncio
3import base64
4import importlib.util
5import json
6import os
7from pathlib import Path
8import random
9import struct
10import sys
11import tempfile
12import threading
13import unittest
14import zlib
15from types import SimpleNamespace
16from unittest.mock import AsyncMock, Mock, patch
17
18
19spec = importlib.util.spec_from_file_location("vm_screen", Path(__file__).with_name("vm-screen.py"))
20screen = importlib.util.module_from_spec(spec)
21spec.loader.exec_module(screen)
22
23
24class Writer:
25 def __init__(self):
26 self.data = bytearray()
27 self.closed = False
28 self.transport = self
29
30 def write(self, data):
31 self.data.extend(data)
32
33 async def drain(self):
34 pass
35
36 def get_write_buffer_size(self):
37 return len(self.data)
38
39 def is_closing(self):
40 return self.closed
41
42 def close(self):
43 self.closed = True
44
45
46class LifecycleTests(unittest.IsolatedAsyncioTestCase):
47 async def test_sigterm_closes_connected_clients(self):
48 with tempfile.TemporaryDirectory() as directory:
49 socket = str(Path(directory) / "screen.sock")
50 process = await asyncio.create_subprocess_exec(sys.executable, str(Path(__file__).with_name("vm-screen.py")),
51 env={**os.environ, "STUDIO_VM_SCREEN_SOCKET": socket})
52 writers = []
53 try:
54 async with asyncio.timeout(5):
55 while not Path(socket).exists():
56 await asyncio.sleep(0.01)
57 for _ in range(4):
58 _, writer = await asyncio.open_unix_connection(socket)
59 writers.append(writer)
60 reader, writer = await asyncio.open_unix_connection(socket)
61 writers.append(writer)
62 self.assertTrue(json.loads(await reader.readline())["fatal"])
63 process.terminate()
64 self.assertEqual(await asyncio.wait_for(process.wait(), 5), 0)
65 finally:
66 for writer in writers:
67 writer.close()
68 await writer.wait_closed()
69 if process.returncode is None:
70 process.kill()
71 await process.wait()
72
73
74class DesktopTests(unittest.IsolatedAsyncioTestCase):
75 def setUp(self):
76 self.reader, self.writer = asyncio.StreamReader(), Writer()
77 self.messages = []
78 self.desktop = screen.Desktop(self.reader, self.writer, self.messages.append)
79 self.desktop.resize(4, 2)
80
81 async def update(self, rectangles):
82 self.reader.feed_data(struct.pack("!BBH", 0, 0, len(rectangles)) + b"".join(rectangles))
83 self.reader.feed_eof()
84 self.presented = self.resized = 0
85
86 def present():
87 self.presented += 1
88
89 def resize():
90 self.resized += 1
91
92 with self.assertRaises(asyncio.IncompleteReadError):
93 await self.desktop.read(present, resize)
94
95 async def test_handshake(self):
96 self.reader.feed_data(b"RFB 003.008\n\x01\x01\0\0\0\0" + struct.pack("!HH16sI", 4, 2, bytes(16), 4) + b"test")
97 await self.desktop.start()
98 self.assertEqual(self.writer.data[:14], b"RFB 003.008\n\x01\x01")
99 self.assertEqual(self.writer.data[-10:], struct.pack("!BBHHHH", 3, 0, 0, 0, 4, 2))
100
101 async def test_raw_and_overlapping_copy_present_once(self):
102 pixels = b"".join(bytes([value, 0, 0, 0]) for value in range(1, 9))
103 await self.update([
104 struct.pack("!HHHHi", 0, 0, 4, 2, 0) + pixels,
105 struct.pack("!HHHHiHH", 1, 0, 3, 2, 1, 0, 0),
106 ])
107 self.assertEqual(list(self.desktop.pixels[::4]), [1, 1, 2, 3, 5, 5, 6, 7])
108 self.assertEqual(self.presented, 1)
109
110 async def test_idle_and_cursor_do_not_encode(self):
111 await self.update([struct.pack("!HHHHi", 0, 0, 2, 1, -239) + bytes([1, 2, 3, 0, 4, 5, 6, 0, 128])])
112 self.assertEqual(self.presented, 0)
113 encoded = base64.b64decode(self.messages[0]["image"].split(",")[1])
114 length = struct.unpack("!I", encoded[33:37])[0]
115 self.assertEqual(zlib.decompress(encoded[41:41 + length]), bytes([0, 3, 2, 1, 255, 6, 5, 4, 0]))
116
117 async def test_resize_and_out_of_bounds(self):
118 await self.update([struct.pack("!HHHHi", 0, 0, 2, 2, -223)])
119 self.assertEqual((self.desktop.width, self.desktop.height, self.resized), (2, 2, 1))
120 self.reader = asyncio.StreamReader()
121 self.desktop.reader = self.reader
122 with self.assertRaisesRegex(ValueError, "outside"):
123 await self.update([struct.pack("!HHHHi", 1, 0, 2, 1, 0)])
124 for width, height in [(0, 2), (4097, 1), (65535, 65535)]:
125 with self.assertRaises(ValueError):
126 self.desktop.resize(width, height)
127
128 def test_input_validation_and_release(self):
129 self.desktop.input({"type": "key", "key": 65507, "down": True})
130 self.desktop.input({"type": "pointer", "x": -1, "y": 10, "buttons": 1})
131 self.desktop.input({"type": "release"})
132 self.assertFalse(self.desktop.keys)
133 self.assertEqual(self.writer.data[-6:], struct.pack("!BBHH", 5, 0, 0, 1))
134 for message in [{"type": "key", "key": True, "down": True}, {"type": "pointer", "x": 1, "y": 1, "buttons": 256}]:
135 with self.assertRaises(ValueError):
136 self.desktop.input(message)
137
138
139class WindowTests(unittest.TestCase):
140 def window(self, id, x=0, y=0, width=2, height=2):
141 return dict(id=id, x=x, y=y, width=width, height=height,
142 caption_left=x, caption_top=y, caption_right=x + width, caption_bottom=y + 1)
143
144 def surface(self, atlas, id, data, x=0, y=0, width=2, height=2):
145 atlas.update(dict(id=id, width=2, height=2, x=x, y=y, rect_width=width, rect_height=height), data)
146
147 def test_partial_surfaces_and_geometry_reuse(self):
148 atlas = screen.WindowDesktop(4, 2, [self.window(1), self.window(2)])
149 self.assertFalse(atlas.complete)
150 with self.assertRaisesRegex(ValueError, "complete first"):
151 self.surface(atlas, 1, bytes(4), width=1, height=1)
152 self.surface(atlas, 1, b"a" * 16)
153 self.surface(atlas, 2, b"b" * 16)
154 self.assertTrue(atlas.complete)
155 self.surface(atlas, 1, b"c" * 4, x=1, y=1, width=1, height=1)
156 self.assertEqual(atlas.surfaces[1][2], b"a" * 12 + b"c" * 4)
157 moved = screen.WindowDesktop(4, 2, [self.window(2), self.window(1, x=1)], atlas)
158 self.assertFalse(moved.repacked)
159 self.assertIs(moved.pixels, atlas.pixels)
160 self.assertEqual(moved.pixels, b"a" * 8 + b"b" * 8 + b"a" * 4 + b"c" * 4 + b"b" * 8)
161 removed = screen.WindowDesktop(4, 2, [self.window(1)], moved)
162 self.assertTrue(removed.repacked)
163 self.assertTrue(removed.complete)
164 self.assertEqual(removed.pixels[16:24], b"a" * 4 + b"c" * 4)
165
166 def test_untrusted_window_bounds_and_atlas_budget(self):
167 for windows in [[self.window(0)], [self.window(1), self.window(1)],
168 [self.window(1, x=4097)], [self.window(1, width=0)]]:
169 with self.subTest(windows=windows), self.assertRaises(ValueError):
170 screen.WindowDesktop(4, 2, windows)
171 with self.assertRaises(OverflowError):
172 screen.WindowDesktop(4096, 1, [self.window(1, width=4096, height=2048), self.window(2, width=4096, height=2048)])
173 atlas = screen.WindowDesktop(4, 2, [self.window(1)])
174 for values, data in [(dict(id=99, width=2, height=2, x=0, y=0, rect_width=2, rect_height=2), bytes(16)),
175 (dict(id=1, width=2, height=2, x=1, y=1, rect_width=2, rect_height=2), bytes(16))]:
176 with self.assertRaises(ValueError):
177 atlas.update(values, data)
178
179
180class GuestTests(unittest.IsolatedAsyncioTestCase):
181 def setUp(self):
182 self.guest = screen.GuestDesktop("fixture")
183 self.messages = []
184 self.presented = self.resized = 0
185 self.viewer = Mock(desktop=SimpleNamespace(cursor={"type": "cursor", "source": "qemu"}),
186 writer=Writer(), send=self.messages.append, resize=self.resize, present=self.present)
187 self.viewer.source = self.guest
188 self.guest.screens = {self.viewer}
189
190 def present(self):
191 self.presented += 1
192
193 def resize(self):
194 self.resized += 1
195
196 async def test_stop_drains_packet_before_closing_channel(self):
197 reader, writer = asyncio.StreamReader(), Writer()
198 writer.get_extra_info = Mock(return_value=Mock(getsockopt=Mock(return_value=struct.pack("3i", 1, 0, 0))))
199 writer.wait_closed = AsyncMock()
200 reader.feed_data(struct.pack("!I", 2) + b"{}")
201 reader.feed_data(struct.pack("!4sII2I", b"SGV1", screen.GUEST_PROTOCOL["packets"]["capabilities"]["id"], 8, 1, 1))
202 reader.feed_data(struct.pack("!4sII6I", b"SGV1", screen.GUEST_PROTOCOL["packets"]["screen"]["id"], 40, 2, 2, 0, 0, 2, 2) + bytes(8))
203 screen.GUESTS[self.guest.name] = self.guest
204 with patch.object(asyncio, "open_unix_connection", AsyncMock(return_value=(reader, writer))):
205 task = asyncio.create_task(self.guest.run())
206 self.guest.task = task
207 async with asyncio.timeout(1):
208 while self.guest.writer is None or self.guest.windows_available is None:
209 await asyncio.sleep(0)
210 self.guest.stop()
211 self.guest.request(refresh=True)
212 stop = struct.pack("!4sII", b"SGV1", screen.GUEST_PROTOCOL["packets"]["stop"]["id"], 0)
213 await asyncio.sleep(0)
214 self.assertFalse(writer.closed)
215 self.assertFalse(task.done())
216 reader.feed_data(bytes(8) + stop)
217 await asyncio.wait_for(task, 1)
218 self.assertTrue(writer.closed)
219 self.assertEqual(self.presented, 2)
220 self.assertNotIn(self.guest.name, screen.GUESTS)
221
222 async def test_stop_during_broker_negotiation(self):
223 reader, writer = asyncio.StreamReader(), Writer()
224 writer.get_extra_info = Mock(return_value=Mock(getsockopt=Mock(return_value=struct.pack("3i", 1, 0, 0))))
225 writer.wait_closed = AsyncMock()
226 stop = struct.pack("!4sII", b"SGV1", screen.GUEST_PROTOCOL["packets"]["stop"]["id"], 0)
227 with patch.object(asyncio, "open_unix_connection", AsyncMock(return_value=(reader, writer))) as connect:
228 task = asyncio.create_task(self.guest.run())
229 self.guest.task = task
230 async with asyncio.timeout(1):
231 while not writer.data:
232 await asyncio.sleep(0)
233 self.guest.stop()
234 self.assertIsNone(self.guest.writer)
235 self.assertFalse(writer.closed)
236 reader.feed_data(struct.pack("!I", 2) + b"{}" + struct.pack("!4sII2I", b"SGV1", screen.GUEST_PROTOCOL["packets"]["capabilities"]["id"], 8, 1, 1) + stop)
237 await asyncio.wait_for(task, 1)
238 connect.assert_awaited_once()
239 self.assertTrue(writer.closed)
240 self.assertTrue(writer.data.endswith(stop))
241
242 async def test_stop_before_reader_starts(self):
243 self.guest.stop()
244 with patch.object(asyncio, "open_unix_connection", AsyncMock()) as connect:
245 await self.guest.run()
246 connect.assert_not_awaited()
247
248 async def test_unsolicited_stop_is_rejected(self):
249 with self.assertRaisesRegex(ValueError, "Unexpected"):
250 await self.packets(("stop", [], b""))
251
252 async def packets(self, *packets):
253 reader = asyncio.StreamReader()
254 for name, words, data in packets:
255 kind = screen.GUEST_PROTOCOL["packets"][name]["id"]
256 payload = struct.pack("!" + "I" * len(words), *words) + data
257 reader.feed_data(struct.pack("!4sII", b"SGV1", kind, len(payload)) + payload)
258 reader.feed_eof()
259 with self.assertRaises(asyncio.IncompleteReadError):
260 await self.guest.read(reader)
261
262 async def test_dirty_rectangles_and_cursor_do_not_reencode_video(self):
263 await self.packets(("screen", [4, 2, 0, 0, 4, 2], bytes(32)),
264 ("screen", [4, 2, 2, 1, 1, 1], b"\x01\x02\x03\0"),
265 ("cursor", [7, 1, 1, 0, 0, 2], struct.pack("!I", 17) + bytes([0,0,0,0,7]) + struct.pack("!I", 33) + bytes([1,2,3,255,0])),
266 ("pointer", [7, 0xffffffff, 9, 1], b""))
267 self.assertEqual(self.guest.pixels[24:28], b"\x01\x02\x03\0")
268 self.assertEqual((self.presented, self.resized), (2, 1))
269 self.assertEqual([message["duration"] for message in self.messages if message["type"] == "cursor"], [17, 33])
270 self.assertIn("invert", next(message for message in self.messages if message["type"] == "cursor"))
271 self.assertEqual(self.guest.pointer["x"], -1)
272
273 async def test_suspend_restores_fallback_and_requires_full_refresh(self):
274 await self.packets(("screen", [2, 2, 0, 0, 2, 2], bytes(16)), ("suspend", [], b""))
275 self.assertFalse(self.guest.pixels)
276 self.assertEqual(self.messages[-1]["source"], "qemu")
277 with self.assertRaisesRegex(ValueError, "complete"):
278 await self.packets(("screen", [2, 2, 0, 0, 1, 1], bytes(4)))
279 self.viewer.desktop.cursor = None
280 await self.packets(("screen", [1, 1, 0, 0, 1, 1], bytes(4)), ("suspend", [], b""))
281 self.assertTrue(self.messages[-1]["fallback"])
282
283 async def test_untrusted_guest_packets_are_bounded(self):
284 for packet in [("screen", [4097, 1, 0, 0, 1, 1], bytes(4)),
285 ("screen", [2, 2, 1, 1, 2, 2], bytes(16)),
286 ("cursor", [1, 1, 1, 1, 0, 1], struct.pack("!I", 100) + bytes(5)),
287 ("cursor", [1, 1, 1, 0, 0, 1], struct.pack("!I", 0) + bytes(5)),
288 ("pointer", [1, 0, 0, 2], b""), ("suspend", [], b"x")]:
289 with self.subTest(packet=packet), self.assertRaises(ValueError):
290 await self.packets(packet)
291 reader = asyncio.StreamReader()
292 reader.feed_data(struct.pack("!4sII", b"SGV1", 1, 0xffffffff))
293 with self.assertRaises(ValueError):
294 await self.guest.read(reader)
295
296 async def test_viewer_close_during_cursor_fanout(self):
297 await self.packets(("screen", [1, 1, 0, 0, 1, 1], bytes(4)))
298 self.viewer.send = lambda _: self.guest.screens.discard(self.viewer)
299 await self.packets(("cursor", [1, 1, 1, 0, 0, 2], (struct.pack("!I", 100) + bytes(5)) * 2))
300 self.assertFalse(self.guest.screens)
301 self.assertEqual(len(self.guest.cursor), 2)
302
303 async def test_capture_pauses_without_closing_guest_channel(self):
304 self.guest.writer = Writer()
305 self.guest.windows_available = False
306 self.guest.request()
307 self.assertEqual(bytes(self.guest.writer.data), struct.pack("!4sIII", b"SGV1", 12, 4, 1))
308 self.guest.writer.data.clear()
309 self.guest.screens.clear()
310 self.guest.request()
311 self.assertEqual(bytes(self.guest.writer.data), struct.pack("!4sIII", b"SGV1", 12, 4, 0))
312 self.assertFalse(self.guest.writer.closed)
313
314 async def test_missing_heartbeat_times_out_but_does_not_encode(self):
315 await self.packets(("screen", [1, 1, 0, 0, 1, 1], bytes(4)), ("heartbeat", [], b""))
316 self.assertEqual(self.presented, 1)
317 reader = asyncio.StreamReader()
318 with self.assertRaises(TimeoutError):
319 await self.guest.read(reader)
320
321 async def test_capabilities_and_cursor_resend_on_connected_viewer(self):
322 self.guest.writer = Writer()
323 self.viewer.experiment = True
324 await self.packets(("capabilities", [1], b""),
325 ("screen", [1, 1, 0, 0, 1, 1], bytes(4)),
326 ("cursor", [3, 1, 1, 0, 0, 1], struct.pack("!I", 100) + bytes(5)),
327 ("pointer", [3, 4, 5, 1], b""))
328 self.messages.clear()
329 self.guest.sync(self.viewer)
330 self.assertEqual([m["type"] for m in self.messages], ["capabilities", "cursor-position", "cursor"])
331 self.assertEqual(self.guest.writer.data[-4:], struct.pack("!I", 1))
332
333 async def test_window_layout_waits_for_full_surfaces_and_falls_back(self):
334 self.viewer.experiment = True
335 records = [1, 0, 0, 2, 2, 0, 0, 2, 1, 2, 2, 0, 2, 2, 2, 0, 4, 1]
336 await self.packets(("screen", [4, 2, 0, 0, 4, 2], bytes(32)),
337 ("windows", [4, 2, 2], struct.pack("!18I", *records)),
338 ("surface", [1, 2, 2, 0, 0, 2, 2], b"a" * 16))
339 self.assertFalse(self.guest.layers.complete)
340 self.presented = 0
341 await self.packets(("surface", [2, 2, 2, 0, 0, 2, 2], b"b" * 16))
342 self.assertTrue(self.guest.layers.complete)
343 self.assertEqual(self.presented, 1)
344 await self.packets(("windows", [4, 2, 0], b""))
345 self.assertIsNone(self.guest.layers)
346 self.assertEqual(self.messages[-1], {"type": "window-capture", "available": False})
347
348
349class StreamTests(unittest.IsolatedAsyncioTestCase):
350 async def asyncSetUp(self):
351 self.writer = Writer()
352 desktop = screen.Desktop(asyncio.StreamReader(), Writer(), lambda _: None)
353 desktop.resize(320, 200)
354 self.stream = screen.Screen(self.writer, desktop)
355
356 async def asyncTearDown(self):
357 pipeline = self.stream.pipeline.weak_ref()
358 encoder = self.stream.encoder.weak_ref()
359 self.stream.close()
360 self.assertIsNone(pipeline())
361 self.assertIsNone(encoder())
362 self.stream.present()
363 self.stream.resize()
364 self.stream.remote_description(None)
365 self.stream.answer(None)
366
367 async def test_browser_payload_and_sending_direction(self):
368 sdp = "\r\n".join([
369 "v=0", "o=- 1 1 IN IP4 127.0.0.1", "s=-", "t=0 0", "a=group:BUNDLE 0",
370 "m=video 9 UDP/TLS/RTP/SAVPF 96 103", "c=IN IP4 0.0.0.0", "a=mid:0", "a=recvonly",
371 "a=rtcp-mux", "a=ice-ufrag:abcd", "a=ice-pwd:abcdefghijklmnopqrstuvwxyz",
372 "a=setup:actpass", "a=fingerprint:sha-256 " + ":".join(["00"] * 32),
373 "a=rtpmap:96 VP8/90000", "a=rtpmap:103 H264/90000",
374 "a=fmtp:103 packetization-mode=1;profile-level-id=42e01f;level-asymmetry-allowed=1", "",
375 ])
376 self.stream.signal({"type": "offer", "sdp": sdp})
377 async with asyncio.timeout(5):
378 while b'"type":"answer"' not in self.writer.data:
379 await asyncio.sleep(0.01)
380 answer = next(json.loads(line) for line in self.writer.data.splitlines() if json.loads(line)["type"] == "answer")
381 self.assertIn("a=sendonly", answer["sdp"])
382 self.assertIn("a=rtpmap:103 H264/90000", answer["sdp"])
383 self.assertEqual(self.stream.pipeline.get_by_name("pay").get_property("pt"), 103)
384
385 async def test_twcc_and_retransmission_are_negotiated(self):
386 offer = "\r\n".join([
387 "v=0", "o=- 1 1 IN IP4 127.0.0.1", "s=-", "t=0 0", "a=group:BUNDLE 0",
388 "m=video 9 UDP/TLS/RTP/SAVPF 103 104", "c=IN IP4 0.0.0.0", "a=mid:0", "a=recvonly",
389 "a=rtcp-mux", "a=ice-ufrag:abcd", "a=ice-pwd:abcdefghijklmnopqrstuvwxyz",
390 "a=setup:actpass", "a=fingerprint:sha-256 " + ":".join(["00"] * 32),
391 "a=rtpmap:103 H264/90000", "a=rtpmap:104 rtx/90000", "a=fmtp:104 apt=103",
392 "a=fmtp:103 packetization-mode=1;profile-level-id=42e01f;level-asymmetry-allowed=1",
393 "a=rtcp-fb:103 nack", "a=rtcp-fb:103 nack pli", "a=rtcp-fb:103 transport-cc",
394 "a=extmap:3 " + screen.TWCC, "",
395 ])
396 self.stream.signal({"type": "offer", "sdp": offer})
397 async with asyncio.timeout(5):
398 while b'"type":"answer"' not in self.writer.data:
399 await asyncio.sleep(0.01)
400 answer = next(json.loads(line)["sdp"] for line in self.writer.data.splitlines() if json.loads(line)["type"] == "answer")
401 self.assertIn("a=rtpmap:104 rtx/90000", answer)
402 self.assertIn("a=fmtp:104 apt=103", answer)
403 primary = self.stream.pipeline.get_by_name("pay").get_property("ssrc")
404 self.assertIn(f"a=ssrc-group:FID {primary} ", answer)
405 self.assertIn(f"a=ssrc:{primary} msid:", answer)
406 self.assertIn("a=rtcp-fb:103 transport-cc", answer)
407 self.assertIsNotNone(self.stream.bwe)
408 self.stream.bwe.set_property("estimated-bitrate", 1000000)
409 self.assertEqual(self.stream.encoder.get_property("bitrate"), 1000)
410
411 async def test_unreliable_movement_cannot_restore_released_buttons(self):
412 def pointer(sequence, buttons, channel="input"):
413 self.stream.input(json.dumps(dict(type="pointer", x=1, y=1, sequence=sequence, buttons=buttons)), channel)
414 pointer(1, 1)
415 pointer(3, 0, "pointer")
416 self.assertEqual(self.stream.desktop.buttons, 1)
417 pointer(2, 0)
418 self.assertEqual(self.stream.desktop.buttons, 0)
419 before = bytes(self.stream.desktop.writer.data)
420 pointer(2, 1, "pointer")
421 self.assertEqual(self.stream.desktop.writer.data, before)
422 self.stream.input(json.dumps(dict(type="release", sequence=4)))
423 pointer(3, 1, "pointer")
424 self.assertEqual(self.stream.desktop.buttons, 0)
425
426 async def test_stalled_guest_closes_and_ignores_queued_input(self):
427 message = json.dumps({"type": "clipboard", "text": "a" * 65536})
428 for _ in range(64):
429 self.stream.input(message)
430 self.assertTrue(self.writer.closed)
431 self.assertLess(len(self.stream.desktop.writer.data), 327680)
432
433 async def test_rtp_packets_fit_tunneled_network(self):
434 await self.test_twcc_and_retransmission_are_negotiated()
435 Gst = screen.Gst
436 pipeline = self.stream.pipeline
437 pipeline.set_state(Gst.State.NULL)
438 pipeline.remove(self.stream.rtc)
439 sink = Gst.ElementFactory.make("appsink")
440 sink.set_property("sync", False)
441 pipeline.add(sink)
442 pipeline.get_by_name("codec").link(sink)
443 pipeline.set_state(Gst.State.PLAYING)
444 pixels = random.Random(0).randbytes(len(self.stream.desktop.pixels))
445 self.stream.frames.emit("push-buffer", Gst.Buffer.new_wrapped(pixels))
446 packets = []
447 async with asyncio.timeout(5):
448 while True:
449 sample = sink.emit("try-pull-sample", 0)
450 if sample:
451 buffer = sample.get_buffer()
452 packets.append(buffer.get_size())
453 if buffer.extract_dup(1, 1)[0] & 0x80:
454 break
455 else:
456 await asyncio.sleep(0.01)
457 self.assertGreater(len(packets), 1)
458 self.assertLessEqual(max(packets), 1200)
459
460
461if __name__ == "__main__":
462 screen.Gst.init(None)
463 threading.Thread(target=screen.GLib.MainLoop().run, daemon=True).start()
464 unittest.main()