重构CLI应用程序为cappa (#59)

* 重构CLI应用程序为cappa

* typing lint

* 修复xlrd依赖

* 升级cappa为0.10.2

* 更新CLI参数说明

httpfpt/db/redis_db.py

@@ -80,7 +80,7 @@ def delete_prefix(self, prefix: str) -> None:
         for key in self.scan_iter(match=f'{prefix}*'):
             self.delete(key)
 
-    def exists(self, *key: Any) -> int:
+    def exists(self, *key: Any) -> Any:
         """
         判断 redis 数据是否存在
 

httpfpt/utils/case_auto_generator.py

@@ -3,11 +3,10 @@
 import os.path
 from pathlib import Path
 
-import typer
-
 from httpfpt.core.get_conf import PROJECT_NAME
 from httpfpt.core.path_conf import TEST_CASE_PATH
 from httpfpt.utils.file_control import search_all_case_yaml_files, search_all_testcase_files, get_file_property
+from httpfpt.utils.rich_console import console
 
 
 def auto_generate_testcases(rewrite: bool = False) -> None:
@@ -48,10 +47,10 @@ def auto_generate_testcases(rewrite: bool = False) -> None:
         else:
             create_file_root_names.append(root_name)
     if len(create_file_root_names) == 0:
-        typer.secho('😝 用例已经很完善了, 添加新测试用例数据后再来生成吧~', fg='green', bold=True)
+        console.print('😝 用例已经很完善了, 添加新测试用例数据后再来生成吧~')
         return
 
-    typer.secho('⏳ 疯狂自动生成中...', fg='green', bold=True)
+    console.print('⏳ 疯狂自动生成中...')
 
     for create_file_root_name in create_file_root_names:
         for yaml_filename in yaml_filenames:
@@ -96,6 +95,6 @@ def {testcase_func_name}(self, data):
                     Path(case_path).parent.mkdir(parents=True, exist_ok=True)
                 with open(case_path, 'w', encoding='utf-8') as f:
                     f.write(case_code)
-                typer.secho(f'📄 Created: {get_file_property(case_path)[0]}', fg='green', bold=True)
+                console.print(f'📄 Created: {get_file_property(case_path)[0]}')
 
-    typer.secho('✅ 测试用例自动生成完成', fg='green', bold=True)
+    console.print('✅ 测试用例自动生成完成')

httpfpt/utils/data_manage/apifox.py

@@ -4,14 +4,13 @@
 import os
 from typing import Optional, Union
 
-import typer
-
 from httpfpt.common.json_handler import read_json_file
 from httpfpt.common.yaml_handler import write_yaml
 from httpfpt.core.get_conf import PROJECT_NAME
 from httpfpt.core.path_conf import YAML_DATA_PATH
 from httpfpt.utils.data_manage.base_format import format_value
 from httpfpt.utils.file_control import get_file_property
+from httpfpt.utils.rich_console import console
 
 
 class ApiFoxParser:
@@ -56,7 +55,7 @@ def import_apifox_to_yaml(self, source: str, project: Optional[str] = None) -> N
                 else:
                     pass  # todo 二叉树数据解析
             if (len(tag_case) or len(root_case)) > 0:
-                typer.secho('⏳ 奋力导入中...', fg='green', bold=True)
+                console.print('⏳ 奋力导入中...')
             # 写入项目 tag 目录
             if len(tag_case) > 0:
                 for k, v in tag_case.items():
@@ -78,10 +77,9 @@ def import_apifox_to_yaml(self, source: str, project: Optional[str] = None) -> N
                         case_file_data,
                         mode='w',
                     )
+            console.print('✅ 导入 apifox 数据成功')
         except Exception as e:
             raise e
-        else:
-            typer.secho('✅ 导入 apifox 数据成功')
 
     @staticmethod
     def get_apifox_params(value: dict) -> Union[dict, None]:

httpfpt/utils/data_manage/git_repo.py

@@ -3,13 +3,13 @@
 import os
 import shutil
 
-import typer
 from pydantic import ValidationError
 
 from httpfpt.common.yaml_handler import read_yaml
 from httpfpt.schemas.case_data import CaseData
 from httpfpt.utils.file_control import search_all_case_yaml_files
 from httpfpt.utils.pydantic_parser import parse_error
+from httpfpt.utils.rich_console import console
 
 
 class GitRepoPaser:
@@ -23,33 +23,35 @@ def import_git_to_local(src: str) -> None:
         """
 
         if 'https' not in src:
-            raise ValueError('❌ git 仓库克隆地址错误, 请使用 https 地址')
+            raise ValueError('❌ Git 仓库克隆地址错误, 请使用 https 地址')
 
         try:
             online_dir_path = os.path.join(os.path.abspath('./data/test_data'), 'online_test_data')
-            typer.echo(online_dir_path)
+            console.print(online_dir_path)
             if os.path.exists(online_dir_path):
                 shutil.rmtree(online_dir_path)
             os.makedirs(online_dir_path)
             result = os.system(f'cd {online_dir_path} && git clone {src}')
         except Exception as e:
-            raise RuntimeError(f'❌ git 仓库测试数据拉取失败：{e}')
+            raise RuntimeError(f'❌ Git 仓库测试数据拉取失败：{e}')
+        if result == 0:
+            console.print('\n✅ Git 仓库数据文件拉取成功')
         else:
-            if result == 0:
-                typer.secho('✅ git 仓库数据文件拉取成功')
-            else:
-                raise RuntimeError('❌ git 仓库测试数据拉取失败')
+            raise RuntimeError('❌ Git 仓库测试数据拉取失败，请检查 Git 地址是否正确')
 
+        console.print('\n🔥 开始自动验证测试数据结构...')
         all_yaml_file = search_all_case_yaml_files(online_dir_path)
+        if len(all_yaml_file) == 0:
+            raise FileNotFoundError('❌ 未在拉取的 Git 仓库中找到测试用例数据文件，请检查 Git 地址是否正确')
         all_case_data = []
         for file in all_yaml_file:
             read_data = read_yaml(None, filename=file)
             all_case_data.append(read_data)
+        count: int = 0
         for case_data in all_case_data:
             try:
-                count: int = 0
                 CaseData.model_validate(case_data)
             except ValidationError as e:
-                count = parse_error(e)
-            if count > 0:
-                raise ValueError(f'导入 Git 用例数据校验失败，共有 {count} 处错误, 错误详情请查看日志')
+                count += parse_error(e)
+        if count > 0:
+            raise ValueError(f'❌ Git 仓库用例数据校验失败，共有 {count} 处错误, 错误详情请查看日志')

httpfpt/utils/data_manage/openapi.py

@@ -6,14 +6,15 @@
 from typing import Union, Optional
 
 import requests
-import typer
+from rich.prompt import Confirm
 
 from httpfpt.common.json_handler import read_json_file
 from httpfpt.common.yaml_handler import write_yaml
 from httpfpt.core.get_conf import PROJECT_NAME
 from httpfpt.core.path_conf import YAML_DATA_PATH
 from httpfpt.utils.data_manage.base_format import format_value
 from httpfpt.utils.file_control import get_file_property
+from httpfpt.utils.rich_console import console
 from httpfpt.utils.time_control import get_current_timestamp
 
 
@@ -62,7 +63,7 @@ def import_openapi_to_yaml(self, openapi_source: str, project: Optional[str] = N
             # 根目录测试用例数据
             root_case = {}
 
-            is_tag = typer.confirm('是否按openapi标签划分数据存放目录?', default=True)
+            is_tag = Confirm.ask('❓ 是否按 openapi 标签划分数据存放目录?', default=True)
             for url, values in self.data['paths'].items():
                 for method, values_map in values.items():
                     params = self.get_swagger_params(values_map)
@@ -164,15 +165,15 @@ def import_openapi_to_yaml(self, openapi_source: str, project: Optional[str] = N
                                 ]
                             )
                         file_list.append(root_filename)
-                typer.secho('⚠️ 即将创建以下数据文件:', fg='yellow', bold=True)
+                console.print('⚠️ 即将创建以下数据文件:')
                 for i in file_list:
-                    typer.secho(f'{i}', fg='cyan', bold=True)
-                is_force_write = typer.confirm(
-                    text='请检查是否存在同名文件, 此操作将强制覆盖写入所有数据文件, 是否继续执行? (此操作不可逆)', default=False  # noqa: E501
+                    console.print(f'\n\tdata\\test_data\\{i}')
+                is_force_write = Confirm.ask(
+                    '\n👁️ 请检查是否存在同名文件, 此操作将强制覆盖写入所有数据文件, 是否继续执行? (此操作不可逆)', default=False  # noqa: E501
                 )
                 # 强制写入
                 if is_force_write:
-                    typer.secho('⏳ 奋力导入中...', fg='green', bold=True)
+                    console.print('⏳ 奋力导入中...')
                     # 写入项目 tag 目录
                     if len(tag_case) > 0:
                         for k, v in tag_case.items():
@@ -216,8 +217,8 @@ def import_openapi_to_yaml(self, openapi_source: str, project: Optional[str] = N
                             )
                 # 选择写入
                 else:
-                    typer.secho('⚠️ 已取消强制覆写入所有数据文件', fg='yellow', bold=True)
-                    is_next = typer.confirm('是否进行逐一选择创建数据文件吗?', abort=True)
+                    console.print('⚠️ 已取消强制覆写入所有数据文件')
+                    is_next = Confirm.ask('❓ 是否进行逐一选择创建数据文件?', default=True)
                     if is_next:
                         # 写入项目 tag 目录
                         if len(tag_case) > 0:
@@ -233,7 +234,7 @@ def import_openapi_to_yaml(self, openapi_source: str, project: Optional[str] = N
                                         else f'openapi_{k}.yaml',
                                     ]
                                 )
-                                is_write = typer.confirm(text=f'是否需要创建 {tag_filename} 数据文件?', default=True)
+                                is_write = Confirm.ask(f'❓ 是否需要创建 {tag_filename} 数据文件?', default=True)
                                 if is_write:
                                     write_yaml(YAML_DATA_PATH, tag_filename, case_file_data, mode='w')
                         # 写入项目根目录
@@ -251,7 +252,7 @@ def import_openapi_to_yaml(self, openapi_source: str, project: Optional[str] = N
                                         if not openapi_source.startswith('http')
                                         else f'openapi_{k}.yaml'
                                     )
-                                is_write = typer.confirm(text=f'是否需要创建 {root_filename} 数据文件?', default=True)
+                                is_write = Confirm.ask(f'❓ 是否需要创建 {root_filename} 数据文件?', default=True)
                                 if is_write:
                                     case_file_data = {'config': config, 'test_steps': v}
                                     write_yaml(
@@ -260,10 +261,9 @@ def import_openapi_to_yaml(self, openapi_source: str, project: Optional[str] = N
                                         case_file_data,
                                         mode='w',
                                     )
+            console.print('✅ 导入 openapi 数据成功')
         except Exception as e:
             raise e
-        else:
-            typer.secho('✅ 导入 openapi 数据成功', fg='green', bold=True)
 
     def get_swagger_data(self, openapi_source: str) -> None:
         """
@@ -361,10 +361,10 @@ def get_swagger_request_data(self, value: Union[dict, str]) -> Union[dict, None]
             if len(value['parameters']) > 0:
                 for i in value['parameters']:
                     if i.get('type') is None:
-                        data[i['name']] = format_value('object')
+                        data[i['name']] = format_value('object')  # type: ignore
                     else:
                         if i.get('type') != 'file':
-                            data[i['name']] = format_value(i.get('type', 'object'))
+                            data[i['name']] = format_value(i.get('type', 'object'))  # type: ignore
             return data if len(data) > 0 else None
         else:
             if not isinstance(value, dict):

httpfpt/utils/rich_console.py

@@ -0,0 +1,5 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+from rich import get_console
+
+console = get_console()

requirements.txt

@@ -1,30 +1,27 @@
 allure-pytest==2.13.2
 cache3==0.2.1
+cappa==0.10.2
 cryptography==41.0.4
 dbutils==3.0.2
 dirty_equals==0.6.0
 faker==13.15.0
 httpx==0.23.0
 jinja2==3.1.2
 loguru==0.6.0
-openpyxl==3.0.7
 pre-commit==3.2.2
 pydantic==2.3.0
 pymysql==0.9.3
-pyright==1.1.315
+pyright==1.1.332
 pytest==7.4.1
 pytest-html==4.0.0
 pytest-metadata==3.0.0
 pytest-pretty==1.2.0
 pytest-xdist==2.5.0
-python-dateutil==2.8.2
 python-dotenv==0.20.0
 python-jsonpath==0.9.0
 pyyaml==6.0.1
-redis==4.5.4
+redis[hiredis]==5.0.1
 requests==2.31.0
+rich==13.6.0
 rtoml==0.9.0
-ruff==0.0.262
-text-unidecode==1.3
-typer[all]==0.9.0
 xlrd==1.2.0