Files
ewoooc/tests/test_phase3f_cleanup_contracts.py
OoO d88dcc8f75
All checks were successful
CD Pipeline / deploy (push) Successful in 1m45s
fix(devops): 清理舊端口與危險 compose 操作
2026-04-30 14:24:53 +08:00

140 lines
4.2 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)
def test_tracked_backup_artifacts_stay_removed():
forbidden_artifacts = [
"app.py.backup_login_required",
"vendor_stockout_list.html.backup",
"database/edm_dashboard.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 == []