xctest register_callback _XCT_didFinishExecutingTestPlan

demo/installation_proxy.py

@@ -26,7 +26,7 @@
 import logging
 import sys
 
-from .afc import AFCClient
+from ios_device.servers.afc import AFCClient
 sys.path.append(os.getcwd())
 
 from ios_device.util.lockdown import LockdownClient
@@ -65,8 +65,9 @@ def watch_completion(self, handler=None, *args):
                     handler(completion, *args)
                 self.logger.info("%s %% Complete", z.get("PercentComplete"))
             if z.get("Status") == "Complete":
+                self.logger.info("Success")
                 return z.get("Status")
-        return "Error"
+        return Exception("Install Error")
 
     def send_cmd_for_bid(self, bid, cmd="Archive", options=None, handler=None, *args):
         cmd = {"Command": cmd,
@@ -179,4 +180,4 @@ def close(self):
 
 
 if __name__ == '__main__':
-    installation_proxy().get_apps()
+    installation_proxy().install('/Users/chenpeijie/Library/Developer/Xcode/DerivedData/WebDriverAgent-aiaejbhejflmruhkhsmeebmhbaia/Build/Products/Debug-iphoneos/Payload.ipa')

demo/instrument_demo/xcuitest.py

@@ -2,7 +2,7 @@
 import threading
 from distutils.version import LooseVersion
 
-from ios_device.servers.DTXSever import DTXServerRPCRawObj
+from ios_device.servers.DTXSever import DTXServerRPCRawObj, DTXEnum
 from ios_device.servers.InstallationProxy import InstallationProxy
 from ios_device.servers.Instrument import InstrumentServer
 from ios_device.servers.house_arrest import HouseArrestClient
@@ -41,12 +41,12 @@ def _callback(res):
                 "dtxproxy:XCTestManager_IDEInterface:XCTestManager_DaemonConnectionInterface",
                 "_IDE_initiateControlSessionWithProtocolVersion:", DTXServerRPCRawObj(XCODE_VERSION)).parsed
             logging.info("result: %s", result)
-        ManagerdLockdown1.register_callback(":finished:", lambda _: quit_event.set())
+        ManagerdLockdown1.register_callback("DTXEnum.FINISHED", lambda _: quit_event.set())
         ManagerdLockdown1.register_unhandled_callback(_callback)
 
         ManagerdLockdown2 = TestManagerdLockdown(self.lockdown).init()
         ManagerdLockdown2._make_channel("dtxproxy:XCTestManager_IDEInterface:XCTestManager_DaemonConnectionInterface")
-        ManagerdLockdown2.register_callback(":finished:", lambda _: quit_event.set())
+        ManagerdLockdown2.register_callback("DTXEnum.FINISHED", lambda _: quit_event.set())
         ManagerdLockdown2.register_unhandled_callback(_callback)
 
         _start_flag = threading.Event()
@@ -70,6 +70,7 @@ def _show_log_message(res):
 
         ManagerdLockdown2.register_callback('_XCT_testBundleReadyWithProtocolVersion:minimumVersion:', _start_executing)
         ManagerdLockdown2.register_callback('_XCT_logDebugMessage:', _show_log_message)
