Files
ewoooc/tests/test_phase3f_cleanup_contracts.py
OoO e73cd6e6a3
All checks were successful
CD Pipeline / deploy (push) Successful in 1m36s
fix(stability): 補強 scheduler 例外記錄
2026-04-30 10:28:37 +08:00

80 lines
2.4 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 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_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 = {
"AUTO_FIX_ENABLED",
"CODE_REVIEW_AUTO_FIX_ENABLED",
"ELEPHANT_ALPHA_ALLOWED_SSH_HOSTS",
"ELEPHANT_ALPHA_URL",
"EMBEDDING_HOST",
"EMBEDDING_TIMEOUT",
"GUNICORN_TIMEOUT",
"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",
"N8N_HOST",
"N8N_PASSWORD",
"N8N_PROTOCOL",
"N8N_USER",
"N8N_WEBHOOK_BASE_URL",
"OLLAMA_EMBED_TIMEOUT",
"TELEGRAM_CHAT_ID",
"WEB_CONCURRENCY",
}
assert expected_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)