一、前置说明
在Appium的1.6.0版本中引入了AppiumService类，可以很方便的通过该类来管理Appium服务器的启动和停止。经过测试，使用该类的实例执行关闭server时，并没有释放端口号，会导致第二次启动时失败。另外，使用该类启动server，不能在窗口中实时显示日志，不方便调试。因此，可以重写 AppiumService 类的 start 方法和 stop 方法，实现相应功能。
二、代码实现
import logging from appium.webdriver.appium_service import * from appium.webdriver.appium_service import AppiumService as OriginalServer from libs import path, file_util from libs.cmd_util import cmd_executor as cmd logger = logging.getLogger(__name__) class AppiumService(OriginalServer): def __init__(self, port=4723, log_file_path=None, tail_log=False): super().__init__() self.port = port self.log_file_path = log_file_path # 用于控制是否开启cmd窗口，实时展示日志内容，当调试代码时建议开启 self.tail_log = tail_log if not self.log_file_path: self.log_file_path = os.path.join(path.get_log_dir(), f'appium_server_{port}.log') if os.path.exists(self.log_file_path): os.remove(self.log_file_path) self.default_start_args = ['-p', str(self.port), '-g', self.log_file_path, '--session-override', '--log-timestamp', '--session-override', '--local-timezone', '--allow-insecure', 'chromedriver_autodownload' ] def start(self, **kwargs: Any) -> sp.Popen: self.stop() env = kwargs['env'] if 'env' in kwargs else None node: str = kwargs.get('node') or get_node() npm: str = kwargs.get('npm') or get_npm() main_script: str = kwargs.get('main_script') or get_main_script(node, npm) # A workaround for https://github.com/appium/python-client/issues/534 default_std = sp.DEVNULL if sys.platform == 'win32' else sp.PIPE stdout = kwargs['stdout'] if 'stdout' in kwargs else default_std stderr = kwargs['stderr'] if 'stderr' in kwargs else default_std timeout_ms = int(kwargs['timeout_ms']) if 'timeout_ms' in kwargs else STARTUP_TIMEOUT_MS args: List[str] = [node, main_script] if 'args' in kwargs: args.extend(kwargs['args']) # ==================添加这段逻辑（开始）：设置默认启动参数======================= if self.port: args.extend(self.default_start_args) # ==================添加这段逻辑（结束）：设置默认启动参数======================= self._cmd = args self._process = sp.Popen(args=args, stdout=stdout, stderr=stderr, env=env) error_msg: Optional[str] = None startup_failure_msg = ( 'Appium server process is unable to start. Make sure proper values have been ' f'provided to \'node\' ({node}), \'npm\' ({npm}) and \'main_script\' ({main_script}) ' f'method arguments.' ) if timeout_ms > 0: status_url_path = make_status_url(args) try: if not self._poll_status(parse_host(args), parse_port(args), status_url_path, timeout_ms): error_msg = ( f'Appium server has started but is not listening on {status_url_path} ' f'within {timeout_ms}ms timeout. Make sure proper values have been provided ' f'to --base-path, --address and --port process arguments.' ) except AppiumStartupError: error_msg = startup_failure_msg elif not self.is_running: error_msg = startup_failure_msg if error_msg is not None: if stderr == sp.PIPE and self._process.stderr is not None: err_output = self._process.stderr.read() if err_output: error_msg += f'\nOriginal error: {str(err_output)}' self.stop() raise AppiumServiceError(error_msg) # ==================添加这段逻辑（开始）：是否启动cmd窗口跟踪log日志======================= if self.tail_log: file_util.tail_log_file(self.log_file_path) # ==================添加这段逻辑（结束）：是否启动cmd窗口跟踪log日志======================= return self._process def stop(self) -> bool: status = super().stop() # 经过实测，原stop()方法执行之后，进程仍被占用，所以这里添加一个关闭进程的逻辑 cmd.kill_process_by_port(self.port) return status if __name__ == '__main__': logging.basicConfig(level=logging.INFO) service = AppiumService(port=4723, tail_log=True) service.start()
三、Demo验证
执行代码，顺利启动appium server，并启动了一个命令行窗口，实时展示了server的日志。
