【原创】Invoke用例分析

Nextgen的测试自动化系统(以下称为NextgenTA)使用Invoke来作为命令行工具。Invoke 是一个 Python 库，用于构建和执行任务。它提供了一种简单且灵活的方式来定义、组织和执行命令行任务，是 Fabric 的现代替代品。由于Fabric的设计比较老旧且依赖 SSH，这导致了很多使用场景受限，仅适用于远程服务器管理。为了提供一个更加通用和现代的任务执行工具，Invoke 便应运而生了。

项目：

https://github.com/pyinvoke/invoke

以下将以NextgenTA的常用任务为例，解析NextgenTA对Invoke的使用。

功能

1. 任务定义：

   以NexgenTA常用的一些任务为例。

   • 使用 Python 函数定义任务，通过装饰器将这些函数标记为任务。
   • 支持为任务添加描述和参数，参数可以有默认值，也可以是必需的。

   以下载模拟器任务为例：

   @task(default=True, aliases=["dl", "download"])
   def download_interactive(c, include_inactive=False, skip_tools=False):
       """Interactive download of nextgen SW simulators (actively being developed), middleware and tools
   
       Answer questions "Y/N" which simulator variants to wipeoff, download and unzip in proper place.
       "No" means don't touch the current simulator.
   
       Simulators which are not actively being developed has "active: False" directive in the Invoke.yaml file.
       If for some reason, you want such simulator to be dowloaded, you can use option: --include-inactive
   
       Default artifacts are the latest _green_ from "default" branch (invoke.yaml contains the defaults).
   
       If you need to download an specific build and/or branch of an artifact, create custom `invoke.yaml`
       file with similar data and use `-f invoke-custom.yaml` to override default download parameters.
       """
       hwcompat_excludes = set()
   
       from rich.console import Console
       from rich.panel import Panel
       from rich.prompt import Confirm
   
       console = Console()
       text = f"[yellow]Destination directory: [b]{DEST_DIR}[/b]"
       console.print(Panel(text))
   
       if not include_inactive:
           inactive_hwcompats = list(iter_inactive_hwcompats(c))
           for variant in c.HWCOMPATS.keys():
               if variant in inactive_hwcompats:
                   hwcompat_excludes.add(variant)
   
       if not Confirm.ask("Download all HW variants?", default=True):
           for variant in c.HWCOMPATS.keys():
               if variant not in hwcompat_excludes and not Confirm.ask(
                   f"Download HW variant {variant}?", default=True
               ):
                   hwcompat_excludes.add(variant)
   
       if not Confirm.ask("Continue?", default=True):
           return
   
       mao = wbtool = suuntoplus = sds = not skip_tools
       dl_list = list(
           _iter_teamcity_downloads(
               c,
               hwcompat_excludes,
               simu=True,
               sds=sds,
               mao=mao,
               wbtool=wbtool,
               suuntoplus=suuntoplus,
           )
       )
   
       loop = asyncio.get_event_loop()
       try:
           console.rule("[bold red] Empty directories")
           with console.status("[bold green]Cleaning directories..."):
               loop.run_until_complete(empty_dir_task(dl_list))
           console.rule("[bold red] Download builds")
           loop.run_until_complete(download_task(dl_list))
           console.rule("[bold red] Unzip archives")
           with console.status("[bold green]Unzipping..."):
               loop.run_until_complete(unzip_task(dl_list))
       finally:
           loop.close()
       console.print(":thumbs_up: [green]Done")
   
   

   在该任务中：

   @task 装饰器是任务标志，其参数中 aliases=["dl", "download"]设置任务别名，所以在执行命令时，以下三种写法都正确:

   pipenv run invoke builds

   pipenv run invoke builds.dl

   pipenv run invoke builds.download

   函数 def download_interactive(c, include_inactive=False, skip_tools=False): 是任务的具体定义，从上到下依次为命令行美化，命令行交互，以及异步下载部分。

2. 任务组织：

   • 可以将任务组织在模块中，并通过命名空间进行管理。
   • 支持分层命名空间，便于大型项目的任务组织。

   from invoke import Collection
   
   ns = Collection()
   ns.add_collection(builds)
   

3. 命令行接口：

   • 提供一个强大的命令行接口来执行任务。
   • 支持自动生成帮助信息，方便用户了解可用的任务和参数。

   $ pipenv run inv builds --help
   Loading .env environment variables...
   Usage: inv[oke] [--core-opts] builds.download-interactive [--options] [other tasks here ...]
   
   Docstring:
     Interactive download of nextgen SW simulators (actively being developed), middleware and tools
   
     Answer questions "Y/N" which simulator variants to wipeoff, download and unzip in proper place.
     "No" means don't touch the current simulator.
   
     Simulators which are not actively being developed has "active: False" directive in the Invoke.yaml file.
     If for some reason, you want such simulator to be dowloaded, you can use option: --include-inactive
   
     Default artifacts are the latest _green_ from "default" branch (invoke.yaml contains the defaults).
   
     If you need to download an specific build and/or branch of an artifact, create custom `invoke.yaml`
     file with similar data and use `-f invoke-custom.yaml` to override default download parameters.
   
   Options:
     -i, --include-inactive
     -s, --skip-tools
   

4. 参数处理：

   • 支持位置参数和关键字参数，参数可以是字符串、整数、布尔值等。
   • 支持命令行参数解析和类型转换。

   @task(default=True, aliases=["dl", "download"])
   def download_interactive(c, include_inactive=False, skip_tools=False):
       """Interactive download of nextgen SW simulators (actively being developed), middleware and tools
       ...
   

5. 任务依赖：

   • 支持任务之间的依赖关系，一个任务可以其他任务，确保任务按顺序执行。（以停止sds为例）

   **@task(post=[status])**
   def stop(c):
       """Terminate SDS Application Server"""
       stop_server()
       time.sleep(1)
   

6. 运行命令：

   • 提供便捷的方法在任务中运行系统命令，通过 run 方法执行 shell 命令，并获取输出。（以robot.clean任务为例）

   @task
   def clean(c):
       """Wipe out old Robot results from filesystem"""
       path = "results"
       if os.path.isdir(path):
           shutil.rmtree(path)
       os.mkdir(path)
   
       if os.path.isdir(DEBUGLOGS_DIR):
           logging.shutdown()
           shutil.rmtree(DEBUGLOGS_DIR)
       os.mkdir(DEBUGLOGS_DIR)
       **c.run(f"git restore {DEBUGLOGS_DIR}")**
   
       Console().print(
           f":thumbs_up: [green]Robot [i]outputdir[/i] ([b]{path}[/b]) is empty"
       )
   
   

7. 配置管理：

   • 支持全局配置、本地配置和环境变量配置。
   • 配置可以通过 .invoke.yaml 文件进行管理，方便定制和扩展: 以下是Por模拟器下载在invoke.ymal的配置内容

       ...
       
       M:
         project: Products_Porvoo_VisualStudio2015Application
         test_builds:
           high: Testing_AcceptanceTests_DockerizedTests_ShortTests_AcceptanceTestsPorvooHighImpactDocker
           low: Testing_AcceptanceTests_DockerizedTests_Long_AcceptanceTestsPorvooLowImpactDocker
         branch: develop
         build: ".lastSuccessful"
         target_dir: "{basedir}/simulator/hwcompat_M"
         filename: simulator.zip
        
        ...
   

8. 并行执行：

   • 支持任务并行执行，可以通过 Threading 或 Multiprocessing 实现并行任务处理，提高效率。