From b9fbf357a4cd55ecfcd98ed0abb0dd4bce0f6c7b Mon Sep 17 00:00:00 2001 From: strepsiades Date: Wed, 11 Mar 2026 11:40:23 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E4=BA=86=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=B7=A5=E5=85=B7=E7=94=A8=E4=BA=8E=E5=88=9D?= =?UTF-8?q?=E5=A7=8B=E5=8C=96=E6=B5=8B=E8=AF=95=E7=8E=AF=E5=A2=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- README.md | 4 + docs/testing_guide.md | 189 +++++ .../config/datasource_config.py | 2 +- .../domain/supplier/sync_strategy.py | 2 +- tests/README.md | 1 + tests/unit/test_supplier_sync_strategy.py | 6 + tools/analyze_persist_mismatches.py | 414 +++++++++++ tools/remote_env.example.yaml | 44 ++ tools/remote_env.py | 699 ++++++++++++++++++ 9 files changed, 1359 insertions(+), 2 deletions(-) create mode 100644 docs/testing_guide.md create mode 100644 tests/unit/test_supplier_sync_strategy.py create mode 100644 tools/analyze_persist_mismatches.py create mode 100644 tools/remote_env.example.yaml create mode 100644 tools/remote_env.py diff --git a/README.md b/README.md index 7096f20..04fab82 100644 --- a/README.md +++ b/README.md @@ -81,10 +81,13 @@ ### 测试 - 全量：`pytest tests -q` - 说明：`tests/README.md` +- 远程环境初始化工具：`python tools/remote_env.py --config tools/remote_env.example.yaml init|run|stop|clear` +- 测试总览：`docs/testing_guide.md` ### 其他工具 - UI 调试：`python -m ui.main --host 127.0.0.1 --port 8765` - 运维与排障：`docs/operations.md` +- 远程测试环境：`tools/remote_env.py` ## 文档导航 @@ -96,5 +99,6 @@ - 库化与接入：`docs/library_integration_guide.md` - Collection 查询与升级：`docs/collection_query_and_upgrade_plan.md` - 运维迭代：`docs/operations.md` +- 测试指南：`docs/testing_guide.md` 历史资料已归档到 `docs/archive/`。 diff --git a/docs/testing_guide.md b/docs/testing_guide.md new file mode 100644 index 0000000..1340772 --- /dev/null +++ b/docs/testing_guide.md @@ -0,0 +1,189 @@ +# 测试指南 + +本文档说明三件事： +- 单元测试怎么跑。 +- 集成测试怎么组织。 +- 测试辅助工具如何使用，尤其是远程 ECM 环境初始化工具。 + +## 1. 测试分层 + +### 1.1 单元测试 + +位置：`tests/unit/` + +目标： +- 校验状态机不变量、事件约束、策略辅助逻辑。 +- 校验 datasource/handler 工具链、配置严格性、导入面稳定性。 +- 对具体 domain 的 update/create 行为做快速回归。 + +常用命令： +- 全量单元测试：`pytest tests/unit -q` +- 指定文件：`pytest tests/unit/test_update_ops.py -q` + +适用场景： +- 修改 `sync_state_machine/` 公共逻辑。 +- 修改某个 domain 的 handler/strategy。 +- 校验 schema 相关逻辑没有引入行为回退。 + +### 1.2 集成测试 + +位置：`tests/integration/` + +目标： +- 覆盖 pipeline 级行为，包括 load/bind/create/update/poll/persist。 +- 校验多节点、多轮次、mock datasource 与 UI smoke 场景。 +- 验证运行配置、状态机配置和工具链连通性。 + +常用命令： +- 全量集成测试：`pytest tests/integration -q` +- UI 烟测：`pytest tests/integration/test_ui_api_smoke.py -q` +- 工具链专项：`pytest tests/integration/test_toolchain_config_and_graph.py -q` + +适用场景： +- 修改 pipeline、collection、状态机执行路径。 +- 修改持久化或跨 domain 依赖行为。 +- 修改运行配置解析、注册表装配等入口逻辑。 + +### 1.3 脚本化大规模回归 + +位置：`scripts/auto_test_domains/` + +目标： +- 用“多轮数据集 + 报告”的方式覆盖复杂业务 domain。 +- 适合真实接口联调、回归复现、差异收敛分析。 + +特点： +- 这类脚本不直接纳入 pytest。 +- 它们更偏向测试编排与问题定位，而不是轻量级快速回归。 + +相关文档： +- `docs/auto_test_spec/agent_auto_test_method.md` + +## 2. 远程环境初始化工具 + +位置：`tools/remote_env.py` + +这个工具是给真实 ECM 环境联调用的最小控制器，提供 4 个命令： +- `clear`：停止服务、恢复 `.env`、清空自动测试专用 CouchDB 数据库。 +- `init`：停止服务，按 JSONL 数据目录执行 `run.py restore-database` 恢复测试数据库。 +- `run`：启动远程 ECM 服务，不阻塞当前命令。 +- `stop`：关闭远程 ECM 服务，并恢复原始 `.env`。 + +它不是 pipeline 执行器，只负责测试环境的基础生命周期管理。 + +### 2.1 配置项 + +示例配置：`tools/remote_env.example.yaml` + +核心配置： +- `couchdb.base_url` / `couchdb.database` / `couchdb.username` / `couchdb.password` +- `service.ecm_root` +- `service.python_path` +- `data.data_dir` +- `env_file.path` +- `env_file.overrides` + +按命令的最低要求： +- `run` / `stop`：只要求 `service.*` 可用。 +- `clear`：要求 `service.*` 和 `couchdb.*` 可用。 +- `init`：要求 `service.*`、`couchdb.*`、`data.*` 全部可用。 + +可选配置： +- `service.start_command` +- `service.restore_command` +- `service.ready_url` +- `service.log_file` +- `service.env` +- `env_file.restore_on_stop` + +支持占位符： +- `${PROJECT_ROOT}`：当前仓库根目录。 +- `${CONFIG_DIR}`：配置文件所在目录。 +- `${ECM_ROOT}`：`service.ecm_root` 的值。 +- `${PYTHON}`：`service.python_path` 的值。 +- `${DATA_DIR}`：`data.data_dir` 的值。 +- `${COUCHDB_DATABASE}`：`couchdb.database` 的值。 +- `${COUCHDB_USERNAME}`：`couchdb.username` 的值。 +- `${COUCHDB_PASSWORD}`：`couchdb.password` 的值。 +- `${COUCHDB_URL}`：`couchdb.url` 的值。 +- `${COUCHDB_PORT}`：`couchdb.port` 的值。 + +### 2.2 基本用法 + +1. 复制示例配置并填入真实路径。 +2. 先执行 `init` 恢复测试数据库。 +3. 再执行 `run` 启动远程服务。 +4. 测试结束后执行 `stop`。 + +命令示例： + +```bash +python tools/remote_env.py --config tools/remote_env.example.yaml init +python tools/remote_env.py --config tools/remote_env.example.yaml run +python tools/remote_env.py --config tools/remote_env.example.yaml stop +python tools/remote_env.py --config tools/remote_env.example.yaml clear +``` + +### 2.3 命令行覆盖 + +如果只想临时覆盖少数字段，可以直接传命令行参数： + +```bash +python tools/remote_env.py \ + --config tools/remote_env.example.yaml \ + --db-name test_push_auto_tmp \ + --ecm-root /data/ecm_app \ + --python-path /data/venv/bin/python \ + init +``` + +支持覆盖的字段： +- `--db-name` +- `--db-user` +- `--db-password` +- `--couchdb-base-url` +- `--couchdb-url` +- `--couchdb-port` +- `--ecm-root` +- `--python-path` +- `--data-dir` +- `--ready-url` + +### 2.4 行为约束 + +- `clear` 和 `init` 会先执行停服，避免服务仍占用 CouchDB 连接。 +- `init` 不会自动执行 `run`；数据库恢复和服务启动分开控制。 +- `run` 如果检测到远端仓库下已有 `run.py` 进程在运行，会直接返回 `already_running`。 +- `.env` 只覆盖 `env_file.overrides` 中声明的键，未声明的键会原样保留。 +- 第一次覆盖 `.env` 时会自动备份，`stop` 和 `clear` 默认会恢复原始 `.env`。 +- 进程状态写到 `runtime/remote_env/state.json`，不需要手工配置 pid 文件。 + +## 3. 推荐测试流程 + +### 3.1 本地开发回归 + +1. 先跑 `pytest tests/unit -q`。 +2. 如果改动影响 pipeline，再跑目标集成测试。 +3. 如果改动影响真实接口流程，再使用 `tools/remote_env.py` 管理远程环境。 + +### 3.2 真实接口联调 + +1. 用 `init` 恢复 ECM 测试库。 +2. 用 `run` 拉起 ECM 服务。 +3. 执行同步脚本或联调脚本。 +4. 查看 `runtime/remote_env/` 下的状态文件与日志。 +5. 完成后执行 `stop`。 + +### 3.3 复杂 domain 回归 + +1. 用 `scripts/auto_test_domains/` 中的 case 脚本准备多轮数据。 +2. 先缩小到单个 domain 复现。 +3. 确认差异与日志后，再扩大到聚合回归。 + +## 4. 相关文件 + +- `tests/README.md` +- `tools/remote_env.py` +- `tools/remote_env.example.yaml` +- `docs/operations.md` +- `docs/auto_test_spec/agent_auto_test_method.md` \ No newline at end of file diff --git a/sync_state_machine/config/datasource_config.py b/sync_state_machine/config/datasource_config.py index 86a2e62..c6c8607 100644 --- a/sync_state_machine/config/datasource_config.py +++ b/sync_state_machine/config/datasource_config.py @@ -38,7 +38,7 @@ class ApiDataSourceConfig(BaseDataSourceConfig): poll_interval: float = 0.5 api_rate_limit_max_requests: int = 10 api_rate_limit_window_seconds: float = 10.0 - target_project_ids: List[str] = Field(min_length=1) + target_project_ids: List[str] = Field(default_factory=list) DataSourceConfig = Annotated[ diff --git a/sync_state_machine/domain/supplier/sync_strategy.py b/sync_state_machine/domain/supplier/sync_strategy.py index 83c6dc2..807b979 100644 --- a/sync_state_machine/domain/supplier/sync_strategy.py +++ b/sync_state_machine/domain/supplier/sync_strategy.py @@ -17,7 +17,7 @@ class SupplierSyncStrategy(DefaultSyncStrategy[SupplierDetailResponse]): default_config = StrategyConfig( auto_bind=True, - auto_bind_fields=["code"], + auto_bind_fields=["credit_code"], depend_fields={}, local_orphan_action=OrphanAction.NONE, remote_orphan_action=OrphanAction.CREATE_LOCAL, diff --git a/tests/README.md b/tests/README.md index 9238cd7..fb4a430 100644 --- a/tests/README.md +++ b/tests/README.md @@ -11,6 +11,7 @@ - 集成：`pytest tests/integration -q` - 工具链专项：`pytest tests/integration/test_toolchain_config_and_graph.py -q` - UI 专项：`pytest tests/integration/test_ui_api_smoke.py -q` +- 测试总览与远程环境工具：`docs/testing_guide.md` ## 设计原则 - 不绕过 schema/状态机校验；配置错误必须显式失败。 diff --git a/tests/unit/test_supplier_sync_strategy.py b/tests/unit/test_supplier_sync_strategy.py new file mode 100644 index 0000000..57d1357 --- /dev/null +++ b/tests/unit/test_supplier_sync_strategy.py @@ -0,0 +1,6 @@ +from sync_state_machine.domain.supplier.sync_strategy import SupplierSyncStrategy + + +def test_supplier_sync_strategy_defaults_to_credit_code_auto_bind(): + assert SupplierSyncStrategy.default_config.auto_bind is True + assert SupplierSyncStrategy.default_config.auto_bind_fields == ["credit_code"] \ No newline at end of file diff --git a/tools/analyze_persist_mismatches.py b/tools/analyze_persist_mismatches.py new file mode 100644 index 0000000..4d2a694 --- /dev/null +++ b/tools/analyze_persist_mismatches.py @@ -0,0 +1,414 @@ +from __future__ import annotations + +import argparse +import json +import random +import sqlite3 +import sys +from collections import Counter +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +REASON_DESCRIPTIONS = { + "hierarchy_shape": "层级结构不一致：一侧是顶层/短路径，另一侧保留了父级、完整 path 或 parent 信息。", + "region_enrichment": "地区信息补全程度不一致：远端带国家、省市名称或完整 region_path，本地缺失或更短。", + "audit_identity": "审计人字段不一致：远端保留了 created_by/updated_by，本地为空或使用不同来源。", + "audit_time_format": "时间格式或更新时间不一致：本地常用 ISO 时间，远端常用空格分隔时间，且更新时间可能已被远端二次维护。", + "company_nature_semantics": "company_nature 语义不一致：本地是较保守的单值/少值，远端是人工维护后的多选性质。", + "sort_policy": "排序策略不一致：本地默认值或来源字段与远端现存排序值不同。", + "null_empty_normalization": "空值归一化不一致：本地用 null，远端用空字符串或空数组。", +} + + +@dataclass +class PersistedNode: + collection_id: str + node_id: str + node_type: str + data_id: str | None + bind_data_id: str | None + action: str | None + status: str | None + binding_status: str | None + error: str | None + sync_log: str + data: Any + origin_data: Any + context: Any + + +@dataclass +class BoundPair: + node_type: str + local_id: str + remote_id: str + local_node: PersistedNode + remote_node: PersistedNode + diffs: list[tuple[str, Any, Any]] + + +def _classify_reasons(pair: BoundPair) -> list[str]: + diff_paths = {path for path, _, _ in pair.diffs} + reasons: list[str] = [] + + if diff_paths & {"parent_id", "parent_name", "path", "path_name"}: + reasons.append("hierarchy_shape") + + if diff_paths & {"country", "country_name", "province_name", "city_name", "region_path", "province", "city"}: + reasons.append("region_enrichment") + + if diff_paths & {"created_by", "updated_by", "created_by_name", "updated_by_name"}: + reasons.append("audit_identity") + + if diff_paths & {"created_at", "updated_at"}: + reasons.append("audit_time_format") + + if "company_nature" in diff_paths: + reasons.append("company_nature_semantics") + + if "sort" in diff_paths: + reasons.append("sort_policy") + + if any( + (local_value is None and remote_value in ("", [])) or (remote_value is None and local_value in ("", [])) + for _, local_value, remote_value in pair.diffs + ): + reasons.append("null_empty_normalization") + + return reasons + + +def _build_reason_summary(mismatches: list[BoundPair]) -> str: + counter: Counter[str] = Counter() + for pair in mismatches: + counter.update(_classify_reasons(pair)) + + if not counter: + return "No mismatch reasons inferred." + + lines = ["Likely mismatch reasons:"] + for reason, count in counter.most_common(): + description = REASON_DESCRIPTIONS.get(reason, reason) + lines.append(f" - {reason}: {count}") + lines.append(f" {description}") + return "\n".join(lines) + + +def _build_business_conclusion(node_type: str, mismatches: list[BoundPair]) -> str: + if not mismatches: + return "No mismatched bound pairs found." + + reason_counts: Counter[str] = Counter() + for pair in mismatches: + reason_counts.update(_classify_reasons(pair)) + + if node_type == "company": + conclusions = [ + "company 当前不一致主要不是绑定错，而是本地 company handler 输出更扁平。", + ] + if reason_counts["hierarchy_shape"]: + conclusions.append("远端 company 已维护集团层级，本地只保留当前公司自身，导致 parent_id/path/path_name 不一致。") + if reason_counts["region_enrichment"]: + conclusions.append("远端 company 还维护了省市编码与名称，本地目前未补这些字段。") + if reason_counts["sort_policy"]: + conclusions.append("sort 来自两套不同维护规则，本地默认值与远端现存排序不一致。") + return "\n".join(f"- {item}" for item in conclusions) + + if node_type == "supplier": + conclusions = [ + "supplier 当前大多数绑定能命中，但远端 supplier 是一套人工维护后更完整的数据快照。", + ] + if reason_counts["company_nature_semantics"]: + conclusions.append("本地 company_nature 往往只有单值，远端 supplier 常是多选性质，因此语义层面天然不一致。") + if reason_counts["region_enrichment"]: + conclusions.append("远端 supplier 补了国家层、地区名称和完整 region_path，本地只有编码级最小信息。") + if reason_counts["audit_identity"] or reason_counts["audit_time_format"]: + conclusions.append("远端保存了创建/更新操作者和远端维护时间，本地导出使用的是 DEPM 审计字段，时间格式也不同。") + if reason_counts["hierarchy_shape"] or reason_counts["null_empty_normalization"]: + conclusions.append("parent_id/path 的空值和路径表达也不同，本地更偏同步输入形态，远端更偏展示/运营形态。") + return "\n".join(f"- {item}" for item in conclusions) + + return "\n".join(f"- {REASON_DESCRIPTIONS.get(reason, reason)}" for reason, _ in reason_counts.most_common()) + + +def _parse_json(value: str | None) -> Any: + if not value: + return None + try: + return json.loads(value) + except json.JSONDecodeError: + return value + + +def _format_json(value: Any) -> str: + return json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + + +def _short_repr(value: Any, *, max_len: int = 160) -> str: + text = repr(value) + if len(text) <= max_len: + return text + return text[: max_len - 3] + "..." + + +def _diff_values(local_value: Any, remote_value: Any, *, path: str = "") -> list[tuple[str, Any, Any]]: + diffs: list[tuple[str, Any, Any]] = [] + + if type(local_value) is not type(remote_value): + diffs.append((path or "", local_value, remote_value)) + return diffs + + if isinstance(local_value, dict): + keys = sorted(set(local_value.keys()) | set(remote_value.keys())) + for key in keys: + child_path = f"{path}.{key}" if path else str(key) + if key not in local_value: + diffs.append((child_path, None, remote_value[key])) + continue + if key not in remote_value: + diffs.append((child_path, local_value[key], None)) + continue + diffs.extend(_diff_values(local_value[key], remote_value[key], path=child_path)) + return diffs + + if isinstance(local_value, list): + if local_value != remote_value: + diffs.append((path or "", local_value, remote_value)) + return diffs + + if local_value != remote_value: + diffs.append((path or "", local_value, remote_value)) + return diffs + + +def _load_nodes(conn: sqlite3.Connection, node_type: str) -> dict[tuple[str, str], PersistedNode]: + cur = conn.cursor() + cur.execute( + """ + SELECT collection_id, node_id, node_type, data_id, bind_data_id, action, status, + binding_status, error, sync_log, data, origin_data, context + FROM nodes + WHERE node_type = ? + """, + (node_type,), + ) + + rows: dict[tuple[str, str], PersistedNode] = {} + for row in cur.fetchall(): + node = PersistedNode( + collection_id=row[0], + node_id=row[1], + node_type=row[2], + data_id=row[3], + bind_data_id=row[4], + action=row[5], + status=row[6], + binding_status=row[7], + error=row[8], + sync_log=row[9] or "", + data=_parse_json(row[10]), + origin_data=_parse_json(row[11]), + context=_parse_json(row[12]), + ) + rows[(node.collection_id, node.node_id)] = node + return rows + + +def _load_bound_pairs(conn: sqlite3.Connection, node_type: str) -> list[BoundPair]: + nodes = _load_nodes(conn, node_type) + cur = conn.cursor() + cur.execute( + "SELECT node_type, local_id, remote_id FROM bindings WHERE node_type = ? AND remote_id IS NOT NULL", + (node_type,), + ) + + pairs: list[BoundPair] = [] + for bound_node_type, local_id, remote_id in cur.fetchall(): + local_node = nodes.get(("local", local_id)) + remote_node = nodes.get(("remote", remote_id)) + if local_node is None or remote_node is None: + continue + local_data = local_node.data or {} + remote_data = remote_node.data or {} + diffs = _diff_values(local_data, remote_data) + pairs.append( + BoundPair( + node_type=bound_node_type, + local_id=local_id, + remote_id=remote_id, + local_node=local_node, + remote_node=remote_node, + diffs=diffs, + ) + ) + return pairs + + +def _build_summary(pairs: list[BoundPair]) -> str: + total = len(pairs) + mismatches = [pair for pair in pairs if pair.diffs] + matched = total - len(mismatches) + diff_counter: Counter[str] = Counter() + for pair in mismatches: + for path, _, _ in pair.diffs: + diff_counter[path] += 1 + + lines = [ + f"Bound pairs: {total}", + f"Matched pairs: {matched}", + f"Mismatched pairs: {len(mismatches)}", + ] + if diff_counter: + lines.append("Top diff fields:") + for field, count in diff_counter.most_common(20): + lines.append(f" - {field}: {count}") + lines.append(_build_reason_summary(mismatches)) + return "\n".join(lines) + + +def _format_pair(pair: BoundPair, *, max_diff_items: int) -> str: + diff_lines = [] + for path, local_value, remote_value in pair.diffs[:max_diff_items]: + diff_lines.append( + f" - {path}: local={_short_repr(local_value)} | remote={_short_repr(remote_value)}" + ) + if len(pair.diffs) > max_diff_items: + diff_lines.append(f" - ... {len(pair.diffs) - max_diff_items} more diff fields") + + return "\n".join( + [ + f"node_type={pair.node_type}", + f"local.node_id={pair.local_node.node_id}", + f"local.data_id={pair.local_node.data_id}", + f"local.bind_data_id={pair.local_node.bind_data_id}", + f"local.action={pair.local_node.action} local.status={pair.local_node.status} local.binding_status={pair.local_node.binding_status}", + f"local.error={pair.local_node.error}", + f"remote.node_id={pair.remote_node.node_id}", + f"remote.data_id={pair.remote_node.data_id}", + f"remote.bind_data_id={pair.remote_node.bind_data_id}", + f"remote.action={pair.remote_node.action} remote.status={pair.remote_node.status} remote.binding_status={pair.remote_node.binding_status}", + f"remote.error={pair.remote_node.error}", + "Reason tags:", + *(f" - {reason}: {REASON_DESCRIPTIONS.get(reason, reason)}" for reason in _classify_reasons(pair)), + "Diff fields:", + *diff_lines, + "", + "Local sync_log:", + pair.local_node.sync_log or "", + "", + "Remote sync_log:", + pair.remote_node.sync_log or "", + "", + "Local context:", + _format_json(pair.local_node.context), + "", + "Remote context:", + _format_json(pair.remote_node.context), + "", + "Local data:", + _format_json(pair.local_node.data), + "", + "Remote data:", + _format_json(pair.remote_node.data), + "", + "Local origin_data:", + _format_json(pair.local_node.origin_data), + "", + "Remote origin_data:", + _format_json(pair.remote_node.origin_data), + ] + ) + + +def build_report( + *, + db_path: Path, + node_type: str, + limit: int, + seed: int, + max_diff_items: int, +) -> str: + conn = sqlite3.connect(db_path) + try: + pairs = _load_bound_pairs(conn, node_type) + finally: + conn.close() + + mismatches = [pair for pair in pairs if pair.diffs] + rng = random.Random(seed) + sampled = list(mismatches) + rng.shuffle(sampled) + sampled = sampled[:limit] + + sections = [ + f"# Persist Mismatch Report\n", + f"db_path: {db_path}", + f"node_type: {node_type}", + f"sample_limit: {limit}", + f"seed: {seed}", + "", + "## Summary", + _build_summary(pairs), + "", + "## Business Conclusion", + _build_business_conclusion(node_type, mismatches), + ] + + if not sampled: + sections.extend(["", "## Sampled Mismatches", "No mismatched bound pairs found."]) + return "\n".join(sections) + + sections.append("") + sections.append("## Sampled Mismatches") + for index, pair in enumerate(sampled, start=1): + sections.extend( + [ + "", + f"### Sample {index}", + _format_pair(pair, max_diff_items=max_diff_items), + ] + ) + return "\n".join(sections) + + +def parse_args(argv: list[str]) -> argparse.Namespace: + parser = argparse.ArgumentParser(description="Analyze mismatched bound pairs in a persisted sync SQLite DB") + parser.add_argument("db_path", help="Path to persisted sqlite db, e.g. backend/runtime/ecm_sync/.../depm_to_ecm.db") + parser.add_argument("--node-type", required=True, help="Node type to analyze, e.g. company or supplier") + parser.add_argument("--limit", type=int, default=3, help="Number of mismatched bound pairs to sample") + parser.add_argument("--seed", type=int, default=11, help="Random seed for sampling") + parser.add_argument("--max-diff-items", type=int, default=20, help="Max diff fields to print per sampled pair") + parser.add_argument("--output", help="Optional output file path. If omitted, prints to stdout.") + return parser.parse_args(argv) + + +def main(argv: list[str] | None = None) -> int: + args = parse_args(argv or sys.argv[1:]) + db_path = Path(args.db_path).resolve() + if not db_path.exists(): + print(f"DB file not found: {db_path}", file=sys.stderr) + return 1 + + report = build_report( + db_path=db_path, + node_type=args.node_type, + limit=max(0, args.limit), + seed=args.seed, + max_diff_items=max(1, args.max_diff_items), + ) + + if args.output: + output_path = Path(args.output).resolve() + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(report, encoding="utf-8") + print(f"Report written to: {output_path}") + return 0 + + print(report) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/tools/remote_env.example.yaml b/tools/remote_env.example.yaml new file mode 100644 index 0000000..715bf53 --- /dev/null +++ b/tools/remote_env.example.yaml @@ -0,0 +1,44 @@ +couchdb: + enabled: true + base_url: http://localhost:5984 + url: http://127.0.0.1 + port: "5984" + database: test_push_auto + username: admin + password: password123 + recreate_after_delete: false + +data: + data_dir: ${PROJECT_ROOT}/working_local_datasource/filtered + +service: + ecm_root: /Users/zyl/my_projects/GroupSideProjectManagementSystem + python_path: /Users/zyl/my_projects/GroupSideProjectManagementSystem/.venv/bin/python + cwd: ${ECM_ROOT} + start_command: + - ${PYTHON} + - run.py + restore_command: + - ${PYTHON} + - run.py + - restore-database + - -d + - ${COUCHDB_DATABASE} + - -s + - ${DATA_DIR} + env: {} + ready_url: http://localhost:8000/docs + ready_timeout_seconds: 120 + ready_interval_seconds: 2 + log_file: ${PROJECT_ROOT}/runtime/remote_env/ecm_remote.log + +env_file: + enabled: true + path: ${ECM_ROOT}/.env + restore_on_stop: true + overrides: + COUCHDB_DBNAME: ${COUCHDB_DATABASE} + COUCHDB_USER: ${COUCHDB_USERNAME} + COUCHDB_PASSWORD: ${COUCHDB_PASSWORD} + COUCHDB_URL: ${COUCHDB_URL} + COUCHDB_PORT: ${COUCHDB_PORT} \ No newline at end of file diff --git a/tools/remote_env.py b/tools/remote_env.py new file mode 100644 index 0000000..cb25525 --- /dev/null +++ b/tools/remote_env.py @@ -0,0 +1,699 @@ +from __future__ import annotations + +import argparse +import base64 +import json +import os +import shutil +import signal +import subprocess +import sys +import time +from copy import deepcopy +from pathlib import Path +from typing import Any +from urllib.parse import urlparse +from urllib import error, request + +import yaml + + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +STATE_DIR = PROJECT_ROOT / "runtime" / "remote_env" +STATE_FILE = STATE_DIR / "state.json" + + +DEFAULT_CONFIG: dict[str, Any] = { + "couchdb": { + "enabled": True, + "base_url": "http://localhost:5984", + "url": "http://127.0.0.1", + "port": "5984", + "database": "test_push_auto", + "username": "admin", + "password": "password123", + "recreate_after_delete": False, + }, + "data": { + "data_dir": "${PROJECT_ROOT}/working_local_datasource/filtered", + }, + "service": { + "ecm_root": "/Users/zyl/my_projects/GroupSideProjectManagementSystem", + "python_path": "/Users/zyl/my_projects/GroupSideProjectManagementSystem/.venv/bin/python", + "cwd": "${ECM_ROOT}", + "start_command": ["${PYTHON}", "run.py"], + "restore_command": [ + "${PYTHON}", + "run.py", + "restore-database", + "-d", + "${COUCHDB_DATABASE}", + "-s", + "${DATA_DIR}", + ], + "env": {}, + "ready_url": "http://localhost:8000/docs", + "ready_timeout_seconds": 120, + "ready_interval_seconds": 2, + "log_file": "${PROJECT_ROOT}/runtime/remote_env/ecm_remote.log", + }, + "env_file": { + "enabled": True, + "path": "${ECM_ROOT}/.env", + "restore_on_stop": True, + "overrides": { + "COUCHDB_DBNAME": "${COUCHDB_DATABASE}", + "COUCHDB_USER": "${COUCHDB_USERNAME}", + "COUCHDB_PASSWORD": "${COUCHDB_PASSWORD}", + "COUCHDB_URL": "${COUCHDB_URL}", + "COUCHDB_PORT": "${COUCHDB_PORT}", + }, + }, +} + + +def _deep_merge(base: dict[str, Any], override: dict[str, Any]) -> dict[str, Any]: + merged = deepcopy(base) + for key, value in override.items(): + if isinstance(value, dict) and isinstance(merged.get(key), dict): + merged[key] = _deep_merge(merged[key], value) + else: + merged[key] = value + return merged + + +def _replace_placeholders(value: Any, placeholders: dict[str, str]) -> Any: + if isinstance(value, str): + result = value + for _ in range(5): + previous = result + for key, replacement in placeholders.items(): + result = result.replace(f"${{{key}}}", replacement) + if result == previous: + break + return result + if isinstance(value, list): + return [_replace_placeholders(item, placeholders) for item in value] + if isinstance(value, dict): + return {key: _replace_placeholders(item, placeholders) for key, item in value.items()} + return value + + +def _ensure_parent(path: str | Path) -> Path: + resolved = Path(path) + resolved.parent.mkdir(parents=True, exist_ok=True) + return resolved + + +def _load_raw_config(config_path: str | Path) -> tuple[dict[str, Any], Path]: + path = Path(config_path) + if not path.is_absolute(): + path = PROJECT_ROOT / path + payload = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + if not isinstance(payload, dict): + raise ValueError(f"Config root must be a mapping: {path}") + return payload, path + + +def _apply_cli_overrides(config: dict[str, Any], args: argparse.Namespace) -> dict[str, Any]: + result = deepcopy(config) + couchdb = result.setdefault("couchdb", {}) + service = result.setdefault("service", {}) + data = result.setdefault("data", {}) + + if args.db_name is not None: + couchdb["database"] = args.db_name + if args.db_user is not None: + couchdb["username"] = args.db_user + if args.db_password is not None: + couchdb["password"] = args.db_password + if args.couchdb_base_url is not None: + couchdb["base_url"] = args.couchdb_base_url + if args.couchdb_url is not None: + couchdb["url"] = args.couchdb_url + if args.couchdb_port is not None: + couchdb["port"] = str(args.couchdb_port) + if args.ecm_root is not None: + service["ecm_root"] = args.ecm_root + if args.python_path is not None: + service["python_path"] = args.python_path + if args.data_dir is not None: + data["data_dir"] = args.data_dir + if args.ready_url is not None: + service["ready_url"] = args.ready_url + return result + + +def load_config(config_path: str | Path, args: argparse.Namespace) -> tuple[dict[str, Any], Path]: + raw_config, resolved_path = _load_raw_config(config_path) + merged = _deep_merge(DEFAULT_CONFIG, raw_config) + merged = _apply_cli_overrides(merged, args) + + config_dir = resolved_path.parent.resolve() + service = merged.setdefault("service", {}) + couchdb = merged.setdefault("couchdb", {}) + data = merged.setdefault("data", {}) + + placeholders = { + "PROJECT_ROOT": str(PROJECT_ROOT), + "CONFIG_DIR": str(config_dir), + "ECM_ROOT": str(service.get("ecm_root", "")), + "PYTHON": str(service.get("python_path", "")), + "DATA_DIR": str(data.get("data_dir", "")), + "COUCHDB_DATABASE": str(couchdb.get("database", "")), + "COUCHDB_USERNAME": str(couchdb.get("username", "")), + "COUCHDB_PASSWORD": str(couchdb.get("password", "")), + "COUCHDB_URL": str(couchdb.get("url", "")), + "COUCHDB_PORT": str(couchdb.get("port", "")), + } + + resolved = _replace_placeholders(merged, placeholders) + service = resolved.setdefault("service", {}) + couchdb = resolved.setdefault("couchdb", {}) + data = resolved.setdefault("data", {}) + env_file = resolved.setdefault("env_file", {}) + + for key in ("ecm_root", "python_path", "cwd", "log_file"): + if service.get(key): + service[key] = str(Path(service[key]).expanduser().resolve()) + if data.get("data_dir"): + data["data_dir"] = str(Path(data["data_dir"]).expanduser().resolve()) + if env_file.get("path"): + env_file["path"] = str(Path(env_file["path"]).expanduser().resolve()) + + service["ready_timeout_seconds"] = int(service.get("ready_timeout_seconds", 120)) + service["ready_interval_seconds"] = int(service.get("ready_interval_seconds", 2)) + return resolved, resolved_path + + +def validate_config(config: dict[str, Any], command: str) -> None: + couchdb = config.get("couchdb", {}) + data = config.get("data", {}) + service = config.get("service", {}) + env_file = config.get("env_file", {}) + + required_service = [ + (service.get("ecm_root"), "service.ecm_root"), + (service.get("python_path"), "service.python_path"), + ] + missing_service = [name for value, name in required_service if value in (None, "")] + if missing_service: + raise ValueError("Missing required config fields: " + ", ".join(missing_service)) + + if not Path(service["ecm_root"]).exists(): + raise FileNotFoundError(f"ECM root does not exist: {service['ecm_root']}") + if not Path(service["python_path"]).exists(): + raise FileNotFoundError(f"Python path does not exist: {service['python_path']}") + + if not isinstance(service.get("start_command"), list) or not service.get("start_command"): + raise ValueError("service.start_command must be a non-empty list") + if not isinstance(service.get("restore_command"), list) or not service.get("restore_command"): + raise ValueError("service.restore_command must be a non-empty list") + + if env_file.get("enabled", True): + env_path = Path(env_file.get("path", "")) + if not env_path.exists(): + raise FileNotFoundError(f"env_file.path does not exist: {env_path}") + + if command in {"clear", "init"} and couchdb.get("enabled", True): + required_couchdb = [ + (couchdb.get("base_url"), "couchdb.base_url"), + (couchdb.get("database"), "couchdb.database"), + ] + missing_couchdb = [name for value, name in required_couchdb if value in (None, "")] + if missing_couchdb: + raise ValueError("Missing required config fields: " + ", ".join(missing_couchdb)) + + if command == "init": + if data.get("data_dir") in (None, ""): + raise ValueError("Missing required config fields: data.data_dir") + if not Path(data["data_dir"]).exists(): + raise FileNotFoundError(f"Data directory does not exist: {data['data_dir']}") + + +def _auth_header(username: str, password: str) -> dict[str, str]: + if not username: + return {} + token = base64.b64encode(f"{username}:{password}".encode("utf-8")).decode("ascii") + return {"Authorization": f"Basic {token}"} + + +def _delete_couchdb_database(base_url: str, database: str, username: str = "", password: str = "") -> int: + url = f"{base_url.rstrip('/')}/{database}" + req = request.Request(url, method="DELETE", headers=_auth_header(username, password)) + try: + with request.urlopen(req, timeout=10) as response: + return response.status + except error.HTTPError as exc: + if exc.code == 404: + return 404 + raise + + +def _create_couchdb_database(base_url: str, database: str, username: str = "", password: str = "") -> int: + url = f"{base_url.rstrip('/')}/{database}" + req = request.Request(url, method="PUT", headers=_auth_header(username, password)) + with request.urlopen(req, timeout=10) as response: + return response.status + + +def _run_command(command: list[str], *, cwd: str | Path, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + return subprocess.run( + command, + cwd=str(cwd), + text=True, + capture_output=True, + check=False, + env=env, + ) + + +def _run_or_raise(command: list[str], *, cwd: str | Path, env: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]: + result = _run_command(command, cwd=cwd, env=env) + if result.returncode != 0: + raise RuntimeError( + "Command failed\n" + f"command={command}\n" + f"stdout:\n{result.stdout}\n\n" + f"stderr:\n{result.stderr}" + ) + return result + + +def _find_backend_pgids(backend_root_dir: str | Path) -> list[int]: + backend_root = str(Path(backend_root_dir).resolve()) + result = _run_command(["ps", "-ax", "-o", "pid=,pgid=,command="], cwd=PROJECT_ROOT) + if result.returncode != 0: + return [] + + pgids: list[int] = [] + for line in result.stdout.splitlines(): + raw = line.strip() + if not raw or backend_root not in raw or " run.py" not in raw: + continue + parts = raw.split(None, 2) + if len(parts) < 3: + continue + try: + pgid = int(parts[1]) + except ValueError: + continue + if pgid not in pgids: + pgids.append(pgid) + return pgids + + +def _port_from_ready_url(url: str | None) -> int | None: + if not url: + return None + parsed = urlparse(url) + if parsed.port is not None: + return int(parsed.port) + if parsed.scheme == "https": + return 443 + if parsed.scheme == "http": + return 80 + return None + + +def _find_port_listener_pgids(port: int | None) -> list[int]: + if port is None: + return [] + result = _run_command(["lsof", "-nP", f"-iTCP:{port}", "-sTCP:LISTEN", "-t"], cwd=PROJECT_ROOT) + if result.returncode not in (0, 1): + return [] + + pgids: list[int] = [] + for line in result.stdout.splitlines(): + raw = line.strip() + if not raw: + continue + try: + pid = int(raw) + pgid = os.getpgid(pid) + except (ValueError, ProcessLookupError, PermissionError): + continue + if pgid not in pgids: + pgids.append(pgid) + return pgids + + +def _find_managed_pgids(service: dict[str, Any]) -> list[int]: + pgids: list[int] = [] + for pgid in _find_backend_pgids(service["ecm_root"]): + if pgid not in pgids: + pgids.append(pgid) + for pgid in _find_port_listener_pgids(_port_from_ready_url(service.get("ready_url"))): + if pgid not in pgids: + pgids.append(pgid) + return pgids + + +def _kill_process_group(pid: int) -> bool: + try: + os.killpg(pid, signal.SIGTERM) + time.sleep(1) + return True + except PermissionError: + try: + os.kill(pid, signal.SIGTERM) + time.sleep(1) + return True + except (PermissionError, ProcessLookupError): + return False + except ProcessLookupError: + return False + + +def _wait_http_ready(url: str, timeout_seconds: int, interval_seconds: int) -> None: + deadline = time.time() + timeout_seconds + last_error = "" + while time.time() < deadline: + try: + with request.urlopen(url, timeout=5) as response: + if 200 <= response.status < 500: + return + except Exception as exc: # noqa: BLE001 + last_error = str(exc) + time.sleep(interval_seconds) + raise TimeoutError(f"Service not ready: {url}; last_error={last_error}") + + +def _load_state() -> dict[str, Any]: + if not STATE_FILE.exists(): + return {} + try: + return json.loads(STATE_FILE.read_text(encoding="utf-8")) + except json.JSONDecodeError: + return {} + + +def _save_state(state: dict[str, Any]) -> None: + STATE_DIR.mkdir(parents=True, exist_ok=True) + STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8") + + +def _clear_state() -> None: + STATE_FILE.unlink(missing_ok=True) + + +def _parse_env_line(line: str) -> tuple[str, str] | None: + stripped = line.strip() + if not stripped or stripped.startswith("#") or "=" not in stripped: + return None + key, value = stripped.split("=", 1) + return key.strip(), value + + +def _read_env_file(path: Path) -> list[str]: + if not path.exists(): + return [] + return path.read_text(encoding="utf-8").splitlines() + + +def _write_env_file(path: Path, lines: list[str]) -> None: + _ensure_parent(path) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + +def _apply_env_overlay(config: dict[str, Any]) -> dict[str, Any]: + env_config = config.get("env_file", {}) + if not env_config.get("enabled", True): + return {"applied": False, "reason": "disabled"} + + env_path = Path(env_config["path"]) + overrides = {str(key): str(value) for key, value in (env_config.get("overrides") or {}).items()} + state = _load_state() + env_state = state.setdefault("env_file", {}) + + if not env_state.get("backup_path"): + backup_dir = STATE_DIR / "env_backup" + backup_dir.mkdir(parents=True, exist_ok=True) + backup_path = backup_dir / env_path.name + shutil.copy2(env_path, backup_path) + env_state["backup_path"] = str(backup_path) + env_state["path"] = str(env_path) + + lines = _read_env_file(env_path) + key_to_index: dict[str, int] = {} + for index, line in enumerate(lines): + parsed = _parse_env_line(line) + if parsed is None: + continue + key_to_index[parsed[0]] = index + + for key, value in overrides.items(): + new_line = f"{key}={value}" + if key in key_to_index: + lines[key_to_index[key]] = new_line + else: + lines.append(new_line) + + _write_env_file(env_path, lines) + _save_state(state) + return { + "applied": True, + "path": str(env_path), + "overrides": sorted(overrides.keys()), + "backup_path": env_state["backup_path"], + } + + +def _restore_env_overlay(config: dict[str, Any]) -> dict[str, Any]: + env_config = config.get("env_file", {}) + if not env_config.get("enabled", True) or not env_config.get("restore_on_stop", True): + return {"restored": False, "reason": "disabled"} + + state = _load_state() + env_state = state.get("env_file") or {} + backup_path = env_state.get("backup_path") + env_path = env_state.get("path") or env_config.get("path") + if not backup_path or not env_path: + return {"restored": False, "reason": "no_backup"} + + backup = Path(backup_path) + target = Path(env_path) + if not backup.exists(): + state.pop("env_file", None) + _save_state(state) + return {"restored": False, "reason": "backup_missing"} + + shutil.copy2(backup, target) + backup.unlink(missing_ok=True) + state.pop("env_file", None) + if state: + _save_state(state) + else: + _clear_state() + return {"restored": True, "path": str(target)} + + +def _build_process_env(config: dict[str, Any]) -> dict[str, str]: + env = os.environ.copy() + env.update({key: str(value) for key, value in (config.get("service", {}).get("env") or {}).items()}) + return env + + +def _state_pid() -> int | None: + state = _load_state() + pid = state.get("pid") + return int(pid) if isinstance(pid, int) else None + + +def _set_state_pid(pid: int | None) -> None: + state = _load_state() + if pid is None: + state.pop("pid", None) + else: + state["pid"] = pid + if state: + _save_state(state) + else: + _clear_state() + + +def clear_database(config: dict[str, Any]) -> dict[str, Any]: + couchdb = config["couchdb"] + if not couchdb.get("enabled", True): + return {"enabled": False} + + delete_status = _delete_couchdb_database( + couchdb["base_url"], + couchdb["database"], + couchdb.get("username", ""), + couchdb.get("password", ""), + ) + create_status = None + if couchdb.get("recreate_after_delete", False): + create_status = _create_couchdb_database( + couchdb["base_url"], + couchdb["database"], + couchdb.get("username", ""), + couchdb.get("password", ""), + ) + return { + "database": couchdb["database"], + "delete_status": delete_status, + "create_status": create_status, + } + + +def import_init_data(config: dict[str, Any]) -> dict[str, Any]: + service = config["service"] + result = _run_or_raise(service["restore_command"], cwd=service["cwd"], env=_build_process_env(config)) + return { + "data_dir": config["data"]["data_dir"], + "restore_command": service["restore_command"], + "stdout_tail": result.stdout[-2000:], + "stderr_tail": result.stderr[-2000:], + } + + +def stop_service(config: dict[str, Any], *, restore_env: bool = True) -> dict[str, Any]: + service = config["service"] + killed: list[int] = [] + pid = _state_pid() + if pid is not None and _kill_process_group(pid): + killed.append(pid) + for pgid in _find_managed_pgids(service): + if pgid in killed: + continue + if _kill_process_group(pgid): + killed.append(pgid) + _set_state_pid(None) + env_restore = _restore_env_overlay(config) if restore_env else {"restored": False, "reason": "skipped"} + return { + "stopped": bool(killed), + "killed_pgids": killed, + "env_file": env_restore, + } + + +def run_service(config: dict[str, Any]) -> dict[str, Any]: + service = config["service"] + existing_pgids = _find_managed_pgids(service) + if existing_pgids: + return { + "started": False, + "reason": "already_running", + "pgids": existing_pgids, + "log_file": service["log_file"], + } + + env_result = _apply_env_overlay(config) + log_path = _ensure_parent(service["log_file"]) + handle = open(log_path, "a", encoding="utf-8") + process = subprocess.Popen( + service["start_command"], + cwd=service["cwd"], + stdout=handle, + stderr=subprocess.STDOUT, + text=True, + start_new_session=True, + env=_build_process_env(config), + ) + _set_state_pid(process.pid) + _wait_http_ready( + str(service["ready_url"]), + int(service.get("ready_timeout_seconds", 120)), + int(service.get("ready_interval_seconds", 2)), + ) + return { + "started": True, + "pid": process.pid, + "log_file": str(log_path), + "cwd": service["cwd"], + "command": service["start_command"], + "ready_url": service["ready_url"], + "env_file": env_result, + } + + +def do_clear(config: dict[str, Any]) -> dict[str, Any]: + stop_result = stop_service(config, restore_env=True) + clear_result = clear_database(config) + return { + "command": "clear", + "stop": stop_result, + "database": clear_result, + } + + +def do_init(config: dict[str, Any]) -> dict[str, Any]: + stop_result = stop_service(config, restore_env=False) + env_result = _apply_env_overlay(config) + clear_result = clear_database(config) + init_result = import_init_data(config) + return { + "command": "init", + "stop": stop_result, + "env_file": env_result, + "database": clear_result, + "init": init_result, + } + + +def do_run(config: dict[str, Any]) -> dict[str, Any]: + return { + "command": "run", + "service": run_service(config), + } + + +def do_stop(config: dict[str, Any]) -> dict[str, Any]: + return { + "command": "stop", + "service": stop_service(config, restore_env=True), + } + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description="远程 ECM 测试环境管理工具") + parser.add_argument( + "command", + choices=["clear", "init", "run", "stop"], + help="clear: 清空 CouchDB 测试库；init: 通过 run.py restore-database 恢复数据；run: 后台启动远程服务；stop: 停止远程服务并恢复 .env", + ) + parser.add_argument("--config", default="tools/remote_env.example.yaml", help="YAML 配置文件路径") + parser.add_argument("--db-name", help="覆盖 couchdb.database") + parser.add_argument("--db-user", help="覆盖 couchdb.username") + parser.add_argument("--db-password", help="覆盖 couchdb.password") + parser.add_argument("--couchdb-base-url", help="覆盖 couchdb.base_url") + parser.add_argument("--couchdb-url", help="覆盖 couchdb.url") + parser.add_argument("--couchdb-port", help="覆盖 couchdb.port") + parser.add_argument("--ecm-root", help="覆盖 service.ecm_root") + parser.add_argument("--python-path", help="覆盖 service.python_path") + parser.add_argument("--data-dir", help="覆盖 data.data_dir") + parser.add_argument("--ready-url", help="覆盖 service.ready_url") + return parser + + +def main(argv: list[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + config, config_path = load_config(args.config, args) + validate_config(config, args.command) + + if args.command == "clear": + result = do_clear(config) + elif args.command == "init": + result = do_init(config) + elif args.command == "run": + result = do_run(config) + elif args.command == "stop": + result = do_stop(config) + else: + raise AssertionError(f"Unhandled command: {args.command}") + + result["config_path"] = str(config_path) + result["state_file"] = str(STATE_FILE) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 + + +if __name__ == "__main__": + try: + raise SystemExit(main()) + except Exception as exc: # noqa: BLE001 + print(json.dumps({"error": str(exc)}, ensure_ascii=False, indent=2), file=sys.stderr) + raise SystemExit(1) \ No newline at end of file