同步

etl_billiards/config/defaults.py

@@ -1,5 +1,5 @@
 # -*- coding: utf-8 -*-
-"""配置默认值"""
+"""配置默认值定义"""
 
 DEFAULTS = {
     "app": {
@@ -21,7 +21,7 @@ DEFAULTS = {
             "timezone": "Asia/Taipei",
             "statement_timeout_ms": 30000,
             "lock_timeout_ms": 5000,
-            "idle_in_tx_timeout_ms": 600000
+            "idle_in_tx_timeout_ms": 600000,
         },
     },
     "api": {
@@ -37,9 +37,19 @@ DEFAULTS = {
     },
     "run": {
         "tasks": [
-            "PRODUCTS", "TABLES", "MEMBERS", "ASSISTANTS", "PACKAGES_DEF",
+            "PRODUCTS",
-            "ORDERS", "PAYMENTS", "REFUNDS", "COUPON_USAGE", "INVENTORY_CHANGE",
+            "TABLES",
-            "TOPUPS", "TABLE_DISCOUNT", "ASSISTANT_ABOLISH",
+            "MEMBERS",
+            "ASSISTANTS",
+            "PACKAGES_DEF",
+            "ORDERS",
+            "PAYMENTS",
+            "REFUNDS",
+            "COUPON_USAGE",
+            "INVENTORY_CHANGE",
+            "TOPUPS",
+            "TABLE_DISCOUNT",
+            "ASSISTANT_ABOLISH",
             "LEDGER",
         ],
         "window_minutes": {
@@ -76,17 +86,14 @@ DEFAULTS = {
         "redact_keys": ["token", "password", "Authorization"],
         "echo_token_in_logs": False,
     },
-<<<<<<< HEAD
-=======
     "testing": {
-        # ONLINE: 正常实时ETL；OFFLINE: 读取归档JSON做T/L
+        # ONLINE: 正常实时 ETL；OFFLINE: 读取归档 JSON 做 T/L
-        "mode": "OFFLINE",
+        "mode": "ONLINE",
-        # 离线归档JSON所在目录
+        # 离线归档 JSON 所在目录（测试/离线回放使用）
         "json_archive_dir": "",
-        # 测试运行时用于生成/复制临时JSON的目录
+        # 测试运行时用于生成/复制临时 JSON 的目录
         "temp_json_dir": "",
     },
->>>>>>> main
 }
 
 # 任务代码常量

etl_billiards/scripts/run_tests.py

@@ -21,6 +21,10 @@ import pytest
 
 PROJECT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
 
+# 确保项目根目录在 sys.path，便于 tests 内部 import config / tasks 等模块
+if PROJECT_ROOT not in sys.path:
+    sys.path.insert(0, PROJECT_ROOT)
+
 SUITE_MAP: Dict[str, str] = {
     "online": "tests/unit/test_etl_tasks_online.py",
     "offline": "tests/unit/test_etl_tasks_offline.py",

etl_billiards/scripts/test_presets.py

@@ -1,53 +1,48 @@
 # -*- coding: utf-8 -*-
-"""测试命令“仓库”，集中维护 run_tests.py 的预置组合。
+"""测试命令仓库：集中维护 run_tests.py 的常用组合，并支持一键执行。
 
-支持的参数键（可在 PRESETS 中自由组合）：
+参数键说明（可在 PRESETS 中任意叠加）：
 
 1. suite
-   - 类型：列表
+   类型：列表；值：["online"], ["offline"], ["integration"] 等。
-   - 作用：引用 run_tests 中的预置套件，值可为 online / offline / integration 等。
+   含义：引用 run_tests 内置测试套件。online=在线模式；offline=离线模式；integration=数据库集成测试。
-   - 用法：["online"] 仅跑在线模式；["online","offline"] 同时跑两套；["integration"] 跑数据库集成测试。
+   用法：["online","offline"] 表示一次执行两套；["integration"] 仅跑数据库相关用例。
 
 2. tests
-   - 类型：列表
+   类型：列表；示例：["tests/unit/test_config.py"]。
-   - 作用：传入任意 pytest 目标路径，适合补充临时/自定义测试文件。
+   含义：自定义的 pytest 目标路径，适合补充临时/个别测试。
-   - 用法：["tests/unit/test_config.py","tests/unit/test_parsers.py"]。
 
 3. mode
-   - 类型：字符串
+   类型：字符串；取值："ONLINE" 或 "OFFLINE"。
-   - 取值：ONLINE 或 OFFLINE。
+   含义：覆盖 TEST_MODE；ONLINE 走 API 全流程，OFFLINE 读取 JSON 归档执行 Transform + Load。
-   - 作用：覆盖 TEST_MODE；ONLINE 走 API 全流程，OFFLINE 读取 JSON 归档执行 T+L。
 
 4. db_dsn
-   - 类型：字符串
+   类型：字符串；示例：postgresql://user:pwd@host:5432/testdb。
-   - 作用：设置 TEST_DB_DSN，指定真实 PostgreSQL 连接；缺省时测试引擎使用伪 DB，仅记录写入不触库。
+   含义：设置 TEST_DB_DSN，使用真实 PostgreSQL 连接；不设置则使用伪 DB（仅记录操作，不落库）。
-   - 示例：postgresql://user:pwd@localhost:5432/testdb。
 
 5. json_archive / json_temp
-   - 类型：字符串
+   类型：字符串；示例："tests/testdata_json"、"C:/tmp/json"。
-   - 作用：离线模式的 JSON 归档目录 / 临时输出目录。
+   含义：离线模式所需的归档输入目录 / 临时输出目录。未设置时沿用 .env 或默认配置。
-   - 说明：不设置时沿用 .env 或默认值；仅在 OFFLINE 模式需要关注。
 
 6. keyword
-   - 类型：字符串
+   类型：字符串；示例："ORDERS"。
-   - 作用：等价 pytest -k，用于筛选测试名/节点。
+   含义：等价 pytest -k，可筛选测试名/节点，只运行包含该关键字的用例。
-   - 示例："ORDERS" 可只运行包含该关键字的测试函数。
 
 7. pytest_args
-   - 类型：字符串
+   类型：字符串；示例："-vv --maxfail=1"。
-   - 作用：附加 pytest 命令行参数。
+   含义：追加 pytest 命令行参数，用于控制日志、失败策略等。
-   - 示例："-vv --maxfail=1 --disable-warnings"。
 
 8. env
-   - 类型：列表
+   类型：列表；示例：["STORE_ID=123","API_TOKEN=xxx"]。
-   - 作用：追加环境变量，形如 ["STORE_ID=123","API_TOKEN=xxx"]，会在 run_tests 内透传给 os.environ。
+   含义：额外的环境变量，在调用 run_tests 前注入到 os.environ。
 
 9. preset_meta
-   - 类型：字符串
+   类型：字符串；仅用于描述场景，不会传给 run_tests（纯注释）。
-   - 作用：纯注释信息，便于描述该预置组合的用途，不会传递给 run_tests。
 
-运行方式建议直接 F5（或 `python scripts/test_presets.py`），脚本将读取 AUTO_RUN_PRESETS 中的配置依次执行。
+使用方式：
-如需临时指定其它预置，可传入 `--preset xxx`；`--list` 用于查看所有参数说明和预置详情。
+- 直接 F5 或 `python scripts/test_presets.py`：读取 AUTO_RUN_PRESETS 的预置并顺序执行。
+- `python scripts/test_presets.py --preset offline_realdb`：临时指定要运行的组合。
+- `python scripts/test_presets.py --list`：查看参数说明及所有预置详情。
 """
 
 from __future__ import annotations
@@ -60,60 +55,46 @@ from typing import List
 
 RUN_TESTS_SCRIPT = os.path.join(os.path.dirname(__file__), "run_tests.py")
 
-AUTO_RUN_PRESETS = ["online_orders"]
+# 默认自动运行的预置（可自定义顺序）
-
+AUTO_RUN_PRESETS = ["offline_realdb"]
-# PRESETS = {
-#     "online_orders": {
-#         "suite": ["online"],
-#         "mode": "ONLINE",
-#         "keyword": "ORDERS",
-#         "pytest_args": "-vv",
-#         "preset_meta": "在线模式，仅跑订单任务，输出更详细日志",
-#     },
-#     "offline_realdb": {
-#         "suite": ["offline"],
-#         "mode": "OFFLINE",
-#         "db_dsn": "postgresql://user:pwd@localhost:5432/testdb",
-#         "json_archive": "tests/testdata_json",
-#         "preset_meta": "离线模式 + 真实测试库，用预置 JSON 回放全量任务",
-#     },
-#     "integration_db": {
-#         "suite": ["integration"],
-#         "db_dsn": "postgresql://user:pwd@localhost:5432/testdb",
-#         "preset_meta": "仅跑数据库连接/操作相关的集成测试",
-#     },
-# }
 
 PRESETS = {
+    "online_orders": {
+        "suite": ["online"],
+        "mode": "ONLINE",
+        "keyword": "ORDERS",
+        "pytest_args": "-vv",
+        "preset_meta": "在线模式，仅跑订单任务并输出详细日志",
+    },
     "offline_realdb": {
         "suite": ["offline"],
         "mode": "OFFLINE",
         "db_dsn": "postgresql://local-Python:Neo-local-1991125@100.64.0.4:5432/LLZQ-test",
         "json_archive": "tests/testdata_json",
-        "preset_meta": "离线模式 + 真实测试库，用预置 JSON 回放全量任务",
+        "keyword": "ORDERS",
+        "preset_meta": "离线模式 + 真实测试库，用预置 JSON 回放并写入测试库",
     },
 }
 
 
-
+def print_parameter_help() -> None:
-def print_parameter_help():
+    print("=== 参数键说明 ===")
-    print("可用参数键说明：")
+    print("suite        : 预置套件列表，如 ['online','offline']")
-    print("  suite         -> 预置测试套件列表，如 ['online','offline']")
+    print("tests        : 自定义 pytest 路径列表")
-    print("  tests         -> 自定义测试文件路径列表")
+    print("mode         : TEST_MODE（ONLINE/ OFFLINE）")
-    print("  mode          -> TEST_MODE，ONLINE / OFFLINE")
+    print("db_dsn       : TEST_DB_DSN，连接真实 PostgreSQL")
-    print("  db_dsn        -> TEST_DB_DSN，连接真实 PostgreSQL")
+    print("json_archive : TEST_JSON_ARCHIVE_DIR，离线模式输入目录")
-    print("  json_archive  -> TEST_JSON_ARCHIVE_DIR，离线 JSON 目录")
+    print("json_temp    : TEST_JSON_TEMP_DIR，离线模式临时目录")
-    print("  json_temp     -> TEST_JSON_TEMP_DIR，离线临时目录")
+    print("keyword      : pytest -k 过滤关键字")
-    print("  keyword       -> pytest -k 过滤关键字")
+    print("pytest_args  : 额外 pytest 参数（字符串）")
-    print("  pytest_args   -> 额外 pytest 参数（单个字符串）")
+    print("env          : 附加环境变量，例如 ['KEY=VALUE']")
-    print("  env           -> 附加环境变量，形如 ['KEY=VALUE']")
+    print("preset_meta  : 仅用于注释说明")
-    print("  preset_meta   -> 注释说明，不会传给 run_tests")
     print()
 
 
-def print_presets():
+def print_presets() -> None:
     if not PRESETS:
-        print("当前没有定义任何预置命令，可自行在 PRESETS 中添加。")
+        print("当前未定义任何预置，请在 PRESETS 中添加。")
         return
     for idx, (name, payload) in enumerate(PRESETS.items(), start=1):
         comment = payload.get("preset_meta", "")
@@ -127,13 +108,32 @@ def print_presets():
         print()
 
 
-def run_presets(preset_names: List[str], dry_run: bool):
+def resolve_targets(requested: List[str] | None) -> List[str]:
-    cmds = []
+    if not PRESETS:
+        raise SystemExit("Pre-sets 为空，请先在 PRESETS 中定义测试组合。")
+
+    def valid(names: List[str]) -> List[str]:
+        return [name for name in names if name in PRESETS]
+
+    if requested:
+        candidates = valid(requested)
+        missing = [name for name in requested if name not in PRESETS]
+        if missing:
+            print(f"警告：忽略未定义的预置 {missing}")
+        if candidates:
+            return candidates
+
+    auto = valid(AUTO_RUN_PRESETS)
+    if auto:
+        return auto
+
+    # 兜底：全部预置
+    return list(PRESETS.keys())
+
+
+def run_presets(preset_names: List[str], dry_run: bool) -> None:
     for name in preset_names:
         cmd = [sys.executable, RUN_TESTS_SCRIPT, "--preset", name]
-        cmds.append(cmd)
-
-    for cmd in cmds:
         printable = " ".join(cmd)
         if dry_run:
             print(f"[Dry-Run] {printable}")
@@ -142,11 +142,11 @@ def run_presets(preset_names: List[str], dry_run: bool):
             subprocess.run(cmd, check=False)
 
 
-def main():
+def main() -> None:
-    parser = argparse.ArgumentParser(description="测试预置仓库（在此集中配置并运行测试组合）")
+    parser = argparse.ArgumentParser(description="测试预置仓库（集中配置即可批量触发 run_tests）")
-    parser.add_argument("--preset", choices=sorted(PRESETS.keys()), nargs="+", help="直接指定要运行的预置命令")
+    parser.add_argument("--preset", choices=sorted(PRESETS.keys()), nargs="+", help="指定要运行的预置命令")
-    parser.add_argument("--list", action="store_true", help="仅列出参数键和所有预置命令")
+    parser.add_argument("--list", action="store_true", help="仅列出参数说明与所有预置")
-    parser.add_argument("--dry-run", action="store_true", help="仅打印将要执行的命令，而不真正运行")
+    parser.add_argument("--dry-run", action="store_true", help="仅打印命令，不执行 pytest")
     args = parser.parse_args()
 
     if args.list:
@@ -154,12 +154,8 @@ def main():
         print_presets()
         return
 
-    if args.preset:
+    targets = resolve_targets(args.preset)
-        target = args.preset
+    run_presets(targets, dry_run=args.dry_run)
-    else:
-        target = AUTO_RUN_PRESETS or list(PRESETS.keys())
-
-    run_presets(target, dry_run=args.dry_run)
 
 
 if __name__ == "__main__":