feat(rtos): 首次发布手机真机调试桥与S2探针

phone-debug-bridge/tests/test_ui.py

@@ -0,0 +1,118 @@
+import json
+import base64
+import struct
+from pathlib import Path
+import sys
+import tempfile
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import urlsplit
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+from ui import Inspector, Phone, save_png, imported_watch_capture, jpeg_info, health_screenshot
+
+
+class UIHandler(BaseHTTPRequestHandler):
+    def log_message(self, *_): pass
+
+    def do_GET(self):
+        self.server.paths.append(self.path)
+        if self.path == '/snapshot.json':
+            value = {'frames': [{'snapshot': 'frame.png'}]}
+            body = json.dumps(value).encode()
+        else:
+            body = b'broken PNG'
+        self.send_response(200)
+        self.send_header('Content-Length', str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+    def do_POST(self):
+        self.server.paths.append(self.path)
+        body = b'{}'
+        self.send_response(200)
+        self.send_header('Content-Length', str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+class UiTests(unittest.TestCase):
+    @staticmethod
+    def jpeg_header(width=480, height=464):
+        # Minimal marker stream exercises metadata parsing, not a JPEG decoder.
+        return b'\xff\xd8\xff\xc0' + struct.pack('>HBHHB', 8, 8, height, width, 0) + b'\xff\xd9'
+
+    def test_watch_jpeg_dimensions_and_truncation(self):
+        data = self.jpeg_header()
+        self.assertEqual((jpeg_info(data)['width'], jpeg_info(data)['height']), (480, 464))
+        with self.assertRaises(ValueError): jpeg_info(data[:-1])
+
+    def test_official_numbered_jpeg_is_exported_with_real_extension(self):
+        data = self.jpeg_header()
+        class HealthPhone:
+            queries = 0
+            find = staticmethod(Phone.find)
+            def unlocked(self): pass
+            def click(self, selector): pass
+            def tree(self):
+                return '<hierarchy><node resource-id="com.heytap.health:id/device_settings_tv_screen_shot" enabled="true" bounds="[0,0][10,10]"/></hierarchy>'
+            def run(self, *args, binary=False):
+                if binary:
+                    self.path = args[-1]
+                    return data
+                if 'content query' in args[-1]:
+                    self.queries += 1
+                    return '' if self.queries == 1 else 'Row: 0 _id=9, _data=/storage/emulated/0/Pictures/com.heytap.health/screenshots/watch_screen_shot (2).jpeg, width=480, height=464, date_modified=100'
+                return ''
+        phone = HealthPhone()
+        with tempfile.TemporaryDirectory() as tmp:
+            result = health_screenshot(phone, Path(tmp) / 'watch.png')
+            self.assertEqual(Path(result['file']).suffix, '.jpg')
+            self.assertEqual(Path(result['file']).read_bytes(), data)
+            self.assertTrue(phone.path.endswith('watch_screen_shot (2).jpeg'))
+
+    def test_failed_inspector_image_always_resumes_runtime(self):
+        server = ThreadingHTTPServer(('127.0.0.1', 0), UIHandler)
+        server.paths = []
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            client = Inspector('http://127.0.0.1:' + str(server.server_port))
+            with tempfile.TemporaryDirectory() as tmp:
+                dest = Path(tmp) / 'capture.png'
+                with self.assertRaises(ValueError): client.screenshot(dest)
+                self.assertFalse(dest.exists())
+            self.assertEqual(server.paths[-1], '/control?pause=false')
+        finally:
+            server.shutdown(); server.server_close(); thread.join()
+
+    def test_phone_ambiguous_selector_is_rejected(self):
+        xml = '<hierarchy><node text="安装" bounds="[0,0][20,20]"/><node text="安装" bounds="[30,0][50,20]"/></hierarchy>'
+        with self.assertRaises(ValueError): Phone.find(xml, ('text', '安装'))
+
+    def test_phone_hidden_node_is_not_selected(self):
+        self.assertIsNone(Phone.find('<hierarchy><node text="隐藏" bounds="[0,0][0,0]"/></hierarchy>', ('text', '隐藏')))
+        self.assertIsNone(Phone.find('<hierarchy><node text="隐藏" visible-to-user="false" bounds="[0,0][20,20]"/></hierarchy>', ('text', '隐藏')))
+
+    def test_import_integrity_failure_does_not_write_capture(self):
+        class ImportedPhone:
+            def run(self, *args, binary=False):
+                return base64.b64decode('iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+aL1sAAAAASUVORK5CYII=') if binary else '{"sha256":"wrong"}'
+        with tempfile.TemporaryDirectory() as tmp:
+            dest = Path(tmp) / 'watch.png'
+            with self.assertRaises(ValueError): imported_watch_capture(ImportedPhone(), dest)
+            self.assertFalse(dest.exists())
+
+    def test_inspector_is_explicit_loopback(self):
+        for url in ('https://example.com', 'http://10.0.0.1', 'http://user:secret@localhost'):
+            with self.assertRaises(ValueError): Inspector(url)
+
+    def test_no_phone_image_fallback_on_capture_failure(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            dest = Path(tmp) / 'watch.png'
+            with self.assertRaises(ValueError): save_png(dest, b'<html>failure</html>')
+            self.assertFalse(dest.exists())
+
+
+if __name__ == '__main__': unittest.main()