+        ManagerdLockdown2.register_callback('_XCT_didFinishExecutingTestPlan', lambda _: quit_event.set())
 
         result = ManagerdLockdown2.call('dtxproxy:XCTestManager_IDEInterface:XCTestManager_DaemonConnectionInterface',
                                         '_IDE_initiateSessionWithIdentifier:forClient:atPath:protocolVersion:',
@@ -145,7 +146,7 @@ def _show_log_message(res):
         conn.call('com.apple.instruments.server.services.processcontrol', "startObservingPid:", DTXServerRPCRawObj(pid))
 
         if quit_event:
-            conn.register_callback(':finished:', lambda _: quit_event.set())
+            conn.register_callback(DTXEnum.FINISHED, lambda _: quit_event.set())
 
         if self.lockdown.ios_version > LooseVersion('12.0'):
             identifier = '_IDE_authorizeTestSessionWithProcessID:'
@@ -165,6 +166,9 @@ def _show_log_message(res):
         while not quit_event.wait(.1):
             pass
         logging.warning("xctrunner quited")
+        conn.deinit()
+        ManagerdLockdown2.deinit()
+        ManagerdLockdown1.deinit()
 
 
 if __name__ == '__main__':

ios_device/servers/DTXSever.py

@@ -1,3 +1,4 @@
+import enum
 import struct
 import time
 import traceback
@@ -15,6 +16,12 @@
 log = logging.getLogger(__name__)
 
 
+class DTXEnum(str, enum.Enum):
+    NOTIFICATION = "notification:"
+    FINISHED = "finished:"
+    OTHER = "other:"
+
+
 class DTXFragment:
 
     def __init__(self, buf):
@@ -253,6 +260,16 @@ def stop(self):
         if self._recv_thread:
             self._recv_thread = None
 
+    def _run_callbacks(self, event_name, data):
+        """
+        Returns:
+            if called
+        """
+        func = self._callbacks.get(event_name)
+        if func:
+            func(data)
+            return True
+
     def register_callback(self, selector, callback: typing.Callable):
         """
         注册回调, 接受 servers server 到 client 的远程调用
@@ -339,46 +356,52 @@ def _call(self, sync: bool, channel_id: int, selector: str, *auxiliaries):
 
     def _receiver(self):
         last_none = 0
-        while self._running:
-            dtx = self._is.recv_dtx(self._cli, 2)  # s
-            if dtx is None:  # 长时间没有回调则抛出错误
-                cur = time.time()
-                if cur - last_none < 0.1:
-                    break
-                last_none = cur
-                continue
-            self._next_identifier = max(self._next_identifier, dtx.identifier + 1)
-            wait_key = (dtx.channel_code, dtx.identifier)
-            if wait_key in self._sync_waits:
-                param = self._sync_waits[wait_key]
-                param['result'] = dtx
-                param['event'].set()
-            elif 2 ** 32 - dtx.channel_code in self._channel_callbacks:
-                try:
-                    self._channel_callbacks[2 ** 32 - dtx.channel_code](DTXServerRPCResult(dtx))
-                except:
-                    traceback.print_exc()
-            else:
-                try:
-                    selector = selector_to_pyobject(dtx.get_selector())
-                except:
-                    selector = None
-                try:
-                    if selector and isinstance(selector, str) and selector in self._callbacks:
-                        self._callbacks[selector](DTXServerRPCResult(dtx))
-                    elif self._unhanled_callback:
-                        self._unhanled_callback(DTXServerRPCResult(dtx))
-                except:
-                    traceback.print_exc()
-                if dtx.expects_reply:
-                    reply = dtx.new_reply()
-                    reply.set_selector(b'\00' * 16)
-                    reply._payload_header.flags = 0x3
-                    self._is.send_dtx(self._cli, reply)
-        self._receiver_exiting = True  # to block incoming calls
-        for wait_key in self._sync_waits:
-            self._sync_waits[wait_key]['result'] = InstrumentServiceConnectionLost
-            self._sync_waits[wait_key]['event'].set()
+        try:
+            while self._running:
+                dtx = self._is.recv_dtx(self._cli, 2)  # s
+                if dtx is None:  # 长时间没有回调则抛出错误
+                    cur = time.time()
+                    if cur - last_none < 0.1:
+                        raise Exception('dtx socket close')
+                    last_none = cur
+                    continue
+                self._next_identifier = max(self._next_identifier, dtx.identifier + 1)
+                wait_key = (dtx.channel_code, dtx.identifier)
+                if wait_key in self._sync_waits:
+                    param = self._sync_waits[wait_key]
+                    param['result'] = dtx
+                    param['event'].set()
+                elif 2 ** 32 - dtx.channel_code in self._channel_callbacks:
+                    try:
+                        self._channel_callbacks[2 ** 32 - dtx.channel_code](DTXServerRPCResult(dtx))
+                    except:
+                        traceback.print_exc()
+                else:
+                    try:
+                        selector = selector_to_pyobject(dtx.get_selector())
+                    except:
+                        selector = None
+                    try:
+                        if selector and isinstance(selector, str) and selector in self._callbacks:
+                            self._callbacks[selector](DTXServerRPCResult(dtx))
+                        elif self._unhanled_callback:
+                            self._unhanled_callback(DTXServerRPCResult(dtx))
+                    except:
+                        traceback.print_exc()
+                    if dtx.expects_reply:
+                        reply = dtx.new_reply()
+                        reply.set_selector(b'\00' * 16)
+                        reply._payload_header.flags = 0x3
+                        self._is.send_dtx(self._cli, reply)
+            self._receiver_exiting = True  # to block incoming calls
+            for wait_key in self._sync_waits:
+                self._sync_waits[wait_key]['result'] = InstrumentServiceConnectionLost
+                self._sync_waits[wait_key]['event'].set()
+        except Exception as E:
+            log.error(E)
+            self._run_callbacks(DTXEnum.NOTIFICATION, None)
+            self._run_callbacks(DTXEnum.FINISHED, None)
+
 
 
 def pre_call(rpc):

ios_device/util/plist_service.py

@@ -72,6 +72,7 @@ def recv_plist(self) -> Optional[Dict[str, Any]]:
             return None
         if payload.startswith(b'bplist00'):
             data = plistlib.loads(payload)
+            log.debug(f'接收 Plist data: {data}')
             return data
         elif payload.startswith(b'<?xml'):
             payload = HARDWARE_PLATFORM_SUB('', payload.decode('utf-8')).encode('utf-8')