重写 AppiumService 类，添加默认启动参数，并实时显示启动日志-编程知识

重写 AppiumService 类，添加默认启动参数，并实时显示启动日志

一、前置说明

在Appium的1.6.0版本中引入了AppiumService类，可以很方便的通过该类来管理Appium服务器的启动和停止。经过测试，使用该类的实例执行关闭server时，并没有释放端口号，会导致第二次启动时失败。另外，使用该类启动server，不能在窗口中实时显示日志，不方便调试。因此，可以重写 AppiumService 类的 start 方法和 stop 方法，实现相应功能。

二、代码实现

import loggingfrom appium.webdriver.appium_service import *
from appium.webdriver.appium_service import AppiumService as OriginalServerfrom libs import path, file_util
from libs.cmd_util import cmd_executor as cmdlogger = logging.getLogger(__name__)class AppiumService(OriginalServer):def __init__(self, port=4723, log_file_path=None, tail_log=False):super().__init__()self.port = portself.log_file_path = log_file_path# 用于控制是否开启cmd窗口，实时展示日志内容，当调试代码时建议开启self.tail_log = tail_logif 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 Nonenode: 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/534default_std = sp.DEVNULL if sys.platform == 'win32' else sp.PIPEstdout = kwargs['stdout'] if 'stdout' in kwargs else default_stdstderr = kwargs['stderr'] if 'stderr' in kwargs else default_stdtimeout_ms = int(kwargs['timeout_ms']) if 'timeout_ms' in kwargs else STARTUP_TIMEOUT_MSargs: List[str] = [node, main_script]if 'args' in kwargs:args.extend(kwargs['args'])# ==================添加这段逻辑（开始）：设置默认启动参数=======================if self.port:args.extend(self.default_start_args)# ==================添加这段逻辑（结束）：设置默认启动参数=======================self._cmd = argsself._process = sp.Popen(args=args, stdout=stdout, stderr=stderr, env=env)error_msg: Optional[str] = Nonestartup_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_msgelif not self.is_running:error_msg = startup_failure_msgif 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._processdef stop(self) -> bool:status = super().stop()# 经过实测，原stop()方法执行之后，进程仍被占用，所以这里添加一个关闭进程的逻辑cmd.kill_process_by_port(self.port)return statusif __name__ == '__main__':logging.basicConfig(level=logging.INFO)service = AppiumService(port=4723, tail_log=True)service.start()