48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
|
from pathlib import Path
|
||
|
|
|
||
|
|
|
||
|
|
ROOT = Path(__file__).resolve().parents[2]
|
||
|
|
|
||
|
|
|
||
|
|
def read_text(path: str) -> str:
|
||
|
|
return (ROOT / path).read_text(encoding="utf-8")
|
||
|
|
|
||
|
|
|
||
|
|
def test_admin_ops_backend_contract_is_guarded_and_profile_aware():
|
||
|
|
source = read_text("apps/web/vue3/main.py")
|
||
|
|
|
||
|
|
assert 'class AdminOpsRequest(BaseModel)' in source
|
||
|
|
assert 'confirm: str' in source
|
||
|
|
assert '"restart-celery"' in source
|
||
|
|
assert '"restart-project"' in source
|
||
|
|
assert '@app.post("/api/admin/restart-celery")' in source
|
||
|
|
assert '@app.post("/api/admin/restart-project")' in source
|
||
|
|
assert 'BBAV5_ADMIN_OPS_TOKEN' in source
|
||
|
|
assert '_validate_admin_ops_request(request, payload, "restart-celery")' in source
|
||
|
|
assert '_validate_admin_ops_request(request, payload, "restart-project")' in source
|
||
|
|
|
||
|
|
assert 'active_tasks_present' in source
|
||
|
|
assert '_find_celery_worker_pids(profile)' in source
|
||
|
|
assert '_terminate_celery_worker_pids(processes_before)' in source
|
||
|
|
assert '_run_celery_start_script(profile)' in source
|
||
|
|
assert 'overview["ops"] = _current_celery_profile()' in source
|
||
|
|
assert '"project_restart_configured"' in source
|
||
|
|
assert '"not_configured"' in source
|
||
|
|
|
||
|
|
|
||
|
|
def test_runtime_monitor_frontend_exposes_admin_ops_controls():
|
||
|
|
source = read_text("apps/web/vue3/index.html")
|
||
|
|
|
||
|
|
assert "重启 Celery" in source
|
||
|
|
assert "重启项目" in source
|
||
|
|
assert "runtimeOverview.ops = data.ops || {}" in source
|
||
|
|
assert "const runtimeOps = computed(() => runtimeOverview.ops || {})" in source
|
||
|
|
assert "const projectRestartAvailable = computed(() => !!runtimeOps.value.project_restart_configured)" in source
|
||
|
|
assert "const celeryOpsLoading = ref(false)" in source
|
||
|
|
assert "const projectOpsLoading = ref(false)" in source
|
||
|
|
assert "postAdminOps('/api/admin/restart-celery', 'restart-celery')" in source
|
||
|
|
assert "postAdminOps('/api/admin/restart-project', 'restart-project')" in source
|
||
|
|
assert "ElMessageBox.confirm" in source
|
||
|
|
assert source.index("重启 Celery") < source.index('v-if="runtimeMonitorExpanded"')
|
||
|
|
assert source.index("重启项目") < source.index('v-if="runtimeMonitorExpanded"')
|