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

这篇具有很好参考价值的文章主要介绍了重写 AppiumService 类，添加默认启动参数，并实时显示启动日志。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

一、前置说明

在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()