119 行
5.2 KiB
Python
119 行
5.2 KiB
Python
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()
|