增加了一个测试工具用于初始化测试环境

This commit is contained in:
strepsiades
2026-03-11 11:40:23 +08:00
parent 1f68bd5ca8
commit b9fbf357a4
9 changed files with 1359 additions and 2 deletions
+414
View File
@@ -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 "<root>", 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 "<root>", local_value, remote_value))
return diffs
if local_value != remote_value:
diffs.append((path or "<root>", 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 "<empty>",
"",
"Remote sync_log:",
pair.remote_node.sync_log or "<empty>",
"",
"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())
+44
View File
@@ -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}
+699
View File
@@ -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)