308 lines
10 KiB
Python
308 lines
10 KiB
Python
import re
|
|
from pathlib import Path
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
|
|
|
|
def _env_example_keys() -> set[str]:
|
|
keys = set()
|
|
for raw_line in (ROOT / ".env.example").read_text(encoding="utf-8").splitlines():
|
|
line = raw_line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
key, _value = line.split("=", 1)
|
|
keys.add(key.strip())
|
|
return keys
|
|
|
|
|
|
def _runtime_env_keys_from_code() -> set[str]:
|
|
pattern = re.compile(r"""os\.(?:getenv|environ\.get)\(\s*['"]([A-Z][A-Z0-9_]+)['"]""")
|
|
scan_paths = [
|
|
ROOT / "app.py",
|
|
ROOT / "config.py",
|
|
ROOT / "scheduler.py",
|
|
ROOT / "run_scheduler.py",
|
|
ROOT / "routes",
|
|
ROOT / "services",
|
|
ROOT / "utils",
|
|
]
|
|
|
|
keys = set()
|
|
for scan_path in scan_paths:
|
|
paths = scan_path.rglob("*.py") if scan_path.is_dir() else [scan_path]
|
|
for path in paths:
|
|
if not path.exists() or "__pycache__" in path.parts:
|
|
continue
|
|
content = path.read_text(encoding="utf-8", errors="ignore")
|
|
keys.update(match.group(1) for match in pattern.finditer(content))
|
|
return keys
|
|
|
|
|
|
def _compose_env_keys() -> set[str]:
|
|
pattern = re.compile(r"""\$\{([A-Z][A-Z0-9_]+)(?::-[^}]*)?\}""")
|
|
keys = set()
|
|
for path in ROOT.glob("docker-compose*.yml"):
|
|
keys.update(pattern.findall(path.read_text(encoding="utf-8", errors="ignore")))
|
|
return keys
|
|
|
|
|
|
def test_phase3f_orphan_ai_services_stay_removed():
|
|
orphan_services = [
|
|
"services/elephant_alpha_decision_router.py",
|
|
"services/telegram_ai_integration.py",
|
|
"services/watcher_agent.py",
|
|
]
|
|
|
|
assert [path for path in orphan_services if (ROOT / path).exists()] == []
|
|
|
|
|
|
def test_app_py_stays_blueprint_only_for_routes():
|
|
app_source = (ROOT / "app.py").read_text(encoding="utf-8")
|
|
|
|
assert "@app.route" not in app_source
|
|
|
|
|
|
def test_app_py_does_not_start_with_stale_restart_todo_banner():
|
|
app_source = (ROOT / "app.py").read_text(encoding="utf-8")
|
|
|
|
assert not app_source.startswith("# ================= TODO LIST")
|
|
assert "重開機後請依序執行" not in app_source
|
|
|
|
|
|
def test_active_guides_do_not_point_to_removed_ai_services():
|
|
active_guides = [
|
|
ROOT / "docs" / "ELEPHANT_ALPHA_SETUP.md",
|
|
]
|
|
removed_modules = [
|
|
"services/telegram_ai_integration.py",
|
|
"services/watcher_agent.py",
|
|
]
|
|
|
|
for guide in active_guides:
|
|
content = guide.read_text(encoding="utf-8")
|
|
assert "Modify routing behavior in `services/elephant_alpha_decision_router.py`" not in content
|
|
for module_path in removed_modules:
|
|
assert module_path not in content
|
|
|
|
|
|
def test_env_example_documents_runtime_and_ai_automation_variables():
|
|
expected_keys = {
|
|
"AI_CALL_LOGGING_ENABLED",
|
|
"AUTO_FIX_ENABLED",
|
|
"AWAITING_REVIEW_PUSH_BATCH",
|
|
"CODE_REVIEW_AUTO_FIX_ENABLED",
|
|
"CODE_REVIEW_OLLAMA_MODEL",
|
|
"CODE_REVIEW_OLLAMA_TIMEOUT",
|
|
"COST_THROTTLE_ENABLED",
|
|
"COST_THROTTLE_PROJECT_RATIO",
|
|
"COST_UNTHROTTLE_PROJECT_RATIO",
|
|
"DEEPSEEK_API_KEY",
|
|
"DEEPSEEK_BASE_URL",
|
|
"DEEPSEEK_DIRECT_ENABLED",
|
|
"DEEPSEEK_MODEL",
|
|
"DEEPSEEK_TIMEOUT",
|
|
"ELEPHANT_ALPHA_ALLOWED_SSH_HOSTS",
|
|
"ELEPHANT_ALPHA_URL",
|
|
"EMBEDDING_HOST",
|
|
"EMBEDDING_TIMEOUT",
|
|
"GUNICORN_TIMEOUT",
|
|
"GUNICORN_THREADS",
|
|
"GUNICORN_WORKER_CLASS",
|
|
"LINE_ENABLED",
|
|
"MOMO_AI_AUTOMATION_SMOKE_HISTORY",
|
|
"MOMO_AI_AUTOMATION_SMOKE_HISTORY_LIMIT",
|
|
"MOMO_ALLOW_INSECURE_INTERNAL_WEBHOOK_FOR_DEV",
|
|
"MOMO_EVENT_ROUTER_DEFAULT_DEDUP_SEC",
|
|
"MOMO_EVENT_ROUTER_QUEUE",
|
|
"MOMO_EVENT_ROUTER_REPLAY_LIMIT",
|
|
"MOMO_EVENT_ROUTER_REPLAY_ON_SUCCESS",
|
|
"MCP_FILESYSTEM_URL",
|
|
"MCP_FIRECRAWL_URL",
|
|
"MCP_CACHE_TTL_SEC",
|
|
"MCP_MAX_RESULT_BYTES",
|
|
"MCP_OMNISEARCH_URL",
|
|
"MCP_POSTGRES_URL",
|
|
"MCP_ROUTER_ENABLED",
|
|
"MCP_TIMEOUT_SEC",
|
|
"MODEL_ROUTER_ENABLED",
|
|
"N8N_HOST",
|
|
"N8N_PASSWORD",
|
|
"N8N_PROTOCOL",
|
|
"N8N_USER",
|
|
"N8N_WEBHOOK_BASE_URL",
|
|
"NEMOTRON_OLLAMA_FIRST",
|
|
"NEMOTRON_OLLAMA_MODEL",
|
|
"NEMOTRON_OLLAMA_TIMEOUT",
|
|
"OLLAMA_EMBED_TIMEOUT",
|
|
"OPENCLAW_ADMIN_USER_IDS",
|
|
"OPENCLAW_AGENT_DISPATCH",
|
|
"OPENCLAW_AGENT_DISPATCH_CMDS",
|
|
"OPENCLAW_ALLOW_PRIVATE_WITHOUT_WHITELIST",
|
|
"OPENCLAW_DAILY_HERMES_TEMPLATE",
|
|
"OPENCLAW_OLLAMA_MODEL",
|
|
"OPENCLAW_PPT_CACHE_TTL_HOURS",
|
|
"OPENCLAW_IMAGE_GEMINI_MODEL",
|
|
"OPENCLAW_IMAGE_OLLAMA_TIMEOUT",
|
|
"OPENCLAW_IMAGE_VISION_MODEL",
|
|
"OPENCLAW_QA_OLLAMA_FIRST",
|
|
"OPENCLAW_QA_OLLAMA_MODEL",
|
|
"OPENCLAW_QA_OLLAMA_TIMEOUT",
|
|
"PPT_VISION_ENABLED",
|
|
"PPT_VISION_MODEL",
|
|
"PPT_VISION_TIMEOUT",
|
|
"PROMOTION_PENDING_BATCH_SIZE",
|
|
"RAG_ENABLED",
|
|
"RAG_DEFAULT_THRESHOLD",
|
|
"RAG_DEFAULT_TOP_K",
|
|
"RAG_EMBED_DIM",
|
|
"RAG_EMBED_MODEL",
|
|
"RAG_EMBED_NORMALIZE",
|
|
"TELEGRAM_ADMIN_CHAT_ID",
|
|
"TELEGRAM_BOT_USERNAME",
|
|
"TELEGRAM_CHAT_ID",
|
|
"WEB_CONCURRENCY",
|
|
}
|
|
|
|
assert expected_keys <= _env_example_keys()
|
|
|
|
|
|
def test_env_example_documents_runtime_os_env_keys():
|
|
internal_runtime_keys = {
|
|
"MOMO_ALLOW_INSECURE_CONFIG_FOR_TESTS",
|
|
"PYTEST_CURRENT_TEST",
|
|
}
|
|
|
|
assert _runtime_env_keys_from_code() - internal_runtime_keys <= _env_example_keys()
|
|
|
|
|
|
def test_env_example_documents_docker_compose_variables():
|
|
assert _compose_env_keys() <= _env_example_keys()
|
|
|
|
|
|
def test_scheduler_does_not_silently_swallow_exceptions():
|
|
scheduler_source = (ROOT / "scheduler.py").read_text(encoding="utf-8")
|
|
|
|
assert "except:" not in scheduler_source
|
|
assert not re.search(r"except(?: Exception)?[^\n]*:\n\s+pass(?:\s|#|$)", scheduler_source)
|
|
|
|
|
|
def test_v2_flagged_silent_exception_sites_stay_logged():
|
|
openclaw_source = (ROOT / "routes" / "openclaw_bot_routes.py").read_text(encoding="utf-8")
|
|
notification_source = (ROOT / "services" / "notification_manager.py").read_text(encoding="utf-8")
|
|
|
|
assert "trend chart temp cleanup failed" in openclaw_source
|
|
assert "trend agg chart temp cleanup failed" in openclaw_source
|
|
assert "[Notification] 讀取 public_url 設定失敗" in notification_source
|
|
|
|
|
|
def test_v2_flagged_imports_are_removed_or_active():
|
|
app_source = (ROOT / "app.py").read_text(encoding="utf-8")
|
|
scheduler_source = (ROOT / "scheduler.py").read_text(encoding="utf-8")
|
|
|
|
for module_name in ["math", "hashlib", "zipfile", "io", "traceback"]:
|
|
assert not re.search(rf"^\s*import\s+{module_name}\b", app_source, re.MULTILINE)
|
|
assert "EXCEL_EXPORT_DIR" not in app_source
|
|
assert "DATABASE_TYPE" not in app_source
|
|
|
|
assert re.search(r"^\s*import\s+schedule\b", app_source, re.MULTILINE)
|
|
assert "schedule.run_pending()" in app_source
|
|
assert "schedule.every(" in app_source
|
|
assert not re.search(r"^\s*import\s+schedule\b", scheduler_source, re.MULTILINE)
|
|
|
|
|
|
def test_tracked_backup_artifacts_stay_removed():
|
|
forbidden_artifacts = [
|
|
"app.py.backup_login_required",
|
|
"vendor_stockout_list.html.backup",
|
|
"database/edm_dashboard.html",
|
|
"templates/list.html",
|
|
"tests/main_test.py",
|
|
]
|
|
|
|
assert [path for path in forbidden_artifacts if (ROOT / path).exists()] == []
|
|
|
|
|
|
def test_active_code_no_longer_references_legacy_5888_port():
|
|
active_paths = [
|
|
ROOT / "app.py",
|
|
ROOT / "tests",
|
|
ROOT / "AUTO_IMPORT_README.md",
|
|
ROOT / "GOOGLE_DRIVE_SETUP.md",
|
|
ROOT / "start_momo.command",
|
|
ROOT / "scripts" / "archive" / "check_email_status.py",
|
|
]
|
|
|
|
offenders = []
|
|
for active_path in active_paths:
|
|
paths = active_path.rglob("*") if active_path.is_dir() else [active_path]
|
|
for path in paths:
|
|
if (
|
|
not path.is_file()
|
|
or path == Path(__file__).resolve()
|
|
or "__pycache__" in path.parts
|
|
or path.suffix == ".pyc"
|
|
):
|
|
continue
|
|
content = path.read_text(encoding="utf-8", errors="ignore")
|
|
if "5888" in content:
|
|
offenders.append(str(path.relative_to(ROOT)))
|
|
|
|
assert offenders == []
|
|
|
|
|
|
def test_executable_scripts_do_not_use_remove_orphans():
|
|
script_paths = [
|
|
ROOT / "scripts",
|
|
ROOT / "docker",
|
|
ROOT / ".gitea" / "workflows",
|
|
]
|
|
|
|
offenders = []
|
|
for script_root in script_paths:
|
|
if not script_root.exists():
|
|
continue
|
|
for path in script_root.rglob("*"):
|
|
if not path.is_file():
|
|
continue
|
|
content = path.read_text(encoding="utf-8", errors="ignore")
|
|
if "--remove-orphans" in content or "docker compose down" in content:
|
|
offenders.append(str(path.relative_to(ROOT)))
|
|
|
|
assert offenders == []
|
|
|
|
|
|
def test_gitea_cd_uses_ewoooc_dedicated_runner_label():
|
|
workflow = (ROOT / ".gitea" / "workflows" / "cd.yaml").read_text(encoding="utf-8")
|
|
|
|
assert "runs-on: ewoooc-host" in workflow
|
|
assert "runs-on: ubuntu-latest" not in workflow
|
|
assert "runs-on: ubuntu-22.04" not in workflow
|
|
assert "runs-on: ubuntu-24.04" not in workflow
|
|
|
|
|
|
def test_devops_handbook_uses_current_docker_runtime_commands():
|
|
handbook = (ROOT / "docs" / "guides" / "devops_handbook.md").read_text(encoding="utf-8")
|
|
|
|
assert "kubectl" not in handbook
|
|
assert "docker restart momo-pro-system" not in handbook
|
|
assert "docker compose build --no-cache momo-pro-system" not in handbook
|
|
assert "docker kill -s HUP momo-pro-system" in handbook
|
|
assert "docker compose build --no-cache momo-app" in handbook
|
|
assert "docker compose up -d --no-deps --force-recreate momo-app" in handbook
|
|
assert "--remove-orphans" in handbook
|
|
assert "禁止" in handbook
|
|
|
|
|
|
def test_deployment_docs_cover_jump_host_known_hosts_repair():
|
|
sop = (ROOT / "docs" / "guides" / "deployment_sop.md").read_text(encoding="utf-8")
|
|
handbook = (ROOT / "docs" / "guides" / "devops_handbook.md").read_text(encoding="utf-8")
|
|
|
|
for content in (sop, handbook):
|
|
assert "Host key verification failed" in content
|
|
assert "ssh-keygen -R 192.168.0.188" in content
|
|
assert "ssh-keyscan -H 192.168.0.188" in content
|
|
assert "StrictHostKeyChecking=no" in content
|
|
assert "不要" in content
|
|
assert "momo_scp_smoke.txt" in sop
|